如何为名称前缀为archive_*的S3桶自动配置Lambda事件触发规则

方案核心逻辑 #

通过AWS EventBridge监听S3桶的创建事件，触发专门的配置Lambda自动完成触发规则绑定，无需手动操作。注意需要先开启当前区域的CloudTrail跟踪，否则EventBridge无法捕获S3桶创建的API事件。

前置准备：修改原有元数据采集Lambda的资源策略 #

首先给你现有的写入DynamoDB的Lambda添加权限，允许所有符合archive_*前缀的S3桶调用它，避免新桶触发时报权限错误。
在Lambda控制台的「配置-权限-资源策略」中添加如下语句，替换<你的元数据采集Lambda的ARN>为你现有Lambda的ARN即可：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "s3.amazonaws.com"
            },
            "Action": "lambda:InvokeFunction",
            "Resource": "<你的元数据采集Lambda的ARN>",
            "Condition": {
                "StringLike": {
                    "aws:SourceArn": "arn:aws:s3:::archive_*"
                }
            }
        }
    ]
}

步骤1：创建自动配置用的Lambda函数 #

新建一个Python运行时的Lambda，作用是收到新桶创建事件后，自动给符合命名规则的桶绑定触发规则，代码如下：

import boto3
import re

# 替换为你自己的元数据采集Lambda的ARN
TARGET_LAMBDA_ARN = "arn:aws:lambda:<区域>:<账号ID>:function:<你的元数据采集Lambda名称>"
# 桶名匹配规则
BUCKET_PATTERN = r'^archive_.*$'

s3_client = boto3.client('s3')

def lambda_handler(event, context):
    # 从EventBridge事件中提取新桶名称
    bucket_name = event['detail']['requestParameters']['bucketName']
    # 检查桶名是否符合规则
    if not re.match(BUCKET_PATTERN, bucket_name):
        print(f"桶{bucket_name}不符合命名规则，跳过配置")
        return
    
    # 给桶添加S3事件通知规则
    try:
        s3_client.put_bucket_notification_configuration(
            Bucket=bucket_name,
            NotificationConfiguration={
                'LambdaFunctionConfigurations': [
                    {
                        'LambdaFunctionArn': TARGET_LAMBDA_ARN,
                        'Events': [
                            's3:ObjectCreated:*',
                            's3:ObjectRemoved:Delete'
                        ]
                    }
                ]
            }
        )
        print(f"已成功为桶{bucket_name}配置Lambda触发规则")
    except Exception as e:
        print(f"配置桶{bucket_name}失败：{str(e)}")
        raise e

给这个配置Lambda添加IAM权限，权限策略如下：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:PutBucketNotificationConfiguration"
            ],
            "Resource": "arn:aws:s3:::archive_*"
        }
    ]
}

提示：Lambda默认创建的执行角色已经包含CloudWatch日志输出权限，无需额外配置。

步骤2：配置EventBridge触发规则 #

打开EventBridge控制台，创建新规则：

• 规则类型选择「事件模式」
• 事件源选择「AWS服务」->「S3」->「AWS API Call via CloudTrail」-> 匹配事件「CreateBucket」
• 目标选择刚才创建的配置Lambda，保存即可。

可选：批量处理已存在的archive_*桶 #

如果你已经有多个符合命名规则的桶需要批量配置，可以本地运行如下代码一次性完成配置：

import boto3
import re

s3 = boto3.client('s3')
PATTERN = r'^archive_.*$'
# 替换为你自己的元数据采集Lambda的ARN
TARGET_LAMBDA_ARN = "arn:aws:lambda:<区域>:<账号ID>:function:<你的元数据采集Lambda名称>"

for bucket in s3.list_buckets()['Buckets']:
    name = bucket['Name']
    if re.match(PATTERN, name):
        s3.put_bucket_notification_configuration(
            Bucket=name,
            NotificationConfiguration={
                'LambdaFunctionConfigurations': [
                    {
                        'LambdaFunctionArn': TARGET_LAMBDA_ARN,
                        'Events': ['s3:ObjectCreated:*', 's3:ObjectRemoved:Delete']
                    }
                ]
            }
        )
        print(f"已配置桶{name}")

方案验证 #

配置完成后，新建一个名称为archive_test的S3桶，等待几秒后查看桶的「属性-事件通知」，即可看到自动创建的Lambda触发规则。

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