如何限制AWS CodePipeline仅同时运行1次，完成后再执行下一次？

嘿，这个问题我之前帮不少开发者解决过——AWS CodePipeline本身确实没有内置的开关来限制并行运行次数，但我们可以通过两种可靠的方法实现「同一时间仅运行一次」的需求，不管当前管道成功还是失败，都得等它结束才能启动下一次。

方法1：用Lambda+CloudWatch Events阻止新执行（推荐） #

这个方法的思路是：当有人触发管道运行时，自动检查当前是否已有正在运行的执行，如果有，直接终止新触发的那次。

步骤分解： #

1. 创建检查Lambda函数
   写一个Python Lambda函数，用来查询目标管道的所有执行状态，判断是否有InProgress的运行中实例。如果有多个运行中的执行（刚触发的那个是最新的），就调用API终止它。

   示例代码：

   import boto3
   
   codepipeline = boto3.client('codepipeline')
   
   def lambda_handler(event, context):
       # 从事件中获取管道名称
       pipeline_name = event['detail']['pipeline']
       # 获取管道的所有执行记录
       executions = codepipeline.list_pipeline_executions(pipelineName=pipeline_name)['pipelineExecutionSummaries']
       
       # 筛选出状态为InProgress的执行
       running_executions = [e for e in executions if e['status'] == 'InProgress']
       
       # 如果运行中的执行超过1个（包含刚触发的这个），就终止最新的那个
       if len(running_executions) > 1:
           latest_execution = max(executions, key=lambda x: x['startTime'])
           codepipeline.stop_pipeline_execution(
               pipelineName=pipeline_name,
               pipelineExecutionId=latest_execution['pipelineExecutionId'],
               reason='阻止并行执行：已有管道在运行'
           )
   

2. 配置Lambda权限
   给这个Lambda的IAM角色添加以下权限策略，确保它能查询管道执行状态和终止执行：

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Effect": "Allow",
               "Action": [
                   "codepipeline:ListPipelineExecutions",
                   "codepipeline:StopPipelineExecution"
               ],
               "Resource": "arn:aws:codepipeline:你的区域:你的账号ID:管道名称"
           }
       ]
   }
   

3. 设置CloudWatch Events触发器
   创建一个CloudWatch Events规则，触发条件选择「事件源为AWS CodePipeline」，事件类型选择StartPipelineExecution，然后把目标设置为刚才创建的Lambda函数。这样每次管道被触发时，Lambda都会自动检查并处理。

方法2：在管道开头加预执行检查阶段 #

这种方法是在管道的最前端添加一个自定义检查阶段，如果检测到有其他运行中的执行，就让当前管道直接失败终止。

步骤分解： #

1. 添加预检查阶段
   打开你的CodePipeline，在所有现有阶段的最前面添加一个新阶段，命名比如CheckRunningExecution。

2. 配置Lambda动作
   在这个新阶段里添加一个Lambda类型的动作，配置刚才创建的检查Lambda函数（逻辑和方法1类似，但要排除当前执行本身）。

   示例代码：

   import boto3
   import json
   
   codepipeline = boto3.client('codepipeline')
   
   def lambda_handler(event, context):
       # 从用户参数中获取管道名称（配置动作时要填这个参数）
       pipeline_name = event['CodePipeline.job']['data']['actionConfiguration']['configuration']['UserParameters']
       # 获取当前执行的ID，用来排除自己
       current_execution_id = event['CodePipeline.job']['data']['pipelineExecutionId']
       
       executions = codepipeline.list_pipeline_executions(pipelineName=pipeline_name)['pipelineExecutionSummaries']
       # 筛选出除了当前执行外的运行中实例
       running_executions = [e for e in executions if e['status'] == 'InProgress' and e['pipelineExecutionId'] != current_execution_id]
       
       if running_executions:
           # 有其他运行中的执行，标记当前任务失败
           codepipeline.put_job_failure_result(
               jobId=event['CodePipeline.job']['id'],
               failureDetails={
                   'message': '已有管道在运行，终止当前执行',
                   'type': 'JobFailed'
               }
           )
       else:
           # 没有运行中的，标记任务成功，继续后续阶段
           codepipeline.put_job_success_result(
               jobId=event['CodePipeline.job']['id']
           )
   

3. 配置动作参数
   在Lambda动作的「用户参数」里填入你的管道名称，这样函数就能知道要检查哪个管道。

两种方法的区别 #

• 方法1会直接阻止新管道的启动（或者说刚启动就终止），在管道的执行历史里会显示被终止的记录。
• 方法2会让新管道先启动，但在第一个阶段就失败终止，适合需要保留触发记录的场景。

不管用哪种方法，测试的时候可以手动触发两次管道，看看是否只有一次在运行，另一次被终止。

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