Add source & target details to CDK

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

@@ -1,5 +1,6 @@
 {
   "engineVersion": "OS_2.9",
+  "targetClusterVersion": "OS_2.9",
   "domainName": "os-service-domain",
   "tlsSecurityPolicy": "TLS_1_2",
   "enforceHTTPS": true,
@@ -15,4 +16,4 @@
   "trafficReplayerServiceEnabled": false,
   "otelCollectorEnabled": true,
   "dpPipelineTemplatePath": "./dp_pipeline_template.yaml"
-}
+}

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

@@ -7,6 +7,7 @@ import { ICertificate } from "aws-cdk-lib/aws-certificatemanager";
 import { IStringParameter, StringParameter } from "aws-cdk-lib/aws-ssm";
 import * as forge from 'node-forge';
 import * as yargs from 'yargs';
+import { ClusterYaml } from "./migration-services-yaml";
 
 
 // parseAndMergeArgs, see @common-utilities.test.ts for an example of different cases
@@ -314,3 +315,125 @@ export enum MigrationSSMParameter {
     TRAFFIC_STREAM_SOURCE_ACCESS_SECURITY_GROUP_ID = 'trafficStreamSourceAccessSecurityGroupId',
     VPC_ID = 'vpcId',
 }
+
+
+export class ClusterNoAuth {};
+
+export class ClusterSigV4Auth {
+    region?: string;
+    serviceSigningName?: string;
+    constructor({region, serviceSigningName: service}: {region: string, serviceSigningName: string}) {
+        this.region = region;
+        this.serviceSigningName = service;
+    }
+}
+
+export class ClusterBasicAuth {
+    username: string;
+    password?: string;
+    password_from_secret_arn?: string;
+
+    constructor({
+        username,
+        password,
+        password_from_secret_arn,
+    }: {
+        username: string;
+        password?: string;
+        password_from_secret_arn?: string;
+    }) {
+        this.username = username;
+        this.password = password;
+        this.password_from_secret_arn = password_from_secret_arn;
+
+        // Validation: Exactly one of password or password_from_secret_arn must be provided
+        if ((password && password_from_secret_arn) || (!password && !password_from_secret_arn)) {
+            throw new Error('Exactly one of password or password_from_secret_arn must be provided');
+        }
+    }
+}
+
+export class ClusterAuth {
+    basicAuth?: ClusterBasicAuth
+    noAuth?: ClusterNoAuth
+    sigv4?: ClusterSigV4Auth
+
+    constructor({basicAuth, noAuth, sigv4}: {basicAuth?: ClusterBasicAuth, noAuth?: ClusterNoAuth, sigv4?: ClusterSigV4Auth}) {
+        this.basicAuth = basicAuth;
+        this.noAuth = noAuth;
+        this.sigv4 = sigv4;
+    }
+
+    validate() {
+        const numDefined = (this.basicAuth? 1 : 0) + (this.noAuth? 1 : 0) + (this.sigv4? 1 : 0)
+        if (numDefined != 1) {
+            throw new Error(`Exactly one authentication method can be defined. ${numDefined} are currently set.`)
+        }
+    }
+
+    toDict() {
+        if (this.basicAuth) {
+            return {basic_auth: this.basicAuth};
+        }
+        if (this.noAuth) {
+            return {no_auth: ""};
+        }
+        if (this.sigv4) {
+            return {sigv4: this.sigv4};
+        }
+        return {};
+    }
+}
+
+function getBasicClusterAuth(basicAuthObject: { [key: string]: any }): ClusterBasicAuth {
+    // Destructure and validate the input object
+    const { username, password, passwordFromSecretArn } = basicAuthObject;
+    // Ensure the required 'username' field is present
+    if (typeof username !== 'string' || !username) {
+        throw new Error('Invalid input: "username" must be a non-empty string');
+    }
+    // Ensure that exactly one of 'password' or 'passwordFromSecretArn' is provided
+    const hasPassword = typeof password === 'string' && password.trim() !== '';
+    const hasPasswordFromSecretArn = typeof passwordFromSecretArn === 'string' && passwordFromSecretArn.trim() !== '';
+    if ((hasPassword && hasPasswordFromSecretArn) || (!hasPassword && !hasPasswordFromSecretArn)) {
+        throw new Error('Exactly one of "password" or "passwordFromSecretArn" must be provided');
+    }
+    return new ClusterBasicAuth({
+        username,
+        password: hasPassword ? password : undefined,
+        password_from_secret_arn: hasPasswordFromSecretArn ? passwordFromSecretArn : undefined,
+    });
+}
+
+function getSigV4ClusterAuth(sigv4AuthObject: { [key: string]: any }): ClusterSigV4Auth {
+    // Destructure and validate the input object
+    const { serviceSigningName, region } = sigv4AuthObject;
+
+    // Create and return the ClusterSigV4Auth object
+    return new ClusterSigV4Auth({serviceSigningName, region});
+}
+
+// Function to parse and validate auth object
+function parseAuth(json: any): ClusterAuth | null {
+    if (json.type === 'basic' && typeof json.username === 'string' && (typeof json.password === 'string' || typeof json.passwordFromSecretArn === 'string') && !(typeof json.password === 'string' && typeof json.passwordFromSecretArn === 'string')) {
+        return new ClusterAuth({basicAuth: getBasicClusterAuth(json)});
+    } else if (json.type === 'sigv4' && typeof json.region === 'string' && typeof json.serviceSigningName === 'string') {
+        return new ClusterAuth({sigv4: getSigV4ClusterAuth(json)});
+    } else if (json.type === 'none') {
+        return new ClusterAuth({noAuth: new ClusterNoAuth()});
+    }
+    return null; // Invalid auth type
+}
+
+export function parseClusterDefinition(json: any): ClusterYaml {
+    const endpoint = json.endpoint
+    const version = json.version
+    if (!endpoint) {
+        throw new Error('Missing required field in cluster definition: endpoint')
+    }
+    const auth = parseAuth(json.auth)
+    if (!auth) {
+        throw new Error(`Invalid auth type when parsing cluster definition: ${json.auth.type}`)
+    }
+    return new ClusterYaml({endpoint, version, auth})
+}

deployment/cdk/opensearch-service-migration/lib/migration-services-yaml.ts

@@ -1,34 +1,26 @@
+import { EngineVersion } from 'aws-cdk-lib/aws-opensearchservice';
+import { ClusterAuth } from './common-utilities';
 import * as yaml from 'yaml';
 
-export class ClusterBasicAuth {
-    username: string;
-    password?: string;
-    password_from_secret_arn?: string;
-
-    constructor({
-        username,
-        password,
-        password_from_secret_arn,
-    }: {
-        username: string;
-        password?: string;
-        password_from_secret_arn?: string;
-    }) {
-        this.username = username;
-        this.password = password;
-        this.password_from_secret_arn = password_from_secret_arn;
-
-        // Validation: Exactly one of password or password_from_secret_arn must be provided
-        if ((password && password_from_secret_arn) || (!password && !password_from_secret_arn)) {
-            throw new Error('Exactly one of password or password_from_secret_arn must be provided');
-        }
-    }
-}
-
 export class ClusterYaml {
     endpoint: string = '';
-    no_auth?: string | null;
-    basic_auth?: ClusterBasicAuth | null;
+    version?: EngineVersion;
+    auth: ClusterAuth;
+
+    constructor({endpoint, auth, version} : {endpoint: string, auth: ClusterAuth, version?: EngineVersion}) {
+        this.endpoint = endpoint;
+        this.auth = auth;
+        this.version = version;
+    }
+    toDict() {
+        return {
+            endpoint: this.endpoint,
+            ...this.auth.toDict(),
+            // TODO: figure out how version should be incorporated
+            // https://opensearch.atlassian.net/browse/MIGRATIONS-1951
+            // version: this.version?.version
+        };
+    }
 }
 
 export class MetricsSourceYaml {
@@ -142,8 +134,8 @@ export class ServicesYaml {
 
     stringify(): string {
         return yaml.stringify({
-            source_cluster: this.source_cluster,
-            target_cluster: this.target_cluster,
+            source_cluster: this.source_cluster?.toDict(),
+            target_cluster: this.target_cluster?.toDict(),
             metrics_source: this.metrics_source,
             backfill: this.backfill?.toDict(),
             snapshot: this.snapshot?.toDict(),

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

@@ -14,7 +14,8 @@ 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";
-import { ClusterBasicAuth, ClusterYaml } from "./migration-services-yaml";
+import { ClusterYaml } from "./migration-services-yaml";
+import { ClusterAuth, ClusterBasicAuth, ClusterNoAuth } from "./common-utilities"
 import { MigrationSSMParameter, createMigrationStringParameter, getMigrationStringParameterValue } from "./common-utilities";
 
 
@@ -115,15 +116,15 @@ export class OpenSearchDomainStack extends Stack {
     }
   }
 
-  generateTargetClusterYaml(domain: Domain, adminUserName: string | undefined, adminUserSecret: ISecret|undefined) {
-    let targetCluster = new ClusterYaml()
-    targetCluster.endpoint = `https://${domain.domainEndpoint}:443`;
+  generateTargetClusterYaml(domain: Domain, adminUserName: string | undefined, adminUserSecret: ISecret|undefined, version: EngineVersion) {
+    let clusterAuth = new ClusterAuth({});
     if (adminUserName) {
-        targetCluster.basic_auth = new ClusterBasicAuth({ username: adminUserName, password_from_secret_arn: adminUserSecret?.secretArn })
+      clusterAuth.basicAuth = new ClusterBasicAuth({ username: adminUserName, password_from_secret_arn: adminUserSecret?.secretArn })
     } else {
-      targetCluster.no_auth = ''
+      clusterAuth.noAuth = new ClusterNoAuth();
     }
-    this.targetClusterYaml = targetCluster;
+     this.targetClusterYaml = new ClusterYaml({endpoint: `https://${domain.domainEndpoint}:443`, auth: clusterAuth, version})
+
   }
 
   constructor(scope: Construct, id: string, props: OpensearchDomainStackProps) {
@@ -229,6 +230,6 @@ export class OpenSearchDomainStack extends Stack {
     });
 
     this.createSSMParameters(domain, adminUserName, adminUserSecret, props.stage, deployId)
-    this.generateTargetClusterYaml(domain, adminUserName, adminUserSecret)
+    this.generateTargetClusterYaml(domain, adminUserName, adminUserSecret, props.version)
   }
 }

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

@@ -16,7 +16,7 @@ import {
 import {StreamingSourceType} from "../streaming-source-type";
 import {LogGroup, RetentionDays} from "aws-cdk-lib/aws-logs";
 import {Fn, RemovalPolicy} from "aws-cdk-lib";
-import {MetadataMigrationYaml, ServicesYaml} from "../migration-services-yaml";
+import {ClusterYaml, MetadataMigrationYaml, ServicesYaml} from "../migration-services-yaml";
 import {ELBTargetGroup, MigrationServiceCore} from "./migration-service-core";
 import { OtelCollectorSidecar } from "./migration-otel-collector-sidecar";
 import { SharedLogFileSystem } from "../components/shared-log-file-system";
@@ -33,7 +33,7 @@ export interface MigrationConsoleProps extends StackPropsExt {
     readonly targetGroups?: ELBTargetGroup[],
     readonly servicesYaml: ServicesYaml,
     readonly otelCollectorEnabled?: boolean,
-    readonly sourceClusterDisabled?: boolean,
+    readonly sourceCluster?: ClusterYaml,
 }
 
 export class MigrationConsoleStack extends MigrationServiceCore {
@@ -151,10 +151,6 @@ export class MigrationConsoleStack extends MigrationServiceCore {
             ...props,
             parameter: MigrationSSMParameter.OS_CLUSTER_ENDPOINT,
         });
-        const sourceClusterEndpoint = props.sourceClusterDisabled ? null : getMigrationStringParameterValue(this, {
-            ...props,
-            parameter: MigrationSSMParameter.SOURCE_CLUSTER_ENDPOINT,
-        });
         const brokerEndpoints = props.streamingSourceType != StreamingSourceType.DISABLED ?
             getMigrationStringParameterValue(this, {
                 ...props,
@@ -236,18 +232,15 @@ export class MigrationConsoleStack extends MigrationServiceCore {
             ]
         })
 
-        const getSecretsPolicy = props.servicesYaml.target_cluster.basic_auth?.password_from_secret_arn ?
-            getTargetPasswordAccessPolicy(props.servicesYaml.target_cluster.basic_auth.password_from_secret_arn) : null;
+        const getTargetSecretsPolicy = props.servicesYaml.target_cluster.auth.basicAuth?.password_from_secret_arn ?
+            getTargetPasswordAccessPolicy(props.servicesYaml.target_cluster.auth.basicAuth?.password_from_secret_arn) : null;
+
+        const getSourceSecretsPolicy = props.servicesYaml.source_cluster?.auth.basicAuth?.password_from_secret_arn ?
+            getTargetPasswordAccessPolicy(props.servicesYaml.source_cluster?.auth.basicAuth?.password_from_secret_arn) : null;
 
         // Upload the services.yaml file to Parameter Store
         let servicesYaml = props.servicesYaml
-        if (!props.sourceClusterDisabled && sourceClusterEndpoint) {
-            servicesYaml.source_cluster = {
-                'endpoint': sourceClusterEndpoint,
-                // TODO: We're not currently supporting auth here, this may need to be handled on the migration console
-                'no_auth': ''
-            }
-        }
+        servicesYaml.source_cluster = props.sourceCluster
         servicesYaml.metadata_migration = new MetadataMigrationYaml();
         if (props.otelCollectorEnabled) {
             const otelSidecarEndpoint = OtelCollectorSidecar.getOtelLocalhostEndpoint();
@@ -277,7 +270,9 @@ export class MigrationConsoleStack extends MigrationServiceCore {
         const openSearchServerlessPolicy = createOpenSearchServerlessIAMAccessPolicy(this.partition, this.region, this.account)
         let servicePolicies = [sharedLogFileSystem.asPolicyStatement(), openSearchPolicy, openSearchServerlessPolicy, ecsUpdateServicePolicy, clusterTasksPolicy,
             listTasksPolicy, artifactS3PublishPolicy, describeVPCPolicy, getSSMParamsPolicy, getMetricsPolicy,
-            ...(getSecretsPolicy ? [getSecretsPolicy] : []) // only add secrets policy if it's non-null
+            // only add secrets policies if they're non-null
+            ...(getTargetSecretsPolicy ? [getTargetSecretsPolicy] : []),
+            ...(getSourceSecretsPolicy ? [getSourceSecretsPolicy] : [])
         ]
         if (props.streamingSourceType === StreamingSourceType.AWS_MSK) {
             const mskAdminPolicies = this.createMSKAdminIAMPolicies(props.stage, props.defaultDeployId)

deployment/cdk/opensearch-service-migration/lib/service-stacks/reindex-from-snapshot-stack.ts

@@ -11,9 +11,10 @@ import {
     createOpenSearchServerlessIAMAccessPolicy,
     getTargetPasswordAccessPolicy,
     getMigrationStringParameterValue,
-    parseAndMergeArgs
+    parseAndMergeArgs,
+    ClusterAuth
 } from "../common-utilities";
-import { ClusterYaml, RFSBackfillYaml, SnapshotYaml } from "../migration-services-yaml";
+import { RFSBackfillYaml, SnapshotYaml } from "../migration-services-yaml";
 import { OtelCollectorSidecar } from "./migration-otel-collector-sidecar";
 import { SharedLogFileSystem } from "../components/shared-log-file-system";
 
@@ -23,7 +24,7 @@ export interface ReindexFromSnapshotProps extends StackPropsExt {
     readonly fargateCpuArch: CpuArchitecture,
     readonly extraArgs?: string,
     readonly otelCollectorEnabled: boolean,
-    readonly clusterAuthDetails: ClusterYaml
+    readonly clusterAuthDetails: ClusterAuth
 }
 
 export class ReindexFromSnapshotStack extends MigrationServiceCore {
@@ -68,24 +69,25 @@ export class ReindexFromSnapshotStack extends MigrationServiceCore {
         });
         const s3Uri = `s3://migration-artifacts-${this.account}-${props.stage}-${this.region}/rfs-snapshot-repo`;
         let rfsCommand = `/rfs-app/runJavaWithClasspath.sh com.rfs.RfsMigrateDocuments --s3-local-dir /tmp/s3_files --s3-repo-uri ${s3Uri} --s3-region ${this.region} --snapshot-name rfs-snapshot --lucene-dir '/lucene' --target-host ${osClusterEndpoint}`
+        rfsCommand = props.clusterAuthDetails.sigv4 ? rfsCommand.concat(`--target-aws-service-signing-name ${props.clusterAuthDetails.sigv4.serviceSigningName} --target-aws-region ${props.clusterAuthDetails.sigv4.region}`) : rfsCommand
         rfsCommand = props.otelCollectorEnabled ? rfsCommand.concat(` --otel-collector-endpoint ${OtelCollectorSidecar.getOtelLocalhostEndpoint()}`) : rfsCommand
         rfsCommand = parseAndMergeArgs(rfsCommand, props.extraArgs);
 
         let targetUser = "";
         let targetPassword = "";
         let targetPasswordArn = "";
-        if (props.clusterAuthDetails.basic_auth) {
-            targetUser = props.clusterAuthDetails.basic_auth.username,
-            targetPassword = props.clusterAuthDetails.basic_auth.password? props.clusterAuthDetails.basic_auth.password : "",
-            targetPasswordArn = props.clusterAuthDetails.basic_auth.password_from_secret_arn? props.clusterAuthDetails.basic_auth.password_from_secret_arn : ""
+        if (props.clusterAuthDetails.basicAuth) {
+            targetUser = props.clusterAuthDetails.basicAuth.username,
+            targetPassword = props.clusterAuthDetails.basicAuth.password || "",
+            targetPasswordArn = props.clusterAuthDetails.basicAuth.password_from_secret_arn || ""
         };
         const sharedLogFileSystem = new SharedLogFileSystem(this, props.stage, props.defaultDeployId);
         const openSearchPolicy = createOpenSearchIAMAccessPolicy(this.partition, this.region, this.account);
         const openSearchServerlessPolicy = createOpenSearchServerlessIAMAccessPolicy(this.partition, this.region, this.account);
         let servicePolicies = [sharedLogFileSystem.asPolicyStatement(), artifactS3PublishPolicy, openSearchPolicy, openSearchServerlessPolicy];
 
-        const getSecretsPolicy = props.clusterAuthDetails.basic_auth?.password_from_secret_arn ?
-            getTargetPasswordAccessPolicy(props.clusterAuthDetails.basic_auth.password_from_secret_arn) : null;
+        const getSecretsPolicy = props.clusterAuthDetails.basicAuth?.password_from_secret_arn ?
+            getTargetPasswordAccessPolicy(props.clusterAuthDetails.basicAuth.password_from_secret_arn) : null;
         if (getSecretsPolicy) {
             servicePolicies.push(getSecretsPolicy);
         }