Merge branch 'main' into ReplayerRetries

DocumentsFromSnapshotMigration/src/test/resources/log4j2.properties

@@ -1,5 +1,27 @@
 status = ERROR
 
+property.logsDir = ${env:SHARED_LOGS_DIR_PATH:-./logs}
+property.failedLoggerFileNamePrefix = ${logsDir}/${hostName}/failedRequests/failedRequests
+
+appenders = console, FailedRequests
+
+appender.FailedRequests.type = RollingRandomAccessFile
+appender.FailedRequests.name = FailedRequests
+appender.FailedRequests.fileName = ${failedLoggerFileNamePrefix}.log
+appender.FailedRequests.filePattern = ${failedLoggerFileNamePrefix}-%d{yyyy-MM-dd-HH:mm}{UTC}-%i.log.gz
+appender.FailedRequests.layout.type = PatternLayout
+appender.FailedRequests.layout.pattern = %m%n
+appender.FailedRequests.policies.type = Policies
+appender.FailedRequests.policies.size.type = SizeBasedTriggeringPolicy
+appender.FailedRequests.policies.size.size = 10 MB
+appender.FailedRequests.strategy.type = DefaultRolloverStrategy
+appender.FailedRequests.immediateFlush = false
+
+logger.FailedRequestsLogger.name = FailedRequestsLogger
+logger.FailedRequestsLogger.level = info
+logger.FailedRequestsLogger.additivity = false
+logger.FailedRequestsLogger.appenderRef.FailedRequests.ref = FailedRequests
+
 appender.console.type = Console
 appender.console.name = Console
 appender.console.target = SYSTEM_OUT

RFS/src/main/resources/log4j2.properties

@@ -1,6 +1,27 @@
 status = WARN
 
-appenders = console
+property.logsDir = ${env:SHARED_LOGS_DIR_PATH:-./logs}
+property.failedLoggerFileNamePrefix = ${logsDir}/${hostName}/failedRequests/failedRequests
+
+appenders = console, FailedRequests
+
+appender.FailedRequests.type = RollingRandomAccessFile
+appender.FailedRequests.name = FailedRequests
+appender.FailedRequests.fileName = ${failedLoggerFileNamePrefix}.log
+appender.FailedRequests.filePattern = ${failedLoggerFileNamePrefix}-%d{yyyy-MM-dd-HH:mm}{UTC}-%i.log.gz
+appender.FailedRequests.layout.type = PatternLayout
+appender.FailedRequests.layout.pattern = %m%n
+appender.FailedRequests.policies.type = Policies
+appender.FailedRequests.policies.size.type = SizeBasedTriggeringPolicy
+appender.FailedRequests.policies.size.size = 10 MB
+appender.FailedRequests.strategy.type = DefaultRolloverStrategy
+appender.FailedRequests.immediateFlush = false
+
+logger.FailedRequestsLogger.name = FailedRequestsLogger
+logger.FailedRequestsLogger.level = info
+logger.FailedRequestsLogger.additivity = false
+logger.FailedRequestsLogger.appenderRef.FailedRequests.ref = FailedRequests
+
 
 appender.console.type = Console
 appender.console.name = Console

TrafficCapture/dockerSolution/src/main/docker/docker-compose.yml

@@ -70,9 +70,9 @@ services:
     networks:
       - migrations
     volumes:
-      - sharedReplayerOutput:/shared-replayer-output
+      - sharedLogsVolume:/shared-logs-output
     environment:
-      - TUPLE_DIR_PATH=/shared-replayer-output/traffic-replayer-default
+      - SHARED_LOGS_DIR_PATH=/shared-logs-output/traffic-replayer-default
     depends_on:
       kafka:
         condition: service_started
@@ -96,7 +96,7 @@ services:
     networks:
       - migrations
     volumes:
-      - sharedReplayerOutput:/shared-replayer-output
+      - sharedLogsVolume:/shared-logs-output
       - ./migrationConsole/lib/console_link/services.yaml:/etc/migration_services.yaml
       # this is a convenience thing for testing -- it should be removed before this makes it to prod.
       # - ./migrationConsole/lib/console_link:/root/lib/console_link
@@ -113,7 +113,7 @@ services:
 volumes:
   sharedComparatorSqlResults:
     driver: local
-  sharedReplayerOutput:
+  sharedLogsVolume:
     driver: local
   grafana_data:
     driver: local

TrafficCapture/trafficReplayer/src/main/resources/log4j2.properties

@@ -1,6 +1,7 @@
 status = warn
 
-property.tupleDir = ${env:TUPLE_DIR_PATH:-./logs/tuples}
+property.tupleDir = ${env:SHARED_LOGS_DIR_PATH:-./logs}
+property.tuplesFileNamePrefix = ${logsDir}/${hostName}/tuples/tuples
 
 appenders = console, ReplayerLogFile, OUTPUT_TUPLES, TRANSACTION_SUMMARIES, TRANSACTION_SUMMARIES_LOGFILE, ALL_ACTIVE_WORK_MONITOR_LOGFILE
 
@@ -27,8 +28,8 @@ appender.ReplayerLogFile.immediateFlush = false
 
 appender.OUTPUT_TUPLES.type = RollingRandomAccessFile
 appender.OUTPUT_TUPLES.name = OUTPUT_TUPLES
-appender.OUTPUT_TUPLES.fileName = ${tupleDir}/tuples.log
-appender.OUTPUT_TUPLES.filePattern = ${tupleDir}/tuples-%d{yyyy-MM-dd-HH:mm}{UTC}-%i.log.gz
+appender.OUTPUT_TUPLES.fileName = ${tuplesFileNamePrefix}.log
+appender.OUTPUT_TUPLES.filePattern = ${tuplesFileNamePrefix}-%d{yyyy-MM-dd-HH:mm}{UTC}-%i.log.gz
 appender.OUTPUT_TUPLES.layout.type = PatternLayout
 appender.OUTPUT_TUPLES.layout.pattern = %m%n
 appender.OUTPUT_TUPLES.policies.type = Policies

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

@@ -308,8 +308,8 @@ export enum MigrationSSMParameter {
     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',
+    SHARED_LOGS_SECURITY_GROUP_ID = 'sharedLogsSecurityGroupId',
+    SHARED_LOGS_EFS_ID = 'sharedLogsEfsId',
     SOURCE_CLUSTER_ENDPOINT = 'sourceClusterEndpoint',
     SERVICE_SECURITY_GROUP_ID = 'serviceSecurityGroupId',
     SERVICES_YAML_FILE = 'servicesYamlFile',

deployment/cdk/opensearch-service-migration/lib/components/shared-log-file-system.ts

@@ -0,0 +1,49 @@
+import { MountPoint, Volume } from 'aws-cdk-lib/aws-ecs';
+import { Effect, PolicyStatement } from 'aws-cdk-lib/aws-iam';
+import { MigrationSSMParameter, getMigrationStringParameterValue } from '../common-utilities';
+import { MigrationServiceCore } from '../service-stacks';
+
+
+export class SharedLogFileSystem {
+
+    readonly volumeId: string;
+    readonly volumeName = "sharedLogsVolume";
+    public readonly mountPointPath = "/shared-logs-output";
+    constructor(private stack: MigrationServiceCore, stage: string, defaultDeployId: string) {
+        this.volumeId = getMigrationStringParameterValue(stack, {
+            stage,
+            defaultDeployId,
+            parameter: MigrationSSMParameter.SHARED_LOGS_EFS_ID,
+        })
+    }
+
+    asVolume(): Volume {
+        return {
+            name: this.volumeName,
+            efsVolumeConfiguration: {
+                fileSystemId: this.volumeId,
+                transitEncryption: "ENABLED"
+            }
+        };
+    }
+
+    asMountPoint(): MountPoint {
+        return {
+            containerPath: this.mountPointPath,
+            readOnly: false,
+            sourceVolume: this.volumeName
+        };
+    }
+
+    asPolicyStatement(): PolicyStatement {
+        const sharedLogFileSystemArn = `arn:${this.stack.partition}:elasticfilesystem:${this.stack.region}:${this.stack.account}:file-system/${this.volumeId}`
+        return new PolicyStatement({
+            effect: Effect.ALLOW,
+            resources: [sharedLogFileSystemArn],
+            actions: [
+                "elasticfilesystem:ClientMount",
+                "elasticfilesystem:ClientWrite"
+            ]
+        });
+    }
+}

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

@@ -173,26 +173,26 @@ export class MigrationAssistanceStack extends Stack {
             this.createMSKResources(props, streamingSecurityGroup)
         }
 
-        const replayerOutputSG = new SecurityGroup(this, 'replayerOutputSG', {
+        const sharedLogsSG = new SecurityGroup(this, 'sharedLogsSG', {
             vpc: props.vpc,
             allowAllOutbound: false,
         });
-        replayerOutputSG.addIngressRule(replayerOutputSG, Port.allTraffic());
+        sharedLogsSG.addIngressRule(sharedLogsSG, Port.allTraffic());
 
-        createMigrationStringParameter(this, replayerOutputSG.securityGroupId, {
+        createMigrationStringParameter(this, sharedLogsSG.securityGroupId, {
             ...props,
-            parameter: MigrationSSMParameter.REPLAYER_OUTPUT_ACCESS_SECURITY_GROUP_ID
+            parameter: MigrationSSMParameter.SHARED_LOGS_SECURITY_GROUP_ID
         });
 
         // Create an EFS file system for Traffic Replayer output
-        const replayerOutputEFS = new FileSystem(this, 'replayerOutputEFS', {
+        const sharedLogsEFS = new FileSystem(this, 'sharedLogsEFS', {
             vpc: props.vpc,
-            securityGroup: replayerOutputSG,
+            securityGroup: sharedLogsSG,
             removalPolicy: replayerEFSRemovalPolicy
         });
-        createMigrationStringParameter(this, replayerOutputEFS.fileSystemId, {
+        createMigrationStringParameter(this, sharedLogsEFS.fileSystemId, {
             ...props,
-            parameter: MigrationSSMParameter.REPLAYER_OUTPUT_EFS_ID
+            parameter: MigrationSSMParameter.SHARED_LOGS_EFS_ID
         });
 
         const serviceSecurityGroup = new SecurityGroup(this, 'serviceSecurityGroup', {

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

@@ -19,6 +19,7 @@ import {Fn, RemovalPolicy} from "aws-cdk-lib";
 import {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";
 
 export interface MigrationConsoleProps extends StackPropsExt {
     readonly migrationsSolutionVersion: string,
@@ -135,7 +136,7 @@ export class MigrationConsoleStack extends MigrationServiceCore {
             { id: "serviceSG", param: MigrationSSMParameter.SERVICE_SECURITY_GROUP_ID },
             { id: "trafficStreamSourceAccessSG", param: MigrationSSMParameter.TRAFFIC_STREAM_SOURCE_ACCESS_SECURITY_GROUP_ID },
             { id: "defaultDomainAccessSG", param: MigrationSSMParameter.OS_ACCESS_SECURITY_GROUP_ID },
-            { id: "replayerOutputAccessSG", param: MigrationSSMParameter.REPLAYER_OUTPUT_ACCESS_SECURITY_GROUP_ID }
+            { id: "sharedLogsAccessSG", param: MigrationSSMParameter.SHARED_LOGS_SECURITY_GROUP_ID }
         ].map(({ id, param }) =>
             SecurityGroup.fromSecurityGroupId(this, id, getMigrationStringParameterValue(this, {
                 ...props,
@@ -160,32 +161,8 @@ export class MigrationConsoleStack extends MigrationServiceCore {
                 parameter: MigrationSSMParameter.KAFKA_BROKERS,
             }) : "";
 
-        const volumeName = "sharedReplayerOutputVolume"
-        const volumeId = getMigrationStringParameterValue(this, {
-            ...props,
-            parameter: MigrationSSMParameter.REPLAYER_OUTPUT_EFS_ID,
-        });
-        const replayerOutputEFSVolume: Volume = {
-            name: volumeName,
-            efsVolumeConfiguration: {
-                fileSystemId: volumeId,
-                transitEncryption: "ENABLED"
-            }
-        };
-        const replayerOutputMountPoint: MountPoint = {
-            containerPath: "/shared-replayer-output",
-            readOnly: false,
-            sourceVolume: volumeName
-        }
-        const replayerOutputEFSArn = `arn:${this.partition}:elasticfilesystem:${this.region}:${this.account}:file-system/${volumeId}`
-        const replayerOutputMountPolicy = new PolicyStatement( {
-            effect: Effect.ALLOW,
-            resources: [replayerOutputEFSArn],
-            actions: [
-                "elasticfilesystem:ClientMount",
-                "elasticfilesystem:ClientWrite"
-            ]
-        })
+        const sharedLogFileSystem = new SharedLogFileSystem(this, props.stage, props.defaultDeployId);
+
 
         const ecsClusterArn = `arn:${this.partition}:ecs:${this.region}:${this.account}:service/migration-${props.stage}-ecs-cluster`
         const allReplayerServiceArn = `${ecsClusterArn}/migration-${props.stage}-traffic-replayer*`
@@ -293,11 +270,12 @@ export class MigrationConsoleStack extends MigrationServiceCore {
             "MIGRATION_SOLUTION_VERSION": props.migrationsSolutionVersion,
             "MIGRATION_SERVICES_YAML_PARAMETER": parameter.parameterName,
             "MIGRATION_SERVICES_YAML_HASH": hashStringSHA256(servicesYaml.stringify()),
+            "SHARED_LOGS_DIR_PATH": `${sharedLogFileSystem.mountPointPath}/migration-console-${props.defaultDeployId}`,
         }
 
         const openSearchPolicy = createOpenSearchIAMAccessPolicy(this.partition, this.region, this.account)
         const openSearchServerlessPolicy = createOpenSearchServerlessIAMAccessPolicy(this.partition, this.region, this.account)
-        let servicePolicies = [replayerOutputMountPolicy, openSearchPolicy, openSearchServerlessPolicy, ecsUpdateServicePolicy, clusterTasksPolicy,
+        let servicePolicies = [sharedLogFileSystem.asPolicyStatement(), openSearchPolicy, openSearchServerlessPolicy, ecsUpdateServicePolicy, clusterTasksPolicy,
             listTasksPolicy, artifactS3PublishPolicy, describeVPCPolicy, getSSMParamsPolicy, getMetricsPolicy,
             ...(getSecretsPolicy ? [getSecretsPolicy] : []) // only add secrets policy if it's non-null
         ]
@@ -388,8 +366,8 @@ export class MigrationConsoleStack extends MigrationServiceCore {
             securityGroups: securityGroups,
             portMappings: servicePortMappings,
             dockerImageCommand: imageCommand,
-            volumes: [replayerOutputEFSVolume],
-            mountPoints: [replayerOutputMountPoint],
+            volumes: [sharedLogFileSystem.asVolume()],
+            mountPoints: [sharedLogFileSystem.asMountPoint()],
             environment: environment,
             taskRolePolicies: servicePolicies,
             cpuArchitecture: props.fargateCpuArch,

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

@@ -15,6 +15,7 @@ import {
 } from "../common-utilities";
 import { ClusterYaml, RFSBackfillYaml, SnapshotYaml } from "../migration-services-yaml";
 import { OtelCollectorSidecar } from "./migration-otel-collector-sidecar";
+import { SharedLogFileSystem } from "../components/shared-log-file-system";
 
 
 export interface ReindexFromSnapshotProps extends StackPropsExt {
@@ -41,6 +42,10 @@ export class ReindexFromSnapshotStack extends MigrationServiceCore {
                 ...props,
                 parameter: MigrationSSMParameter.OS_ACCESS_SECURITY_GROUP_ID,
             })),
+            SecurityGroup.fromSecurityGroupId(this, "sharedLogsAccessSG", getMigrationStringParameterValue(this, {
+                ...props,
+                parameter: MigrationSSMParameter.SHARED_LOGS_SECURITY_GROUP_ID,
+            })),
         ]
 
         const artifactS3Arn = getMigrationStringParameterValue(this, {
@@ -74,10 +79,10 @@ export class ReindexFromSnapshotStack extends MigrationServiceCore {
             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 : ""
         };
-
+        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 = [artifactS3PublishPolicy, openSearchPolicy, openSearchServerlessPolicy];
+        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;
@@ -91,6 +96,8 @@ export class ReindexFromSnapshotStack extends MigrationServiceCore {
             dockerDirectoryPath: join(__dirname, "../../../../../", "DocumentsFromSnapshotMigration/docker"),
             dockerImageCommand: ['/bin/sh', '-c', "/rfs-app/entrypoint.sh"],
             securityGroups: securityGroups,
+            volumes: [sharedLogFileSystem.asVolume()],
+            mountPoints: [sharedLogFileSystem.asMountPoint()],
             taskRolePolicies: servicePolicies,
             cpuArchitecture: props.fargateCpuArch,
             taskCpuUnits: 2048,
@@ -101,6 +108,7 @@ export class ReindexFromSnapshotStack extends MigrationServiceCore {
                 "RFS_TARGET_USER": targetUser,
                 "RFS_TARGET_PASSWORD": targetPassword,
                 "RFS_TARGET_PASSWORD_ARN": targetPasswordArn,
+                "SHARED_LOGS_DIR_PATH": `${sharedLogFileSystem.mountPointPath}/reindex-from-snapshot-${props.defaultDeployId}`,
             },
             ...props
         });