如何通过CDK部署仅运行一次的ECS独立任务

针对你需要在部署期间执行ECS独立迁移任务的需求，以下是两种实用的实现方案，均基于TypeScript CDK：

方案1：部署阶段自动执行（自定义资源） #

这种方案会在CDK栈创建/更新时自动触发Fargate迁移任务，任务失败则部署终止，完全匹配你"部署期间必须执行"的要求。

示例代码 #

import * as cdk from 'aws-cdk-lib';
import * as ecs from 'aws-cdk-lib/aws-ecs';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as cr from 'aws-cdk-lib/custom-resources';
import { Construct } from 'constructs';

export class MigrationStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // 引用已有的ECS集群
    const cluster = ecs.Cluster.fromClusterAttributes(this, 'ExistingCluster', {
      clusterName: 'MyECSCluster',
      vpc: cdk.aws_ec2.Vpc.fromLookup(this, 'ExistingVpc', { vpcId: 'vpc-xxxxxx' }),
    });

    // 引用已有的迁移任务定义
    const migrationTaskDef = ecs.FargateTaskDefinition.fromTaskDefinitionArn(
      this,
      'ExistingMigrationTaskDef',
      'arn:aws:ecs:your-region:your-account-id:task-definition/app-migrations:1'
    );

    // 创建执行ECS任务的Lambda角色
    const migrationExecutionRole = new iam.Role(this, 'MigrationExecutionRole', {
      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
    });

    // 赋予角色ECS操作权限
    migrationExecutionRole.addToPolicy(new iam.PolicyStatement({
      actions: ['ecs:RunTask', 'ecs:DescribeTasks', 'ecs:StopTask'],
      resources: [migrationTaskDef.taskDefinitionArn],
    }));

    // 赋予角色VPC资源访问权限
    migrationExecutionRole.addToPolicy(new iam.PolicyStatement({
      actions: ['ec2:DescribeSubnets', 'ec2:DescribeSecurityGroups'],
      resources: ['*'],
    }));

    // 创建自定义资源提供者（Lambda触发ECS任务）
    const ecsRunTaskProvider = new cr.Provider(this, 'EcsRunTaskProvider', {
      onEventHandler: new cdk.aws_lambda.Function(this, 'EcsRunTaskHandler', {
        runtime: cdk.aws_lambda.Runtime.NODEJS_18_X,
        handler: 'index.handler',
        code: cdk.aws_lambda.Code.fromInline(`
const AWS = require('aws-sdk');
const ecs = new AWS.ECS();

exports.handler = async (event) => {
  if (event.RequestType === 'Create' || event.RequestType === 'Update') {
    const params = {
      cluster: 'MyECSCluster',
      taskDefinition: '${migrationTaskDef.taskDefinitionArn}',
      launchType: 'FARGATE',
      networkConfiguration: {
        awsvpcConfiguration: {
          subnets: ['subnet-xxxx', 'subnet-yyyy'],
          securityGroups: ['sg-xxxxxxxxxxx'],
          assignPublicIp: 'ENABLED' // 无公网需求可设为DISABLED
        }
      },
      count: 1
    };

    try {
      const runTaskRes = await ecs.runTask(params).promise();
      const taskArn = runTaskRes.tasks[0].taskArn;

      // 等待任务执行完成
      await ecs.waitFor('tasksStopped', {
        cluster: 'MyECSCluster',
        tasks: [taskArn]
      }).promise();

      // 检查任务退出码
      const describeRes = await ecs.describeTasks({
        cluster: 'MyECSCluster',
        tasks: [taskArn]
      }).promise();
      const exitCode = describeRes.tasks[0].containers[0].exitCode;

      if (exitCode !== 0) throw new Error(\`迁移失败，退出码：\${exitCode}\`);

      return { Status: 'SUCCESS', PhysicalResourceId: taskArn };
    } catch (err) {
      return { Status: 'FAILED', Reason: err.message };
    }
  } else {
    // 删除栈时无需执行任务，直接返回成功
    return { Status: 'SUCCESS', PhysicalResourceId: event.PhysicalResourceId };
  }
};
        `),
        role: migrationExecutionRole,
        timeout: cdk.Duration.minutes(15), // 根据迁移耗时调整
      }),
    });

    // 创建自定义资源，触发迁移任务
    new cdk.CustomResource(this, 'MigrationTask', {
      serviceToken: ecsRunTaskProvider.serviceToken,
    });
  }
}

关键说明 #

• 自定义资源会绑定CDK部署生命周期，创建/更新栈时自动触发迁移
• Lambda会等待任务完成并检查退出码，迁移失败则部署中断
• 根据VPC配置调整assignPublicIp，若使用NAT网关可设为DISABLED

方案2：Step Functions编排（复杂场景适用） #

如果你的迁移需要重试、分支逻辑或与其他部署步骤联动，可以使用Step Functions的EcsRunTask任务。

示例代码 #

import * as cdk from 'aws-cdk-lib';
import * as ecs from 'aws-cdk-lib/aws-ecs';
import * as sfn from 'aws-cdk-lib/aws-stepfunctions';
import * as tasks from 'aws-cdk-lib/aws-stepfunctions-tasks';
import { Construct } from 'constructs';

export class MigrationStepFunctionStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const cluster = ecs.Cluster.fromClusterAttributes(this, 'ExistingCluster', {
      clusterName: 'MyECSCluster',
      vpc: cdk.aws_ec2.Vpc.fromLookup(this, 'ExistingVpc', { vpcId: 'vpc-xxxxxx' }),
    });

    const migrationTaskDef = ecs.FargateTaskDefinition.fromTaskDefinitionArn(
      this,
      'ExistingMigrationTaskDef',
      'arn:aws:ecs:your-region:your-account-id:task-definition/app-migrations:1'
    );

    // 定义ECS迁移任务步骤
    const runMigration = new tasks.EcsRunTask(this, 'RunMigration', {
      cluster,
      taskDefinition: migrationTaskDef,
      launchType: tasks.LaunchType.FARGATE,
      networkConfiguration: {
        subnets: cdk.aws_ec2.SubnetSelection.fromSubnetIds({
          subnetIds: ['subnet-xxxx', 'subnet-yyyy'],
        }),
        securityGroups: [cdk.aws_ec2.SecurityGroup.fromSecurityGroupId(this, 'MigrationSG', 'sg-xxxxxxxxxxx')],
        assignPublicIp: cdk.aws_ec2.AssignPublicIp.ENABLED,
      },
      integrationPattern: sfn.IntegrationPattern.RUN_JOB, // 等待任务完成
    });

    // 添加重试逻辑（可选）
    runMigration.addRetry({
      maxAttempts: 2,
      errors: ['ECS.TaskFailed'],
      interval: cdk.Duration.minutes(1),
    });

    // 创建状态机
    new sfn.StateMachine(this, 'MigrationStateMachine', {
      definitionBody: sfn.DefinitionBody.fromChainable(runMigration),
      timeout: cdk.Duration.minutes(30),
    });

    // 若需部署时自动触发，可结合自定义资源调用状态机
  }
}

关键说明 #

• IntegrationPattern.RUN_JOB会让Step Functions等待任务执行完成
• 可通过addRetry添加重试逻辑，处理数据库连接超时等临时错误
• 如需部署时自动执行，需额外添加自定义资源触发状态机

通用注意事项 #

• 确保迁移任务定义的IAM角色拥有数据库访问权限（如RDS、DynamoDB权限）
• 调整Lambda/状态机的超时时间，匹配你的迁移任务耗时
• 测试阶段可手动触发Lambda或状态机，验证迁移逻辑正常

内容的提问来源于stack exchange，提问作者quentino