Migration Analytics Services to CDK

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

@@ -0,0 +1,5 @@
+FROM public.ecr.aws/a0w2c5q7/otelcol-with-opensearch:amd-latest
+
+COPY ./otel-config-cdk.yml /etc/otel-config.yml
+# RUN apt-get update && apt-get install file -y
+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: "https://${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/default-values.json

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

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

@@ -11,6 +11,7 @@ import {StringParameter} from "aws-cdk-lib/aws-ssm";
 export interface NetworkStackProps extends StackPropsExt {
     readonly vpcId?: string
     readonly availabilityZoneCount?: number
+    readonly migrationAnalyticsEnabled?: boolean
 }
 
 export class NetworkStack extends Stack {
@@ -76,6 +77,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
+                });
+            }
         }
 
     }

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

@@ -46,19 +46,24 @@ 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) {
+
+  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: domain.domainEndpoint
     });
-
+    
     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,12 +87,12 @@ 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
 
@@ -165,7 +170,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,13 @@ 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`],
+            // TODO: add otel collector endpoint
+            dockerImageCommand: ['/bin/sh', '-c', command],
             securityGroups: securityGroups,
             taskRolePolicies: [mskClusterConnectPolicy, mskTopicProducerPolicy],
             portMappings: [servicePort],

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

@@ -0,0 +1,102 @@
+import {StackPropsExt} from "../stack-composer";
+import {
+  BastionHostLinux,
+  BlockDeviceVolume,
+  MachineImage,
+  Port,
+  SecurityGroup,
+  IVpc,
+} from "aws-cdk-lib/aws-ec2";
+import {PortMapping, Protocol, ServiceConnectService} from "aws-cdk-lib/aws-ecs";
+import {Construct} from "constructs";
+import {join} from "path";
+import {MigrationServiceCore} from "./migration-service-core";
+import {StringParameter} from "aws-cdk-lib/aws-ssm";
+
+export interface MigrationAnalyticsProps extends StackPropsExt {
+    readonly vpc: IVpc,
+}
+
+// The MigrationAnalyticsStack consists of the OpenTelemetry Collector ECS container & an
+// Bastion host to allow access to the opensearch dashboard.
+export class MigrationAnalyticsStack extends MigrationServiceCore {
+
+    constructor(scope: Construct, id: string, props: MigrationAnalyticsProps) {
+        super(scope, id, props)
+
+        // Bastion Security Group
+        const bastionSecurityGroup = new SecurityGroup(
+          this,
+          "analyticsDashboardBastionSecurityGroup",
+          {
+            vpc: props.vpc,
+            allowAllOutbound: true,
+            securityGroupName: "analyticsDashboardBastionSecurityGroup",
+          }
+        );
+        let securityGroups = [
+            SecurityGroup.fromSecurityGroupId(this, "serviceConnectSG", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/serviceConnectSecurityGroupId`)),
+            SecurityGroup.fromSecurityGroupId(this, "migrationAnalyticsSGId", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/analyticsDomainSGId`)),
+            bastionSecurityGroup
+        ]
+        securityGroups[1].addIngressRule(bastionSecurityGroup, Port.tcp(443))
+
+        // Bastion host to access Opensearch Dashboards
+        new BastionHostLinux(this, "AnalyticsDashboardBastionHost", {
+          vpc: props.vpc,
+          securityGroup: bastionSecurityGroup,
+          machineImage: MachineImage.latestAmazonLinux2023(),
+          blockDevices: [
+            {
+              deviceName: "/dev/xvda",
+              volume: BlockDeviceVolume.ebs(10, {
+                encrypted: true,
+              }),
+            },
+          ],
+        });
+
+        // Port Mappings for collector and health check
+        const otelCollectorPort: PortMapping = {
+          name: "otel-collector-connect",
+          hostPort: 4317,
+          containerPort: 4317,
+          protocol: Protocol.TCP
+        }
+        const otelHealthCheckPort: PortMapping = {
+          name: "otel-healthcheck-connect",
+          hostPort: 13133,
+          containerPort: 13133,
+          protocol: Protocol.TCP
+        }
+        const serviceConnectServiceCollector: ServiceConnectService = {
+          portMappingName: "otel-collector-connect",
+          port: 4317,
+          dnsName: "otel-collector"
+        }
+        const serviceConnectServiceHealthCheck: ServiceConnectService = {
+          portMappingName: "otel-healthcheck-connect",
+          port: 13133,
+          dnsName: "otel-healthcheck"
+        }
+
+        const analyticsDomainEndpoint = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/analyticsDomainEndpoint`)
+
+        this.createService({
+            serviceName: `otel-collector`,
+            dockerFilePath: join(__dirname, "../../../../../", "TrafficCapture/dockerSolution/src/main/docker/otelcol"),
+            securityGroups: securityGroups,
+            taskCpuUnits: 1024,
+            taskMemoryLimitMiB: 4096,
+            portMappings: [otelCollectorPort, otelHealthCheckPort],
+            serviceConnectServices: [serviceConnectServiceCollector, serviceConnectServiceHealthCheck],
+            environment: {
+              "ANALYTICS_DOMAIN_ENDPOINT": analyticsDomainEndpoint
+            },
+            ...props
+        });
+
+
+    }
+
+}

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

@@ -10,7 +10,8 @@ import {Effect, PolicyStatement} from "aws-cdk-lib/aws-iam";
 
 export interface MigrationConsoleProps extends StackPropsExt {
     readonly vpc: IVpc,
-    readonly fetchMigrationEnabled: boolean
+    readonly fetchMigrationEnabled: boolean,
+    readonly migrationAnalyticsEnabled: boolean
 }
 
 export class MigrationConsoleStack extends MigrationServiceCore {
@@ -21,8 +22,12 @@ export class MigrationConsoleStack extends MigrationServiceCore {
             SecurityGroup.fromSecurityGroupId(this, "serviceConnectSG", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/serviceConnectSecurityGroupId`)),
             SecurityGroup.fromSecurityGroupId(this, "mskAccessSG", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/mskAccessSecurityGroupId`)),
             SecurityGroup.fromSecurityGroupId(this, "defaultDomainAccessSG", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/osAccessSecurityGroupId`)),
-            SecurityGroup.fromSecurityGroupId(this, "replayerOutputAccessSG", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/replayerOutputAccessSecurityGroupId`))
+            SecurityGroup.fromSecurityGroupId(this, "replayerOutputAccessSG", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/replayerOutputAccessSecurityGroupId`)),
         ]
+        if (props.migrationAnalyticsEnabled) {
+            securityGroups.push(SecurityGroup.fromSecurityGroupId(this, "migrationAnalyticsSGId", StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/analyticsDomainSGId`)))
+        }
+
         const osClusterEndpoint = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/osClusterEndpoint`)
         const brokerEndpoints = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/mskBrokers`);
 

deployment/cdk/opensearch-service-migration/lib/service-stacks/traffic-replayer-stack.ts

@@ -14,7 +14,8 @@ export interface TrafficReplayerProps extends StackPropsExt {
     readonly addOnMigrationId?: string,
     readonly customTargetEndpoint?: string,
     readonly customKafkaGroupId?: string,
-    readonly extraArgs?: string
+    readonly extraArgs?: string,
+    readonly analyticsServiceEnabled?: boolean
 }
 
 export class TrafficReplayerStack extends MigrationServiceCore {
@@ -93,11 +94,13 @@ export class TrafficReplayerStack extends MigrationServiceCore {
         const osClusterEndpoint = props.customTargetEndpoint ? props.customTargetEndpoint : `https://${cdkDomainEndpoint}:443`
         const brokerEndpoints = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/mskBrokers`);
         const groupId = props.customKafkaGroupId ? props.customKafkaGroupId : `logging-group-${deployId}`
+        // TODO: add otel collector endpoint
         let replayerCommand = `/runJavaWithClasspath.sh org.opensearch.migrations.replay.TrafficReplayer ${osClusterEndpoint} --insecure --kafka-traffic-brokers ${brokerEndpoints} --kafka-traffic-topic logging-traffic-topic --kafka-traffic-group-id ${groupId} --kafka-traffic-enable-msk-auth`
         if (props.enableClusterFGACAuth) {
             const osUserAndSecret = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${deployId}/osUserAndSecretArn`);
             replayerCommand = replayerCommand.concat(` --auth-header-user-and-secret ${osUserAndSecret}`)
         }
+        replayerCommand = props.analyticsServiceEnabled ? replayerCommand.concat(" --otelCollectorEndpoint http://otel-collector:4317") : replayerCommand
         replayerCommand = props.extraArgs ? replayerCommand.concat(` ${props.extraArgs}`) : replayerCommand
         this.createService({
             serviceName: `traffic-replayer-${deployId}`,