如何排查数千个Lambda中被限流的具体函数？

一、CloudWatch控制台高效查询方式 #

无需逐个输入函数名，通过以下步骤快速定位限流函数：

• 进入CloudWatch指标页面，选择AWS/Lambda命名空间下的Throttles指标
• 在维度筛选栏选择FunctionName，点击「添加全部值」（若函数数量过多，可通过搜索框批量匹配）
• 切换到图表视图，设置统计类型为Sum或Maximum，时间范围锁定到你发现限流的时段
• 点击图表右上角的「排序」按钮，按指标值从高到低排列，非零值的函数即为被限流的目标
• 进阶用法：使用指标数学表达式精准筛选，示例表达式：

  FILTER(Lambda.Throttles, "$.FunctionName", "*")
  

  再设置阈值为0，仅显示超过阈值的函数，进一步缩小范围

二、boto3脚本自动化排查 #

通过脚本批量遍历所有Lambda函数，自动统计指定时间内的限流次数，直接输出结果：

import boto3
from datetime import datetime, timedelta

def get_throttled_lambda_functions():
    cloudwatch = boto3.client('cloudwatch')
    lambda_client = boto3.client('lambda')
    
    # 自定义查询时间范围，示例为过去24小时
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    throttled_functions = []
    # 分页获取所有Lambda函数，避免API调用限制
    paginator = lambda_client.get_paginator('list_functions')
    
    for page in paginator.paginate():
        for function in page['Functions']:
            func_name = function['FunctionName']
            # 查询该函数的Throttles指标总和
            metric_response = cloudwatch.get_metric_statistics(
                Namespace='AWS/Lambda',
                MetricName='Throttles',
                Dimensions=[{'Name': 'FunctionName', 'Value': func_name}],
                StartTime=start_time,
                EndTime=end_time,
                Period=3600,  # 按小时统计，可根据需求调整为300（5分钟）等
                Statistics=['Sum'],
                Unit='Count'
            )
            
            total_throttles = sum(dp['Sum'] for dp in metric_response['Datapoints'])
            if total_throttles > 0:
                last_throttle = max(dp['Timestamp'] for dp in metric_response['Datapoints']) if metric_response['Datapoints'] else None
                throttled_functions.append({
                    '函数名': func_name,
                    '总限流次数': int(total_throttles),
                    '最后一次限流时间': last_throttle.strftime('%Y-%m-%d %H:%M:%S UTC') if last_throttle else '无记录'
                })
    
    # 按限流次数降序排序
    throttled_functions.sort(key=lambda x: x['总限流次数'], reverse=True)
    
    # 输出结果
    print("=== 被限流的Lambda函数 ===")
    for func in throttled_functions:
        print(f"函数名: {func['函数名']} | 总限流次数: {func['总限流次数']} | 最后一次限流: {func['最后一次限流时间']}")

if __name__ == "__main__":
    get_throttled_lambda_functions()

脚本使用说明 #

• 确保已安装boto3并配置好AWS凭证（环境变量、~/.aws/credentials文件或IAM角色）
• 可调整Period（统计周期）和时间范围参数，适配你的排查场景
• 自动处理分页，支持数千个Lambda函数的批量查询

三、提前预防：批量配置CloudWatch告警 #

通过CloudFormation或AWS CLI批量给Lambda函数配置Throttles指标告警，限流发生时直接触发通知（如SNS邮件），避免事后排查的繁琐。

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