12. Terminal project: a web access-log analyzer

Goal: read nginx/Apache "combined" access logs (the format you met in OS Ch 34), and report requests per status code, the slowest paths, the top client IPs, and the error rate per minute. This is the first tool in the guide you might actually run on a server.

A log line looks like this:

203.0.113.9 - - [04/Oct/2026:10:15:32 +0000] "GET /api/orders?id=7 HTTP/1.1" 200 512 "-" "curl/8.7" 0.042

# labs/logstat.py
"""Summarize a combined-format access log (with request time as the last field)."""
import argparse
import re
import sys
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime
from collections.abc import Iterable, Iterator

LINE = re.compile(
    r'(?P<ip>\S+) \S+ \S+ \[(?P<ts>[^\]]+)\] '
    r'"(?P<method>[A-Z]+) (?P<path>\S+) [^"]*" '
    r'(?P<status>\d{3}) (?P<size>\d+|-) "[^"]*" "[^"]*"(?: (?P<rt>[\d.]+))?'
)

@dataclass(frozen=True, slots=True)
class Hit:
    ip: str
    ts: datetime
    method: str
    path: str
    status: int
    seconds: float | None

def parse(lines: Iterable[str]) -> Iterator[Hit]:
    bad = 0
    for line in lines:
        m = LINE.match(line)
        if not m:
            bad += 1
            continue
        yield Hit(
            ip=m["ip"],
            ts=datetime.strptime(m["ts"], "%d/%b/%Y:%H:%M:%S %z"),
            method=m["method"],
            path=m["path"].split("?", 1)[0],     # group /api/orders?id=7 with /api/orders
            status=int(m["status"]),
            seconds=float(m["rt"]) if m["rt"] else None,
        )
    if bad:
        print(f"logstat: skipped {bad} unparseable lines", file=sys.stderr)

def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("log", type=argparse.FileType("r", encoding="utf-8", errors="replace"))
    p.add_argument("--top", type=int, default=5)
    args = p.parse_args(argv)

    status: Counter[int] = Counter()
    ips: Counter[str] = Counter()
    times: defaultdict[str, list[float]] = defaultdict(list)
    errors_per_min: Counter[str] = Counter()
    total = 0

    for hit in parse(args.log):
        total += 1
        status[hit.status] += 1
        ips[hit.ip] += 1
        if hit.seconds is not None:
            times[hit.path].append(hit.seconds)
        if hit.status >= 500:
            errors_per_min[hit.ts.strftime("%H:%M")] += 1

    if total == 0:
        print("no requests parsed", file=sys.stderr)
        return 1

    print(f"{total} requests")
    print("\nstatus codes:")
    for code, n in sorted(status.items()):
        print(f"  {code}  {n:>8}  {n / total:6.1%}")

    print(f"\ntop {args.top} client IPs:")
    for ip, n in ips.most_common(args.top):
        print(f"  {ip:<15} {n:>8}")

    print(f"\nslowest paths by p95 (min 5 requests):")
    def p95(xs: list[float]) -> float:
        xs = sorted(xs)
        return xs[min(len(xs) - 1, int(0.95 * len(xs)))]
    ranked = sorted(((p95(v), k, len(v)) for k, v in times.items() if len(v) >= 5), reverse=True)
    for t, path, n in ranked[: args.top]:
        print(f"  {t * 1000:8.1f} ms  {path}  ({n} requests)")

    if errors_per_min:
        print("\nminutes with 5xx errors:")
        for minute, n in sorted(errors_per_min.items()):
            print(f"  {minute}  {'#' * min(n, 60)} {n}")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

Generate a test log so you can run it anywhere:

# labs/fakelog.py: write a synthetic access log to stdout
import random
from datetime import datetime, timedelta, timezone

random.seed(7)
paths = ["/", "/api/orders", "/api/users", "/login", "/static/app.js"]
start = datetime(2026, 10, 4, 10, 0, tzinfo=timezone.utc)
for i in range(5000):
    ts = start + timedelta(seconds=i * 0.7)
    path = random.choice(paths)
    status = random.choices([200, 304, 404, 500, 503], weights=[80, 8, 6, 4, 2])[0]
    rt = random.lognormvariate(-3, 0.8) * (8 if path == "/api/orders" else 1)
    ip = f"203.0.113.{random.randint(1, 40)}"
    print(f'{ip} - - [{ts:%d/%b/%Y:%H:%M:%S %z}] "GET {path} HTTP/1.1" {status} 512 "-" "curl/8.7" {rt:.3f}')

$ uv run labs/fakelog.py > access.log
$ uv run labs/logstat.py access.log --top 3
5000 requests

status codes:
  200      4007   80.1%
  ...
slowest paths by p95 (min 5 requests):
     ...  ms  /api/orders  (...)

What to notice:

• A compiled regex with named groups is the standard way to parse semi-structured text. Test it against real lines first; log formats drift.
• Percentiles, not averages. The networking guide Ch 45 explains why averages lie about latency.
• @dataclass(frozen=True, slots=True) gives a small, immutable record type. Chapter 15 explains both flags.
• The parser is a generator, so the analyzer streams. On a 20 GB log on a server with 2 GB of RAM, this design is the difference between a report and an OOM kill.

Extend it: accept .gz files transparently (gzip.open); add --since 10:30; flag IPs with more than 20 401s in a minute (the seed of the detector in Ch 52); load the same file with pandas in Ch 54 and compare.