如何使用boto3获取CloudWatch日志事件中的JSON数据

核心逻辑 #

你现有代码只完成了日志组、日志流名称的拉取，要拿到日志里的JSON内容，还需要补上三个核心环节：

• 补全CloudWatch Logs客户端初始化，修复原代码未定义client的问题
• 处理接口分页逻辑，全量拉取每个日志流下的实际日志事件
• 从日志事件的message字段中提取合法JSON片段，做容错解析避免非JSON日志导致程序崩溃

注意：CloudWatch所有列表类接口（查询日志组、日志流、日志事件）都有返回条数/大小限制，不处理分页会出现数据漏拉的问题。原代码遍历日志组时固定传入logGroupName[0]查询流，会导致除第一个日志组外的其他日志组流信息拉取错误。

完整可运行代码 #

import boto3
import json

# 初始化CloudWatch Logs客户端
boto3.setup_default_session(region_name="us-west-2")
client = boto3.client('logs')

# 拉取所有日志组，若已明确目标日志组可跳过此段，直接给target_log_group赋值即可
logGroupName = []
lg_paginator = client.get_paginator('describe_log_groups')
for page in lg_paginator.paginate():
    for lg in page['logGroups']:
        logGroupName.append(lg['logGroupName'])

# 配置目标日志组，替换为你需要提取内容的实际日志组名称
target_log_group = logGroupName[0]

# 拉取目标日志组下的所有日志流
log_streams = []
ls_paginator = client.get_paginator('describe_log_streams')
for page in ls_paginator.paginate(logGroupName=target_log_group, orderBy='LastEventTime', descending=True):
    for ls in page['logStreams']:
        log_streams.append(ls['logStreamName'])

# 拉取所有日志事件，解析内嵌JSON
parsed_result = []
for stream_name in log_streams:
    event_paginator = client.get_paginator('get_log_events')
    # 分页拉取单流内全量日志
    for event_page in event_paginator.paginate(
        logGroupName=target_log_group,
        logStreamName=stream_name,
        startFromHead=True
    ):
        for event in event_page['events']:
            raw_msg = event['message'].strip()
            try:
                # 适配带非JSON前缀的日志场景，自动定位JSON起止位置
                json_start_idx = raw_msg.find('{')
                json_end_idx = raw_msg.rfind('}') + 1
                if json_start_idx != -1 and json_end_idx > json_start_idx:
                    log_data = json.loads(raw_msg[json_start_idx:json_end_idx])
                    parsed_result.append(log_data)
            except json.JSONDecodeError:
                # 非JSON格式日志直接跳过，可按需添加错误日志记录
                continue

# 打印解析结果，可按需替换为存储、后续处理逻辑
for log_item in parsed_result:
    print(log_item)

使用说明 #

• 如果明确要查询指定时间范围内的日志，可以在get_log_events的分页参数中添加startTime、endTime参数，注意时间单位为毫秒级Unix时间戳
• 如果日志有固定格式前缀，可直接调整JSON截取逻辑，比如固定前缀长度为N时直接切片raw_msg[N:]做解析即可
• 日志量级较大时，建议替换为filter_log_events接口，支持按关键词、时间范围过滤后拉取，执行效率更高

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