Python

Defaults:

• Use the standard library unless noted. Every snippet here has been run.
• Return an exit code from main() and send errors to stderr.
• Set a timeout on every network call.
• Retry only idempotent requests, and only on network errors, 5xx and 429.
• Measure durations with time.monotonic(), never time.time() - the wall clock can jump.
• Every snippet has a Go equivalent under the same heading.

Script skeleton

Start every script from this; cron and CI act on the exit code.

import argparse
import sys


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="What this does.")
    p.add_argument("path")
    p.add_argument("--top", type=int, default=10)
    p.add_argument("--verbose", action="store_true")
    return p.parse_args()


def main() -> int:
    args = parse_args()
    try:
        with open(args.path) as f:
            for line in f:  # streams: fine for huge files
                ...
    except OSError as e:
        print(f"error: {e}", file=sys.stderr)
        return 1
    return 0


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

Parse key=value log lines

Use split("=", 1) so values containing = survive. A pair without = raises ValueError.

def parse_line(line: str) -> dict[str, str]:
    """Turn 'ts k=v k=v' into a dict. Raises ValueError on bad input."""
    ts, *pairs = line.split()
    fields = dict(p.split("=", 1) for p in pairs)
    fields["ts"] = ts
    return fields

Skip bad lines by catching only the errors you expect, so real bugs still surface:

bad = 0
for line in f:
    if not line.strip():
        continue
    try:
        fields = parse_line(line)
        status = int(fields["status"])
    except (ValueError, KeyError):
        bad += 1
        continue

Count and group

from collections import Counter, defaultdict

counts = Counter()
counts["payments"] += 1
counts.most_common(3)              # [("payments", 1)]

latencies = defaultdict(list)
latencies["payments"].append(134)  # no KeyError for a new key

A small record type

Use default_factory, never a mutable default like = [] - every instance shares it.

from dataclasses import dataclass, field


@dataclass
class Stats:
    count: int = 0
    errors: int = 0
    latencies: list[int] = field(default_factory=list)

    @property
    def error_rate(self) -> float:
        return self.errors / self.count if self.count else 0.0

Print an aligned table

Spec	Effect
<12	Left-align to 12 characters
>7	Right-align to 7 characters
.1%	0.5 becomes 50.0%

rows = sorted(stats.items(), key=lambda kv: kv[1].error_rate, reverse=True)
print(f"{'service':<12} {'count':>7} {'err%':>7}")
for name, s in rows:
    print(f"{name:<12} {s.count:>7} {s.error_rate:>7.1%}")

Latency percentiles

Use method="inclusive" - the default can report a p99 slower than your slowest request. quantiles needs at least two values, hence the single-value guard.

import statistics


def percentile(values: list[int], pct: int) -> float:
    if len(values) == 1:
        return float(values[0])
    return statistics.quantiles(values, n=100, method="inclusive")[pct - 1]


p50 = statistics.median(values)
p99 = percentile(values, 99)

HTTP request with a timeout

Without a timeout, a hung server hangs your script forever. Catch HTTPError before URLError; it is a subclass.

import urllib.error
import urllib.request


def fetch_status(url: str, timeout: float) -> tuple[int | None, str]:
    try:
        with urllib.request.urlopen(url, timeout=timeout) as resp:
            return resp.status, str(resp.status)
    except urllib.error.HTTPError as e:  # 4xx/5xx still carry a code
        return e.code, str(e.code)
    except (urllib.error.URLError, TimeoutError, OSError) as e:
        return None, f"error: {getattr(e, 'reason', e)}"

Retry with exponential backoff and jitter

Never retry other 4xx or non-idempotent requests - a retried POST can charge a customer twice. Jitter stops every client retrying at the same moment.

import random
import time


def is_retryable(status: int | None) -> bool:
    if status is None:  # timeout or connection error
        return True
    return status >= 500 or status == 429


def check(url: str, retries: int = 3, timeout: float = 5.0,
          base_delay: float = 1.0) -> bool:
    for attempt in range(retries + 1):
        status, _ = fetch_status(url, timeout)
        if status is not None and 200 <= status < 400:
            return True
        if not is_retryable(status) or attempt == retries:
            return False
        delay = base_delay * 2 ** attempt                 # 1s, 2s, 4s
        time.sleep(delay + random.uniform(0, delay / 2))  # jitter
    return False

Run checks in parallel

Use threads for network-bound work; the GIL is released while waiting on I/O. Wrap fut.result() - it re-raises the worker’s exception, and one bad URL must not stop the rest.

from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=10) as pool:
    futures = {pool.submit(check, url): url for url in urls}
    for fut in as_completed(futures):  # in finish order
        url = futures[fut]
        try:
            healthy = fut.result()
        except Exception as e:
            healthy = False
            print(f"{url} crashed: {e}")
        print("OK  " if healthy else "FAIL", url)

Find OOMKilled containers

Uses the official client (pip install kubernetes). Runs in a pod or on a laptop; needs get and list on pods.

from kubernetes import client, config

try:
    config.load_incluster_config()  # inside a pod
except config.ConfigException:
    config.load_kube_config()       # on a laptop

v1 = client.CoreV1Api()
for pod in v1.list_pod_for_all_namespaces().items:
    for cs in pod.status.container_statuses or []:  # None before start
        term = cs.last_state.terminated if cs.last_state else None
        if term and term.reason == "OOMKilled":
            print(pod.metadata.namespace, pod.metadata.name,
                  cs.name, f"restarts={cs.restart_count}")

Why this matters: the app that was starved of memory its whole life.

Read files: plain, gzip, stdin, JSON lines, CSV

Stream lines; never f.read() a log file. Route every input through one helper so all tools accept the same things: plain files, .gz, and - for stdin. errors="replace" stops one bad byte from killing a run.

import gzip
import sys
from collections.abc import Iterator
from typing import TextIO


def open_text(path: str) -> TextIO:
    """Open a plain file, a .gz file, or stdin ("-") as text."""
    if path == "-":
        return sys.stdin
    if path.endswith(".gz"):
        return gzip.open(path, "rt", encoding="utf-8", errors="replace")
    return open(path, encoding="utf-8", errors="replace")


def read_lines(paths: list[str]) -> Iterator[str]:
    """Stream lines from every path in order, one file open at a time."""
    for path in paths:
        f = open_text(path)
        try:
            yield from f
        finally:
            if f is not sys.stdin:
                f.close()

import csv
import json

# Lines from many files, .gz or stdin ("-"), one file open at a time
for line in read_lines(["sample.log", "more.jsonl.gz"]):
    pass

# JSON lines: one object per line; skip what doesn't parse
records, bad = [], 0
for line in read_lines(["more.jsonl.gz"]):
    if not line.strip():
        continue
    try:
        obj = json.loads(line)
    except json.JSONDecodeError:
        bad += 1
        continue
    if isinstance(obj, dict):  # valid JSON can still be a list or a number
        records.append(obj)
    else:
        bad += 1
print(len(records), bad)

# CSV with a header row; newline="" is required by the csv module
with open("report.csv", newline="", encoding="utf-8") as f:
    for row in csv.DictReader(f):
        print(row["service"], int(row["status"]))

• Check the JSON type. json.loads accepts any JSON value, so a line like [1,2,3] parses fine but isn’t a record.
• Open CSV with newline="". The csv module handles line endings itself; without it, quoted fields containing newlines break.

Full script: log analyzer

Per-service request count, error rate (5xx), p50 and p99 latency. Accepts key=value or JSON lines, plain or gzipped, from files or stdin.

python3 log_analyzer.py app.log app.log.1.gz
kubectl logs deploy/api | python3 log_analyzer.py --service payments
python3 log_analyzer.py --json /var/log/app/*.gz | jq '.services[0]'

service        count    err%      p50      p99
auth               2   50.0%     1050     1992
payments           2   50.0%      267      397
search             1    0.0%       55       55
skipped 2 malformed line(s)

#!/usr/bin/env python3
"""Summarise access logs per service: requests, error rate, p50 and p99 latency.

Reads key=value or JSON lines, plain or gzipped, from files or stdin.
Exit codes: 0 ok, 2 input could not be read.
"""
import argparse
import gzip
import json
import statistics
import sys
from collections import defaultdict
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field
from typing import TextIO


@dataclass
class Stats:
    count: int = 0
    errors: int = 0
    latencies: list[float] = field(default_factory=list)

    @property
    def error_rate(self) -> float:
        return self.errors / self.count if self.count else 0.0


def open_text(path: str) -> TextIO:
    """Open a plain file, a .gz file, or stdin ("-") as text."""
    if path == "-":
        return sys.stdin
    if path.endswith(".gz"):
        return gzip.open(path, "rt", encoding="utf-8", errors="replace")
    return open(path, encoding="utf-8", errors="replace")


def read_lines(paths: list[str]) -> Iterator[str]:
    """Stream lines from every path in order, one file open at a time."""
    for path in paths:
        f = open_text(path)
        try:
            yield from f
        finally:
            if f is not sys.stdin:
                f.close()


def parse_line(line: str) -> dict:
    """Parse a JSON object or 'ts k=v k=v'. Raises ValueError on bad input."""
    line = line.strip()
    if line.startswith("{"):
        obj = json.loads(line)  # JSONDecodeError is a ValueError
        if not isinstance(obj, dict):
            raise ValueError("JSON line is not an object")
        return obj
    ts, *pairs = line.split()
    fields = dict(p.split("=", 1) for p in pairs)
    fields["ts"] = ts
    return fields


def analyze(lines: Iterable[str], only: str | None = None) -> tuple[dict[str, Stats], int]:
    stats: dict[str, Stats] = defaultdict(Stats)
    bad = 0
    for line in lines:
        if not line.strip():
            continue
        try:
            f = parse_line(line)
            service = str(f["service"])
            status = int(f["status"])
            latency = float(f["latency_ms"])
        except (ValueError, KeyError, TypeError):
            bad += 1
            continue
        if only and service != only:
            continue
        s = stats[service]
        s.count += 1
        s.errors += status >= 500
        s.latencies.append(latency)
    return stats, bad


def percentile(values: list[float], pct: int) -> float:
    if len(values) == 1:
        return values[0]
    # "inclusive" keeps results within the observed values
    return statistics.quantiles(values, n=100, method="inclusive")[pct - 1]


def summarize(stats: dict[str, Stats], top: int) -> list[dict]:
    ordered = sorted(stats.items(), key=lambda kv: (-kv[1].error_rate, -kv[1].count, kv[0]))
    return [
        {
            "service": name,
            "count": s.count,
            "error_rate": round(s.error_rate, 4),
            "p50_ms": round(percentile(s.latencies, 50), 1),
            "p99_ms": round(percentile(s.latencies, 99), 1),
        }
        for name, s in ordered[:top]
    ]


def print_table(rows: list[dict]) -> None:
    print(f"{'service':<12} {'count':>7} {'err%':>7} {'p50':>8} {'p99':>8}")
    for r in rows:
        print(f"{r['service']:<12} {r['count']:>7} {r['error_rate']:>7.1%} "
              f"{r['p50_ms']:>8.0f} {r['p99_ms']:>8.0f}")


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Summarise access logs per service.")
    p.add_argument("paths", nargs="*", default=["-"], help="log files (.gz ok); - for stdin")
    p.add_argument("--service", help="only this service")
    p.add_argument("--top", type=int, default=20, help="max rows (default 20)")
    p.add_argument("--json", action="store_true", help="JSON output for scripts")
    return p.parse_args()


def main() -> int:
    args = parse_args()
    try:
        stats, bad = analyze(read_lines(args.paths), args.service)
    except (OSError, EOFError) as e:  # missing file, unreadable or truncated .gz
        print(f"error: {e}", file=sys.stderr)
        return 2
    rows = summarize(stats, args.top)
    if args.json:
        json.dump({"services": rows, "malformed_lines": bad}, sys.stdout, indent=2)
        print()
    else:
        print_table(rows)
    if bad:
        print(f"skipped {bad} malformed line(s)", file=sys.stderr)
    return 0


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

• Exit codes: 0 ran, 2 couldn’t read input. Malformed lines are counted and reported on stderr, not fatal.
• analyze takes any iterable of lines, not a path, so tests pass a list and nothing touches the disk.
• Memory grows with request count, because every latency is kept for the percentiles. For very large inputs, switch to fixed latency buckets (a histogram).

Unit-test the analyzer

Make logic take data, not filenames, so tests need no files. Run with python -m pytest.

import pytest
from log_analyzer import analyze, parse_line


def test_parse_good_line():
    f = parse_line("2026-10-13T14:02:11Z service=auth status=200 latency_ms=90")
    assert f["service"] == "auth"
    assert f["status"] == "200"


def test_parse_bad_line():
    with pytest.raises(ValueError):
        parse_line("garbage line here")


def test_analyze_counts_errors_and_skips_bad_lines():
    lines = [
        "t service=auth status=503 latency_ms=10",
        '{"service": "auth", "status": 200, "latency_ms": 20}',
        "garbage",
        "",
    ]
    stats, bad = analyze(lines)
    assert stats["auth"].count == 2
    assert stats["auth"].error_rate == 0.5
    assert bad == 1