能否通过Cloudwatch警报检测AWS Lambda连续5次执行失败？

一、关于现成指标 #

Lambda默认提供的Errors指标是时间段聚合型（比如5分钟内的错误总数），它无法追踪单次调用的执行顺序，也没法区分错误类型（第三方服务失败/输入异常），完全满足不了你的需求，必须通过自定义指标或日志处理来实现。

二、实现连续5次第三方失败的告警 #

要实现仅在连续5次调用因第三方服务中断失败时告警，需要分两步处理：

1. 在Lambda中区分错误类型并上报自定义指标 #

在Lambda函数的错误捕获逻辑中，明确区分两种失败场景：

• 当判定为第三方服务中断（比如调用第三方API返回5xx错误、超时、特定错误码等），向CloudWatch上报自定义指标（比如命名为LambdaThirdPartyStatus），值设为1；
• 当调用成功或因输入数据异常失败时，上报该指标值为0（输入异常也可以选择不上报，但建议统一上报，保证数据点的连续性）。

上报指标的代码示例（Python）：

import boto3

cloudwatch = boto3.client('cloudwatch')

def lambda_handler(event, context):
    try:
        # 业务逻辑：调用第三方服务等
        # ...
        # 成功时上报指标
        cloudwatch.put_metric_data(
            Namespace='CustomLambdaMetrics',
            MetricData=[
                {
                    'MetricName': 'LambdaThirdPartyStatus',
                    'Value': 0,
                    'Unit': 'Count'
                }
            ]
        )
        return {"status": "success"}
    except Exception as e:
        # 判断是否为第三方服务错误
        if is_third_party_error(e):
            # 第三方失败时上报1
            cloudwatch.put_metric_data(
                Namespace='CustomLambdaMetrics',
                MetricData=[
                    {
                        'MetricName': 'LambdaThirdPartyStatus',
                        'Value': 1,
                        'Unit': 'Count'
                    }
                ]
            )
        else:
            # 输入异常时上报0
            cloudwatch.put_metric_data(
                Namespace='CustomLambdaMetrics',
                MetricData=[
                    {
                        'MetricName': 'LambdaThirdPartyStatus',
                        'Value': 0,
                        'Unit': 'Count'
                    }
                ]
            )
        raise e

def is_third_party_error(error):
    # 根据实际场景判断，比如检查错误信息、HTTP状态码等
    return "third-party-service-unavailable" in str(error)

2. 创建CloudWatch告警检测连续5次失败 #

CloudWatch标准告警无法直接检测连续N个数据点的状态，但可以通过Metric Math实现：

1. 打开CloudWatch控制台，进入「告警」→「创建告警」；
2. 选择你的自定义指标CustomLambdaMetrics/LambdaThirdPartyStatus；
3. 切换到「数学表达式」标签，输入以下表达式来检查最近5个数据点是否全部为1：

   IF(AND(m1, OFFSET(m1, 1), OFFSET(m1, 2), OFFSET(m1, 3), OFFSET(m1, 4)), 1, 0)
   

   （其中m1是你选择的原始指标变量名，可根据实际调整）
4. 设置告警条件：当该数学表达式的返回值等于1时触发告警；
5. 配置通知渠道（比如SNS主题，绑定你的邮箱/短信），完成告警创建。

这个表达式的逻辑是：检查当前数据点和前4个数据点是否都为1（即连续5次第三方服务失败），如果满足则返回1，触发告警。

三、替代方案：基于日志的指标过滤 #

如果不想在代码中硬编码上报指标，也可以通过CloudWatch日志来实现：

1. 在Lambda中，当第三方服务失败时，打印包含特定标记的日志（比如"error_category": "third_party_failure"）；
2. 创建CloudWatch日志指标过滤器，匹配该标记，生成一个计数指标（每次匹配到就累加1）；
3. 利用Metric Math类似的逻辑，检测该计数指标在最近5个连续评估周期内都有值（即连续5次调用都触发了第三方错误），进而触发告警。

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