用 Python 写了个系统日志聚合分析脚本，ELK 太重，这个够轻

一个人的博客，服务器流量不大，上 ELK 纯属大炮打蚊子。 但日志分散在各处，出问题的时候翻日志翻到眼瞎。最后我用 Python 写了个轻量级的日志聚合脚本，200 行代码搞定，跑了一年，分享一下。

为什么不上 ELK？

不是不知道 ELK 好，是成本摆在那里：

• 内存 4G 起步，我轻量服务器总共才 2G
• 日志量一天也就几十 MB， Elasticsearch 索引开销比日志本身还大
• 维护成本：版本升级、索引生命周期管理、权限配置，一个人搞不定

核心需求其实很简单： 把 Nginx 错误日志、应用日志、系统日志拢到一块，按错误类型聚合，每天发一份摘要到微信/邮件。出问题的时候知道该先看哪。

脚本设计思路

整体架构分三层：

1. 采集层：按配置读取多个日志文件，支持 tail -f 实时模式和批量读取历史模式

2. 解析层：用正则提取时间、级别、IP、URL、错误信息，按指纹聚合去重

3. 输出层：生成 Markdown 摘要，推送到企业微信机器人/邮件

关键设计：不存储原始日志，只存聚合后的统计。一条异常出现 100 次，输出里只出现一次，附带次数和首次/末次时间。

核心代码

1. 配置定义（dataclasses，比字典清晰）

from dataclasses import dataclass, field
from typing import List, Optional
import re
from datetime import datetime, timedelta

@dataclass
class LogSource:
    name: str           # 如 "nginx_error"
    path: str           # /var/log/nginx/error.log
    pattern: str        # 正则表达式
    time_format: str    # 时间解析格式
    tail: bool = True   # 是否实时追踪

@dataclass
class LogEntry:
    timestamp: datetime
    level: str          # ERROR / WARN / INFO
    ip: Optional[str]
    url: Optional[str]
    message: str
    fingerprint: str    # 去重指纹，取 message 主干

@dataclass
class AggregateResult:
    fingerprint: str
    count: int
    level: str
    first_at: datetime
    last_at: datetime
    samples: List[str] = field(default_factory=list)

用 dataclass 的好处是：字段类型一目了然，IDE 自动补全，配置校验时不用写一堆 if key in dict。

2. 日志解析（Nginx 错误日志为例）

NGINX_ERROR_RE = re.compile(
    r'^(?P\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) '
    r'\[(?P\w+)\] '
    r'(?P\d+#\d+): '
    r'\*\d+ (?P.+?), '
    r'client: (?P[\d.]+), '
    r'server: [^,]+, '
    r'request: "(?P\w+) (?P\S+).*", '
    r'host: "(?P[^"]+)"',
    re.MULTILINE
)

def parse_nginx_error(line: str) -> Optional[LogEntry]:
    m = NGINX_ERROR_RE.match(line)
    if not m:
        return None
    # 生成指纹：去掉 IP 和时间戳，保留错误类型和 URL 路径
    msg = m.group('message')
    fp = re.sub(r'\d+\.\d+\.\d+\.\d+', '', msg)
    fp = re.sub(r'/\d+', '/', fp)  # 把数字 ID 泛化
    return LogEntry(
        timestamp=datetime.strptime(m.group('time'), '%Y/%m/%d %H:%M:%S'),
        level=m.group('level'),
        ip=m.group('ip'),
        url=m.group('url'),
        message=msg,
        fingerprint=fp
    )

指纹生成是关键。 如果直接用原始 message 做 key，”404 from 192.168.1.5″ 和 “404 from 192.168.1.6” 会被当成两条。把 IP 和 ID 泛化后，同类错误才能聚合。

3. 聚合统计（用 defaultdict，比 dict.setdefault 舒服）

from collections import defaultdict

def aggregate(entries: List[LogEntry]) -> List[AggregateResult]:
    groups = defaultdict(lambda: {
        'count': 0, 'first': None, 'last': None,
        'level': '', 'samples': []
    })
    for e in entries:
        g = groups[e.fingerprint]
        g['count'] += 1
        g['level'] = e.level
        if g['first'] is None or e.timestamp < g['first']:
            g['first'] = e.timestamp
        if g['last'] is None or e.timestamp > g['last']:
            g['last'] = e.timestamp
        if len(g['samples']) < 3:
            g['samples'].append(f"{e.ip} → {e.url}")

    return sorted([
        AggregateResult(
            fingerprint=fp,
            count=v['count'],
            level=v['level'],
            first_at=v['first'],
            last_at=v['last'],
            samples=v['samples']
        )
        for fp, v in groups.items()
    ], key=lambda x: x.count, reverse=True)

取 Top 3 samples 是为了摘要里能看到具体案例，而不是只看泛化的指纹。

4. 实时采集（用 `subprocess` 调 `tail -f`）

import subprocess
from threading import Thread
from queue import Queue, Empty

def tail_f(path: str, queue: Queue, stop_event):
    proc = subprocess.Popen(
        ['tail', '-f', '-n', '0', path],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        text=True
    )
    while not stop_event.is_set():
        try:
            line = proc.stdout.readline()
            if line:
                queue.put(line)
        except Exception:
            break
    proc.terminate()

# 主循环：每 5 分钟聚合一次推送
queue = Queue()
stop_event = threading.Event()
Thread(target=tail_f, args=(LOG_PATH, queue, stop_event), daemon=True).start()

buffer = []
while True:
    try:
        line = queue.get(timeout=300)  # 5分钟
        entry = parse_nginx_error(line)
        if entry:
            buffer.append(entry)
    except Empty:
        if buffer:
            results = aggregate(buffer)
            send_report(results)  # 推送到企业微信
            buffer.clear()

踩坑经验（5条）

1. 日志轮转切割时 tail -f 会断

- logrotate 的 copytruncate 模式不会移动文件描述符，但如果用了 create 模式，tail -f 会跟丢新文件。解决：改用 tail -F（大写 F），它会自动跟踪同名新文件。

2. 编码地狱：GBK、UTF-8、BOM 混用

- 有些旧应用日志是 GBK 编码，Python 默认 UTF-8 解码会崩。解决：用 chardet 预检编码，或者直接用 errors='replace' 兜底，至少保证不中断采集。

3. 正则性能：大日志文件千万别用 re.findall

- 逐行 match 和全文 findall 性能差 10 倍以上。如果日志量突然暴增（比如被攻击），findall 会让 CPU 打满。逐行处理是流式采集的唯一正确姿势。

4. 指纹泛化过度会漏掉真正不同的问题

- 刚开始我把所有数字都替换成 ，结果 "404" 和 "500" 也变成了 ，导致两类错误被当成一种。解决：指纹泛化要克制，只替换变化部分（IP、ID、时间戳），保留状态码、错误类型等关键信息。

5. 推送频率太高会被企业微信机器人限流

- 企业微信群机器人限制每分钟 20 条。如果日志异常暴涨，会触发限流。解决：加合并窗口（5 分钟聚合一次），异常量超过阈值再推送，普通 INFO 级别只做本地落盘不推送。

效果

跑了一年，这个脚本每天推送一条摘要消息，格式如下：

[日志日报] 2026-08-01
━━━━━━━━━━━━━━
ERROR × 12 | PHP Fatal error: Allowed memory...
  首次: 03:15 | 末次: 22:47
  样例: 10.0.0.5 → /wp-json/wp/v2/media

WARN × 45 | upstream timed out (110...)
  首次: 09:22 | 末次: 18:10
  样例: 10.0.0.12 → /api/slow-endpoint

INFO × 3 | SSL_do_handshake() failed
  ...

比 ELK 轻了不止一个量级，但核心信息一条没落。 一个人的服务器，够用就好。

下一步优化

• 接入 SQLite 做历史趋势查询（当前只存当天，不存历史）
• 异常自动降噪：连续出现的同类错误，只报第一次和恢复通知
• 支持更多日志格式：JSON 格式的应用日志（直接用 `json.loads` 解析，比正则快）

代码放 GitHub 了，有需要自取。