把日志说的话变成数字

目标

只用标准库，构建从访问日志和部署 CSV 中提取运维指标（条数、状态分布、热门路径、分位数、最糟糕的 1 分钟、按服务的失败率）的工具。

为什么重要

生产服务器上既没有 pandas，也没有互联网，但有 csv、Counter、statistics、datetime。逐行读取的解析器、分位数、带时区的时间比较——把这三样练熟，无论日志是 5 千行还是 50 万行，都能用同一个工具给出答案。素材是 /opt/fixtures/pyops/access.log（nginx combined 加上最后一列处理时间（秒），从 2026-09-10 02:00 到 05:00 KST，03:12 到 03:17 是故障时段）和 /opt/fixtures/pyops/deploys.csv（service,version,deployed_at,duration_s,status）。

步骤

1. 创建 /root/pyops/data/logstat.py。logstat.py count <로그>（占位符为日志）用一行 lines=<n> parsed=<m> 输出总行数和用正则解析成功的行数。解析时提取时间、路径、状态码和处理时间。
2. logstat.py status <로그>（占位符为日志）用一行 2xx=<n> 3xx=<n> 4xx=<n> 5xx=<n> 输出按状态码类别的条数（Counter）。
3. logstat.py top <로그> --n 5（占位符为日志）按请求数从多到少，每行一个，以 <건수> <경로>（占位符依次为条数与路径）的格式输出请求数最多的前 n 个路径（most_common）。
4. logstat.py latency <로그>（占位符为日志）用 p50=<x> p95=<x> p99=<x> 输出处理时间的 p50、p95、p99。取 statistics.quantiles(times, n=100)（默认 method）的 [49]、[94]、[98]，四舍五入到小数点后第三位。
5. logstat.py worst <로그>（占位符为日志）用 worst_minute=<YYYY-MM-DDTHH:MM> count=<n> 输出 5xx 最多的 1 分钟区间。时间用 strptime 构造 aware datetime，并把秒调整为 0 后归并。保持日志的时区（+0900）原样写出。
6. 创建 /root/pyops/data/deploys.py。deploys.py summary <csv> 对每个服务，按服务名称顺序，每行输出一条 service=<이름> total=<n> failed=<m> fail_rate=<0.000> median_s=<x>（占位符为名称）（DictReader、statistics.median，fail_rate 取小数点后第三位）。
7. 给 logstat.py status <로그> --since <ISO> --until <ISO>（占位符为日志与 ISO 时间）加上时间过滤。两个值是形如 2026-09-10T03:00:00+09:00 的带时区 ISO 8601，只统计满足 since ≤ ts < until 的行。
8. logstat.py report <로그> --json（占位符为日志）向标准输出输出一个包含上述全部指标的 JSON 对象。键是 lines、parsed、status（按类别的对象）、top（[[경로, 건수], ...]，占位符依次为路径与条数，共 5 个）、latency（p50、p95、p99）、worst_minute（{"minute": ..., "count": ...}）。把它保存到 /root/pyops/data/report.json。

参考

• 正则示例：\[(?P<ts>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) [^"]*" (?P<status>\d{3}) \d+ "[^"]*" "[^"]*" (?P<rt>[\d.]+)$
• 时间：datetime.strptime(ts, "%d/%b/%Y:%H:%M:%S %z")，过滤输入：datetime.fromisoformat(s)
• 常见错误：用平均值汇报延迟，比较 naive 和 aware 时间，不统计解析失败的行。

逐行读取并统计解析成功数

创建 /root/pyops/data/logstat.py。logstat.py count <로그>（占位符为日志）输出 lines=<n> parsed=<m>。用能提取时间、路径、状态和处理时间的正则来解析。

文件用 for line in f 逐行读取，用 re.compile 编译好的模式做 search，结果为 None 就跳过，但 lines 要计数。也可以直接使用参考部分的正则。

按状态码类别的条数

logstat.py status <로그>（占位符为日志）输出一行 2xx=<n> 3xx=<n> 4xx=<n> 5xx=<n>。

把 f"{status // 100}xx" 作为键放进 Counter。不存在的类别也必须输出 0（dict.get(k, 0)）。

请求最多的前 n 个路径

logstat.py top <로그> --n 5（占位符为日志）按从多到少，每行一个输出 <건수> <경로>（占位符依次为条数与路径）。

Counter.most_common(n) 会按从多到少返回（值，条数）列表。输出顺序是先条数，后路径。

不用平均值，而用分位数

logstat.py latency <로그>（占位符为日志）输出 p50=<x> p95=<x> p99=<x>。取 statistics.quantiles(times, n=100) 的 [49]、[94]、[98]，四舍五入到小数点后第三位。

n=100 时有 99 个切分点，所以索引 49、94、98 分别是 p50、p95、p99。method 请保持默认值（exclusive）。

5xx 最多的 1 分钟

logstat.py worst <로그>（占位符为日志）输出 worst_minute=<YYYY-MM-DDTHH:MM> count=<n>。时间保持日志的时区原样，把秒调整为 0，按 1 分钟为单位归并。

对 strptime 使用 %z，就会得到 aware datetime。把 ts.replace(second=0, microsecond=0) 作为 Counter 的键，然后看 most_common(1)。输出用 strftime("%Y-%m-%dT%H:%M")。

部署 CSV 中按服务的失败率

创建 /root/pyops/data/deploys.py。deploys.py summary <csv> 按服务名称顺序，每行输出一条 service=<이름> total=<n> failed=<m> fail_rate=<0.000> median_s=<x>（占位符为名称）。

csv.DictReader 会为每行返回一个字典，值全都是字符串。把 duration_s 收集到 defaultdict(list) 中，再用 statistics.median 计算。fail_rate 是 failed / total，取小数点后第三位。

用带时区的时间来过滤

logstat.py status <로그> --since <ISO> --until <ISO>（占位符为日志与 ISO 时间）只统计满足 since ≤ ts < until 的行。两个值是带时区的 ISO 8601（例如 2026-09-10T03:00:00+09:00）。

如果用 argparse 的 type=datetime.fromisoformat 来接收，得到的就是 aware datetime。与 naive 值比较会出现 TypeError，所以要确认输入中有时区。

把全部内容做成 JSON 报告

logstat.py report <로그> --json（占位符为日志）输出一个包含 lines、parsed、status、top（5 个，[路径, 条数]）、latency、worst_minute（{minute, count}）的 JSON 对象，并把它保存到 /root/pyops/data/report.json。

把 datetime 放进 json.dumps 会失败——要先转成字符串再放进去。用 logstat.py report ... --json > /root/pyops/data/report.json 保存。