如何通过Lambda+EventBridge定期重建CDK创建的CloudFormation栈及Boto3部署模板？

当然可以！这两个需求都能轻松实现，我来一步步给你拆解具体方案：

核心思路很简单：通过EventBridge按30天的周期触发Lambda函数，Lambda函数里调用Boto3先删除现有栈，等删除完成后再重新创建新栈。

1. 创建Lambda函数 #

首先得给Lambda分配足够的IAM权限，至少需要cloudformation:DeleteStack和cloudformation:CreateStack权限；如果你的栈涉及EC2、IAM角色等其他资源，还要给Lambda对应资源的操作权限（比如EC2的实例创建权限、IAM角色的创建权限等）。

下面是Python版本的Lambda示例代码，逻辑是先删栈（处理栈不存在的情况），再重建：

import boto3

cfn_client = boto3.client('cloudformation')
TARGET_STACK_NAME = '你的目标栈名称'
CFN_TEMPLATE_S3_URL = 'https://你的S3桶.s3.区域.amazonaws.com/模板文件路径.yaml'  # 比如CDK部署时生成的模板URL

def lambda_handler(event, context):
    # 第一步：删除现有栈
    try:
        cfn_client.delete_stack(StackName=TARGET_STACK_NAME)
        print(f"启动栈 {TARGET_STACK_NAME} 删除流程")
        # 等待栈删除完成，避免直接创建导致冲突
        delete_waiter = cfn_client.get_waiter('stack_delete_complete')
        delete_waiter.wait(StackName=TARGET_STACK_NAME)
        print(f"栈 {TARGET_STACK_NAME} 已成功删除")
    except cfn_client.exceptions.ClientError as e:
        if "does not exist" in str(e):
            print(f"栈 {TARGET_STACK_NAME} 不存在，直接执行创建")
        else:
            # 其他错误直接抛出，让Lambda重试或者告警
            raise e

    # 第二步：重新创建栈
    try:
        create_response = cfn_client.create_stack(
            StackName=TARGET_STACK_NAME,
            TemplateURL=CFN_TEMPLATE_S3_URL,
            # 如果你的栈需要参数，在这里添加
            Parameters=[
                {
                    'ParameterKey': 'InstanceType',
                    'ParameterValue': 't2.micro'
                }
            ],
            # 如果栈包含IAM资源，必须添加这个权限声明
            Capabilities=['CAPABILITY_NAMED_IAM']
        )
        print(f"栈创建请求已提交，栈ID: {create_response['StackId']}")
        # 可选：等待创建完成，方便日志追踪
        create_waiter = cfn_client.get_waiter('stack_create_complete')
        create_waiter.wait(StackName=TARGET_STACK_NAME)
        print(f"栈 {TARGET_STACK_NAME} 已成功重建")
    except Exception as e:
        print(f"栈创建失败: {str(e)}")
        raise e

2. 创建EventBridge触发规则 #

进入EventBridge控制台，创建一个新规则：

• 规则类型选择按计划
• 触发频率可以直接用rate(30 days)（比Cron表达式更直观），或者用Cron表达式比如0 0 1 */2 *（每两个月的1号触发，近似30天周期）
• 目标选择你刚才创建的Lambda函数，配置好触发即可

这个操作其实刚才Lambda里已经用到了，单独拎出来详细说下细节：

核心步骤 #

1. 准备模板并上传到S3：把你的CloudFormation模板（比如CDK synth生成的模板）上传到S3桶，确保执行Boto3代码的实体（Lambda角色、本地IAM用户等）有s3:GetObject权限访问这个模板文件（可以通过Bucket Policy或者IAM策略配置）。

2. Boto3代码实现
   下面是本地运行或者集成到其他服务的Python代码示例：

import boto3

def deploy_cfn_stack_from_s3():
    cfn_client = boto3.client('cloudformation')
    stack_name = '你的栈名称'
    # S3模板的HTTPS URL，格式是https://<桶名>.s3.<区域>.amazonaws.com/<模板路径>
    template_s3_url = 'https://my-cfn-templates.s3.us-east-1.amazonaws.com/ec2-instance-stack.yaml'

    # 定义栈的参数（根据你的模板需求调整）
    stack_params = [
        {
            'ParameterKey': 'KeyPairName',
            'ParameterValue': 'my-ec2-key-pair'
        },
        {
            'ParameterKey': 'VpcId',
            'ParameterValue': 'vpc-12345678'
        }
    ]

    # 发起栈创建请求
    try:
        response = cfn_client.create_stack(
            StackName=stack_name,
            TemplateURL=template_s3_url,
            Parameters=stack_params,
            # 栈包含IAM资源时必须声明的权限
            Capabilities=['CAPABILITY_IAM', 'CAPABILITY_NAMED_IAM'],
            # 可选：给栈添加标签
            Tags=[
                {
                    'Key': 'Environment',
                    'Value': 'Staging'
                }
            ]
        )
        print(f"栈创建请求提交成功，栈ID: {response['StackId']}")

        # 等待栈创建完成
        waiter = cfn_client.get_waiter('stack_create_complete')
        waiter.wait(StackName=stack_name)
        print(f"栈 {stack_name} 部署完成！")
    except cfn_client.exceptions.AlreadyExistsException:
        print(f"栈 {stack_name} 已存在，若要更新请使用update_stack方法")
    except Exception as e:
        print(f"部署失败: {str(e)}")
        raise e

if __name__ == "__main__":
    deploy_cfn_stack_from_s3()

注意事项 #

• 如果需要更新现有栈，把create_stack换成update_stack即可，但要确保栈处于CREATE_COMPLETE或UPDATE_COMPLETE状态
• 确保执行代码的实体有足够的权限：除了CloudFormation的操作权限，还要有模板中涉及资源的创建/修改权限
• S3模板URL必须是HTTPS格式，不能用S3的HTTP或私有访问路径

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