Migration Analytics Services to CDK

TrafficCapture/dockerSolution/src/main/docker/otelcol/Dockerfile

@@ -0,0 +1,4 @@
+FROM public.ecr.aws/a0w2c5q7/otelcol-with-opensearch:amd-latest
+
+COPY ./otel-config-cdk.yml /etc/otel-config.yml
+ENTRYPOINT ["./otelcontribcol", "--config", "/etc/otel-config.yml"]

TrafficCapture/dockerSolution/src/main/docker/otelcol/otel-config-cdk.yml

@@ -0,0 +1,48 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+
+processors:
+  batch:
+  attributes:
+    # This processor is currently renaming two attributes
+    # that are prefixed with `log4j.context_data.` to the base attribute name
+    # to make queries within OpenSearch clearer. Both the `insert from_attribute`
+    # and the `delete` actions will fail silently if the attribute is not present,
+    # which means that these are safe for events that both do and don't have these
+    # attributes. This pattern should be extended to all of our standard attributes.
+    actions:
+      - key: event
+        from_attribute: log4j.context_data.event
+        action: insert
+      - key: log4j.context_data.event
+        action: delete
+      - key: channel_id
+        from_attribute: log4j.context_data.channel_id
+        action: insert
+      - key: log4j.context_data.channel_id
+        action: delete
+
+extensions:
+  health_check:
+
+exporters:
+  opensearch:
+    namespace: migrations
+    http:
+      endpoint: "${ANALYTICS_DOMAIN_ENDPOINT}"
+  logging:
+    verbosity: detailed
+  debug:
+
+service:
+  extensions: [health_check]
+  telemetry:
+    logs:
+      level: "debug"
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [attributes]
+      exporters: [logging, debug, opensearch]

deployment/cdk/opensearch-service-migration/README.md

@@ -177,6 +177,20 @@ echo $FETCH_MIGRATION_COMMAND
 
 The pipeline configuration file can be viewed (and updated) via AWS Secrets Manager.
 
+## Accessing the Migration Analytics Domain
+
+The analytics domain receives metrics and events from the Capture Proxy and Replayer (if configured) and allows a user to visualize the progress and success of their migration.
+
+The domain & dashboard are only accessible from within the VPC, but a BastionHost is optionally set up within the VPC that allows a user to use Session Manager to make the dashboard avaiable locally via port forwarding.
+
+For the Bastion Host to be available, add `"migrationAnalyticsBastionEnabled": true` to cdk.context.json and redeploy at least the MigrationAnalytics stack.
+
+Run the `accessAnalyticsDashboard` script, and then open https://localhost:8157/_dashboards to view your dashboard.
+```shell
+# ./accessAnalyticsDashboard.sh STAGE REGION
+./accessAnalyticsDashboard.sh dev us-east-1
+```
+
 
 ## Tearing down CDK
 To remove all the CDK stack(s) which get created during a deployment we can execute a command similar to below

deployment/cdk/opensearch-service-migration/accessAnalyticsDashboard.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+
+# Example usage: ./accessAnalyticsDashboard.sh dev us-east-1
+
+stage=$1
+region=$2
+
+export AWS_DEFAULT_REGION=$region
+
+bastion_id=$(aws ec2 describe-instances --filters Name=instance-state-name,Values=running Name=tag-key,Values=migration_deployment Name=tag:Name,Values=BastionHost Name=tag:aws:cloudformation:stack-name,Values=OSMigrations-${stage}-${region}-MigrationAnalytics | jq --raw-output '.Reservations[0].Instances[0].InstanceId')
+
+domain_endpoint=$(aws opensearch describe-domains --domain-names migration-analytics-domain | jq --raw-output '.DomainStatusList[0].Endpoints.vpc')
+
+JSON_STRING=$( jq -n -c\
+                  --arg port "443" \
+                  --arg localPort "8157" \
+                  --arg host "$domain_endpoint" \
+                  '{portNumber: [$port], localPortNumber: [$localPort], host: [$host]}' )
+
+echo "Access the Analytics Dashboard at https://localhost:8157/_dashboards"
+
+aws ssm start-session --target $bastion_id --document-name AWS-StartPortForwardingSessionToRemoteHost --parameters "${JSON_STRING}"

deployment/cdk/opensearch-service-migration/default-values.json

@@ -10,5 +10,7 @@
   "migrationConsoleServiceEnabled": true,
   "captureProxyESServiceEnabled": true,
   "trafficReplayerServiceEnabled": true,
+  "migrationAnalyticsServiceEnabled": true,
+  "migrationAnalyticsBastionEnabled": false,
   "dpPipelineTemplatePath": "./dp_pipeline_template.yaml"
 }

deployment/cdk/opensearch-service-migration/lib/network-stack.ts

@@ -9,8 +9,9 @@ import {StackPropsExt} from "./stack-composer";
 import {StringParameter} from "aws-cdk-lib/aws-ssm";
 
 export interface NetworkStackProps extends StackPropsExt {
-    readonly vpcId?: string,
-    readonly availabilityZoneCount?: number,
+    readonly vpcId?: string
+    readonly availabilityZoneCount?: number
+    readonly migrationAnalyticsEnabled?: boolean
     readonly targetClusterEndpoint?: string
 }
 
@@ -104,6 +105,19 @@ export class NetworkStack extends Stack {
                 parameterName: `/migration/${props.stage}/${props.defaultDeployId}/osAccessSecurityGroupId`,
                 stringValue: defaultSecurityGroup.securityGroupId
             });
+
+            if (props.migrationAnalyticsEnabled) {
+                const analyticsSecurityGroup = new SecurityGroup(this, 'migrationAnalyticsSG', {
+                    vpc: this.vpc
+                });
+                analyticsSecurityGroup.addIngressRule(analyticsSecurityGroup, Port.allTraffic());
+
+                new StringParameter(this, 'SSMParameterMigrationAnalyticsSGId', {
+                    description: 'Migration Assistant parameter for analytics domain access security group id',
+                    parameterName: `/migration/${props.stage}/${props.defaultDeployId}/analyticsDomainSGId`,
+                    stringValue: analyticsSecurityGroup.securityGroupId
+                });
+            }
         }
 
         if (props.targetClusterEndpoint) {

deployment/cdk/opensearch-service-migration/lib/opensearch-domain-stack.ts

@@ -9,7 +9,7 @@ import {
 import {Domain, EngineVersion, TLSSecurityPolicy, ZoneAwarenessConfig} from "aws-cdk-lib/aws-opensearchservice";
 import {RemovalPolicy, SecretValue, Stack} from "aws-cdk-lib";
 import {IKey, Key} from "aws-cdk-lib/aws-kms";
-import {PolicyStatement} from "aws-cdk-lib/aws-iam";
+import {AnyPrincipal, Effect, PolicyStatement} from "aws-cdk-lib/aws-iam";
 import {ILogGroup, LogGroup} from "aws-cdk-lib/aws-logs";
 import {ISecret, Secret} from "aws-cdk-lib/aws-secretsmanager";
 import {StackPropsExt} from "./stack-composer";
@@ -25,7 +25,8 @@ export interface OpensearchDomainStackProps extends StackPropsExt {
   readonly dedicatedManagerNodeCount?: number,
   readonly warmInstanceType?: string,
   readonly warmNodes?: number
-  readonly accessPolicies?: PolicyStatement[],
+  readonly accessPolicyJson?: object,
+  readonly openAccessPolicyEnabled?: boolean
   readonly useUnsignedBasicAuth?: boolean,
   readonly fineGrainedManagerUserARN?: string,
   readonly fineGrainedManagerUserName?: string,
@@ -36,7 +37,7 @@ export interface OpensearchDomainStackProps extends StackPropsExt {
   readonly ebsEnabled?: boolean,
   readonly ebsIops?: number,
   readonly ebsVolumeSize?: number,
-  readonly ebsVolumeType?: EbsDeviceVolumeType,
+  readonly ebsVolumeTypeName?: string,
   readonly encryptionAtRestEnabled?: boolean,
   readonly encryptionAtRestKmsKeyARN?: string,
   readonly appLogEnabled?: boolean,
@@ -46,19 +47,62 @@ export interface OpensearchDomainStackProps extends StackPropsExt {
   readonly vpcSubnetIds?: string[],
   readonly vpcSecurityGroupIds?: string[],
   readonly availabilityZoneCount?: number,
-  readonly domainRemovalPolicy?: RemovalPolicy
+  readonly domainRemovalPolicy?: RemovalPolicy,
+  readonly domainAccessSecurityGroupParameter?: string,
+  readonly endpointParameterName?: string
+
 }
 
 
 export class OpenSearchDomainStack extends Stack {
 
-  createSSMParameters(domain: Domain, adminUserName: string|undefined, adminUserSecret: ISecret|undefined, stage: string, deployId: string) {
+  getEbsVolumeType(ebsVolumeTypeName: string) : EbsDeviceVolumeType|undefined {
+    const ebsVolumeType: EbsDeviceVolumeType|undefined = ebsVolumeTypeName ? EbsDeviceVolumeType[ebsVolumeTypeName as keyof typeof EbsDeviceVolumeType] : undefined
+        if (ebsVolumeTypeName && !ebsVolumeType) {
+            throw new Error("Provided ebsVolumeType does not match a selectable option, for reference https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.aws_ec2.EbsDeviceVolumeType.html")
+        }
+    return ebsVolumeType
+  }
+
+  createOpenAccessPolicy(domainName: string) {
+      const openPolicy = new PolicyStatement({
+          effect: Effect.ALLOW,
+          principals: [new AnyPrincipal()],
+          actions: ["es:*"],
+          resources: [`arn:aws:es:${this.region}:${this.account}:domain/${domainName}/*`]
+      })
+      return openPolicy
+    }
+
+    parseAccessPolicies(jsonObject: { [x: string]: any; }): PolicyStatement[] {
+      let accessPolicies: PolicyStatement[] = []
+      const statements = jsonObject['Statement']
+      if (!statements || statements.length < 1) {
+          throw new Error ("Provided accessPolicies JSON must have the 'Statement' element present and not be empty, for reference https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_statement.html")
+      }
+      // Access policies can provide a single Statement block or an array of Statement blocks
+      if (Array.isArray(statements)) {
+          for (let statementBlock of statements) {
+              const statement = PolicyStatement.fromJson(statementBlock)
+              accessPolicies.push(statement)
+          }
+      }
+      else {
+          const statement = PolicyStatement.fromJson(statements)
+          accessPolicies.push(statement)
+      }
+      return accessPolicies
+  }
+
+  createSSMParameters(domain: Domain, endpointParameterName: string|undefined, adminUserName: string|undefined, adminUserSecret: ISecret|undefined, stage: string, deployId: string) {
+
+    const endpointParameter = endpointParameterName ?? "osClusterEndpoint"
     new StringParameter(this, 'SSMParameterOpenSearchEndpoint', {
       description: 'OpenSearch migration parameter for OpenSearch endpoint',
-      parameterName: `/migration/${stage}/${deployId}/osClusterEndpoint`,
+      parameterName: `/migration/${stage}/${deployId}/${endpointParameter}`,
       stringValue: `https://${domain.domainEndpoint}:443`
     });
-
+    
     if (domain.masterUserPassword && !adminUserSecret) {
       console.log(`An OpenSearch domain fine-grained access control user was configured without an existing Secrets Manager secret, will not create SSM Parameter: /migration/${stage}/${deployId}/osUserAndSecret`)
     }
@@ -82,15 +126,14 @@ export class OpenSearchDomainStack extends Stack {
     let adminUserSecret: ISecret|undefined = props.fineGrainedManagerUserSecretManagerKeyARN ?
         Secret.fromSecretCompleteArn(this, "managerSecret", props.fineGrainedManagerUserSecretManagerKeyARN) : undefined
 
-
     const appLG: ILogGroup|undefined = props.appLogGroup && props.appLogEnabled ?
         LogGroup.fromLogGroupArn(this, "appLogGroup", props.appLogGroup) : undefined
 
+    const domainAccessSecurityGroupParameter = props.domainAccessSecurityGroupParameter ?? "osAccessSecurityGroupId"
     const defaultOSClusterAccessGroup = SecurityGroup.fromSecurityGroupId(this, "defaultDomainAccessSG",
-        StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/osAccessSecurityGroupId`))
+        StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/${domainAccessSecurityGroupParameter}`))
 
     // Map objects from props
-
     let adminUserName: string|undefined = props.fineGrainedManagerUserName
     // Enable demo mode setting
     if (props.enableDemoAdmin) {
@@ -123,10 +166,19 @@ export class OpenSearchDomainStack extends Stack {
       }
     }
 
+    const ebsVolumeType = props.ebsVolumeTypeName ? this.getEbsVolumeType(props.ebsVolumeTypeName) : undefined
+
+    let accessPolicies: PolicyStatement[] | undefined
+    if (props.openAccessPolicyEnabled) {
+      accessPolicies = [this.createOpenAccessPolicy(props.domainName)]
+  } else {
+      accessPolicies = props.accessPolicyJson ? this.parseAccessPolicies(props.accessPolicyJson) : undefined
+  }
+
     const domain = new Domain(this, 'Domain', {
       version: props.version,
       domainName: props.domainName,
-      accessPolicies: props.accessPolicies,
+      accessPolicies: accessPolicies,
       useUnsignedBasicAuth: props.useUnsignedBasicAuth,
       capacity: {
         dataNodeInstanceType: props.dataNodeInstanceType,
@@ -152,7 +204,7 @@ export class OpenSearchDomainStack extends Stack {
         enabled: props.ebsEnabled,
         iops: props.ebsIops,
         volumeSize: props.ebsVolumeSize,
-        volumeType: props.ebsVolumeType
+        volumeType: ebsVolumeType
       },
       logging: {
         appLogEnabled: props.appLogEnabled,
@@ -165,6 +217,7 @@ export class OpenSearchDomainStack extends Stack {
       removalPolicy: props.domainRemovalPolicy
     });
 
-    this.createSSMParameters(domain, adminUserName, adminUserSecret, props.stage, deployId)
+    this.createSSMParameters(domain, props.endpointParameterName, adminUserName, adminUserSecret, props.stage, deployId)
+
   }
 }

deployment/cdk/opensearch-service-migration/lib/service-stacks/capture-proxy-es-stack.ts

@@ -11,6 +11,7 @@ import {StringParameter} from "aws-cdk-lib/aws-ssm";
 
 export interface CaptureProxyESProps extends StackPropsExt {
     readonly vpc: IVpc,
+    readonly analyticsServiceEnabled: boolean
 }
 
 /**
@@ -71,10 +72,13 @@ export class CaptureProxyESStack extends MigrationServiceCore {
         })
 
         const brokerEndpoints = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/mskBrokers`);
+        let command = `/usr/local/bin/docker-entrypoint.sh eswrapper & /runJavaWithClasspath.sh org.opensearch.migrations.trafficcapture.proxyserver.CaptureProxy  --kafkaConnection ${brokerEndpoints} --enableMSKAuth --destinationUri https://localhost:19200 --insecureDestination --listenPort 9200 --sslConfigFile /usr/share/elasticsearch/config/proxy_tls.yml`
+        command = props.analyticsServiceEnabled ? command.concat(" --otelCollectorEndpoint http://otel-collector:4317") : command
         this.createService({
             serviceName: "capture-proxy-es",
             dockerFilePath: join(__dirname, "../../../../../", "TrafficCapture/dockerSolution/build/docker/trafficCaptureProxyServer"),
-            dockerImageCommand: ['/bin/sh', '-c', `/usr/local/bin/docker-entrypoint.sh eswrapper & /runJavaWithClasspath.sh org.opensearch.migrations.trafficcapture.proxyserver.CaptureProxy  --kafkaConnection ${brokerEndpoints} --enableMSKAuth --destinationUri https://localhost:19200 --insecureDestination --listenPort 9200 --sslConfigFile /usr/share/elasticsearch/config/proxy_tls.yml & wait -n 1`],
+            // TODO: add otel collector endpoint
+            dockerImageCommand: ['/bin/sh', '-c', command.concat(" & wait -n 1")],
             securityGroups: securityGroups,
             taskRolePolicies: [mskClusterConnectPolicy, mskTopicProducerPolicy],
             portMappings: [servicePort, esServicePort],

deployment/cdk/opensearch-service-migration/lib/service-stacks/capture-proxy-stack.ts

@@ -12,6 +12,7 @@ import {StringParameter} from "aws-cdk-lib/aws-ssm";
 export interface CaptureProxyProps extends StackPropsExt {
     readonly vpc: IVpc,
     readonly customSourceClusterEndpoint?: string
+    readonly analyticsServiceEnabled?: boolean
 }
 
 /**
@@ -62,10 +63,12 @@ export class CaptureProxyStack extends MigrationServiceCore {
 
         const brokerEndpoints = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/mskBrokers`);
         const sourceClusterEndpoint = props.customSourceClusterEndpoint ? props.customSourceClusterEndpoint : "https://elasticsearch:9200"
+        let command = `/runJavaWithClasspath.sh org.opensearch.migrations.trafficcapture.proxyserver.CaptureProxy  --kafkaConnection ${brokerEndpoints} --enableMSKAuth --destinationUri ${sourceClusterEndpoint} --insecureDestination --listenPort 9200 --sslConfigFile /usr/share/elasticsearch/config/proxy_tls.yml`
+        command = props.analyticsServiceEnabled ? command.concat(" --otelCollectorEndpoint http://otel-collector:4317") : command
         this.createService({
             serviceName: "capture-proxy",
             dockerFilePath: join(__dirname, "../../../../../", "TrafficCapture/dockerSolution/build/docker/trafficCaptureProxyServer"),
-            dockerImageCommand: ['/bin/sh', '-c', `/runJavaWithClasspath.sh org.opensearch.migrations.trafficcapture.proxyserver.CaptureProxy  --kafkaConnection ${brokerEndpoints} --enableMSKAuth --destinationUri ${sourceClusterEndpoint} --insecureDestination --listenPort 9200 --sslConfigFile /usr/share/elasticsearch/config/proxy_tls.yml`],
+            dockerImageCommand: ['/bin/sh', '-c', command],
             securityGroups: securityGroups,
             taskRolePolicies: [mskClusterConnectPolicy, mskTopicProducerPolicy],
             portMappings: [servicePort],