如何用boto3从AWS CloudWatch获取所有错误日志？解决start_query超时问题

1. 分批次异步拆分时间范围查询 #

单查询超时的核心原因是时间跨度太大、扫描数据量过多。可以把整体时间范围拆分成多个小窗口（比如按6小时、1天为单位），异步提交多个查询，之后再合并结果。

示例代码逻辑：

import boto3
import time
from datetime import datetime, timedelta

client = boto3.client('logs')

def split_time_range(start_ts, end_ts, window_hours=6):
    # 将大时间范围拆分为多个小时级窗口
    current_start = start_ts
    while current_start < end_ts:
        window_end = min(current_start + timedelta(hours=window_hours).total_seconds() * 1000, end_ts)
        yield (current_start, window_end)
        current_start = window_end

# 遍历目标日志组
for group in your_log_groups_list:
    creation_time = get_log_group_creation_time(group) # 自行实现获取日志组创建时间的逻辑
    last_event_ts = int(datetime.now().timestamp() * 1000)
    start_ts = int(creation_time) * 1000

    # 提交分窗口查询
    query_ids = []
    for window_start, window_end in split_time_range(start_ts, last_event_ts):
        resp = client.start_query(
            logGroupName=group,
            startTime=window_start,
            endTime=window_end,
            queryString='filter @message like /ERROR/'
        )
        query_ids.append(resp['queryId'])
    
    # 轮询查询结果并合并
    for qid in query_ids:
        while True:
            result = client.get_query_results(queryId=qid)
            if result['status'] in ['Complete', 'Failed']:
                if result['status'] == 'Complete':
                    # 这里处理查询到的错误日志结果
                    process_error_logs(result['results'])
                break
            time.sleep(5) # 间隔5秒再轮询

2. 优化查询语句缩小扫描范围 #

• 优先过滤时间或其他字段：先通过@timestamp缩小时间范围，再匹配错误关键词，减少扫描的日志条目：

  filter @timestamp > '2024-05-01' and @message like /ERROR/
  

• 使用精确正则：避免模糊匹配，比如用/^ERROR:/代替/ERROR/，减少误匹配的日志量。

3. 预创建Metric Filter聚合错误日志 #

提前给日志组创建Metric Filter，自动把包含ERROR的日志过滤并聚合到CloudWatch Metrics，甚至可以配置导出到S3。后续无需再跑大范围查询，直接从Metrics获取统计数据，或从S3批量分析原始日志：

创建Metric Filter示例：

client.put_metric_filter(
    logGroupName=group,
    filterName='ErrorLogFilter',
    filterPattern='ERROR',
    metricTransformations=[
        {
            'metricName': 'ErrorLogCount',
            'metricNamespace': 'Custom/ApplicationLogs',
            'metricValue': '1'
        }
    ]
)

4. 异步查询+后台轮询结果 #

提交查询后，无需同步等待，把查询ID存储到DynamoDB或其他存储中，用独立的Lambda函数或定时进程轮询查询状态，获取结果后导出到S3或数据库，避免因长时间等待导致超时。

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