Alb diagrams

deployment/cdk/opensearch-service-migration/.gitignore

@@ -8,3 +8,4 @@ dist
 # CDK asset staging directory
 .cdk.staging
 cdk.out
+certs

deployment/cdk/opensearch-service-migration/cdk.context.json

@@ -11,7 +11,6 @@
     "trafficReplayerEnableClusterFGACAuth": true,
     "captureProxyESServiceEnabled": true,
     "fetchMigrationEnabled": true,
-    "otelCollectorEnabled": true,
-    "sourceClusterEndpoint": "https://capture-proxy-es.migration.dev.local:19200"
+    "otelCollectorEnabled": true
   }
 }

deployment/cdk/opensearch-service-migration/lib/common-utilities.ts

@@ -1,8 +1,10 @@
 import {Effect, PolicyStatement, Role, ServicePrincipal} from "aws-cdk-lib/aws-iam";
 import {Construct} from "constructs";
-import {StringParameter} from "aws-cdk-lib/aws-ssm";
 import {CpuArchitecture} from "aws-cdk-lib/aws-ecs";
 import {RemovalPolicy} from "aws-cdk-lib";
+import { IApplicationLoadBalancer } from "aws-cdk-lib/aws-elasticloadbalancingv2";
+import { ICertificate } from "aws-cdk-lib/aws-certificatemanager";
+import { IStringParameter, StringParameter } from "aws-cdk-lib/aws-ssm";
 
 export function createOpenSearchIAMAccessPolicy(partition: string, region: string, accountId: string): PolicyStatement {
     return new PolicyStatement({
@@ -25,8 +27,8 @@ export function createOpenSearchServerlessIAMAccessPolicy(partition: string, reg
 }
 
 export function createMSKConsumerIAMPolicies(scope: Construct, partition: string, region: string, accountId: string, stage: string, deployId: string): PolicyStatement[] {
-    const mskClusterARN = StringParameter.valueForStringParameter(scope, `/migration/${stage}/${deployId}/mskClusterARN`);
-    const mskClusterName = StringParameter.valueForStringParameter(scope, `/migration/${stage}/${deployId}/mskClusterName`);
+    const mskClusterARN = getMigrationStringParameterValue(scope, { parameter: MigrationSSMParameter.MSK_CLUSTER_ARN, stage, defaultDeployId: deployId });
+    const mskClusterName = getMigrationStringParameterValue(scope, { parameter: MigrationSSMParameter.MSK_CLUSTER_NAME, stage, defaultDeployId: deployId });
     const mskClusterConnectPolicy = new PolicyStatement({
         effect: Effect.ALLOW,
         resources: [mskClusterARN],
@@ -57,8 +59,8 @@ export function createMSKConsumerIAMPolicies(scope: Construct, partition: string
 }
 
 export function createMSKProducerIAMPolicies(scope: Construct, partition: string, region: string, accountId: string, stage: string, deployId: string): PolicyStatement[] {
-    const mskClusterARN = StringParameter.valueForStringParameter(scope, `/migration/${stage}/${deployId}/mskClusterARN`);
-    const mskClusterName = StringParameter.valueForStringParameter(scope, `/migration/${stage}/${deployId}/mskClusterName`);
+    const mskClusterARN = getMigrationStringParameterValue(scope, { parameter: MigrationSSMParameter.MSK_CLUSTER_ARN, stage, defaultDeployId: deployId });
+    const mskClusterName = getMigrationStringParameterValue(scope, { parameter: MigrationSSMParameter.MSK_CLUSTER_NAME, stage, defaultDeployId: deployId });
     const mskClusterConnectPolicy = new PolicyStatement({
         effect: Effect.ALLOW,
         resources: [mskClusterARN],
@@ -149,4 +151,75 @@ export function parseRemovalPolicy(optionName: string, policyNameString?: string
         throw new Error(`Provided '${optionName}' with value '${policyNameString}' does not match a selectable option, for reference https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.RemovalPolicy.html`)
     }
     return policy
-}
+}
+
+
+export type ALBConfig = NewALBListenerConfig;
+
+export interface NewALBListenerConfig {
+    alb: IApplicationLoadBalancer,
+    albListenerCert: ICertificate,
+    albListenerPort?: number,
+}
+
+export function isNewALBListenerConfig(config: ALBConfig): config is NewALBListenerConfig {
+    const parsed = config as NewALBListenerConfig;
+    return parsed.alb !== undefined && parsed.albListenerCert !== undefined;
+}
+
+
+export interface MigrationSSMConfig {
+    parameter: MigrationSSMParameter,
+    stage: string,
+    defaultDeployId: string
+}
+
+export function createMigrationStringParameter(scope: Construct, stringValue: string, props: MigrationSSMConfig) {
+    return new StringParameter(scope, `SSMParameter${props.parameter.charAt(0).toUpperCase() + props.parameter.slice(1)}`, {
+        parameterName: getMigrationStringParameterName(props),
+        stringValue: stringValue,
+        description: `Opensearch migration SSM parameter for ${props.parameter} with stage ${props.stage} and deploy id ${props.defaultDeployId}`,
+    });
+}
+
+export function getMigrationStringParameter(scope: Construct, props: MigrationSSMConfig): IStringParameter {
+    return StringParameter.fromStringParameterName(scope, `SSMParameter${props.parameter.charAt(0).toUpperCase() + props.parameter.slice(1)}`,
+        getMigrationStringParameterName(props));
+}
+
+export function getMigrationStringParameterValue(scope: Construct, props: MigrationSSMConfig): string {
+    return StringParameter.valueForTypedStringParameterV2(scope, getMigrationStringParameterName(props));
+}
+
+export function getCustomStringParameterValue(scope: Construct, parameterName: string): string {
+    return StringParameter.valueForTypedStringParameterV2(scope, parameterName);
+}
+
+export function getMigrationStringParameterName(props: MigrationSSMConfig): string {
+    return `/migration/${props.stage}/${props.defaultDeployId}/${props.parameter}`;
+}
+
+export enum MigrationSSMParameter {
+    ALB_MIGRATION_URL = 'albMigrationUrl',
+    ARTIFACT_S3_ARN = 'artifactS3Arn',
+    CLOUD_MAP_NAMESPACE_ID = 'cloudMapNamespaceId',
+    FETCH_MIGRATION_COMMAND = 'fetchMigrationCommand',
+    FETCH_MIGRATION_TASK_DEF_ARN = 'fetchMigrationTaskDefArn',
+    FETCH_MIGRATION_TASK_EXEC_ROLE_ARN = 'fetchMigrationTaskExecRoleArn',
+    FETCH_MIGRATION_TASK_ROLE_ARN = 'fetchMigrationTaskRoleArn',
+    KAFKA_BROKERS = 'kafkaBrokers',
+    MSK_CLUSTER_ARN = 'mskClusterARN',
+    MSK_CLUSTER_NAME = 'mskClusterName',
+    OS_ACCESS_SECURITY_GROUP_ID = 'osAccessSecurityGroupId',
+    OS_CLUSTER_ENDPOINT = 'osClusterEndpoint',
+    OS_USER_AND_SECRET_ARN = 'osUserAndSecretArn',
+    OSI_PIPELINE_LOG_GROUP_NAME = 'osiPipelineLogGroupName',
+    OSI_PIPELINE_ROLE_ARN = 'osiPipelineRoleArn',
+    REPLAYER_OUTPUT_ACCESS_SECURITY_GROUP_ID = 'replayerOutputAccessSecurityGroupId',
+    REPLAYER_OUTPUT_EFS_ID = 'replayerOutputEfsId',
+    SOURCE_CLUSTER_ENDPOINT = 'sourceClusterEndpoint',
+    SERVICE_SECURITY_GROUP_ID = 'serviceSecurityGroupId',
+    SERVICES_YAML_FILE = 'servicesYamlFile',
+    TRAFFIC_STREAM_SOURCE_ACCESS_SECURITY_GROUP_ID = 'trafficStreamSourceAccessSecurityGroupId',
+    VPC_ID = 'vpcId',
+}

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

@@ -1,4 +1,4 @@
-import {Stack, SecretValue} from "aws-cdk-lib";
+import {SecretValue, Stack} from "aws-cdk-lib";
 import {IVpc} from "aws-cdk-lib/aws-ec2";
 import {Construct} from "constructs";
 import {
@@ -12,35 +12,49 @@ import {Secret as SMSecret} from "aws-cdk-lib/aws-secretsmanager";
 import {join} from "path";
 import {readFileSync} from "fs"
 import {StackPropsExt} from "./stack-composer";
-import {StringParameter} from "aws-cdk-lib/aws-ssm";
 import {
+    MigrationSSMParameter,
     createDefaultECSTaskRole,
+    createMigrationStringParameter,
     createOpenSearchIAMAccessPolicy,
-    createOpenSearchServerlessIAMAccessPolicy
+    createOpenSearchServerlessIAMAccessPolicy,
+    getMigrationStringParameter,
+    getMigrationStringParameterValue
 } from "./common-utilities";
 
 export interface FetchMigrationProps extends StackPropsExt {
     readonly vpc: IVpc,
     readonly dpPipelineTemplatePath: string,
-    readonly sourceEndpoint: string,
     readonly fargateCpuArch: CpuArchitecture
 }
 
 export class FetchMigrationStack extends Stack {
 
     constructor(scope: Construct, id: string, props: FetchMigrationProps) {
         super(scope, id, props);
+        const sourceEndpoint = getMigrationStringParameterValue(this, {
+            ...props,
+            parameter: MigrationSSMParameter.SOURCE_CLUSTER_ENDPOINT,
+        });
 
         // Import required values
-        const targetClusterEndpoint = StringParameter.fromStringParameterName(this, "targetClusterEndpoint", `/migration/${props.stage}/${props.defaultDeployId}/osClusterEndpoint`)
-        const domainAccessGroupId = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/osAccessSecurityGroupId`)
+        const targetClusterEndpoint = getMigrationStringParameter(this, {
+            ...props,
+            parameter: MigrationSSMParameter.OS_CLUSTER_ENDPOINT,
+        });
+        const domainAccessGroupId = getMigrationStringParameterValue(this, {
+            ...props,
+            parameter: MigrationSSMParameter.OS_ACCESS_SECURITY_GROUP_ID,
+        });
         // This SG allows outbound access for ECR access as well as communication with other services in the cluster
-        const serviceGroupId = StringParameter.valueForStringParameter(this, `/migration/${props.stage}/${props.defaultDeployId}/serviceSecurityGroupId`)
-
+        const serviceGroupId = getMigrationStringParameterValue(this, {
+            ...props,
+            parameter: MigrationSSMParameter.SERVICE_SECURITY_GROUP_ID,
+        });
         const ecsCluster = Cluster.fromClusterAttributes(this, 'ecsCluster', {
             clusterName: `migration-${props.stage}-ecs-cluster`,
             vpc: props.vpc
-        })
+        });
 
         const serviceName = "fetch-migration"
         const ecsTaskRole = createDefaultECSTaskRole(this, serviceName)
@@ -60,22 +74,18 @@ export class FetchMigrationStack extends Stack {
             taskRole: ecsTaskRole
         });
 
-        new StringParameter(this, 'SSMParameterFetchMigrationTaskDefArn', {
-            description: 'OpenSearch Migration Parameter for Fetch Migration task definition ARN',
-            parameterName: `/migration/${props.stage}/${props.defaultDeployId}/fetchMigrationTaskDefArn`,
-            stringValue: fetchMigrationFargateTask.taskDefinitionArn
+        createMigrationStringParameter(this, fetchMigrationFargateTask.taskDefinitionArn, {
+            ...props,
+            parameter: MigrationSSMParameter.FETCH_MIGRATION_TASK_DEF_ARN,
         });
-        new StringParameter(this, 'SSMParameterFetchMigrationTaskRoleArn', {
-            description: 'OpenSearch Migration Parameter for Fetch Migration task role ARN',
-            parameterName: `/migration/${props.stage}/${props.defaultDeployId}/fetchMigrationTaskRoleArn`,
-            stringValue: fetchMigrationFargateTask.taskRole.roleArn
+        createMigrationStringParameter(this, fetchMigrationFargateTask.taskRole.roleArn, {
+            ...props,
+            parameter: MigrationSSMParameter.FETCH_MIGRATION_TASK_ROLE_ARN,
         });
-        new StringParameter(this, 'SSMParameterFetchMigrationTaskExecRoleArn', {
-            description: 'OpenSearch Migration Parameter for Fetch Migration task exec role ARN',
-            parameterName: `/migration/${props.stage}/${props.defaultDeployId}/fetchMigrationTaskExecRoleArn`,
-            stringValue: fetchMigrationFargateTask.obtainExecutionRole().roleArn
+        createMigrationStringParameter(this, fetchMigrationFargateTask.obtainExecutionRole().roleArn, {
+            ...props,
+            parameter: MigrationSSMParameter.FETCH_MIGRATION_TASK_EXEC_ROLE_ARN,
         });
-
         // Create Fetch Migration Container
         const fetchMigrationContainer = fetchMigrationFargateTask.addContainer("fetchMigrationContainer", {
             image: ContainerImage.fromAsset(join(__dirname, "../../../..", "FetchMigration")),
@@ -86,7 +96,7 @@ export class FetchMigrationStack extends Stack {
         // Create DP pipeline config from template file
         let dpPipelineData: string = readFileSync(props.dpPipelineTemplatePath, 'utf8');
         // Replace only source cluster host - target host will be overridden by inline env var
-        dpPipelineData = dpPipelineData.replace("<SOURCE_CLUSTER_HOST>", props.sourceEndpoint);
+        dpPipelineData = dpPipelineData.replace("<SOURCE_CLUSTER_HOST>", sourceEndpoint);
         // Base64 encode
         let encodedPipeline = Buffer.from(dpPipelineData).toString("base64");
 
@@ -115,10 +125,9 @@ export class FetchMigrationStack extends Stack {
         executionCommand += ` --cluster ${ecsCluster.clusterName} --launch-type FARGATE`
         executionCommand += ` --network-configuration '${networkConfigString}'`
 
-        new StringParameter(this, 'SSMParameterFetchMigrationRunTaskCommand', {
-            description: 'OpenSearch Migration Parameter for CLI command to kick off the Fetch Migration ECS Task',
-            parameterName: `/migration/${props.stage}/${props.defaultDeployId}/fetchMigrationCommand`,
-            stringValue: executionCommand
+        createMigrationStringParameter(this, executionCommand, {
+            ...props,
+            parameter: MigrationSSMParameter.FETCH_MIGRATION_COMMAND,
         });
     }
 }