如何用Python3 Boto突破限制获取指定CloudWatch日志组的全部日志？

可以实现，AWS官方Python SDK boto3 提供了分页机制，能绕过单次10000条记录的限制，获取日志组的全部日志。以下是两种常用实现方式：

方法一：使用内置分页器（推荐） #

boto3的分页器（Paginator）会自动处理nextToken参数，帮你遍历所有分页结果，无需手动管理分页逻辑。

步骤： #

1. 安装依赖：

pip install boto3

2. 代码示例：

import boto3
import datetime

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

# 配置查询参数
log_group_name = '你的日志组名称'
# 时间范围：毫秒级时间戳，示例为最近7天
start_time = (datetime.datetime.now() - datetime.timedelta(days=7)).timestamp() * 1000
end_time = datetime.datetime.now().timestamp() * 1000

# 创建分页器
paginator = logs_client.get_paginator('filter_log_events')

# 遍历所有分页结果
for page in paginator.paginate(
    logGroupName=log_group_name,
    startTime=start_time,
    endTime=end_time,
    # 可选：添加过滤模式，比如只包含特定关键词的日志
    # filterPattern='ERROR'
):
    # 处理当前页的日志事件
    for event in page['events']:
        print(event['message'])
        # 可将日志写入文件或数据库

方法二：手动循环处理分页令牌 #

如果需要更自定义的分页逻辑（比如中途停止、自定义重试），可以手动检查nextToken并循环调用API：

import boto3
import datetime

logs_client = boto3.client('logs')
log_group_name = '你的日志组名称'
start_time = (datetime.datetime.now() - datetime.timedelta(days=7)).timestamp() * 1000
end_time = datetime.datetime.now().timestamp() * 1000

next_token = None
while True:
    # 调用API，仅当存在nextToken时传入该参数
    api_params = {
        'logGroupName': log_group_name,
        'startTime': start_time,
        'endTime': end_time
    }
    if next_token:
        api_params['nextToken'] = next_token
    
    response = logs_client.filter_log_events(**api_params)
    
    # 处理当前页日志
    for event in response['events']:
        print(event['message'])
    
    # 检查是否还有下一页
    next_token = response.get('nextToken')
    if not next_token:
        break

注意事项 #

• 凭证配置：确保你的Python环境已配置AWS凭证（可通过环境变量AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY，或~/.aws/credentials文件）
• 速率限制：CloudWatch Logs API有调用速率限制，boto3默认会自动重试，但如果遇到ThrottlingException，可添加延迟（如time.sleep(1)）或调整客户端重试配置
• 时间范围优化：尽量缩小查询的时间范围，避免不必要的分页，提升效率

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