使用Python下载指定范围AWS CloudWatch日志时事件缺失的技术问询

我需要从AWS下载符合以下条件的所有CloudWatch日志：

• 特定日志组（log group）
• 特定时间范围

我的实现思路：

• 遍历该日志组下的所有日志流（log stream）
• 遍历每个日志流中的事件，构建完整的日志事件列表

编写的Python代码如下：

import boto3


def overlaps(start1, end1, start2, end2):
    return max(start1, start2) < min(end1, end2)


def load_logs(region, group, start=0, end=2672995600000):
    client = boto3.client('logs', region_name=region)
    paginator = client.get_paginator('describe_log_streams')

    response_iterator = paginator.paginate(logGroupName=group)
    events = []
    for page in response_iterator:
        for log_stream in page["logStreams"]:
            print(f"Stream: {log_stream['logStreamName']}, start: {log_stream['firstEventTimestamp']} end: {log_stream['lastEventTimestamp']}")
            if overlaps(log_stream["firstEventTimestamp"], log_stream["lastEventTimestamp"], start, end):
                print("processing")
                token = None
                while True:
                    event_args = {
                        "logGroupName": group,
                        "logStreamName": log_stream['logStreamName'],
                        "startTime": start,
                        "endTime": end
                    }

                    if token is not None:
                        event_args["nextToken"] = token

                    response = client.get_log_events(**event_args)

                    for event in response["events"]:
                        if start < event["timestamp"] < end:
                            events.append(event)

                    if response["nextBackwardToken"] == token:
                        break
                    else:
                        token = response["nextBackwardToken"]

    print(events)

当传入start=0、end=2672995600000（遥远未来时间）时，仅能下载部分事件，events列表未包含所有日志事件，怀疑是get_log_events的迭代环节存在遗漏，求解决办法。

问题根源 #

你的代码仅使用nextBackwardToken进行反向遍历，但CloudWatch Logs的get_log_events默认是**正向（从头往尾）**获取事件，且分页令牌分为nextForwardToken和nextBackwardToken，只处理反向令牌会导致遗漏部分事件。同时，未指定startFromHead=True时，API可能从中间位置开始返回事件，跳过头部日志。

修正后的代码 #

import boto3

def overlaps(start1, end1, start2, end2):
    return max(start1, start2) < min(end1, end2)

def load_logs(region, group, start=0, end=2672995600000):
    client = boto3.client('logs', region_name=region)
    paginator = client.get_paginator('describe_log_streams')
    response_iterator = paginator.paginate(logGroupName=group)
    events = []

    for page in response_iterator:
        for log_stream in page["logStreams"]:
            print(f"Stream: {log_stream['logStreamName']}, start: {log_stream['firstEventTimestamp']} end: {log_stream['lastEventTimestamp']}")
            if not overlaps(log_stream["firstEventTimestamp"], log_stream["lastEventTimestamp"], start, end):
                continue
            print("processing")

            # 正向遍历所有事件，从日志流头部开始
            forward_token = None
            while True:
                event_args = {
                    "logGroupName": group,
                    "logStreamName": log_stream['logStreamName'],
                    "startTime": start,
                    "endTime": end,
                    "startFromHead": True  # 强制从最早事件开始遍历
                }
                if forward_token:
                    event_args["nextToken"] = forward_token

                response = client.get_log_events(**event_args)
                # 过滤并收集时间范围内的事件
                events.extend([event for event in response["events"] if start <= event["timestamp"] <= end])

                # 检查是否还有更多正向事件
                if response["nextForwardToken"] == forward_token:
                    break
                forward_token = response["nextForwardToken"]

    print(f"共获取到 {len(events)} 条日志事件")
    return events

关键修复点 #

1. 添加startFromHead=True：强制从日志流的最早事件开始遍历，避免API从中间位置返回数据。
2. 使用nextForwardToken迭代：正向遍历对应的分页令牌是nextForwardToken，确保完整获取所有分页事件。
3. 调整时间过滤逻辑：用<=替代<，避免刚好等于start或end的事件被过滤。
4. 简化代码结构：用continue跳过无需处理的日志流，提升代码可读性。

额外建议 #

• 若日志量极大，建议加入批量写入文件的逻辑，避免内存占用过高。
• 可添加限流重试机制（捕获ThrottlingException），应对AWS API的请求限制。

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