用 Python 标准库分析 JSONL 请求日志:5xx 错误率与 P95 延迟

举报
yd_210400787 发表于 2026/10/30 10:19:56 2026/10/30
【摘要】 用 Python 标准库读取 JSONL 请求日志,校验状态码与耗时,统计 5xx 错误率和 nearest-rank P95 延迟;附可运行脚本、样例输出与统计口径说明。

排查接口变慢时,平均耗时往往不够用:少量很慢的请求可能被大量快速请求掩盖。本文用 Python 标准库写一个小脚本,读取每行一个 JSON 对象的请求日志,统计有效请求数、5xx 错误率和 P95 延迟,并把坏数据单独计数。示例只使用本地文件,不需要安装第三方包或上传日志。

准备输入

假设 requests.jsonl 的内容如下。每行是独立 JSON,对应一次请求;status 是 HTTP 状态码,duration_ms 是非负的毫秒耗时。

{"status":200,"duration_ms":12.5}
{"status":200,"duration_ms":40}
{"status":503,"duration_ms":110}
{"status":404,"duration_ms":9}
{"status":200,"duration_ms":25}
bad-json

最后一行故意写错,用来演示脚本如何跳过无效记录。实际日志可能包含路径、请求 ID 等字段,脚本只读取上述两个字段;如果日志含有个人信息,应先按自己的数据处理要求脱敏。

完整脚本

保存为 analyze_jsonl.py:

#!/usr/bin/env python3
"""Summarize request status and latency from a JSONL file."""

import argparse
import json
import math
from pathlib import Path


def analyze(path: Path) -> tuple[int, int, int, list[float]]:
    total = 0
    errors = 0
    invalid = 0
    durations: list[float] = []

    with path.open("r", encoding="utf-8-sig") as source:
        for line_number, line in enumerate(source, start=1):
            if not line.strip():
                continue
            try:
                record = json.loads(line)
                status = record["status"]
                duration = record["duration_ms"]
                if (
                    not isinstance(status, int)
                    or isinstance(status, bool)
                    or not 100 <= status <= 599
                    or not isinstance(duration, (int, float))
                    or isinstance(duration, bool)
                    or not math.isfinite(duration)
                    or duration < 0
                ):
                    raise ValueError("invalid status or duration_ms")
            except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
                invalid += 1
                print(f"跳过第 {line_number} 行:{exc}")
                continue

            total += 1
            errors += status >= 500
            durations.append(float(duration))

    return total, errors, invalid, durations


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("path", type=Path, help="JSONL 日志路径")
    args = parser.parse_args()

    total, errors, invalid, durations = analyze(args.path)
    if total == 0:
        print(f"没有有效记录;无效行:{invalid}")
        return

    durations.sort()
    p95_index = math.ceil(total * 0.95) - 1
    print(f"有效请求:{total}")
    print(f"5xx 请求:{errors},占比:{errors / total:.2%}")
    print(f"P95 延迟:{durations[p95_index]:.1f} ms")
    print(f"无效行:{invalid}")


if __name__ == "__main__":
    main()

在终端运行:

python3 analyze_jsonl.py requests.jsonl

针对上面的样例,实测输出为:

跳过第 6 行:Expecting value: line 1 column 1 (char 0)
有效请求:5
5xx 请求:1,占比:20.00%
P95 延迟:110.0 ms
无效行:1

统计口径要说清楚

脚本把 500~599 视为服务端错误,因此 404 不计入 5xx 错误率。空行直接忽略;JSON 格式错误、缺少字段、字段类型错误、超出范围或负耗时都计为无效行。Python 的 bool 属于 int 的子类,所以这里特意排除 true/false;math.isfinite() 排除了无穷大和 NaN,避免污染排序结果。

P95 采用 nearest-rank(向上取整排名)定义:先把 N 个耗时从小到大排序,再取第 ceil(0.95 × N) 个值。样例有 5 条有效记录,排名为 ceil(4.75)=5,所以 P95 是最大值 110 ms。不同监控系统可能使用插值或直方图估算,小样本下数值可能不同;比较两套结果前应先对齐口径。

这个版本会把所有有效耗时保存在内存中,适合小到中等规模的本地日志。若需要处理海量日志,可以改用流式分桶或专门的分位数算法,但那时 P95 将可能是近似值。生产排查还应按时间窗口、接口和实例分组,否则总体指标可能掩盖局部故障。

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0)

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。