如何基于API Gateway阶段将Lambda授权器请求日志存入独立CloudWatch日志组？

现有代码的问题 #

你当前的代码无法实现目标，原因如下：

• Python logging.getLogger() 仅创建本地日志器实例，不会自动在CloudWatch中创建新的日志组/日志流，Lambda默认会将所有日志输出到固定的/aws/lambda/[函数名]日志组。
• 使用global logger在Lambda并发执行场景下会导致日志混乱，因为Lambda执行环境可能被复用，不同请求的日志会混写到同一个日志器。
• 日志流的命名逻辑错误，CloudWatch日志组和日志流是平级关系，不是嵌套路径格式。

正确实现方案 #

要实现按API Gateway阶段拆分CloudWatch日志组，需要手动对接CloudWatch Logs API，或者使用第三方库简化操作，以下是可行方案：

方案1：使用watchtower库对接CloudWatch Logs

watchtower是官方推荐的Python库，可直接将日志输出到指定CloudWatch日志组。

1. 安装依赖：将watchtower打包到Lambda部署包，或通过Lambda层引入。
2. 修改代码：

import logging
import watchtower
from botocore.exceptions import ClientError

def get_stage_logger(stage_name):
    # 按阶段定义独立日志组
    log_group_name = f"/aws/lambda/apigw-authorizer-{stage_name}"
    logger = logging.getLogger(stage_name)
    logger.setLevel(logging.INFO)
    
    # 避免重复添加处理器（执行环境复用问题）
    if not logger.handlers:
        try:
            # 创建CloudWatch日志处理器
            cw_handler = watchtower.CloudWatchLogHandler(log_group=log_group_name)
            logger.addHandler(cw_handler)
            # 添加控制台输出（可选，方便本地调试）
            console_handler = logging.StreamHandler()
            logger.addHandler(console_handler)
        except ClientError as e:
            # 权限不足时降级到默认日志
            default_logger = logging.getLogger()
            default_logger.error(f"Failed to create stage log group: {e}")
            return default_logger
    return logger

def lambda_handler(event, context):
    stage = event.get("requestContext", {}).get("stage", "unknown")
    logger = get_stage_logger(stage)
    
    logger.info(f"Authorizer request received for stage: {stage}")
    # 后续业务逻辑...
    return {
        "principalId": "user",
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Action": "execute-api:Invoke",
                    "Effect": "Allow",
                    "Resource": event["methodArn"]
                }
            ]
        }
    }

3. 配置权限：确保Lambda执行角色拥有以下CloudWatch Logs权限：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:log-group:/aws/lambda/apigw-authorizer-*:*"
        }
    ]
}

方案2：手动调用CloudWatch Logs API（不依赖第三方库）

如果不想引入外部依赖，可以直接用boto3操作CloudWatch Logs：

import logging
import boto3
import time
from botocore.exceptions import ClientError

logs_client = boto3.client('logs')

def put_stage_log(stage_name, message, level="INFO"):
    log_group_name = f"/aws/lambda/apigw-authorizer-{stage_name}"
    log_stream_name = f"{stage_name}-{int(time.time())}"
    
    try:
        # 创建日志组（不存在则忽略）
        logs_client.create_log_group(logGroupName=log_group_name)
        # 创建日志流（不存在则忽略）
        logs_client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
        
        # 获取下一个序列令牌
        response = logs_client.describe_log_streams(
            logGroupName=log_group_name,
            logStreamNamePrefix=log_stream_name,
            limit=1
        )
        sequence_token = response['logStreams'][0].get('uploadSequenceToken')
        
        # 发送日志事件
        log_event = {
            'logGroupName': log_group_name,
            'logStreamName': log_stream_name,
            'logEvents': [
                {
                    'timestamp': int(time.time() * 1000),
                    'message': f"[{level}] {message}"
                }
            ]
        }
        if sequence_token:
            log_event['sequenceToken'] = sequence_token
        
        logs_client.put_log_events(**log_event)
    except ClientError as e:
        # 降级到默认日志
        default_logger = logging.getLogger()
        default_logger.error(f"Failed to send stage log: {e}")

def lambda_handler(event, context):
    stage = event.get("requestContext", {}).get("stage", "unknown")
    
    put_stage_log(stage, f"Authorizer request received for stage: {stage}")
    # 后续业务逻辑...
    return {
        "principalId": "user",
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Action": "execute-api:Invoke",
                    "Effect": "Allow",
                    "Resource": event["methodArn"]
                }
            ]
        }
    }

关键注意事项 #

• 执行环境复用：Lambda的执行环境可能被重复使用，因此要避免重复添加日志处理器，或在每次请求时创建独立的日志器实例。
• 权限控制：必须确保Lambda角色拥有操作目标日志组的权限，避免因权限不足导致日志无法写入。
• 性能影响：手动调用CloudWatch Logs API会增加Lambda的执行时间，高并发场景下建议使用watchtower的批量日志发送功能。

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