使用Lambda向SNS主题发送告警：每日每设备仅发一次邮件

SNS本身的订阅过滤器或内置去重机制无法满足你「同一设备每日最多一封告警邮件」的需求，因为这类功能只负责消息内容匹配或短时间内的重复消息过滤，没法记录设备告警的历史发送状态。你需要在Lambda处理流程中加入状态存储逻辑，具体方案如下：

核心思路 #

通过DynamoDB记录每个设备+告警类型的最新邮件发送日期，Lambda触发告警时先检查该记录，只有当当天未发送过（或次日重新触发）时，才调用SNS发送邮件，同时更新DynamoDB的状态。

具体实现步骤 #

1. 创建DynamoDB状态表 #

新建一张用于记录告警发送状态的表，结构如下：

• 表名：DeviceAlertSendLog
• 分区键（Partition Key）：device_id（字符串类型，存储设备全局唯一ID）
• 排序键（Sort Key）：alert_id（字符串类型，存储告警类型ID）
• 额外属性：last_sent_date（字符串类型，格式为YYYY-MM-DD，记录最近一次发送邮件的日期）

2. 修改Lambda处理逻辑 #

在Lambda的告警判断环节之后，加入以下逻辑：

import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
sns = boto3.client('sns')

def lambda_handler(event, context):
    # 原有逻辑：处理S3数据、判断是否触发告警
    device_id = "设备全局唯一ID"
    alert_id = "告警类型ID"
    alert_content = "邮件内容，包含设备ID、告警ID、日期等"
    sns_topic_arn = "你的SNS Topic ARN"
    
    # 获取当前日期（YYYY-MM-DD）
    current_date = datetime.now().strftime("%Y-%m-%d")
    
    # 查询DynamoDB中该设备+告警的发送记录
    table = dynamodb.Table('DeviceAlertSendLog')
    response = table.get_item(
        Key={
            'device_id': device_id,
            'alert_id': alert_id
        }
    )
    
    # 判断是否需要发送邮件
    need_send = False
    if 'Item' not in response:
        # 无记录，首次触发告警
        need_send = True
    else:
        last_sent_date = response['Item']['last_sent_date']
        if last_sent_date != current_date:
            # 上次发送日期不是今天，允许发送
            need_send = True
    
    if need_send:
        # 调用SNS发送邮件
        sns.publish(
            TopicArn=sns_topic_arn,
            Message=alert_content,
            Subject=f"设备告警：{device_id} - {alert_id}"
        )
        # 更新DynamoDB记录
        table.put_item(
            Item={
                'device_id': device_id,
                'alert_id': alert_id,
                'last_sent_date': current_date
            }
        )
    
    return {
        'statusCode': 200,
        'body': '处理完成'
    }

3. SNS配置保持原有设置 #

不需要在SNS侧配置特殊的订阅过滤器或去重规则，只需确保Lambda有调用SNS publish接口的权限，且订阅者（邮箱）已确认订阅即可。

关键说明 #

• 去重逻辑：通过device_id + alert_id组合键确保每个设备的每种告警类型独立判断发送状态，避免不同告警类型互相干扰。
• 日期判断：使用YYYY-MM-DD格式的日期字符串，确保次日同一告警触发时会重新发送邮件。
• 成本控制：DynamoDB的读写操作成本极低，这种小数据量的状态存储几乎不会产生额外费用。

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