typestar

py_log_scanner.py in Python

A log scanner with argparse, logging, regex and a summary table.

import argparse
import logging
import re
import sys
from collections import Counter, defaultdict

LINE = re.compile(
    r"(?P<time>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<level>DEBUG|INFO|WARNING|ERROR)\s+"
    r"(?P<logger>[\w.]+):\s+(?P<message>.*)"
)

SAMPLE = """09:14:02 INFO typestar.serve: started on 8080
09:14:07 DEBUG typestar.db: loaded 794 steps
09:15:31 WARNING typestar.ingest: snippet too long: big.py
09:16:02 ERROR typestar.serve: unhandled 500 on /api/results
09:16:03 ERROR typestar.serve: unhandled 500 on /api/results
09:17:44 INFO typestar.serve: 200 on /healthz
not a log line at all
"""

log = logging.getLogger("scanner")


def parse(lines):
    """Yield match dicts, counting the lines that do not fit."""
    skipped = 0
    for line in lines:
        line = line.strip()
        if not line:
            continue
        found = LINE.fullmatch(line)
        if found is None:
            skipped += 1
            log.debug("unparsed: %s", line)
            continue
        yield found.groupdict()
    if skipped:
        log.warning("%d line(s) did not parse", skipped)


def summarize(records):
    levels = Counter()
    per_logger = defaultdict(Counter)
    repeated = Counter()

    for record in records:
        levels[record["level"]] += 1
        per_logger[record["logger"]][record["level"]] += 1
        if record["level"] == "ERROR":
            repeated[record["message"]] += 1
    return levels, per_logger, repeated


def main(argv=None):
    parser = argparse.ArgumentParser(description="summarize a log")
    parser.add_argument("--level", default="INFO",
                        choices=["DEBUG", "INFO", "WARNING"])
    parser.add_argument("--top", type=int, default=3)
    args = parser.parse_args(argv)

    logging.basicConfig(level=getattr(logging, args.level),
                        format="%(levelname)s %(name)s: %(message)s")

    levels, per_logger, repeated = summarize(parse(SAMPLE.splitlines()))

    print(f"{'level':<9}{'count':>6}")
    for level in ("ERROR", "WARNING", "INFO", "DEBUG"):
        if levels[level]:
            print(f"{level:<9}{levels[level]:>6}")

    print("\nby logger")
    for name in sorted(per_logger):
        counts = per_logger[name]
        detail = " ".join(f"{k.lower()}={v}" for k, v in sorted(counts.items()))
        print(f"  {name:<18} {detail}")

    if repeated:
        print("\nrepeated errors")
        for message, count in repeated.most_common(args.top):
            print(f"  {count}x {message}")
    return 0 if not levels["ERROR"] else 1


if __name__ == "__main__":
    sys.exit(main())

How it works

  1. argparse defines the interface, including a level flag.
  2. The regex names its groups, so the code reads them by name.
  3. Counters build the summary in one pass over the lines.

Keywords and builtins used here

The run, in numbers

Lines
86
Characters to type
2239
Tokens
554
Three-star pace
110 tpm

At the three-star pace of 110 tokens a minute, this run takes about 302 seconds.

Type this snippet

Step 3 of 3 in Encore, step 41 of 41 in Domain tools.

← Previous