使用Boto3从AWS CloudWatch获取指定生产环境Lambda函数的调用指标数据

我来帮你拆解并解决这个问题，你之前的脚本之所以没拿到数据，主要是踩了两个常见的CloudWatch Lambda指标的坑，我一步步给你说明：

问题根源分析 #

1. 混淆了Lambda标签与CloudWatch指标维度：你把Lambda函数的STAGE标签当成了CloudWatch指标的维度，但AWS/Lambda的Invocations指标默认只有FunctionName这个核心维度，STAGE是你给Lambda加的自定义标签，并不是指标本身的一部分，所以直接按STAGE维度查询自然拿不到数据。
2. Metric Math查询的时间限制：你尝试的SELECT类Metric Math查询有严格的时间范围限制（通常仅支持过去3小时内的数据），完全覆盖不了30天的查询需求，所以会触发MaxQueryTimeRangeExceed错误。

正确的实现思路 #

要实现和控制台一样的效果，我们需要分两步走：

1. 先批量获取所有带有STAGE=production标签的Lambda函数列表；
2. 针对每个生产环境的函数，按FunctionName维度查询其每日调用量，最后聚合整理结果。

完整可运行的Python脚本 #

import datetime
import boto3
from typing import List, Dict

def get_production_lambda_functions(region: str = "eu-west-1") -> List[str]:
    """获取所有标记为production的Lambda函数名称"""
    lambda_client = boto3.client("lambda", region_name=region)
    prod_function_names = []
    # 用分页器处理函数数量过多的情况，避免遗漏
    paginator = lambda_client.get_paginator("list_functions")
    
    for page in paginator.paginate():
        for func in page["Functions"]:
            func_arn = func["FunctionArn"]
            # 查询当前函数的所有标签
            tags = lambda_client.list_tags(Resource=func_arn)["Tags"]
            if tags.get("STAGE") == "production":
                prod_function_names.append(func["FunctionName"])
    
    return prod_function_names

def get_daily_invocation_data(
    function_names: List[str],
    region: str = "eu-west-1",
    days_back: int = 30
) -> Dict[str, Dict[datetime.datetime, int]]:
    """获取指定Lambda函数过去N天的每日调用量总和"""
    cloudwatch_client = boto3.client("cloudwatch", region_name=region)
    
    # 统一使用UTC时区处理时间，避免和CloudWatch存储的UTC数据出现偏差
    end_time = datetime.datetime.now(datetime.timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
    start_time = end_time - datetime.timedelta(days=days_back)
    
    # 为每个生产函数构造单独的指标查询
    metric_queries = []
    for idx, func_name in enumerate(function_names):
        query_id = f"func_{idx}"
        metric_queries.append({
            "Id": query_id,
            "MetricStat": {
                "Metric": {
                    "Namespace": "AWS/Lambda",
                    "MetricName": "Invocations",
                    "Dimensions": [{"Name": "FunctionName", "Value": func_name}]
                },
                "Period": 86400,  # 1天对应的秒数
                "Stat": "Sum",
                "Unit": "Count"
            },
            "ReturnData": True,
            "Label": func_name  # 用函数名做标签，后续整理结果更直观
        })
    
    # 调用CloudWatch批量查询指标数据
    response = cloudwatch_client.get_metric_data(
        MetricDataQueries=metric_queries,
        StartTime=start_time,
        EndTime=end_time,
        ScanBy="TimestampAscending"
    )
    
    # 把结果整理成易读的格式
    final_result = {}
    for metric_result in response["MetricDataResults"]:
        func_name = metric_result["Label"]
        # 时间戳和调用量一一对应，转成整数（因为调用量是计数，不会有小数）
        final_result[func_name] = {
            ts: int(count) for ts, count in zip(metric_result["Timestamps"], metric_result["Values"])
        }
    
    return final_result

if __name__ == "__main__":
    # 1. 获取生产环境函数列表
    prod_functions = get_production_lambda_functions()
    print(f"共找到{len(prod_functions)}个生产环境Lambda函数")
    
    # 2. 获取过去30天的每日调用数据
    invocation_report = get_daily_invocation_data(prod_functions)
    
    # 示例：打印每个函数的每日调用量
    for func_name, daily_data in invocation_report.items():
        print(f"\n=== {func_name} 调用统计 ===")
        for date, count in daily_data.items():
            print(f"{date.date()}: {count} 次调用")

关键注意事项 #

1. 分页处理：用Lambda的分页器获取函数列表，避免函数数量超过单次API返回上限时遗漏数据；
2. 时区一致性：全程使用UTC时区处理时间，确保和CloudWatch存储的UTC时间对齐，不会出现日期截断的问题；
3. 批量查询限制：CloudWatch的get_metric_data一次最多支持100个查询，如果你的生产函数超过100个，需要把函数列表分成多个批次查询；
4. 数据格式转换：CloudWatch返回的调用量是浮点数，我们可以直接转成整数，因为调用次数是整数计数。

备注：内容来源于stack exchange，提问作者AlexNikolaev94