AWS Lambda（Python）Logger实例异常：有时追加至旧日志而非新建

AWS Lambda的执行环境存在复用机制（热启动）：

• 当函数首次执行或闲置过久后执行，会触发冷启动：创建全新的执行环境，此时你创建的Logger对象是新实例，日志会写入新的CloudWatch日志流。
• 当函数在短时间内重复执行，Lambda会复用之前的执行环境（热启动）：此时你的Logger对象是之前环境中已存在的实例，后续日志会直接追加到同一个日志流中，导致你看到“有时追加至之前日志”的现象。

这种行为和你调用函数的方式无关（测试标签或定时事件），只取决于Lambda是否复用了执行环境。

根据需求，有三种常见处理方式：

1. 区分单次调用的日志（无需新日志流） #

如果不需要强制生成新的日志流，只是想快速区分不同调用的日志，可以在日志中加入Lambda的请求ID（每次调用唯一）。以aws-lambda-powertools的Logger为例：

def lambda_handler(event, context):
    # 在handler内部初始化Logger，并传入当前请求ID
    joblog = Logger(service="xxx", level="INFO", request_id=context.aws_request_id)
    joblog.info("日志内容", extra={"request_id": context.aws_request_id})
    # 后续日志操作

这样即使日志在同一个流中，也能通过request_id快速筛选出单次调用的所有日志。

2. 强制生成独立的CloudWatch日志流 #

如果必须每次调用都生成独立的日志流，可结合Lambda请求ID或时间戳生成唯一日志流名称，再通过boto3操作CloudWatch Logs写入。示例代码：

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

def lambda_handler(event, context):
    log_client = boto3.client('logs')
    log_group_name = "/aws/lambda/your-function-name"  # 替换为你的函数日志组
    # 生成唯一日志流名称
    log_stream_name = f"{context.aws_request_id}-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    # 创建日志流（已存在则忽略错误）
    try:
        log_client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
    except log_client.exceptions.ResourceAlreadyExistsException:
        pass
    
    # 配置Logger写入自定义日志流
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    # 移除默认Lambda日志处理器
    for handler in logger.handlers[:]:
        logger.removeHandler(handler)
    
    # 自定义CloudWatch日志处理器
    class CloudWatchLogHandler(logging.Handler):
        def emit(self, record):
            log_event = {
                'logStreamName': log_stream_name,
                'logEvents': [{
                    'timestamp': int(datetime.now().timestamp() * 1000),
                    'message': self.format(record)
                }]
            }
            try:
                # 处理序列令牌逻辑（简化示例，实际需补充）
                response = log_client.describe_log_streams(
                    logGroupName=log_group_name,
                    logStreamNamePrefix=log_stream_name
                )
                next_token = response['logStreams'][0]['uploadSequenceToken'] if 'uploadSequenceToken' in response['logStreams'][0] else None
                if next_token:
                    log_event['sequenceToken'] = next_token
                log_client.put_log_events(logGroupName=log_group_name, **log_event)
            except ClientError as e:
                pass
    
    logger.addHandler(CloudWatchLogHandler())
    logger.info("xxx")
    return {"statusCode": 200}

注意：这种方式会增加API调用开销，需确保Lambda角色拥有logs:CreateLogStream和logs:PutLogEvents权限。

3. 调整环境复用频率 #

虽然无法直接禁用热启动，但可以将函数的预留并发设为0（默认值），让闲置执行环境快速回收。但这会提升冷启动频率，可能影响执行性能，需根据业务场景权衡。

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