如何通过Python logging模块将CDK部署的Lambda日志定向到指定CloudWatch位置？

实现思路 #

Lambda默认会把日志输出到自动创建的CloudWatch日志组，但如果要定向到你指定的日志资源，需要自定义logging.Handler，通过boto3调用CloudWatch Logs API将日志事件发送到目标位置。

具体代码实现 #

1. 自定义CloudWatch日志Handler #

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

class CloudWatchLogsHandler(logging.Handler):
    def __init__(self, log_group_name, log_stream_name=None):
        super().__init__()
        self.log_group = log_group_name
        self.log_stream = log_stream_name or f"lambda-log-stream-{os.environ.get('AWS_LAMBDA_FUNCTION_NAME', 'unknown')}"
        self.client = boto3.client('logs')
        self.sequence_token = None
        
        # 确保日志流存在
        self._create_log_stream_if_not_exists()

    def _create_log_stream_if_not_exists(self):
        try:
            self.client.create_log_stream(
                logGroupName=self.log_group,
                logStreamName=self.log_stream
            )
        except ClientError as e:
            # 日志流已存在则忽略该异常
            if e.response['Error']['Code'] != 'ResourceAlreadyExistsException':
                raise

    def emit(self, record):
        try:
            # 格式化日志消息
            log_message = self.format(record)
            timestamp = int(record.created * 1000)  # CloudWatch要求毫秒级时间戳
            
            # 构造日志事件
            log_events = [{
                'timestamp': timestamp,
                'message': log_message
            }]
            
            # 发送日志事件
            kwargs = {
                'logGroupName': self.log_group,
                'logStreamName': self.log_stream,
                'logEvents': log_events
            }
            if self.sequence_token:
                kwargs['sequenceToken'] = self.sequence_token
            
            response = self.client.put_log_events(**kwargs)
            # 更新下一次发送所需的sequence token
            self.sequence_token = response.get('nextSequenceToken')
        
        except Exception:
            self.handleError(record)

2. 配置logging并绑定Handler #

在Lambda函数入口处添加如下代码：

import os
import logging

# 初始化日志配置
def setup_logging():
    # 从环境变量获取目标日志组
    log_destination = os.environ.get('LOGS_DESTINATION')
    if not log_destination:
        raise ValueError("环境变量LOGS_DESTINATION未设置")
    
    # 配置根logger
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    
    # 移除默认handler，避免重复输出到默认日志组
    for handler in logger.handlers[:]:
        logger.removeHandler(handler)
    
    # 添加自定义CloudWatch Handler
    cw_handler = CloudWatchLogsHandler(log_group_name=log_destination)
    # 设置日志格式，可按需调整
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    cw_handler.setFormatter(formatter)
    logger.addHandler(cw_handler)

# Lambda处理函数
def lambda_handler(event, context):
    setup_logging()
    logger = logging.getLogger(__name__)
    
    # 测试日志输出
    logger.info("这是一条定向到指定CloudWatch日志组的INFO日志")
    logger.warning("这是一条警告日志")
    
    return {
        'statusCode': 200,
        'body': '日志已发送到指定位置'
    }

关键注意事项 #

• Lambda权限配置：在CDK中需要给Lambda角色添加CloudWatch Logs权限，允许执行logs:CreateLogStream和logs:PutLogEvents操作，示例CDK代码片段：

  from aws_cdk import aws_iam as iam
  
  # 假设lambda_role是你的Lambda角色
  lambda_role.add_to_policy(iam.PolicyStatement(
      actions=[
          "logs:CreateLogStream",
          "logs:PutLogEvents"
      ],
      resources=[f"{os.environ['LOGS_DESTINATION']}*"]
  ))
  

• 日志流命名：示例中用Lambda函数名作为日志流的一部分，可根据业务需求自定义命名规则。
• 批量发送优化：如果日志量较大，建议修改Handler实现批量缓存日志（比如积累N条或定时发送），避免频繁调用API触发限流。
• 异常容错：自定义Handler中加入了异常捕获，避免日志发送失败导致Lambda函数报错中断。

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