# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Runtime resource reporting utilities for dependency verification.
This module provides a small, best-effort diagnostic helper used by the
dependency verification commands (``latest`` and ``oldest``) to dump the
amount of resources still available on the CI runner after each test phase
and for each Python version.
The information collected here is primarily meant to debug situations in
which a long-running CI job stalls or is killed by the orchestrator because
the pod exhausts a resource (disk space, inodes, memory or CPU quota). The
helper intentionally never raises: a diagnostic probe must not be able to
abort the job it is supposed to observe.
:author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
"""
from __future__ import annotations
import subprocess
from pathlib import Path
from typing import Final
from mafw.tools.shell_tools import CONSOLE
from mafw.tools.shell_tools import run_stdout as _run_stdout
RESOURCE_SHELL_PROBES: Final[tuple[tuple[str, str], ...]] = (
('disk space (df -h)', 'df -h'),
('inodes (df -i)', 'df -i'),
)
"""Shell commands used to inspect disk usage, as ``(label, command)`` pairs."""
CGROUP_FILE_PROBES: Final[tuple[tuple[str, str], ...]] = (
('memory.current', '/sys/fs/cgroup/memory.current'),
('memory.max', '/sys/fs/cgroup/memory.max'),
('memory.events', '/sys/fs/cgroup/memory.events'),
('cpu.stat', '/sys/fs/cgroup/cpu.stat'),
)
"""cgroup v2 pseudo-files inspected to report memory and CPU quota usage.
Each entry is a ``(label, path)`` pair. The paths follow the Kubernetes/cgroup v2
layout mounted inside the runner pod. Missing files are reported but never fatal.
"""
[docs]
def _probe_shell(label: str, command: str) -> None:
"""
Run a shell probe and print its output, swallowing any error.
:param label: Human-readable description shown before the output.
:type label: str
:param command: Shell command to execute (e.g. ``df -h``).
:type command: str
"""
try:
output = _run_stdout(command, quiet=True)
except (subprocess.CalledProcessError, OSError) as exc: # pragma: no cover - defensive, CI-only probe
CONSOLE.print(f' [{label}] unavailable: {exc}', markup=False)
return
CONSOLE.print(f' [{label}]', markup=False)
CONSOLE.print(output or ' (no output)', markup=False)
[docs]
def _probe_cgroup_file(label: str, path: str) -> None:
"""
Read a cgroup pseudo-file and print its content, swallowing any error.
:param label: Human-readable description shown before the output.
:type label: str
:param path: Absolute path of the cgroup pseudo-file to read.
:type path: str
"""
file_path = Path(path)
try:
content = file_path.read_text(encoding='utf-8').strip()
except OSError as exc: # pragma: no cover - defensive, cgroup files are CI-only
CONSOLE.print(f' [{label}] unavailable: {exc}', markup=False)
return
CONSOLE.print(f' [{label}]', markup=False)
CONSOLE.print(content or ' (empty)', markup=False)
[docs]
def report_resources(context: str) -> None:
"""
Print a best-effort snapshot of the resources left on the runner.
The snapshot includes disk usage (``df -h`` and ``df -i``) and cgroup v2
memory and CPU statistics. All probes are best-effort: any failure to run
a command or to read a pseudo-file is reported inline and never propagated,
so the diagnostics cannot abort the verification job.
:param context: Short label describing when the snapshot is taken, for
example ``"py3.14 after unit tests"``. It is echoed in the banner so
that consecutive snapshots can be told apart in the CI log.
:type context: str
"""
CONSOLE.print(f'📊 Resource usage — {context}')
for label, command in RESOURCE_SHELL_PROBES:
_probe_shell(label, command)
for label, path in CGROUP_FILE_PROBES:
_probe_cgroup_file(label, path)