66 lines
2.6 KiB
Python
66 lines
2.6 KiB
Python
"""分析 Nginx 访问日志：状态码分布、请求最多的 IP、最慢的接口、5xx 明细。
|
||
|
||
日志格式：Nginx combined 格式末尾追加 $request_time（秒），对应配置：
|
||
log_format main '$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent '
|
||
'"$http_referer" "$http_user_agent" $request_time';
|
||
运行：python analyze.py [日志文件]，默认分析同目录的 sample_access.log
|
||
等价的命令行写法见 Linux 专题课程（awk / sort / uniq）。
|
||
"""
|
||
import re
|
||
import sys
|
||
from collections import Counter, defaultdict
|
||
from pathlib import Path
|
||
|
||
LINE = re.compile(
|
||
r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) [^"]*" '
|
||
r'(?P<status>\d{3}) (?P<bytes>\d+) "[^"]*" "[^"]*" (?P<rt>[\d.]+)'
|
||
)
|
||
|
||
|
||
def parse(path: Path):
|
||
for n, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||
m = LINE.match(line)
|
||
if not m:
|
||
print(f"第 {n} 行格式不匹配，跳过", file=sys.stderr)
|
||
continue
|
||
d = m.groupdict()
|
||
d["path"] = d["path"].split("?")[0] # 去掉查询参数，按接口聚合
|
||
d["rt"] = float(d["rt"])
|
||
yield d
|
||
|
||
|
||
def analyze(rows: list[dict]) -> dict:
|
||
by_api: dict[str, list[float]] = defaultdict(list)
|
||
for r in rows:
|
||
by_api[f'{r["method"]} {r["path"]}'].append(r["rt"])
|
||
slowest = sorted(((max(v), k) for k, v in by_api.items()), reverse=True)[:3]
|
||
return {
|
||
"status": Counter(r["status"] for r in rows),
|
||
"top_ips": Counter(r["ip"] for r in rows).most_common(3),
|
||
"slowest": [(k, t) for t, k in slowest],
|
||
"errors_5xx": [(r["time"], r["status"], r["path"], r["rt"]) for r in rows if r["status"].startswith("5")],
|
||
}
|
||
|
||
|
||
def main() -> None:
|
||
path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(__file__).with_name("sample_access.log")
|
||
rows = list(parse(path))
|
||
res = analyze(rows)
|
||
print(f"共 {len(rows)} 条请求")
|
||
print("状态码分布：", dict(sorted(res["status"].items())))
|
||
print("请求最多的 IP：", res["top_ips"])
|
||
print("最慢的接口（最大耗时）：", res["slowest"])
|
||
print("5xx 明细：")
|
||
for e in res["errors_5xx"]:
|
||
print(" ", *e)
|
||
|
||
if path.name == "sample_access.log": # 用示例日志自检
|
||
assert res["status"]["401"] == 3 and res["status"]["429"] == 1
|
||
assert res["top_ips"][0] == ("198.51.100.7", 4)
|
||
assert res["slowest"][0] == ("GET /api/reports/monthly", 30.001)
|
||
assert [e[1] for e in res["errors_5xx"]] == ["504", "502"]
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|