如何让Python Lambda识别触发它的CloudWatch日志自定义指标过滤器？

嘿，这个问题我之前也碰到过！其实CloudWatch日志订阅触发Lambda的时候，会把触发源的关键信息包含在事件 payload 里，你完全可以从里面提取出是哪个指标过滤器发起的请求。下面是具体的实现方法：

核心思路：利用subscriptionFilterArn字段 #

当CloudWatch的指标过滤器触发Lambda时，Lambda收到的event参数里，awslogs对象下的subscriptionFilterArn字段就是关键——这个ARN的最后一段就是你创建指标过滤器时设置的名称，格式大概是这样的：
arn:aws:logs:us-east-1:123456789012:subscription-filter:/aws/lambda/my-log-group:my-custom-filter

你只需要拆分这个ARN，就能拿到过滤器的名称，然后根据名称分支执行不同逻辑。

Python Lambda代码示例 #

import json

def lambda_handler(event, context):
    # 解析CloudWatch传递的事件数据
    try:
        awslogs_payload = event["awslogs"]
        filter_arn = awslogs_payload["subscriptionFilterArn"]
    except KeyError as e:
        print(f"事件格式异常，缺失关键字段: {e}")
        return {"statusCode": 400, "body": json.dumps("无效的CloudWatch日志事件")}
    
    # 从ARN中提取过滤器名称（ARN最后一段就是过滤器名）
    filter_name = filter_arn.split(":")[-1]
    
    # 根据过滤器名称执行不同操作
    match filter_name:
        case "error-log-filter":
            # 处理错误日志的逻辑
            print("开始处理错误日志过滤器触发的任务")
            # 这里写你的业务代码，比如发送告警、存储错误详情等
        case "access-log-filter":
            # 处理访问日志的逻辑
            print("开始处理访问日志过滤器触发的任务")
            # 比如统计访问量、分析请求路径等
        case _:
            print(f"未匹配到对应逻辑的过滤器: {filter_name}")
    
    return {
        "statusCode": 200,
        "body": json.dumps(f"已处理来自过滤器 {filter_name} 的事件")
    }

额外提示 #

• 如果你担心ARN格式变化（虽然AWS很少改这类核心格式），也可以在创建指标过滤器时，给每个过滤器设置独特的过滤模式，或者在Lambda的环境变量里映射过滤器名称到对应的逻辑，但直接用ARN提取是最直接且无需额外配置的方法。
• 测试的时候，可以手动模拟CloudWatch日志事件（包含正确的subscriptionFilterArn）来验证不同分支的逻辑是否正常执行。

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