feat(scheduler-targets-alpha): SageMakerStartPipelineExecution Target

packages/@aws-cdk/aws-scheduler-targets-alpha/README.md

@@ -34,6 +34,7 @@ The following targets are supported:
 8. `targets.KinesisStreamPutRecord`: [Put a record to an Amazon Kinesis Data Streams](#put-a-record-to-an-amazon-kinesis-data-streams)
 9. `targets.KinesisDataFirehosePutRecord`: [Put a record to a Kinesis Data Firehose](#put-a-record-to-a-kinesis-data-firehose)
 10. `targets.CodePipelineStartPipelineExecution`: [Start a CodePipeline execution](#start-a-codepipeline-execution)
+11. `targets.SageMakerStartPipelineExecution`: [Start a Sagemaker pipeline execution](#start-a-sagemaker-pipeline-execution)
 
 ## Invoke a Lambda function
 
@@ -289,3 +290,26 @@ new Schedule(this, 'Schedule', {
   target: new targets.CodePipelineStartPipelineExecution(pipeline),
 });
 ```
+
+## Start a Sagemaker pipeline execution
+
+Use the `SageMakerStartPipelineExecution` target to start a new execution for a Sagemaker pipeline.
+
+The code snippet below creates an event rule with a Sagemaker pipeline as target which is
+called every hour by Event Bridge Scheduler.
+
+```ts
+import * as sagemaker from 'aws-cdk-lib/aws-sagemaker';
+
+declare const pipeline: sagemaker.CfnPipeline;
+
+new Schedule(this, 'Schedule', {
+  schedule: ScheduleExpression.rate(Duration.minutes(60)),
+  target: new targets.CodePipelineStartPipelineExecution(pipeline, {
+    pipelineParameterList: [{
+      name: 'parameter-name',
+      value: 'parameter-value',
+    }],
+  }),
+});
+```

packages/@aws-cdk/aws-scheduler-targets-alpha/lib/index.ts

@@ -5,6 +5,7 @@ export * from './inspector-start-assessment-run';
 export * from './kinesis-data-firehose-put-record';
 export * from './kinesis-stream-put-record';
 export * from './lambda-invoke';
+export * from './sage-maker-start-pipeline-execution';
 export * from './sns-publish';
 export * from './sqs-send-message';
 export * from './stepfunctions-start-execution';

packages/@aws-cdk/aws-scheduler-targets-alpha/lib/sage-maker-start-pipeline-execution.ts

@@ -0,0 +1,95 @@
+import { ISchedule, IScheduleTarget, ScheduleTargetConfig } from '@aws-cdk/aws-scheduler-alpha';
+import { ArnFormat, Names, Stack } from 'aws-cdk-lib';
+import { IRole, PolicyStatement } from 'aws-cdk-lib/aws-iam';
+import { CfnPipeline } from 'aws-cdk-lib/aws-sagemaker';
+import { ScheduleTargetBase, ScheduleTargetBaseProps } from './target';
+import { sameEnvDimension } from './util';
+
+/**
+ * Properties for a pipeline parameter
+ */
+export interface SageMakerPipelineParameter {
+  /**
+   * Name of parameter to start execution of a SageMaker Model Building Pipeline.
+   */
+  readonly name: string;
+
+  /**
+   * Value of parameter to start execution of a SageMaker Model Building Pipeline.
+   */
+  readonly value: string;
+}
+
+/**
+ * Properties for a SageMaker Target
+ */
+export interface SageMakerStartPipelineExecutionProps extends ScheduleTargetBaseProps {
+  /**
+   * List of parameter names and values to use when executing the SageMaker Model Building Pipeline.
+   *
+   * The length must be between 0 and 200.
+   *
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-scheduler-schedule-sagemakerpipelineparameters.html#cfn-scheduler-schedule-sagemakerpipelineparameters-pipelineparameterlist
+   *
+   * @default - no pipeline parameter list
+   */
+  readonly pipelineParameterList?: SageMakerPipelineParameter[];
+}
+
+/**
+ * Use a SageMaker pipeline as a target for AWS EventBridge Scheduler.
+ */
+export class SageMakerStartPipelineExecution extends ScheduleTargetBase implements IScheduleTarget {
+  private readonly pipelineArn: string;
+  constructor(
+    private readonly pipeline: CfnPipeline,
+    private readonly props: SageMakerStartPipelineExecutionProps = {},
+  ) {
+    const targetArn = Stack.of(pipeline).formatArn({
+      service: 'sagemaker',
+      resource: 'pipeline',
+      resourceName: pipeline.pipelineName,
+      arnFormat: ArnFormat.SLASH_RESOURCE_NAME,
+    });
+    super(props, targetArn);
+    this.pipelineArn = targetArn;
+
+    if (props.pipelineParameterList !== undefined && props.pipelineParameterList.length > 200) {
+      throw new Error(`pipelineParameterList length must be between 0 and 200, got ${props.pipelineParameterList.length}`);
+    }
+  }
+
+  protected addTargetActionToRole(schedule: ISchedule, role: IRole): void {
+    if (!sameEnvDimension(this.pipeline.stack.region, schedule.env.region)) {
+      throw new Error(`Cannot assign pipeline in region ${this.pipeline.stack.region} to the schedule ${Names.nodeUniqueId(schedule.node)} in region ${schedule.env.region}. Both the schedule and the pipeline must be in the same region.`);
+    }
+
+    if (!sameEnvDimension(this.pipeline.stack.account, schedule.env.account)) {
+      throw new Error(`Cannot assign pipeline in account ${this.pipeline.stack.account} to the schedule ${Names.nodeUniqueId(schedule.node)} in account ${schedule.env.region}. Both the schedule and the pipeline must be in the same account.`);
+    }
+
+    if (this.props.role && !sameEnvDimension(this.props.role.env.account, this.pipeline.stack.account)) {
+      throw new Error(`Cannot grant permission to execution role in account ${this.props.role.env.account} to invoke target ${Names.nodeUniqueId(this.pipeline.node)} in account ${this.pipeline.stack.account}. Both the target and the execution role must be in the same account.`);
+    }
+
+    role.addToPrincipalPolicy(new PolicyStatement({
+      actions: ['sagemaker:StartPipelineExecution'],
+      resources: [this.pipelineArn],
+    }));
+  }
+
+  protected bindBaseTargetConfig(_schedule: ISchedule): ScheduleTargetConfig {
+    const sageMakerPipelineParameters = this.props.pipelineParameterList ? {
+      pipelineParameterList: this.props.pipelineParameterList.map(param => {
+        return {
+          name: param.name,
+          value: param.value,
+        };
+      }),
+    } : undefined;
+    return {
+      ...super.bindBaseTargetConfig(_schedule),
+      sageMakerPipelineParameters,
+    };
+  }
+}

packages/@aws-cdk/aws-scheduler-targets-alpha/test/integ.sage-maker-start-pipeline-execution.ts

@@ -0,0 +1,91 @@
+import * as scheduler from '@aws-cdk/aws-scheduler-alpha';
+import { ExpectedResult, IntegTest } from '@aws-cdk/integ-tests-alpha';
+import * as cdk from 'aws-cdk-lib';
+import { ManagedPolicy, Role, ServicePrincipal } from 'aws-cdk-lib/aws-iam';
+import { CfnPipeline } from 'aws-cdk-lib/aws-sagemaker';
+import { SageMakerPipelineParameter, SageMakerStartPipelineExecution } from '../lib';
+import { Bucket } from 'aws-cdk-lib/aws-s3';
+
+const app = new cdk.App();
+const stack = new cdk.Stack(app, 'aws-cdk-scheduler-targets-sagemaker-start-pipeline-execution');
+
+const s3Bucket = new Bucket(stack, 'SageMakerBucket', {
+  removalPolicy: cdk.RemovalPolicy.DESTROY,
+  autoDeleteObjects: true,
+});
+
+const pipelineDefinition = {
+  PipelineDefinitionBody: JSON.stringify({
+    Version: '2020-12-01',
+    Metadata: {},
+    Parameters: [
+      {
+        Name: 'ParameterName',
+        Type: 'String',
+        DefaultValue: 'Value',
+      },
+    ],
+    Steps: [
+      {
+        Name: 'TrainingStep',
+        Type: 'Training',
+        Arguments: {
+          AlgorithmSpecification: {
+            TrainingImage: '382416733822.dkr.ecr.us-east-1.amazonaws.com/linear-learner:1',
+            TrainingInputMode: 'File',
+          },
+          OutputDataConfig: {
+            S3OutputPath: s3Bucket.s3UrlForObject(),
+          },
+          ResourceConfig: {
+            InstanceCount: 1,
+            InstanceType: 'ml.m5.large',
+            VolumeSizeInGB: 10,
+          },
+          StoppingCondition: {
+            MaxRuntimeInSeconds: 3600,
+          },
+        },
+      },
+    ],
+  }),
+};
+
+const pipelineRole = new Role(stack, 'SageMakerPipelineRole', {
+  assumedBy: new ServicePrincipal('sagemaker.amazonaws.com'),
+});
+pipelineRole.addManagedPolicy(ManagedPolicy.fromAwsManagedPolicyName('AmazonSageMakerFullAccess'));
+const pipelineParameterList: SageMakerPipelineParameter[] = [{
+  name: 'ParameterName',
+  value: 'ParameterValue',
+}];
+const pipeline = new CfnPipeline(stack, 'Pipeline', {
+  pipelineName: 'sagemaker-pipeline',
+  pipelineDefinition: pipelineDefinition,
+  roleArn: pipelineRole.roleArn,
+});
+
+new scheduler.Schedule(stack, 'Schedule', {
+  schedule: scheduler.ScheduleExpression.rate(cdk.Duration.minutes(10)),
+  target: new SageMakerStartPipelineExecution(pipeline, {
+    pipelineParameterList,
+  }),
+});
+
+const integrationTest = new IntegTest(app, 'integrationtest-sagemaker-start-pipeline-execution', {
+  testCases: [stack],
+  stackUpdateWorkflow: false, // this would cause the schedule to trigger with the old code
+});
+
+// Verifies that the pipeline run by the scheduler
+integrationTest.assertions.awsApiCall('Sagemaker', 'listPipelineExecutions', {
+  PipelineName: 'sagemaker-pipeline',
+}).assertAtPath(
+  'PipelineExecutionSummaries.0.PipelineExecutionStatus',
+  ExpectedResult.stringLikeRegexp('Succeeded'),
+).waitForAssertions({
+  interval: cdk.Duration.seconds(30),
+  totalTimeout: cdk.Duration.minutes(10),
+});
+
+app.synth();