使用Python解析日志中的JSON请求数据流并排除响应内容

看起来你遇到的核心问题是在实时流式日志中正确分组跨多行的JSON条目——因为tail -0f是逐行输出的，而Request/Response的JSON可能分散在多行，直接按行处理会把完整的JSON拆碎，导致解析失败。下面是一套实用的解决方案，结合生成器和边界识别来处理这个场景：

核心思路 #

1. 识别日志条目边界：每条日志以日期前缀开头（比如2024-05-20 14:30:00），用正则匹配这个前缀作为新条目的起始标记。
2. 流式合并多行条目：用生成器逐行读取输入，累积行直到遇到下一个日期前缀，输出完整的单条日志块。
3. 提取目标内容：对每个完整的日志块，提取日期和Request的JSON，忽略Response部分。

完整代码实现 #

import re
import sys
import json

# 调整这个正则以匹配你的日志日期前缀格式（比如带毫秒/时区的话修改正则）
DATE_PREFIX_REGEX = re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}')

def stream_log_entries(input_stream):
    """生成器：将逐行输入合并为完整的日志条目"""
    current_entry = []
    for raw_line in input_stream:
        line = raw_line.rstrip('\n')  # 保留行内的空格，仅去掉末尾换行符
        
        # 遇到新的日期前缀，输出之前累积的条目（如果存在）
        if DATE_PREFIX_REGEX.match(line):
            if current_entry:
                yield '\n'.join(current_entry)  # 用换行符还原原条目格式
            current_entry = [line]
        else:
            # 属于当前条目的后续行，加入累积列表
            current_entry.append(line)
    
    # 处理最后一条未输出的日志
    if current_entry:
        yield '\n'.join(current_entry)

def parse_target_content():
    """解析每个完整日志条目，提取日期和Request JSON"""
    for entry in stream_log_entries(sys.stdin):
        # 提取日期
        date_match = DATE_PREFIX_REGEX.match(entry)
        if not date_match:
            print(f"跳过无效日志条目：{entry[:50]}...", file=sys.stderr)
            continue
        log_date = date_match.group()

        # 提取Request JSON（这里用状态机处理嵌套大括号，比正则更可靠）
        request_start = entry.find('Request: {')
        if request_start == -1:
            continue  # 没有Request部分的条目直接跳过
        
        # 计数大括号来定位Request JSON的结束位置
        brace_count = 1
        json_end_idx = request_start + len('Request: {')
        while json_end_idx < len(entry) and brace_count > 0:
            char = entry[json_end_idx]
            if char == '{':
                brace_count += 1
            elif char == '}':
                brace_count -= 1
            json_end_idx += 1
        
        if brace_count != 0:
            print(f"[{log_date}] Request JSON格式不完整，跳过", file=sys.stderr)
            continue
        
        # 提取并解析JSON
        request_json_str = entry[request_start + len('Request: '):json_end_idx]
        try:
            request_data = json.loads(request_json_str)
            # 按你需要的格式输出结果，这里是日期+格式化的JSON
            print(f"[{log_date}] 解析到Request：")
            print(json.dumps(request_data, indent=2, ensure_ascii=False))
            print("="*50)
        except json.JSONDecodeError as e:
            print(f"[{log_date}] Request JSON解析失败：{str(e)}", file=sys.stderr)

if __name__ == "__main__":
    parse_target_content()

关键细节说明 #

1. 边界识别的可靠性：用日期前缀作为条目分隔符，比依赖换行或其他标记更准确，因为日志的JSON内容本身会包含换行。
2. 状态机处理JSON：相比正则表达式，用大括号计数的方式能正确处理嵌套的JSON结构，避免正则在复杂JSON下提前终止匹配的问题。
3. 流式内存效率：生成器逐行处理输入，不会一次性加载全部日志，完全适配tail -0f的实时场景。
4. 错误处理：加入了无效条目、不完整JSON、解析失败的错误提示，避免程序崩溃，同时方便排查问题。

使用方法 #

保存上面的代码为parse.py，然后执行：

tail -0f apiserver.log | python parse.py

适配你的日志格式 #

如果你的日志有特殊格式，需要调整以下部分：

• 修改DATE_PREFIX_REGEX的正则，匹配你的实际日期前缀（比如带毫秒：^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}）
• 调整request_start = entry.find('Request: {')中的标记字符串，比如如果日志里是"request": {...}就改成对应的查找内容

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