feat(stepfunctions-tasks): add emr-containers support for calling CreateVirtualCluster, DeleteVirtualCluster, and StartJobRun

packages/@aws-cdk/aws-stepfunctions-tasks/lib/emrcontainers/create-virtual-cluster.ts

@@ -115,7 +115,7 @@ export class EmrContainersEksCreateVirtualCluster extends sfn.TaskStateBase {
     return {
       Resource: integrationResourceArn('emr-containers', 'createVirtualCluster', this.integrationPattern),
       Parameters: sfn.FieldUtils.renderObject({
-        Name: this.props.virtualClusterName ?? `${sfn.JsonPath.stringAt('$$.Execution.Name')}`,
+        Name: this.props.virtualClusterName ?? sfn.JsonPath.stringAt('States.Format(\'{}/{}\', $$.Execution.Name, $$.StateMachine.Name)'),
         ContainerProvider: {
           Id: this.props.eksCluster.clusterName,
           Info: {
@@ -125,15 +125,11 @@ export class EmrContainersEksCreateVirtualCluster extends sfn.TaskStateBase {
           },
           Type: ContainerProviderTypes.EKS,
         },
-        ...(this.props.tags ? this.renderTags(this.props.tags) : undefined),
+        Tags: this.props.tags,
       }),
     };
   };
 
-  private renderTags(tags?: { [key: string]: any } | undefined): { [key: string]: any } {
-    return tags ? { Tags: Object.entries(tags).map(([key, value]) => ({ Key: key, Value: value })) } : {};
-  }
-
   private createPolicyStatements(): iam.PolicyStatement[] {
     return [
       new iam.PolicyStatement({

packages/@aws-cdk/aws-stepfunctions-tasks/lib/emrcontainers/start-job-run.ts

@@ -18,7 +18,7 @@ export interface EmrContainersStartJobRunProps extends sfn.TaskStateBaseProps {
   /**
    * The ID of the virtual cluster where the job will be run
    */
-  readonly virtualClusterId: sfn.TaskInput;
+  readonly virtualCluster: VirtualClusterInput;
 
   /**
    * The name of the job run.
@@ -119,8 +119,8 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
     }
 
     if (props.jobDriver.sparkSubmitJobDriver?.entryPointArguments
-       && (typeof props.jobDriver.sparkSubmitJobDriver.entryPointArguments.value === 'string'
-       && !sfn.JsonPath.isEncodedJsonPath(props.jobDriver.sparkSubmitJobDriver.entryPointArguments.value))) {
+       && typeof props.jobDriver.sparkSubmitJobDriver.entryPointArguments.value === 'string'
+       && !sfn.JsonPath.isEncodedJsonPath(props.jobDriver.sparkSubmitJobDriver.entryPointArguments.value)) {
       throw new Error(`Entry point arguments must be a string array. Received ${typeof props.jobDriver.sparkSubmitJobDriver.entryPointArguments.value}.`);
     }
 
@@ -143,7 +143,7 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
     }
 
     if (props.executionRole === undefined
-      && sfn.JsonPath.isEncodedJsonPath(props.virtualClusterId.value)) {
+      && sfn.JsonPath.isEncodedJsonPath(props.virtualCluster.id)) {
       throw new Error('Execution role cannot be undefined when the virtual cluster ID is not a concrete value. Provide an execution role with the correct trust policy');
     }
 
@@ -164,7 +164,7 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
     return {
       Resource: integrationResourceArn('emr-containers', 'startJobRun', this.integrationPattern),
       Parameters: sfn.FieldUtils.renderObject({
-        VirtualClusterId: this.props.virtualClusterId.value,
+        VirtualClusterId: this.props.virtualCluster.id,
         Name: this.props.jobName,
         ExecutionRoleArn: this.role.roleArn,
         ReleaseLabel: this.props.releaseLabel.label,
@@ -177,20 +177,20 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
         },
         ConfigurationOverrides: {
           ApplicationConfiguration: cdk.listMapper(this.applicationConfigPropertyToJson)(this.props.applicationConfig),
-          MonitoringConfiguration: this.props.monitoring ? {
+          MonitoringConfiguration: {
             CloudWatchMonitoringConfiguration: this.logGroup ? {
-              LogGroupName: this.logGroup?.logGroupName, // automatically generated name https://docs.aws.amazon.com/cdk/api/latest/typescript/api/aws-logs/loggroup.html#aws_logs_LogGroup_synopsis
-              LogStreamNamePrefix: this.props.monitoring?.logStreamNamePrefix,
+              LogGroupName: this.logGroup.logGroupName,
+              LogStreamNamePrefix: this.props.monitoring!.logStreamNamePrefix,
             } : undefined,
             PersistentAppUI: (this.props.monitoring?.persistentAppUI === false)
               ? 'DISABLED'
               : 'ENABLED',
             S3MonitoringConfiguration: this.logBucket ? {
-              LogUri: this.logBucket?.s3UrlForObject(), // automatically generated unique name https://docs.aws.amazon.com/cdk/api/latest/docs/@aws-cdk_aws-s3.Bucket.html#bucketname
+              LogUri: this.logBucket.s3UrlForObject(),
             } : undefined,
-          } : undefined,
+          },
         },
-        ...(this.props.tags ? this.renderTags(this.props.tags) : undefined),
+        Tags: this.props.tags,
       }),
     };
   }
@@ -229,10 +229,6 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
     return Array.isArray(value) && value.every(item => typeof item === 'string');
   }
 
-  private renderTags(tags?: { [key: string]: any } | undefined): { [key: string]: any } {
-    return tags ? { Tags: Object.entries(tags).map(([key, value]) => ({ Key: key, Value: value })) } : {};
-  }
-
   // https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/creating-job-execution-role.html
   private createJobExecutionRole(): iam.Role {
     const jobExecutionRole = new iam.Role(this, 'Job-Execution-Role', {
@@ -296,7 +292,7 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
         service: 'EMRcontainers',
         action: 'describeVirtualCluster',
         parameters: {
-          id: this.props.virtualClusterId.value,
+          id: this.props.virtualCluster.id,
         },
         outputPaths: ['virtualCluster.containerProvider.info.eksInfo.namespace', 'virtualCluster.containerProvider.id'],
         physicalResourceId: cr.PhysicalResourceId.of('id'),
@@ -360,7 +356,7 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
             arnFormat: cdk.ArnFormat.SLASH_RESOURCE_SLASH_RESOURCE_NAME,
             service: 'emr-containers',
             resource: 'virtualclusters',
-            resourceName: sfn.JsonPath.isEncodedJsonPath(this.props.virtualClusterId.value) ? '*' : this.props.virtualClusterId.value, // Need wild card for dynamic start job run https://docs.aws.amazon.com/step-functions/latest/dg/emr-eks-iam.html
+            resourceName: sfn.JsonPath.isEncodedJsonPath(this.props.virtualCluster.id) ? '*' : this.props.virtualCluster.id, // Need wild card for dynamic start job run https://docs.aws.amazon.com/step-functions/latest/dg/emr-eks-iam.html
           }),
         ],
         actions: ['emr-containers:StartJobRun'],
@@ -380,7 +376,7 @@ export class EmrContainersStartJobRun extends sfn.TaskStateBase implements iam.I
               arnFormat: cdk.ArnFormat.SLASH_RESOURCE_SLASH_RESOURCE_NAME,
               service: 'emr-containers',
               resource: 'virtualclusters',
-              resourceName: sfn.JsonPath.isEncodedJsonPath(this.props.virtualClusterId.value) ? '*' : `${this.props.virtualClusterId.value}/jobruns/*`, // Need wild card for dynamic start job run https://docs.aws.amazon.com/step-functions/latest/dg/emr-eks-iam.html
+              resourceName: sfn.JsonPath.isEncodedJsonPath(this.props.virtualCluster.id) ? '*' : `${this.props.virtualCluster.id}/jobruns/*`, // Need wild card for dynamic start job run https://docs.aws.amazon.com/step-functions/latest/dg/emr-eks-iam.html
             }),
           ],
           actions: [

packages/@aws-cdk/aws-stepfunctions-tasks/test/emrcontainers/create-virtual-cluster.test.ts

@@ -18,7 +18,6 @@ describe('Invoke emr-containers CreateVirtualCluster with ', () => {
     // WHEN
     const task = new EmrContainersEksCreateVirtualCluster(stack, 'Task', {
       eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
-      integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
     });
 
     new sfn.StateMachine(stack, 'SM', {
@@ -42,7 +41,7 @@ describe('Invoke emr-containers CreateVirtualCluster with ', () => {
       },
       End: true,
       Parameters: {
-        'Name.$': '$$.Execution.Name',
+        'Name.$': "States.Format('{}/{}', $$.Execution.Name, $$.StateMachine.Name)",
         'ContainerProvider': {
           Id: clusterId,
           Info: {
@@ -80,7 +79,6 @@ describe('Invoke emr-containers CreateVirtualCluster with ', () => {
               Namespace: 'namespace',
             },
           },
-          Type: 'EKS',
         },
       },
     });
@@ -89,23 +87,14 @@ describe('Invoke emr-containers CreateVirtualCluster with ', () => {
   test('clusterId from payload', () => {
     // WHEN
     const task = new EmrContainersEksCreateVirtualCluster(stack, 'Task', {
-      virtualClusterName: emrContainersVirtualClusterName,
       eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromJsonPathAt('$.ClusterId')),
-      integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
     });
 
     // THEN
     expect(stack.resolve(task.toStateJson())).toMatchObject({
       Parameters: {
-        Name: emrContainersVirtualClusterName,
         ContainerProvider: {
           'Id.$': '$.ClusterId',
-          'Info': {
-            EksInfo: {
-              Namespace: 'default',
-            },
-          },
-          'Type': 'EKS',
         },
       },
     });
@@ -118,25 +107,16 @@ describe('Invoke emr-containers CreateVirtualCluster with ', () => {
     });
 
     const task = new EmrContainersEksCreateVirtualCluster(stack, 'Task', {
-      virtualClusterName: emrContainersVirtualClusterName,
       eksCluster: EksClusterInput.fromCluster(eksCluster),
-      integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
     });
 
     // THEN
     expect(stack.resolve(task.toStateJson())).toMatchObject({
       Parameters: {
-        Name: emrContainersVirtualClusterName,
         ContainerProvider: {
           Id: {
             Ref: 'EKSClusterEDAD5FD1',
           },
-          Info: {
-            EksInfo: {
-              Namespace: 'default',
-            },
-          },
-          Type: 'EKS',
         },
       },
     });
@@ -146,11 +126,9 @@ describe('Invoke emr-containers CreateVirtualCluster with ', () => {
     // WHEN
     const task = new EmrContainersEksCreateVirtualCluster(stack, 'Task', {
       eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
-      virtualClusterName: emrContainersVirtualClusterName,
       tags: {
         key: 'value',
       },
-      integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
     });
 
     new sfn.StateMachine(stack, 'SM', {
@@ -160,20 +138,9 @@ describe('Invoke emr-containers CreateVirtualCluster with ', () => {
     // THEN
     expect(stack.resolve(task.toStateJson())).toMatchObject({
       Parameters: {
-        Name: emrContainersVirtualClusterName,
-        ContainerProvider: {
-          Id: clusterId,
-          Info: {
-            EksInfo: {
-              Namespace: 'default',
-            },
-          },
-          Type: 'EKS',
+        Tags: {
+          key: 'value',
         },
-        Tags: [{
-          Key: 'key',
-          Value: 'value',
-        }],
       },
     });
   });
@@ -184,7 +151,6 @@ test('Permitted role actions included for CreateVirtualCluster if service integr
   const task = new EmrContainersEksCreateVirtualCluster(stack, 'Task', {
     virtualClusterName: emrContainersVirtualClusterName,
     eksCluster: EksClusterInput.fromTaskInput(sfn.TaskInput.fromText(clusterId)),
-    integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
   });
 
   new sfn.StateMachine(stack, 'SM', {

packages/@aws-cdk/aws-stepfunctions-tasks/test/emrcontainers/delete-virtual-cluster.test.ts

@@ -16,7 +16,6 @@ describe('Invoke EMR Containers Delete Virtual cluster with ', () => {
     // WHEN
     const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
       virtualClusterId: sfn.TaskInput.fromText(virtualClusterId),
-      integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
     });
 
     // THEN
@@ -74,7 +73,6 @@ describe('Invoke EMR Containers Delete Virtual cluster with ', () => {
     // WHEN
     const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
       virtualClusterId: sfn.TaskInput.fromJsonPathAt('$.VirtualClusterId'),
-      integrationPattern: sfn.IntegrationPattern.RUN_JOB,
     });
 
     // THEN
@@ -91,7 +89,6 @@ describe('Valid policy statements and resources are passed ', () => {
     // WHEN
     const task = new EmrContainersDeleteVirtualCluster(stack, 'Task', {
       virtualClusterId: sfn.TaskInput.fromText(virtualClusterId),
-      integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
     });
 
     new sfn.StateMachine(stack, 'SM', {