日志分析与阈值告警
实时 tail 日志、正则提取指标、超阈值触发告警与摘要报表。
在线演示
0有效日志行
0触发告警窗口
0ERROR / 5xx 条数
0%错误占比
等待操作：粘贴日志或点击“载入示例日志”，再点“分析日志”。
项目简介
服务日志里藏着最真实的运行状态。本页实现一个日志监控告警器：持续 tail 日志文件，用正则提取错误码与耗时指标，窗口内错误数或平均耗时超过阈值时输出告警并生成摘要，可对接企业微信 / 钉钉 webhook。
代码示例：tail 与指标提取
py
import re
import time
from collections import deque
from pathlib import Path
ERROR_RE = re.compile(r"(ERROR|Exception|Traceback)")
LATENCY_RE = re.compile(r"latency=([\d.]+)ms")
WINDOW_SEC = 60
ERROR_THRESHOLD = 5
LATENCY_THRESHOLD = 800.0
def follow(file: Path):
with file.open(encoding="utf-8", errors="ignore") as f:
f.seek(0, 2) # 从文件末尾开始
while True:
line = f.readline()
if line:
yield line
else:
time.sleep(0.2)
def run(log_path: Path):
errors: deque[float] = deque()
latencies: deque[float] = deque()
for line in follow(log_path):
t = time.time()
if ERROR_RE.search(line):
errors.append(t)
m = LATENCY_RE.search(line)
if m:
latencies.append((t, float(m.group(1))))
# 清理窗口外数据
while errors and errors[0] < t - WINDOW_SEC: errors.popleft()
while latencies and latencies[0][0] < t - WINDOW_SEC: latencies.popleft()
avg = sum(v for _, v in latencies) / len(latencies) if latencies else 0
if len(errors) > ERROR_THRESHOLD:
alert(f"{len(errors)} errors in {WINDOW_SEC}s")
errors.clear()
if latencies and avg > LATENCY_THRESHOLD:
alert(f"avg latency {avg:.0f}ms > {LATENCY_THRESHOLD}ms")
latencies.clear()
def alert(msg: str) -> None:
print(f"[ALERT] {time.strftime('%H:%M:%S')} {msg}") # 可替换为 webhook 推送操作步骤
- 按你的日志格式调整
ERROR_RE与LATENCY_RE正则，阈值常量按需修改。 - 运行
python log_alert.py /var/log/app/app.log即可实时监控。 - 把
alert()替换为 requests 调用企业微信 / 钉钉机器人 webhook，实现移动端告警。
完整可运行工程
可直接运行零第三方依赖Python 3.8+
完整工程（非片段）：解析、聚合、告警三段逻辑集中在 log_alert.py，页面上的在线演示与它使用同一套判定规则（级别词优先、再退回状态码），因此演示里看到的结论就是命令行会得到的结论。
log-alert/
├── log_alert.py # 解析 / 聚合 / 告警 + CLI
├── test_log_alert.py # 最小用例回归
├── requirements.txt
└── README.md
├── log_alert.py # 解析 / 聚合 / 告警 + CLI
├── test_log_alert.py # 最小用例回归
├── requirements.txt
└── README.md
log_alert.py
核心实现：parse / level_of / bucketize / run_report，含 --selftest。
python
#!/usr/bin/env python3
"""日志指标提取与阈值告警（零第三方依赖，仅标准库）
用法：
python log_alert.py access.log --threshold 3 --window 60
python log_alert.py app.log --json
tail -n 500 access.log | python log_alert.py -
python log_alert.py --selftest
"""
import argparse
import json
import re
import sys
from datetime import datetime
from pathlib import Path
ACCESS = re.compile(r'^(\S+) \S+ \S+ \[([^\]]+)\] "(\w+) ([^"]*?) [^"]*" (\d{3}) (\d+|-)')
APP_TIME = re.compile(r"^(\d{2}/\w{3}/\d{4})\s+([\d:]+)")
ERR_PAT = re.compile(r"\bERROR\b|\bFATAL\b|Exception|Traceback")
WARN_PAT = re.compile(r"\bWARN(?:ING)?\b")
def level_of(status, line):
"""级别判定顺序：显式级别词 > HTTP 状态码。避免把 "GET /api/errors 200" 误判为错误。"""
if ERR_PAT.search(line):
return "ERROR"
if WARN_PAT.search(line):
return "WARN"
if status is not None:
return "ERROR" if status >= 500 else ("WARN" if status >= 400 else "INFO")
return "INFO"
TIME_RE = re.compile(r"(?test_log_alert.py
unittest 回归：级别判定、误判防护、时间槽聚合、阈值敏感度。
python
"""log_alert.py 最小用例回归（标准库 unittest）"""
import unittest
from log_alert import bucketize, fmt_slot, level_of, parse, run_report, SAMPLE
class TestLogAlert(unittest.TestCase):
def test_level_from_status(self):
self.assertEqual(level_of(500, 'GET /x HTTP/1.1" 500'), "ERROR")
self.assertEqual(level_of(404, 'GET /x HTTP/1.1" 404'), "WARN")
self.assertEqual(level_of(200, 'GET /x HTTP/1.1" 200'), "INFO")
def test_path_keyword_not_misjudged(self):
self.assertEqual(level_of(200, 'GET /api/errors HTTP/1.1" 200'), "INFO")
def test_explicit_level_word(self):
self.assertEqual(level_of(None, "18/Sep/2026 13:57:31 [ERROR] db pool exhausted"), "ERROR")
self.assertEqual(level_of(None, "18/Sep/2026 13:57:33 [WARN] slow query"), "WARN")
def test_parse_counts(self):
p = parse(SAMPLE)
self.assertEqual(len(p["rows"]), 7)
self.assertEqual(p["skipped"], 0)
def test_bucket_and_fmt(self):
b = bucketize(parse(SAMPLE)["rows"])
self.assertEqual(len(b), 3)
self.assertEqual(fmt_slot(13 * 3600 + 55 * 60), "13:55")
def test_alert_threshold_sensitivity(self):
self.assertEqual(len(run_report(SAMPLE, 99, 60)["alerts"]), 0)
self.assertGreaterEqual(len(run_report(SAMPLE, 1, 60)["alerts"]), 1)
if __name__ == "__main__":
unittest.main(verbosity=2)
requirements.txt
依赖清单：仅标准库。
text
# 本工程仅使用 Python 标准库（argparse / re / json / pathlib / datetime）。
# 不需要安装任何第三方包；此文件保留是为了统一 `pip install -r requirements.txt` 流程。
README.md
运行方式、支持的日志格式、退出码约定。
markdown
# 日志指标提取与阈值告警（log-alert）
读日志 → 判级别 → 按时间窗口聚合 → 超阈值告警，全程零第三方依赖。
既适合 `tail -f | python log_alert.py -` 做轻量巡检，也能作为接入 Prometheus 之前的过渡方案。
## 文件说明
| 文件 | 作用 |
|---|---|
| `log_alert.py` | 解析 / 聚合 / 告警核心 + CLI 入口 |
| `test_log_alert.py` | unittest 最小用例回归 |
| `requirements.txt` | 依赖清单（本工程无第三方依赖） |
## 运行
```bash
python -m pip install -r requirements.txt # 无第三方依赖，可跳过
python log_alert.py access.log --threshold 3 --window 60
python log_alert.py access.log --json
tail -n 500 /var/log/nginx/access.log | python log_alert.py -
python log_alert.py --selftest
python test_log_alert.py
```
## 支持的日志格式
1. **访问日志**（Nginx / Apache combined）：
`IP - - [18/Sep/2026:13:55:02 +0800] "POST /api/pay HTTP/1.1" 500 218`
2. **应用日志**（行首为时间戳的文本日志）：
`18/Sep/2026 13:57:31 [ERROR] db pool exhausted`
级别判定优先看显式级别词（ERROR / FATAL / Exception / Traceback、WARN），其次看 HTTP 状态码
（500+ → ERROR，400–499 → WARN，其余 INFO）。因此 `/api/errors` 这类**正常返回 200 的路径不会被误判**。
## 退出码约定
| 退出码 | 含义 |
|---|---|
| 0 | 解析成功且未触发告警 |
| 2 | 触发告警（可被 cron / CI 直接捕获） |
| 1 | 参数或输入错误 |
阈值与窗口通过 `--threshold` / `--window` 传入，默认「60 秒内 ERROR ≥ 3 条」。
运行标注
- 可直接运行：保存 4 个文件到同一目录，执行
python log_alert.py 你的日志.log --threshold 3。 - 需按需修改的配置项：阈值与窗口用
--threshold/--window传参（默认 3 条 / 60 秒）；若日志时间格式与示例不同，改log_alert.py顶部的ACCESS/APP_TIME两个正则即可。不改也能直接跑。 - 配套用法：
tail -n 500 access.log | python log_alert.py -可做滚动巡检；触发告警时退出码为 2。 - 自测：
python log_alert.py --selftest（或python test_log_alert.py），6 个用例应全部通过。
来源参考
实现与部署所依据的官方文档 / 仓库（均可直接访问）：