who_need_help/scripts/summarize-docker-stats.py

#!/usr/bin/env python3
"""Summarize Docker's machine-readable stats without applying pass/fail limits."""

from __future__ import annotations

import argparse
import json
import re
from collections import defaultdict
from decimal import Decimal
from pathlib import Path
from statistics import fmean


UNIT_BYTES = {
    "B": Decimal(1),
    "kB": Decimal(1_000),
    "MB": Decimal(1_000_000),
    "GB": Decimal(1_000_000_000),
    "TB": Decimal(1_000_000_000_000),
    "KiB": Decimal(1_024),
    "MiB": Decimal(1_048_576),
    "GiB": Decimal(1_073_741_824),
    "TiB": Decimal(1_099_511_627_776),
}


def parse_size(value: str) -> int:
    match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)\s*([A-Za-z]+)", value.strip())
    if match is None:
        raise ValueError(f"invalid Docker size value: {value}")
    number, unit = match.groups()
    try:
        multiplier = UNIT_BYTES[unit]
    except KeyError as error:
        raise ValueError(f"unsupported Docker size unit: {unit}") from error
    return int(Decimal(number) * multiplier)


def parse_memory_usage(value: str) -> int:
    used, separator, _limit = value.partition("/")
    if not separator:
        raise ValueError(f"invalid Docker MemUsage value: {value}")
    return parse_size(used)


def parse_cpu(value: str) -> float:
    if not value.endswith("%"):
        raise ValueError(f"invalid Docker CPUPerc value: {value}")
    return float(value.removesuffix("%"))


def summarize(source: Path) -> dict[str, object]:
    samples: dict[str, list[dict[str, object]]] = defaultdict(list)
    observed_at: list[str] = []

    with source.open(encoding="utf-8") as stream:
        for line_number, line in enumerate(stream, start=1):
            if not line.strip():
                continue

            try:
                row = json.loads(line)
                name = str(row["Name"])
                timestamp = str(row["ObservedAt"])
                samples[name].append(
                    {
                        "observed_at": timestamp,
                        "cpu_percent": parse_cpu(str(row["CPUPerc"])),
                        "memory_bytes": parse_memory_usage(str(row["MemUsage"])),
                        "pids": int(row["PIDs"]),
                    }
                )
                observed_at.append(timestamp)
            except (KeyError, TypeError, ValueError, json.JSONDecodeError) as error:
                raise ValueError(f"{source}:{line_number}: {error}") from error

    if not samples:
        raise ValueError(f"{source}: no Docker stats samples")

    containers: list[dict[str, object]] = []

    for name in sorted(samples):
        rows = samples[name]
        cpu_values = [float(row["cpu_percent"]) for row in rows]
        memory_values = [int(row["memory_bytes"]) for row in rows]
        pid_values = [int(row["pids"]) for row in rows]
        window_size = min(10, len(rows))
        window_count = min(10, len(rows))
        memory_windows: list[dict[str, object]] = []

        for window_index in range(window_count):
            start = window_index * len(rows) // window_count
            end = (window_index + 1) * len(rows) // window_count
            window_rows = rows[start:end]
            window_memory = [int(row["memory_bytes"]) for row in window_rows]
            memory_windows.append(
                {
                    "index": window_index + 1,
                    "sample_count": len(window_rows),
                    "first_observed_at": str(window_rows[0]["observed_at"]),
                    "last_observed_at": str(window_rows[-1]["observed_at"]),
                    "average": fmean(window_memory),
                    "minimum": min(window_memory),
                    "maximum": max(window_memory),
                    "first": window_memory[0],
                    "last": window_memory[-1],
                }
            )

        containers.append(
            {
                "name": name,
                "sample_count": len(rows),
                "cpu_percent": {
                    "average": fmean(cpu_values),
                    "minimum": min(cpu_values),
                    "maximum": max(cpu_values),
                },
                "memory_bytes": {
                    "average": fmean(memory_values),
                    "minimum": min(memory_values),
                    "maximum": max(memory_values),
                    "first": memory_values[0],
                    "last": memory_values[-1],
                    "first_ten_average": fmean(memory_values[:window_size]),
                    "last_ten_average": fmean(memory_values[-window_size:]),
                    "first_to_last_delta": memory_values[-1] - memory_values[0],
                    "sequential_windows": memory_windows,
                },
                "pids": {
                    "minimum": min(pid_values),
                    "maximum": max(pid_values),
                    "first": pid_values[0],
                    "last": pid_values[-1],
                },
            }
        )

    return {
        "schema_version": 2,
        "measurement": (
            "docker stats --no-stream; the sampler sleeps one second after "
            "each complete multi-container collection"
        ),
        "thresholds_applied": False,
        "first_observed_at": min(observed_at),
        "last_observed_at": max(observed_at),
        "containers": containers,
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("source", type=Path)
    parser.add_argument("destination", type=Path)
    args = parser.parse_args()

    summary = summarize(args.source)
    args.destination.write_text(
        json.dumps(summary, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()