scripts.flake_report

Reference › scripts › flake_report

Reads NDJSON recovery events from standard input and prints a per-module summary, so that the modules that recover most often stand out. tests/scripts/test_flake_report.py imports it. It takes no options: main passes standard input to process. parse_line reads one event line and returns its timestamp and module, or nothing for a line that is not a usable event, process counts the events per module and prints the summary, and format_date renders timestamps as dates.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/flake_report.py

#!/usr/bin/env python3
"""Read NDJSON recovery events from stdin and print a per-module summary."""

from __future__ import annotations

import json
import sys
from collections import defaultdict
from datetime import UTC, datetime
from typing import TextIO


def parse_line(line: str) -> tuple[float, str] | None:
    """Parse a single NDJSON line. Return (ts, module) or None if malformed."""
    try:
        obj = json.loads(line)
    except json.JSONDecodeError:
        return None

    if not isinstance(obj, dict):
        return None

    ts = obj.get("ts")
    module = obj.get("module")

    # `bool` subclasses `int`, so a bare isinstance(ts, (int, float)) accepts
    # `true`/`false` and silently reports them as 1970-01-01 rather than
    # counting them as malformed. A boolean is not a timestamp.
    if isinstance(ts, bool) or not isinstance(ts, (int, float)):
        return None
    if not isinstance(module, str):
        return None

    # Everything below is a value json.loads ACCEPTS but datetime cannot
    # render, so the isinstance check above passes and the failure surfaces
    # later in format_date — where nothing catches it, killing the whole
    # report over one bad record and breaking process()'s documented
    # "Always returns 0". Measured 2026-08-12, all four reproduced:
    #   {"ts": <400-digit int>} -> OverflowError: int too large to convert
    #   {"ts": NaN}             -> ValueError: Invalid value NaN
    #   {"ts": Infinity}        -> OverflowError: timestamp out of range
    #   {"ts": 1e20}            -> OverflowError: timestamp out of range
    # json.loads accepts NaN/Infinity/-Infinity by default (they are not
    # standard JSON), and Python ints are unbounded.
    try:
        ts_value = float(ts)
    except OverflowError:
        return None

    # Validate by making the SAME call the report will make, rather than
    # guessing a platform range: if it cannot be rendered, it is malformed.
    try:
        datetime.fromtimestamp(ts_value, tz=UTC)
    except OverflowError, OSError, ValueError:
        return None

    return ts_value, module


def format_date(ts: float) -> str:
    """Format a unix timestamp as a UTC date string."""
    dt = datetime.fromtimestamp(ts, tz=UTC)
    return dt.strftime("%Y-%m-%d")


def process(input_stream: TextIO) -> int:
    """Read input, aggregate, and print the report. Always returns 0."""
    modules: dict[str, list[float]] = defaultdict(list)
    skipped = 0

    for line in input_stream:
        line = line.rstrip("\n").rstrip("\r")
        if not line:
            continue

        parsed = parse_line(line)
        if parsed is None:
            skipped += 1
            continue

        ts, module = parsed
        modules[module].append(ts)

    if not modules:
        print("no recoveries")
        if skipped:
            print(f"skipped {skipped} malformed line(s)", file=sys.stderr)
        return 0

    summary = []
    for module, timestamps in modules.items():
        count = len(timestamps)
        first_ts = min(timestamps)
        last_ts = max(timestamps)
        summary.append((count, module, format_date(first_ts), format_date(last_ts)))

    summary.sort(key=lambda item: (-item[0], item[1]))

    for count, module, first_date, last_date in summary:
        print(f"{count}x {module} (first: {first_date}, last: {last_date})")

    if skipped:
        print(f"skipped {skipped} malformed line(s)", file=sys.stderr)

    return 0


def main() -> int:
    """Entry point."""
    return process(sys.stdin)


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