如何将Lambda函数/tmp目录中的日志文件上传至CloudWatch？

没问题，我来帮你搞定这个需求！Lambda函数执行结束后，/tmp目录确实会被AWS清理回收，所以必须在函数退出前完成日志的上传操作。下面是具体的实现步骤和代码示例（以Python为例，其他语言思路一致）：

1. 配置IAM权限 #

首先要确保你的Lambda执行角色拥有CloudWatch Logs的操作权限，否则上传会失败。给角色添加以下权限策略（可以按需缩小资源范围）：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "logs:DescribeLogStreams"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        }
    ]
}

2. 核心实现代码 #

我们可以在函数的finally块中执行日志上传（确保不管业务逻辑成功还是失败，日志都能被上传），同时处理CloudWatch Logs的批次限制（单次最多上传1000条日志事件，总大小不超过1MB）：

import boto3
import time
import os

# 初始化CloudWatch Logs客户端
logs_client = boto3.client('logs')

def upload_logs_to_cloudwatch(log_group_name, log_stream_name, log_file_path):
    """将指定日志文件的内容上传到CloudWatch Logs"""
    # 确保日志组存在，不存在则创建
    try:
        logs_client.create_log_group(logGroupName=log_group_name)
    except logs_client.exceptions.ResourceAlreadyExistsException:
        pass

    # 确保日志流存在，不存在则创建
    try:
        logs_client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
    except logs_client.exceptions.ResourceAlreadyExistsException:
        pass

    # 获取上传所需的序列令牌（首次上传可忽略，后续批次需要）
    sequence_token = None
    try:
        stream_response = logs_client.describe_log_streams(
            logGroupName=log_group_name,
            logStreamNamePrefix=log_stream_name,
            limit=1
        )
        if 'uploadSequenceToken' in stream_response['logStreams'][0]:
            sequence_token = stream_response['logStreams'][0]['uploadSequenceToken']
    except Exception as e:
        print(f"获取序列令牌失败: {str(e)}")

    # 读取日志文件并分批上传
    log_events = []
    with open(log_file_path, 'r') as log_file:
        for line in log_file:
            # 构造CloudWatch日志事件（时间戳需为毫秒级）
            log_event = {
                'timestamp': int(time.time() * 1000),
                'message': line.strip()
            }
            log_events.append(log_event)

            # 达到批次上限时立即上传
            if len(log_events) >= 1000:
                if sequence_token:
                    response = logs_client.put_log_events(
                        logGroupName=log_group_name,
                        logStreamName=log_stream_name,
                        logEvents=log_events,
                        sequenceToken=sequence_token
                    )
                else:
                    response = logs_client.put_log_events(
                        logGroupName=log_group_name,
                        logStreamName=log_stream_name,
                        logEvents=log_events
                    )
                sequence_token = response['nextSequenceToken']
                log_events = []

        # 上传剩余的日志事件
        if log_events:
            if sequence_token:
                logs_client.put_log_events(
                    logGroupName=log_group_name,
                    logStreamName=log_stream_name,
                    logEvents=log_events,
                    sequenceToken=sequence_token
                )
            else:
                logs_client.put_log_events(
                    logGroupName=log_group_name,
                    logStreamName=log_stream_name,
                    logEvents=log_events
                )

def lambda_handler(event, context):
    try:
        # 这里是你的业务逻辑代码
        # 假设外部库将日志写入/tmp/app.log
        # ... 你的业务代码 ...
    finally:
        # 不管执行结果如何，都上传日志
        # 可以用context动态获取函数名，避免硬编码
        log_group_name = f"/aws/lambda/{context.function_name}"
        # 用时间戳作为日志流后缀，避免重复
        log_stream_name = f"{context.function_name}-{int(time.time())}"
        upload_logs_to_cloudwatch(log_group_name, log_stream_name, '/tmp/app.log')

3. 注意事项 #

• 批次限制：CloudWatch Logs的PutLogEvents接口有严格的批次限制，所以代码中做了分批处理，避免触发报错。
• 执行时间：Lambda函数有执行时间上限（默认3秒，最大15分钟），如果日志文件过大，要确保上传操作不会占用过多时间导致函数超时。
• 日志时间戳：CloudWatch要求日志事件的时间戳为毫秒级，代码中用int(time.time() * 1000)生成，你也可以根据日志文件中的时间戳来设置（如果日志里有记录的话）。
• 其他语言适配：如果用Node.js/Java等语言，思路完全一致，只需调用对应AWS SDK的CloudWatch Logs模块即可。

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