如何在S3指定时长内无文件上传时触发Lambda并通过SNS通知

S3本身没有原生的「N小时无上传」触发规则，所以需要用定时触发+遍历检测的组合方案实现需求，具体步骤如下：

步骤1：创建基础资源 #

• 首先创建用于接收告警的SNS主题，提前把需要接收通知的邮箱、手机号等端点订阅到该主题，确认订阅生效
• 创建检测用的Lambda函数，为其配置S3的ListBucket权限、SNS的Publish权限，以及Lambda基础运行权限

步骤2：配置定时触发规则 #

使用Amazon EventBridge创建定时规则，触发频率按你的检测需求设置即可：

• 若检测10小时无上传，可设置每10小时触发一次Lambda
• 若检测1小时无上传，可设置每1小时触发一次Lambda
  触发目标直接选择你刚才创建的Lambda函数即可。如果需要降低告警延迟，也可以把触发频率设置为小于检测阈值，比如检测10小时无上传的场景，设置每1小时触发一次检测。

步骤3：编写Lambda核心检测逻辑 #

核心逻辑为调用S3接口拉取最新文件的最后修改时间，和当前时间做差值判断，符合阈值就触发SNS通知。
Python示例代码片段：

import boto3
from datetime import datetime, timezone

s3 = boto3.client('s3')
sns = boto3.client('sns')
# 替换为你的S3桶名
BUCKET_NAME = 'your-target-bucket'
# 替换为你的SNS主题ARN
SNS_TOPIC_ARN = 'arn:aws:sns:region:account-id:your-topic-name'
# 替换为你的检测阈值，单位为小时
CHECK_THRESHOLD_HOURS = 10

def lambda_handler(event, context):
    # 拉取桶内最新1个对象的信息
    resp = s3.list_objects_v2(Bucket=BUCKET_NAME, MaxKeys=1)
    if 'Contents' not in resp:
        # 桶内无任何文件，直接触发告警
        send_alarm("目标S3桶内无任何文件，未检测到上传行为")
        return
    latest_file = resp['Contents'][0]
    last_modified = latest_file['LastModified']
    now = datetime.now(timezone.utc)
    # 计算距离上次上传的小时数
    diff_hours = (now - last_modified).total_seconds() / 3600
    if diff_hours > CHECK_THRESHOLD_HOURS:
        send_alarm(f"目标S3桶已超过{CHECK_THRESHOLD_HOURS}小时无新文件上传，最近一次上传时间为{last_modified.strftime('%Y-%m-%d %H:%M:%S UTC')}")

def send_alarm(content):
    sns.publish(
        TopicArn=SNS_TOPIC_ARN,
        Subject='S3文件上传告警',
        Message=content
    )

可选优化 #

• 如果只需要检测特定前缀目录的上传情况，调用list_objects_v2时增加Prefix参数指定目标路径即可
• 桶内文件量极大的场景，可以开启S3事件通知，每次上传时自动把时间写入DynamoDB，Lambda检测时直接读DynamoDB的最新时间即可，不需要每次调用S3接口，性能更高
• 可在Lambda中增加告警去重逻辑，避免连续触发重复通知

内容的提问来源于stack exchange，提问作者Seung Hoon Paek