scripts.container_cpu_accounting

Reference › scripts › container_cpu_accounting

Incus container CPU accounting: it reads the cgroup cpuacct core-seconds of the test-fleet containers before and after one run, so a run's CPU cost can be stated. run_tests imports it, tests/scripts/test_container_cpu_accounting.py tests it, and it shares the live-observer plumbing in pytest_live_observer.

Invocation: main has three subcommands. snapshot reads the current usage and prints it as JSON, or writes it to a file with -o. diff compares two snapshot files and prints the delta. run takes a snapshot, runs the given command, takes another and reports the difference. The snapshot commands take a project and a comma-separated container list.

resolve_project picks the Incus project, from the profile or the environment. list_container_pids asks Incus for each container's init process id, cgroup_cpu_dir finds that process's CPU cgroup directory and read_usage_seconds reads the accumulated usage. take_snapshot gathers these into a ContainerCpuSnapshot, which converts to and from a dictionary. diff_report computes per-container and total core-seconds, and persist_diff_report appends the report to an NDJSON file whose location container_cpu_ndjson_path resolves from the environment. A failure to read raises ContainerCpuError.

At a glance

Class diagram

Dependency diagram

API

Classes

ContainerCpuError

Bases: RuntimeError · Line 70

A container's CPU usage could not be resolved or read.

ContainerCpuSnapshot (dataclass)

Decorators: dataclass(frozen=True) · Line 75

One point-in-time read of cumulative CPU usage per container.

Functions

Source

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

#!/usr/bin/env python3
"""Incus container CPU accounting — cgroup ``cpuacct`` core-seconds around one
suite run (``plans/prolog-learning-subsystem-research-2026-08-12.md``,
recommendation #5).

The research record measured the suite's CPU floor at 2,567.7 core-seconds /
6 physical cores = 428.0s, and called every "the suite is at its floor" claim
built on that number -- including its own -- conditioned on one omission:
Memgraph and Redis run on the SAME 6 physical cores as the suite process
itself and are EXCLUDED from the 2,567.7 core-s figure. The true floor is
above 428s by an unmeasured amount. This script is the measurement: two
cgroup reads (before/after a run) turn "unmeasured amount" into a number.

Infra is Incus containers on ``10.53.2.0/24`` for the ``test`` profile (see
``INFRASTRUCTURE.md`` and ``env/test.env``) -- NOT Docker, NOT localhost.
``incus list`` inspects them directly; this script reads the SAME accounting
Incus itself derives, but at microsecond precision (``cpu.stat``'s
``usage_usec``) rather than the whole-second rounding ``incus info`` prints,
because a run's CPU delta is often smaller than one second of rounding error.

Usage
-----
    # Wrap a run end-to-end (recommended): snapshots before/after, reports
    # the delta, and exits with the wrapped command's own exit code.
    python scripts/container_cpu_accounting.py run -- uv run pytest tests/core/threading

    # Or take snapshots manually around an already-running job:
    python scripts/container_cpu_accounting.py snapshot -o /tmp/before.json
    ...  # run the suite yourself
    python scripts/container_cpu_accounting.py snapshot -o /tmp/after.json
    python scripts/container_cpu_accounting.py diff /tmp/before.json /tmp/after.json

Both forms default to the ``memgraph``/``redis`` containers in the
``orchestrata-test`` Incus project (the two the research record names) --
override with ``--containers`` and/or ``--project``/``--profile``.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any

#: Fleet -> Incus project, from INFRASTRUCTURE.md's "Environment fleets"
#: table. Mirrors the mapping every other script in this tree hardcodes
#: per-profile (e.g. scripts/promote_env.py's "orchestrata-test" literal) --
#: there is no single canonical resolver to import, so this is that mapping
#: made local and explicit rather than re-guessed inline.
PROJECT_BY_PROFILE = {
    "prod": "default",
    "test": "orchestrata-test",
    "dev": "orchestrata-dev",
}

#: The two containers the research record names as excluded from the
#: measured 2,567.7 core-s floor because they share the suite's own 6
#: physical cores.
DEFAULT_CONTAINERS = ("memgraph", "redis")

_CGROUP_ROOT = Path("/sys/fs/cgroup")


class ContainerCpuError(RuntimeError):
    """A container's CPU usage could not be resolved or read."""


@dataclass(frozen=True)
class ContainerCpuSnapshot:
    """One point-in-time read of cumulative CPU usage per container."""

    ts: float
    project: str
    usage_seconds: dict[str, float]

    def to_dict(self) -> dict[str, Any]:
        return {
            "ts": self.ts,
            "project": self.project,
            "usage_seconds": self.usage_seconds,
        }

    @staticmethod
    def from_dict(data: Mapping[str, Any]) -> ContainerCpuSnapshot:
        try:
            usage = {str(k): float(v) for k, v in data["usage_seconds"].items()}
            return ContainerCpuSnapshot(
                ts=float(data["ts"]), project=str(data["project"]), usage_seconds=usage
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise ContainerCpuError(f"malformed snapshot: {data!r}") from exc


def resolve_project(
    profile: str | None = None, *, env: Mapping[str, str] | None = None
) -> str:
    """Incus project for ``profile`` (default: ``ORCHESTRATA_ENV_PROFILE``,
    itself defaulting to ``test`` -- the same default ``env/test.env``
    documents for the rest of the suite)."""
    e = env if env is not None else os.environ
    p = profile if profile is not None else e.get("ORCHESTRATA_ENV_PROFILE", "test")
    try:
        return PROJECT_BY_PROFILE[p]
    except KeyError:
        known = ", ".join(sorted(PROJECT_BY_PROFILE))
        raise ContainerCpuError(
            f"unknown env profile {p!r}; expected one of: {known}"
        ) from None


def list_container_pids(
    project: str,
    *,
    env: Mapping[str, str] | None = None,
    timeout: float = 30.0,
) -> dict[str, int]:
    """``{container_name: host_pid}`` for every RUNNING instance in
    ``project``, via ``incus list --format json``'s ``state.pid`` -- the same
    field ``incus info`` derives its own (whole-second) CPU line from.

    ``env``/``timeout`` are dependency-injection seams: production leaves
    both at their defaults (the real process environment, 30s); tests use
    them to trigger the real ``incus`` CLI's own failure modes (an
    unreachable socket, an impossible timeout) without mocking anything.
    """
    try:
        result = subprocess.run(
            ["incus", "--project", project, "list", "--format", "json"],
            capture_output=True,
            text=True,
            check=True,
            timeout=timeout,
            env=dict(env) if env is not None else None,
        )
    except FileNotFoundError as exc:
        raise ContainerCpuError("incus executable not found on PATH") from exc
    except subprocess.CalledProcessError as exc:
        raise ContainerCpuError(
            f"incus list --project {project} failed: {exc.stderr.strip()}"
        ) from exc
    except subprocess.TimeoutExpired as exc:
        raise ContainerCpuError(f"incus list --project {project} timed out") from exc

    try:
        instances = json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        raise ContainerCpuError(f"incus list returned non-JSON output: {exc}") from exc

    pids: dict[str, int] = {}
    for inst in instances:
        if inst.get("status") != "Running":
            continue
        pid = (inst.get("state") or {}).get("pid")
        name = inst.get("name")
        if pid and name:
            pids[str(name)] = int(pid)
    return pids


def cgroup_cpu_dir(pid: int, *, cgroup_root: Path = _CGROUP_ROOT) -> Path:
    """The host cgroup-v2 directory that accounts for the WHOLE container
    owning ``pid`` -- not just that one process's own leaf scope.

    Derived from ``/proc/<pid>/cgroup`` rather than assumed from a naming
    convention: Incus/LXC nests a container's init process one level under a
    directory named ``lxc.payload.<project>_<name>`` (e.g.
    ``.../lxc.payload.orchestrata-test_memgraph/init.scope``); that
    directory -- not the ``init.scope`` leaf under it -- is what accounts for
    every process the container has ever spawned, matching what ``incus
    info``'s own CPU line is computed from.
    """
    cgroup_path = Path(f"/proc/{pid}/cgroup")
    try:
        line = cgroup_path.read_text(encoding="utf-8").strip()
    except OSError as exc:
        raise ContainerCpuError(f"cannot read {cgroup_path}: {exc}") from exc

    # cgroup v2 unified hierarchy: exactly one line, "0::<path>".
    _, sep, rel = line.partition("::")
    if not sep or not rel.startswith("/"):
        raise ContainerCpuError(
            f"pid {pid}: unexpected {cgroup_path} contents {line!r} "
            "(expected a cgroup v2 unified '0::/...' line -- cgroup v1 host?)"
        )
    segments = PurePosixPath(rel).parts[1:]  # drop the leading "/"
    for i, seg in enumerate(segments):
        if seg.startswith("lxc.payload."):
            return cgroup_root.joinpath(*segments[: i + 1])
    raise ContainerCpuError(
        f"pid {pid}: no 'lxc.payload.*' segment in cgroup path {rel!r} "
        "-- not an Incus/LXC container process?"
    )


def read_usage_seconds(cgroup_dir: Path) -> float:
    """Cumulative CPU time this cgroup has consumed, in seconds, from
    ``cpu.stat``'s ``usage_usec`` (cgroup v2's cpuacct-equivalent field --
    microsecond precision, unlike ``incus info``'s whole-second rounding)."""
    stat_path = cgroup_dir / "cpu.stat"
    try:
        text = stat_path.read_text(encoding="utf-8")
    except OSError as exc:
        raise ContainerCpuError(f"cannot read {stat_path}: {exc}") from exc
    for line in text.splitlines():
        key, _, value = line.partition(" ")
        if key == "usage_usec":
            try:
                return int(value) / 1_000_000
            except ValueError as exc:
                raise ContainerCpuError(
                    f"{stat_path}: bad usage_usec {value!r}"
                ) from exc
    raise ContainerCpuError(f"{stat_path} has no usage_usec field")


def take_snapshot(
    project: str,
    containers: Sequence[str] = DEFAULT_CONTAINERS,
    *,
    env: Mapping[str, str] | None = None,
) -> ContainerCpuSnapshot:
    """Read cumulative CPU usage for ``containers`` in ``project`` right now.

    ``env`` is the SAME dependency-injection seam ``list_container_pids``
    already exposes, threaded through rather than re-invented: production
    callers leave it at the default (the real process environment); tests
    pass an explicit ``Mapping`` (e.g. a real, unreachable ``INCUS_SOCKET``)
    so the real ``incus`` CLI's own failure modes can be exercised without
    mutating ``os.environ`` (RUL-002).
    """
    pids = list_container_pids(project, env=env)
    missing = [name for name in containers if name not in pids]
    if missing:
        have = ", ".join(sorted(pids)) or "none"
        raise ContainerCpuError(
            f"container(s) not running in project {project!r}: {', '.join(missing)} "
            f"(running: {have})"
        )
    usage = {
        name: read_usage_seconds(cgroup_cpu_dir(pids[name])) for name in containers
    }
    return ContainerCpuSnapshot(ts=time.time(), project=project, usage_seconds=usage)


def diff_report(
    before: ContainerCpuSnapshot, after: ContainerCpuSnapshot
) -> dict[str, Any]:
    """Per-container and total CPU-seconds consumed between two snapshots.

    Deliberately no averaging/history across MULTIPLE diffs -- like the
    fixture-cost instrument (recommendation #3), this is an n=1 measurement
    around one run, not a learner.
    """
    if before.project != after.project:
        raise ContainerCpuError(
            f"snapshot project mismatch: {before.project!r} vs {after.project!r}"
        )
    names = sorted(set(before.usage_seconds) & set(after.usage_seconds))
    skipped = sorted(set(before.usage_seconds) ^ set(after.usage_seconds))
    core_seconds = {
        name: round(after.usage_seconds[name] - before.usage_seconds[name], 3)
        for name in names
    }
    return {
        "project": before.project,
        "wall_seconds": round(after.ts - before.ts, 3),
        "core_seconds": core_seconds,
        "total_core_seconds": round(sum(core_seconds.values()), 3),
        "skipped_containers": skipped,
    }


# ---------------------------------------------------------------------------
# Persistence (recommendation #5's wiring into scripts/run_tests.py): a
# printed number scrolls away -- exactly the "the research keeps being done
# and deleted" failure the research record complains about for
# recommendation #1. This durably appends the delta so a later question can
# answer itself from reports/, the same discipline
# pytest_live_observer._append_test_durations already follows.
# ---------------------------------------------------------------------------

_CONTAINER_CPU_PATH_ENV = "ORCH_CONTAINER_CPU_PATH"


def container_cpu_ndjson_path(env: Mapping[str, str] | None = None) -> Path:
    """Where container-CPU delta records are appended.

    Mirrors ``pytest_live_observer.durations_ndjson_path``'s override
    convention: ``ORCH_CONTAINER_CPU_PATH`` lets tests redirect output away
    from the real ``reports/`` directory. Default:
    ``<cwd>/reports/container-cpu.ndjson``.
    """
    e = env if env is not None else os.environ
    override = e.get(_CONTAINER_CPU_PATH_ENV)
    if override:
        return Path(override)
    return Path.cwd() / "reports" / "container-cpu.ndjson"


def persist_diff_report(
    report: Mapping[str, Any], *, env: Mapping[str, str] | None = None
) -> None:
    """Append ``report`` (from ``diff_report``) as one NDJSON line, keyed by
    the SAME run identity (``run_id``/``head_commit``/``batch_id``/``lane``,
    each omitted rather than null when unset) that
    ``pytest_live_observer.run_identity`` already stamps on every
    test-duration and fixture-cost record -- recommendation #3's join-key
    convention, reused here rather than inventing a second identity scheme.
    A row written by run_tests.py's OWN process (not a batch subprocess)
    typically carries only ``run_id``/``head_commit``: ``batch_id``/``lane``
    are per-subprocess and never present in the top-level runner's own
    environment.

    Best-effort: never raises, matching every other sink in this project
    (``_emit``, ``_append_test_durations``) -- a write failure must never
    fail the run it observed. The caller (``run_tests.py``) additionally
    treats the WHOLE measurement (snapshot/diff/persist) as pure
    observation with no correctness role; this function's own contract is
    narrower and matches its siblings: only the write step is swallowed.
    """
    # Lazy: keeps this script's own CLI usage (snapshot/diff/run) free of a
    # src/orchestrata_ultimate import when persistence is never invoked --
    # only run_tests.py's wiring pays for it.
    from orchestrata_ultimate.core.threading.pytest_live_observer import run_identity

    e = env if env is not None else os.environ
    record: dict[str, Any] = {
        "event": "container_cpu",
        **report,
        **run_identity(env=e),
    }
    try:
        path = container_cpu_ndjson_path(e)
        path.parent.mkdir(parents=True, exist_ok=True)
        fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
        try:
            os.write(fd, (json.dumps(record) + "\n").encode("utf-8"))
        finally:
            os.close(fd)
    except OSError:
        pass


def _print_report(report: dict[str, Any]) -> None:
    print(json.dumps(report, indent=2, sort_keys=True))
    print(
        f"\ntotal: {report['total_core_seconds']:.3f} core-s over "
        f"{report['wall_seconds']:.3f}s wall ({report['project']})",
        file=sys.stderr,
    )


def _add_common_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--project", help="Incus project (default: derived from --profile)"
    )
    parser.add_argument(
        "--profile",
        help="Env profile (test/dev/prod) selecting the Incus project when --project is not given "
        "(default: $ORCHESTRATA_ENV_PROFILE, itself defaulting to 'test')",
    )
    parser.add_argument(
        "--containers",
        default=",".join(DEFAULT_CONTAINERS),
        help=f"comma-separated container names (default: {','.join(DEFAULT_CONTAINERS)})",
    )


def _resolve_project_arg(args: argparse.Namespace) -> str:
    return args.project if args.project else resolve_project(args.profile)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="container_cpu_accounting.py",
        description="cgroup cpuacct core-seconds for Incus test-fleet containers, before/after one run.",
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_snap = sub.add_parser(
        "snapshot", help="Read current CPU usage and print/save it as JSON."
    )
    _add_common_args(p_snap)
    p_snap.add_argument("-o", "--output", help="write JSON here instead of stdout")

    p_diff = sub.add_parser("diff", help="Compare two snapshot JSON files.")
    p_diff.add_argument("before", help="snapshot JSON taken before the run")
    p_diff.add_argument("after", help="snapshot JSON taken after the run")

    p_run = sub.add_parser(
        "run", help="Snapshot, run a command, snapshot again, report the delta."
    )
    _add_common_args(p_run)
    p_run.add_argument(
        "command",
        nargs=argparse.REMAINDER,
        help="command to run, e.g. -- uv run pytest ...",
    )

    args = parser.parse_args(argv)

    try:
        if args.cmd == "snapshot":
            containers = [c.strip() for c in args.containers.split(",") if c.strip()]
            snap = take_snapshot(_resolve_project_arg(args), containers)
            payload = json.dumps(snap.to_dict(), indent=2, sort_keys=True)
            if args.output:
                Path(args.output).write_text(payload + "\n", encoding="utf-8")
            else:
                print(payload)
            return 0

        if args.cmd == "diff":
            before = ContainerCpuSnapshot.from_dict(
                json.loads(Path(args.before).read_text(encoding="utf-8"))
            )
            after = ContainerCpuSnapshot.from_dict(
                json.loads(Path(args.after).read_text(encoding="utf-8"))
            )
            _print_report(diff_report(before, after))
            return 0

        if args.cmd == "run":
            command = args.command
            if command and command[0] == "--":
                command = command[1:]
            if not command:
                print(
                    "usage: container_cpu_accounting.py run -- <command...>",
                    file=sys.stderr,
                )
                return 2
            containers = [c.strip() for c in args.containers.split(",") if c.strip()]
            project = _resolve_project_arg(args)
            before = take_snapshot(project, containers)
            completed = subprocess.run(command)
            after = take_snapshot(project, containers)
            _print_report(diff_report(before, after))
            return completed.returncode

        return 2
    except ContainerCpuError as exc:
        print(f"container_cpu_accounting: {exc}", file=sys.stderr)
        return 1


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