日志分析脚本

摘要：日志分析的目标不是「看日志」，而是把几万行压缩成几条可行动的信息：哪类错误最多、什么时候开始异常、影响多少用户。本文用 Shell 与 Python 两种方式实现错误聚合、时间窗口统计和 Top N 提取。

适用环境

bash

which awk sort uniq grep
python3 --version
# 假设日志格式：2026-10-09 14:03:11 [ERROR] module message
ls -la /var/log/app/*.log 2>/dev/null || echo "将用样例数据演示"

操作步骤

一、先确定分析维度

text

1. 量级分布  ：各级别（ERROR/WARN/INFO）各多少条
2. 时间趋势  ：按分钟/小时统计，找出异常开始的时间点
3. 错误归类  ：把同类错误归一化后聚合，找出 Top N
4. 影响面    ：涉及多少用户/IP/接口

二、Shell 一行式：快速看分布

bash

LOG=/var/log/app/app.log

# 各级别数量
grep -oE '\b(ERROR|WARN|INFO|DEBUG)\b' "$LOG" | sort | uniq -c | sort -rn

# 按小时统计 ERROR 数量
grep 'ERROR' "$LOG" | awk '{print substr($2,1,2)}' | sort | uniq -c

# 按分钟统计（前 20 名，快速定位尖峰时刻）
grep 'ERROR' "$LOG" | awk '{print substr($2,1,5)}' | sort | uniq -c | sort -rn | head -20

# 提取错误信息的 Top N（去掉时间、IP 等变量后归类）
grep 'ERROR' "$LOG" | sed -E 's/^[0-9-]+ [0-9:]+ //; s/[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+/<IP>/g; s/[0-9]{3,}/<N>/g' \
  | sort | uniq -c | sort -rn | head -10

注意

直接用 uniq -c 统计原始错误行几乎没有意义——因为每行都带时间戳和请求 ID，每条都"只出现一次"。必须先做归一化（把时间、IP、数字替换成占位符）再聚合。

三、只看某个时间窗口

bash

# 提取 14:00-15:00 之间的日志（时间字符串比较，适用于 YYYY-MM-DD HH:MM:SS 格式）
awk '$1=="2026-10-09" && $2>="14:00:00" && $2<="15:00:00"' "$LOG" > /tmp/win.log
wc -l /tmp/win.log

# 最近 N 分钟（用 date 计算起点，兼容跨小时）
SINCE=$(date -d '30 minutes ago' '+%F %T')
awk -v s="$SINCE" '($1" "$2) >= s' "$LOG" | grep ERROR | tail -50

四、Python 版：结构化、可扩展

python

#!/usr/bin/env python3
"""日志分析：按级别统计、按分钟聚合、错误归一化 Top N。"""
import argparse
import re
from collections import Counter, defaultdict
from datetime import datetime

LINE_RE = re.compile(
    r"^(?P<date>\d{4}-\d{2}-\d{2})\s+(?P<time>\d{2}:\d{2}:\d{2})\s+"
    r"\[(?P<level>[A-Z]+)\]\s+(?P<msg>.*)$"
)

# 归一化规则：把易变部分替换为占位符，让同类错误能聚合在一起
NORMALIZE = [
    (re.compile(r"\d+\.\d+\.\d+\.\d+"), "<IP>"),          # IP
    (re.compile(r"\b[0-9a-f]{8,}\b"), "<HEX>"),           # 请求 ID / trace id
    (re.compile(r"\b\d+\b"), "<N>"),                      # 数字
    (re.compile(r"\s+"), " "),
]


def normalize(msg: str) -> str:
    for pat, rep in NORMALIZE:
        msg = pat.sub(rep, msg)
    return msg.strip()[:120]


def parse(path, level_filter=None):
    with open(path, encoding="utf-8", errors="replace") as f:
        for line in f:
            m = LINE_RE.match(line.rstrip("\n"))
            if not m:
                continue
            if level_filter and m.group("level") != level_filter:
                continue
            yield m.groupdict()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("logfile")
    ap.add_argument("--level", default="ERROR", help="只分析该级别")
    ap.add_argument("--top", type=int, default=10, help="Top N 错误")
    args = ap.parse_args()

    level_cnt = Counter()
    minute_cnt = defaultdict(int)
    msg_cnt = Counter()
    first_seen, last_seen = {}, {}

    for rec in parse(args.logfile, args.level):
        level_cnt[rec["level"]] += 1
        minute = rec["time"][:5]
        minute_cnt[minute] += 1
        key = normalize(rec["msg"])
        msg_cnt[key] += 1
        ts = f"{rec['date']} {rec['time']}"
        first_seen.setdefault(key, ts)
        last_seen[key] = ts

    print(f"=== 级别分布（{args.level}）===")
    for lv, n in level_cnt.most_common():
        print(f"{lv:<8} {n}")

    print(f"\n=== 按分钟 Top 10 ===")
    for m, n in sorted(minute_cnt.items(), key=lambda x: -x[1])[:10]:
        print(f"{m}  {n}")

    print(f"\n=== 归一化错误 Top {args.top} ===")
    for msg, n in msg_cnt.most_common(args.top):
        print(f"{n:>6}  首次:{first_seen[msg]}  末次:{last_seen[msg]}")
        print(f"        {msg}")


if __name__ == "__main__":
    main()

五、运行与解读

bash

python3 logstat.py /var/log/app/app.log --level ERROR --top 10

解读要点：

按分钟 Top：异常是从某一分钟突然开始的，还是缓慢上升？前者多为变更/发布导致，后者多为容量或泄漏问题。
归一化 Top：占比最高的那一类通常就是根因，先修它。
首次/末次时间：末次时间仍在增长说明问题还在持续。

六、接入日常巡检

bash

# 每天 8 点出昨天的错误摘要
0 8 * * * /usr/bin/python3 /usr/local/bin/logstat.py /var/log/app/app.log \
  --top 5 | tee /var/log/ops/daily-log-$(date +\%F).txt

验证

[ ] 级别分布数字与 grep -c 抽样核对一致
[ ] 归一化后同类错误能合并（人工构造 3 条只有 IP 不同的错误验证）
[ ] 时间窗口提取的行数与 wc -l 预期一致
[ ] 脚本对不匹配的日志行能跳过而不报错

bash

printf '2026-10-09 10:00:01 [ERROR] conn failed 10.0.0.1\n2026-10-09 10:00:02 [ERROR] conn failed 10.0.0.2\n' > /tmp/t.log
python3 logstat.py /tmp/t.log --top 5 | tail -5

常见坑

不归一化直接聚合：每行都不同，Top N 毫无意义。
日志格式不统一：新旧版本日志格式混在一个文件里，正则匹配率下降，应先按格式分流。
grep 大文件无节制：几十 GB 的日志全量扫描会拖慢机器，应按时间先缩小范围或用 tail -n。
忽略时区：日志时间是 UTC，脚本按本地时间过滤会取错窗口。
把 WARN 一起当 ERROR：会淹没真正的问题，分析时应分级别单独统计。

日志分析脚本 ​

适用环境 ​

操作步骤 ​

一、先确定分析维度 ​

二、Shell 一行式：快速看分布 ​

三、只看某个时间窗口 ​

四、Python 版：结构化、可扩展 ​

五、运行与解读 ​

六、接入日常巡检 ​

验证 ​

常见坑 ​

参考资料 ​

日志分析脚本

适用环境

操作步骤

一、先确定分析维度

二、Shell 一行式：快速看分布

三、只看某个时间窗口

四、Python 版：结构化、可扩展

五、运行与解读

六、接入日常巡检

验证

常见坑

参考资料