Python 运维自动化系列（02）：日志文件几个 G，先别慌——写一个流式日志分析工具

# 反面教材：内存爆炸版
with open("access.log", encoding="utf-8") as f:
    lines = f.readlines()   # 5GB 的文件 = 至少 5GB 内存，机器当场卡死

# 正确：内存占用恒定，文件多大都不怕
with open("access.log", encoding="utf-8") as f:
    for line in f:          # 每次只有一行在内存里
        process(line)

10.0.0.77 - - [28/Sep/2026:10:08:11 +0800] "GET /api/pay HTTP/1.1" 500 3210

import re

LINE_RE = re.compile(
    r'^(?P<ip>\S+) - - \[(?P<ts>[^\]]+)\] "(?P<method>\S+) (?P<url>\S+) \S+" '
    r"(?P<status>\d{3}) (?P<size>\d+)"
)

m = LINE_RE.match(line)
if m:
    print(m["ip"], m["url"], m["status"])

from collections import Counter

status_counter = Counter()
status_counter["500"] += 1
status_counter["500"] += 1
status_counter.most_common(3)   # 自动按次数排序取 Top N

import random
import re
from collections import Counter
from datetime import datetime, timedelta
from pathlib import Path

LOG_PATH = Path(__file__).parent / "fixtures" / "access.log"

IPS = ["10.0.0.12", "10.0.0.33", "172.16.4.9", "192.168.1.105", "10.0.0.77"]
URLS = ["/api/orders", "/api/users", "/static/app.js", "/health", "/api/pay"]
STATUS = [200, 200, 200, 200, 301, 404, 500]
LINE_RE = re.compile(
    r'^(?P<ip>\S+) - - \[(?P<ts>[^\]]+)\] "(?P<method>\S+) (?P<url>\S+) \S+" '
    r"(?P<status>\d{3}) (?P<size>\d+)"
)


def generate(n: int = 300) -> None:
    """真实运维里这一步不存在——日志文件本身就在那里。
    这里用固定随机种子生成一份可复现的仿真日志，方便演示和测试。"""
    LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
    rng = random.Random(2026)
    base = datetime(2026, 9, 28, 10, 0, 0)
    with LOG_PATH.open("w", encoding="utf-8") as f:
        for i in range(n):
            ts = base + timedelta(seconds=rng.randint(0, 3599))
            ip = rng.choice(IPS)
            url = rng.choice(URLS)
            status = rng.choice(STATUS)
            if rng.random() < 0.02:          # 混入 2% 坏行，模拟脏数据
                f.write("这是一个被截断的坏行...\n")
                continue
            f.write(
                f'{ip} - - [{ts.strftime("%d/%b/%Y:%H:%M:%S +0800")}] '
                f'"GET {url} HTTP/1.1" {status} {rng.randint(80, 9000)}\n'
            )


def analyze(path: Path) -> None:
    status_counter: Counter[int] = Counter()
    ip_counter: Counter[str] = Counter()
    url_counter: Counter[str] = Counter()
    error_per_minute: Counter[str] = Counter()
    bad_lines = 0
    total = 0

    for line in path.open(encoding="utf-8"):     # 流式：一次一行
        m = LINE_RE.match(line)
        if not m:                                 # 坏行不崩，计数继续
            bad_lines += 1
            continue
        total += 1
        status_counter[int(m["status"])] += 1
        ip_counter[m["ip"]] += 1
        url_counter[m["url"]] += 1
        if int(m["status"]) >= 500:
            minute = m["ts"].split(":")[1] + ":" + m["ts"].split(":")[2]
            error_per_minute[minute] += 1

    print(f"有效行 {total}，坏行 {bad_lines}")
    print("\n状态码分布 Top 5:")
    for code, n in status_counter.most_common(5):
        print(f"  {code}: {n} 次")
    print("\n请求量 Top 3 IP:")
    for ip, n in ip_counter.most_common(3):
        print(f"  {ip}: {n} 次")
    print("\n请求量 Top 3 URL:")
    for url, n in url_counter.most_common(3):
        print(f"  {url}: {n} 次")
    err_total = sum(error_per_minute.values())
    print(f"\n5xx 错误共 {err_total} 次（错误率 {err_total / total:.1%}）")
    print("5xx 最多的分钟 Top 3:")
    for minute, n in error_per_minute.most_common(3):
        print(f"  {minute}: {n} 次")


if __name__ == "__main__":
    generate()
    print(f"已生成仿真日志: {LOG_PATH}（{LOG_PATH.stat().st_size} 字节）\n")
    analyze(LOG_PATH)

已生成仿真日志: demos\fixtures\access.log（23738 字节）

有效行 295，坏行 5

状态码分布 Top 5:
  200: 161 次
  301: 55 次
  404: 40 次
  500: 39 次

请求量 Top 3 IP:
  10.0.0.77: 68 次
  192.168.1.105: 65 次
  172.16.4.9: 58 次

请求量 Top 3 URL:
  /health: 65 次
  /api/pay: 59 次
  /api/orders: 58 次

5xx 错误共 39 次（错误率 13.2%）
5xx 最多的分钟 Top 3:
  10:08: 3 次
  10:27: 2 次
  10:17: 2 次

Traceback (most recent call last):
  File "demos\log_analysis_demo.py", line 62, in analyze
    if m["status"] >= 500:
TypeError: '>=' not supported between instances of 'str' and 'int'

status_counter[int(m["status"])] += 1
if int(m["status"]) >= 500: