diff --git a/pyproject.toml b/pyproject.toml index 4fdf15925..ffd3a40c2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "marshmallow>=3.20.0", "marshmallow-dataclass>=8.6.0", "mcp>=1.27.0", + "prometheus-client>=0.20.0", "pydantic>=2.5.0", "requests>=2.31.0", "rich>=13.0.0", diff --git a/src/srtctl/core/power/__init__.py b/src/srtctl/core/power/__init__.py new file mode 100644 index 000000000..de50627b3 --- /dev/null +++ b/src/srtctl/core/power/__init__.py @@ -0,0 +1,10 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Raw multinode GPU power artifacts for the ``dcgm-power`` telemetry provider. + +The provider records watts per allocated GPU, the srt-slurm topology needed to +map devices to ``prefill``/``decode``/``agg``, and the exact formal benchmark +window. It never integrates power into energy; that belongs to consumers of the +artifact contract. +""" diff --git a/src/srtctl/core/power/contract.py b/src/srtctl/core/power/contract.py new file mode 100644 index 000000000..c30d1f22a --- /dev/null +++ b/src/srtctl/core/power/contract.py @@ -0,0 +1,149 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Versioned wire format shared by every dcgm-power artifact writer and reader.""" + +from __future__ import annotations + +import json +import math +import os +import tempfile +from contextlib import suppress +from pathlib import Path, PurePosixPath +from typing import Any, TypeGuard + +SCHEMA_VERSION = 1 + +PRODUCER = "srt-slurm.dcgm-power" +POWER_METRIC = "DCGM_FI_DEV_POWER_USAGE" +POWER_UNIT = "W" +POWER_SCOPE = "gpu_device_board_as_reported_by_dcgm" +CLOCK_SOURCE = "head_node_unix_clock" + +MANIFEST_FILENAME = "manifest.json" +SAMPLES_FILENAME = "samples.csv" +WINDOWS_DIRNAME = "windows" + +SAMPLES_HEADER = ( + "schema_version", + "timestamp_unix", + "scrape_seq", + "hostname", + "gpu_index", + "gpu_uuid", + "power_w", +) + +MAX_SAMPLE_GAP_SECONDS = 3.0 + +BENCHMARK_TYPE_SA_BENCH = "sa-bench" + +CONTAINER_LOG_DIR = "/logs" +MEASUREMENT_WINDOW_DIR_ENV = "SRT_MEASUREMENT_WINDOW_DIR" + + +class Reason: + """Stable machine-readable reason codes recorded in artifacts.""" + + EXPORTER_STARTUP_TIMEOUT = "exporter_startup_timeout" + EXPORTER_LAUNCH_FAILED = "exporter_launch_failed" + EXPORTER_EXITED = "exporter_exited" + ENDPOINT_TIMEOUT = "endpoint_timeout" + ENDPOINT_HTTP_ERROR = "endpoint_http_error" + ENDPOINT_PARSE_ERROR = "endpoint_parse_error" + ENDPOINT_RESOLUTION_FAILED = "endpoint_resolution_failed" + POWER_METRIC_MISSING = "power_metric_missing" + DUPLICATE_POWER_METRIC = "duplicate_power_metric" + SAMPLES_CSV_MISSING = "samples_csv_missing" + SAMPLES_CSV_HEADER_MISMATCH = "samples_csv_header_mismatch" + SAMPLES_CSV_MALFORMED = "samples_csv_malformed" + DUPLICATE_SAMPLE_ROW = "duplicate_sample_row" + GPU_INDEX_MISSING = "gpu_index_missing" + GPU_UUID_MISSING = "gpu_uuid_missing" + INVALID_POWER_VALUE = "invalid_power_value" + UNEXPECTED_DEVICE = "unexpected_device" + EXPECTED_DEVICE_MISSING = "expected_device_missing" + GPU_UUID_CHANGED = "gpu_uuid_changed" + MIG_INSTANCE_UNSUPPORTED = "mig_instance_unsupported" + TIMESTAMP_NON_MONOTONIC = "timestamp_non_monotonic" + CONFLICTING_WORKER_ROLES = "conflicting_worker_roles" + CONFLICTING_HET_GROUPS = "conflicting_het_groups" + COLLECTOR_EXCEPTION = "collector_exception" + COLLECTOR_INTERRUPTED = "collector_interrupted" + COLLECTOR_JOIN_TIMEOUT = "collector_join_timeout" + BENCHMARK_CHILD_REAP_TIMEOUT = "benchmark_child_reap_timeout" + MEASUREMENT_WINDOW_MISSING = "measurement_window_missing" + MEASUREMENT_WINDOW_UNEXPECTED = "measurement_window_unexpected" + MEASUREMENT_WINDOW_DUPLICATE = "measurement_window_duplicate" + MEASUREMENT_WINDOW_MALFORMED = "measurement_window_malformed" + MEASUREMENT_WINDOW_ARTIFACT_PATH_INVALID = "measurement_window_artifact_path_invalid" + MEASUREMENT_WINDOW_INCOMPLETE = "measurement_window_incomplete" + MEASUREMENT_WINDOW_RESULT_MISSING = "measurement_window_result_missing" + MEASUREMENT_WINDOW_RESULT_MISMATCH = "measurement_window_result_mismatch" + MEASUREMENT_WINDOW_RESULT_PATH_INVALID = "measurement_window_result_path_invalid" + MEASUREMENT_WINDOW_CLOCK_MISMATCH = "measurement_window_clock_mismatch" + MEASUREMENT_WINDOW_NOT_BRACKETED = "measurement_window_not_bracketed" + SAMPLE_GAP_EXCEEDED = "sample_gap_exceeded" + + +FATAL_LIFECYCLE_REASONS = ( + Reason.EXPORTER_EXITED, + Reason.COLLECTOR_EXCEPTION, + Reason.COLLECTOR_INTERRUPTED, + Reason.COLLECTOR_JOIN_TIMEOUT, + Reason.BENCHMARK_CHILD_REAP_TIMEOUT, +) + + +OPERATIONAL_FAILURE_REASONS = ( + Reason.BENCHMARK_CHILD_REAP_TIMEOUT, + Reason.COLLECTOR_JOIN_TIMEOUT, +) + +STARTUP_FAILURE_REASONS = ( + Reason.EXPORTER_STARTUP_TIMEOUT, + Reason.EXPORTER_LAUNCH_FAILED, + Reason.ENDPOINT_RESOLUTION_FAILED, +) + + +def is_safe_relative_subpath(value: str) -> bool: + """Whether ``value`` is a relative POSIX path that stays below its root.""" + if not value or value.startswith(("/", "~")): + return False + parts = PurePosixPath(value).parts + return bool(parts) and not any(part in ("..", "") for part in parts) + + +def is_finite_number(value: Any) -> TypeGuard[int | float]: + """Whether ``value`` is a finite real number; bools are not numbers here.""" + return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) + + +def dedupe(values: list[str]) -> tuple[str, ...]: + """First-seen-order deduplication for reason-code accumulation.""" + return tuple(dict.fromkeys(values)) + + +def atomic_write_json(path: Path, payload: Any) -> None: + """Replace ``path`` with serialized JSON and leave no partial file behind.""" + path.parent.mkdir(parents=True, exist_ok=True) + fd, temp_path = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", text=True) + try: + handle = os.fdopen(fd, "w", encoding="utf-8") + except BaseException: + with suppress(OSError): + os.close(fd) + Path(temp_path).unlink(missing_ok=True) + raise + + try: + with handle: + handle.write(json.dumps(payload, indent=2, sort_keys=False) + "\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_path, path) + except BaseException: + Path(temp_path).unlink(missing_ok=True) + raise diff --git a/src/srtctl/core/power/manifest.py b/src/srtctl/core/power/manifest.py new file mode 100644 index 000000000..eecf086f0 --- /dev/null +++ b/src/srtctl/core/power/manifest.py @@ -0,0 +1,169 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""``manifest.json``: producer identity, topology, lifecycle state, and validity.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from srtctl import __version__ as PRODUCER_VERSION +from srtctl.core.power.contract import ( + CLOCK_SOURCE, + POWER_METRIC, + POWER_SCOPE, + POWER_UNIT, + PRODUCER, + SCHEMA_VERSION, + dedupe, +) +from srtctl.core.power.samples import ObservedDevice +from srtctl.core.power.topology import ExpectedDevice + +STATUS_STARTING = "starting" +STATUS_RUNNING = "running" +STATUS_COMPLETE = "complete" +STATUS_INCOMPLETE = "incomplete" +STATUS_FAILED = "failed" + +TERMINAL_STATUSES = (STATUS_COMPLETE, STATUS_INCOMPLETE, STATUS_FAILED) + + +@dataclass(frozen=True) +class DcgmExporterIdentity: + """Exactly which exporter image produced the samples. + + ``container_image_sha256`` is ``None`` when the resolved image is not a + regular file (for example a registry URI pulled at srun time). + """ + + container_image_resolved: str + container_image_sha256: str | None + port: int + command: str + + def to_dict(self) -> dict[str, Any]: + return { + "container_image_resolved": self.container_image_resolved, + "container_image_sha256": self.container_image_sha256, + "port": self.port, + "command": self.command, + } + + +@dataclass(frozen=True) +class ExpectedWindow: + """One measured concurrency point the benchmark is expected to record.""" + + benchmark_type: str + concurrency: int + + @property + def key(self) -> tuple[str, int]: + return (self.benchmark_type, self.concurrency) + + def to_dict(self) -> dict[str, Any]: + return {"benchmark_type": self.benchmark_type, "concurrency": self.concurrency} + + +@dataclass(frozen=True) +class WindowValidation: + """Structural coverage audit for one expected window.""" + + benchmark_type: str + concurrency: int + window_file: str | None + power_coverage_valid: bool + reason_codes: tuple[str, ...] = () + per_device_max_sample_gap_seconds: dict[str, float] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "benchmark_type": self.benchmark_type, + "concurrency": self.concurrency, + "window_file": self.window_file, + "power_coverage_valid": self.power_coverage_valid, + "reason_codes": list(self.reason_codes), + "per_device_max_sample_gap_seconds": dict(self.per_device_max_sample_gap_seconds), + } + + +@dataclass(frozen=True) +class ArtifactError: + """A stale, malformed, duplicate, or unsafe artifact file.""" + + path: str + reason_codes: tuple[str, ...] + + def to_dict(self) -> dict[str, Any]: + return {"path": self.path, "reason_codes": list(self.reason_codes)} + + +@dataclass +class PowerManifest: + """The orchestrator-owned manifest for one power session.""" + + job_id: str + run_name: str + sample_interval_seconds: float + request_timeout_seconds: float + required: bool + started_at_unix: float + dcgm_exporter: DcgmExporterIdentity + expected_devices: list[ExpectedDevice] + expected_windows: list[ExpectedWindow] + producer_git_commit: str | None = None + status: str = STATUS_STARTING + stopped_at_unix: float | None = None + publication_valid: bool | None = None + observed_devices: list[ObservedDevice] = field(default_factory=list) + max_scrape_duration_seconds: float | None = None + scrape_count: int = 0 + sample_row_count: int = 0 + window_validations: list[WindowValidation] = field(default_factory=list) + artifact_errors: list[ArtifactError] = field(default_factory=list) + reason_codes: list[str] = field(default_factory=list) + _terminal_committed: bool = field(default=False, init=False, repr=False) + + def mark_terminal(self, *, status: str, stopped_at_unix: float, publication_valid: bool) -> None: + """Freeze lifecycle state. Only ``complete`` may ever publish.""" + if self._terminal_committed or self.status in TERMINAL_STATUSES: + raise RuntimeError(f"manifest is already terminal: {self.status!r}") + if status not in TERMINAL_STATUSES: + raise ValueError(f"not a terminal status: {status!r}") + self.status = status + self.stopped_at_unix = stopped_at_unix + self.publication_valid = publication_valid and status == STATUS_COMPLETE + self._terminal_committed = True + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": SCHEMA_VERSION, + "producer": PRODUCER, + "producer_version": PRODUCER_VERSION, + "producer_git_commit": self.producer_git_commit, + "source_metric": POWER_METRIC, + "unit": POWER_UNIT, + "power_scope": POWER_SCOPE, + "timestamp_source": CLOCK_SOURCE, + "job_id": self.job_id, + "run_name": self.run_name, + "sample_interval_seconds": self.sample_interval_seconds, + "request_timeout_seconds": self.request_timeout_seconds, + "max_scrape_duration_seconds": self.max_scrape_duration_seconds, + "required": self.required, + "started_at_unix": self.started_at_unix, + "stopped_at_unix": self.stopped_at_unix, + "status": self.status, + "publication_valid": self.publication_valid, + "dcgm_exporter": self.dcgm_exporter.to_dict(), + "expected_devices": [device.to_dict() for device in self.expected_devices], + "observed_devices": [device.to_dict() for device in self.observed_devices], + "expected_windows": [window.to_dict() for window in self.expected_windows], + "scrape_count": self.scrape_count, + "sample_row_count": self.sample_row_count, + "window_validations": [validation.to_dict() for validation in self.window_validations], + "artifact_errors": [error.to_dict() for error in self.artifact_errors], + "reason_codes": list(dedupe(self.reason_codes)), + } diff --git a/src/srtctl/core/power/parser.py b/src/srtctl/core/power/parser.py new file mode 100644 index 000000000..7844f8904 --- /dev/null +++ b/src/srtctl/core/power/parser.py @@ -0,0 +1,106 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Strict DCGM exporter power parsing. + +Only ``DCGM_FI_DEV_POWER_USAGE`` is read. Device identity comes from the ``gpu`` +and ``UUID`` labels; the optional ``Hostname`` label is deliberately ignored +because the collector already knows which allocated node it polled. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass + +from prometheus_client.parser import text_string_to_metric_families + +from srtctl.core.power.contract import POWER_METRIC, Reason, dedupe + +_MIG_LABELS = ("GPU_I_ID", "GPU_I_PROFILE") + + +@dataclass(frozen=True) +class PowerReading: + """One physical GPU's power draw within a single scrape.""" + + gpu_index: int + gpu_uuid: str + power_w: float + + +@dataclass(frozen=True) +class ParsedScrape: + """Readings that may be persisted, plus why anything was dropped.""" + + readings: tuple[PowerReading, ...] = () + reason_codes: tuple[str, ...] = () + + +def parse_power_scrape(text: str) -> ParsedScrape: + """Parse one exporter ``/metrics`` body into publishable power readings.""" + reasons: list[str] = [] + try: + families = list(text_string_to_metric_families(text)) + # prometheus-client releases in our supported >=0.20 range have raised + # ValueError, KeyError, and IndexError for malformed exposition. Keep this + # third-party boundary broad while still allowing BaseException control + # flow (for example KeyboardInterrupt) to propagate. + except Exception: # noqa: BLE001 + return ParsedScrape(reason_codes=(Reason.ENDPOINT_PARSE_ERROR,)) + + by_index: dict[int, PowerReading] = {} + duplicated: set[int] = set() + saw_power_sample = False + + for family in families: + for sample in family.samples: + if sample.name != POWER_METRIC: + continue + saw_power_sample = True + labels = sample.labels + + if any(labels.get(label) for label in _MIG_LABELS): + reasons.append(Reason.MIG_INSTANCE_UNSUPPORTED) + continue + + gpu_index = _parse_index(labels.get("gpu")) + if gpu_index is None: + reasons.append(Reason.GPU_INDEX_MISSING) + continue + + gpu_uuid = (labels.get("UUID") or "").strip() + if not gpu_uuid: + reasons.append(Reason.GPU_UUID_MISSING) + continue + + value = sample.value + if not math.isfinite(value) or value < 0: + reasons.append(Reason.INVALID_POWER_VALUE) + continue + + if gpu_index in by_index: + duplicated.add(gpu_index) + continue + by_index[gpu_index] = PowerReading(gpu_index=gpu_index, gpu_uuid=gpu_uuid, power_w=value) + + if duplicated: + reasons.append(Reason.DUPLICATE_POWER_METRIC) + for gpu_index in duplicated: + by_index.pop(gpu_index, None) + + if not saw_power_sample: + reasons.append(Reason.POWER_METRIC_MISSING) + + readings = tuple(by_index[index] for index in sorted(by_index)) + return ParsedScrape(readings=readings, reason_codes=dedupe(reasons)) + + +def _parse_index(raw: str | None) -> int | None: + if raw is None: + return None + try: + value = int(raw) + except ValueError: + return None + return value if value >= 0 else None diff --git a/src/srtctl/core/power/samples.py b/src/srtctl/core/power/samples.py new file mode 100644 index 000000000..2ae6d4b1c --- /dev/null +++ b/src/srtctl/core/power/samples.py @@ -0,0 +1,217 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""``samples.csv`` writer, reader, and observed-device derivation. + +Terminal validation re-reads the persisted bytes rather than trusting in-memory +state, so the reader here is deliberately strict about types and ordering. +""" + +from __future__ import annotations + +import csv +import itertools +import math +from collections.abc import Iterable, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, TextIO + +from srtctl.core.power.contract import SAMPLES_HEADER, SCHEMA_VERSION, Reason, dedupe + + +@dataclass(frozen=True) +class SampleRow: + """One persisted observation of one GPU.""" + + timestamp_unix: float + scrape_seq: int + hostname: str + gpu_index: int + gpu_uuid: str + power_w: float + schema_version: int = SCHEMA_VERSION + + @property + def key(self) -> tuple[int, str, int]: + return (self.scrape_seq, self.hostname, self.gpu_index) + + def to_csv(self) -> list[Any]: + return [ + self.schema_version, + repr(self.timestamp_unix), + self.scrape_seq, + self.hostname, + self.gpu_index, + self.gpu_uuid, + repr(self.power_w), + ] + + +@dataclass(frozen=True) +class ObservedDevice: + """A device as it actually appeared in the persisted samples.""" + + hostname: str + gpu_index: int + gpu_uuids: tuple[str, ...] + first_sample_time_unix: float + last_sample_time_unix: float + sample_times: tuple[float, ...] = field(default=(), repr=False) + + @property + def key(self) -> tuple[str, int]: + return (self.hostname, self.gpu_index) + + def to_dict(self) -> dict[str, Any]: + return { + "hostname": self.hostname, + "gpu_index": self.gpu_index, + "gpu_uuids": list(self.gpu_uuids), + "first_sample_time_unix": self.first_sample_time_unix, + "last_sample_time_unix": self.last_sample_time_unix, + } + + +class SampleWriter: + """Append-only ``samples.csv`` writer owned by the collector thread.""" + + def __init__(self, path: Path): + path.parent.mkdir(parents=True, exist_ok=True) + self.path = path + self.row_count = 0 + handle = open(path, "w", newline="", encoding="utf-8") # noqa: SIM115 + try: + writer = csv.writer(handle) + writer.writerow(SAMPLES_HEADER) + handle.flush() + except BaseException: + handle.close() + raise + self._handle: TextIO | None = handle + self._writer = writer + + @property + def closed(self) -> bool: + return self._handle is None + + def append(self, rows: Iterable[SampleRow]) -> None: + if self._handle is None: + raise ValueError("samples.csv writer is closed") + for row in rows: + self._writer.writerow(row.to_csv()) + self.row_count += 1 + + def flush(self) -> None: + if self._handle is not None: + self._handle.flush() + + def close(self) -> None: + if self._handle is None: + return + self._handle.flush() + self._handle.close() + self._handle = None + + +def read_samples(path: Path) -> tuple[tuple[SampleRow, ...], tuple[str, ...]]: + """Parse persisted samples strictly, returning valid rows and reason codes.""" + if not path.is_file(): + return (), (Reason.SAMPLES_CSV_MISSING,) + + reasons: list[str] = [] + rows: list[SampleRow] = [] + try: + with open(path, newline="", encoding="utf-8") as handle: + reader = csv.reader(handle) + header = next(reader, None) + if header != list(SAMPLES_HEADER): + return (), (Reason.SAMPLES_CSV_HEADER_MISMATCH,) + for raw in reader: + row = _parse_row(raw) + if row is None: + reasons.append(Reason.SAMPLES_CSV_MALFORMED) + continue + rows.append(row) + except (OSError, UnicodeDecodeError, csv.Error): + # NOTE: a corrupt byte or oversized field is malformed data, not a crash. + reasons.append(Reason.SAMPLES_CSV_MALFORMED) + + seen: set[tuple[int, str, int]] = set() + unique: list[SampleRow] = [] + for row in rows: + if row.key in seen: + reasons.append(Reason.DUPLICATE_SAMPLE_ROW) + continue + seen.add(row.key) + unique.append(row) + + if _has_non_monotonic_device(unique): + reasons.append(Reason.TIMESTAMP_NON_MONOTONIC) + + return tuple(unique), dedupe(reasons) + + +def derive_observed_devices(rows: Sequence[SampleRow]) -> list[ObservedDevice]: + """Collapse persisted rows into per-device identity and sample bounds.""" + ordered: dict[tuple[str, int], list[SampleRow]] = {} + for row in rows: + ordered.setdefault((row.hostname, row.gpu_index), []).append(row) + + devices: list[ObservedDevice] = [] + for key in sorted(ordered): + device_rows = sorted(ordered[key], key=lambda row: row.scrape_seq) + times = tuple(row.timestamp_unix for row in device_rows) + devices.append( + ObservedDevice( + hostname=key[0], + gpu_index=key[1], + gpu_uuids=tuple(dict.fromkeys(row.gpu_uuid for row in device_rows)), + first_sample_time_unix=min(times), + last_sample_time_unix=max(times), + sample_times=tuple(sorted(times)), + ) + ) + return devices + + +def _parse_row(raw: list[str]) -> SampleRow | None: + if len(raw) != len(SAMPLES_HEADER): + return None + try: + schema_version = int(raw[0]) + timestamp_unix = float(raw[1]) + scrape_seq = int(raw[2]) + gpu_index = int(raw[4]) + power_w = float(raw[6]) + except ValueError: + return None + + hostname, gpu_uuid = raw[3], raw[5] + if schema_version != SCHEMA_VERSION or not hostname or not gpu_uuid: + return None + if not math.isfinite(timestamp_unix) or not math.isfinite(power_w) or power_w < 0: + return None + if scrape_seq < 0 or gpu_index < 0: + return None + return SampleRow( + timestamp_unix=timestamp_unix, + scrape_seq=scrape_seq, + hostname=hostname, + gpu_index=gpu_index, + gpu_uuid=gpu_uuid, + power_w=power_w, + schema_version=schema_version, + ) + + +def _has_non_monotonic_device(rows: Sequence[SampleRow]) -> bool: + per_device: dict[tuple[str, int], list[SampleRow]] = {} + for row in rows: + per_device.setdefault((row.hostname, row.gpu_index), []).append(row) + for device_rows in per_device.values(): + ordered = sorted(device_rows, key=lambda row: row.scrape_seq) + for previous, current in itertools.pairwise(ordered): + if current.timestamp_unix < previous.timestamp_unix: + return True + return False diff --git a/src/srtctl/core/power/topology.py b/src/srtctl/core/power/topology.py new file mode 100644 index 000000000..c40ee385b --- /dev/null +++ b/src/srtctl/core/power/topology.py @@ -0,0 +1,142 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Expected GPU topology derived from srt-slurm backend-process placement.""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any + +from srtctl.core.power.contract import Reason, dedupe +from srtctl.core.power.samples import ObservedDevice +from srtctl.core.topology import Process + +DeviceKey = tuple[str, int] + +WORKER_ROLES = ("prefill", "decode", "agg") + + +@dataclass(frozen=True) +class DeviceAssignment: + """One backend process's claim on a physical GPU.""" + + worker_role: str + worker_index: int + worker_process: int + het_group: int | None + + def to_dict(self) -> dict[str, Any]: + return { + "worker_role": self.worker_role, + "worker_index": self.worker_index, + "worker_process": self.worker_process, + "het_group": self.het_group, + } + + +@dataclass(frozen=True) +class ExpectedDevice: + """A physical GPU the run is expected to sample.""" + + hostname: str + gpu_index: int + assignments: tuple[DeviceAssignment, ...] + + def __post_init__(self) -> None: + if not self.assignments: + raise ValueError("expected device requires at least one assignment") + + @property + def key(self) -> DeviceKey: + return (self.hostname, self.gpu_index) + + def to_dict(self) -> dict[str, Any]: + return { + "hostname": self.hostname, + "gpu_index": self.gpu_index, + "assignments": [assignment.to_dict() for assignment in self.assignments], + } + + +def build_expected_devices(processes: Sequence[Process]) -> list[ExpectedDevice]: + """Map every allocated backend process onto the GPUs it occupies.""" + grouped: dict[DeviceKey, list[DeviceAssignment]] = {} + for process in processes: + assignment = DeviceAssignment( + worker_role=process.endpoint_mode, + worker_index=process.endpoint_index, + worker_process=process.node_rank, + het_group=process.het_group, + ) + for gpu_index in sorted(process.gpu_indices): + grouped.setdefault((process.node, gpu_index), []).append(assignment) + + return [ + ExpectedDevice(hostname=key[0], gpu_index=key[1], assignments=tuple(assignments)) + for key, assignments in sorted(grouped.items()) + ] + + +def resolve_roles(devices: Sequence[ExpectedDevice]) -> tuple[dict[DeviceKey, str], tuple[str, ...]]: + """Resolve one semantic role per device, or report the conflict.""" + roles: dict[DeviceKey, str] = {} + for device in devices: + distinct = {assignment.worker_role for assignment in device.assignments} + if len(distinct) != 1: + return {}, (Reason.CONFLICTING_WORKER_ROLES,) + roles[device.key] = distinct.pop() + return roles, () + + +def resolve_het_groups(devices: Sequence[ExpectedDevice]) -> tuple[dict[str, int | None], tuple[str, ...]]: + """Resolve one heterogeneous Slurm group per node, or report the conflict.""" + groups: dict[str, int | None] = {} + for device in devices: + distinct = {assignment.het_group for assignment in device.assignments} + if len(distinct) != 1: + return {}, (Reason.CONFLICTING_HET_GROUPS,) + group = distinct.pop() + if device.hostname in groups and groups[device.hostname] != group: + return {}, (Reason.CONFLICTING_HET_GROUPS,) + groups[device.hostname] = group + return groups, () + + +@dataclass(frozen=True) +class DeviceValidation: + """Whether device identity and topology permit publication.""" + + valid: bool + reason_codes: tuple[str, ...] + + +def validate_devices( + expected: Sequence[ExpectedDevice], + observed: Sequence[ObservedDevice], +) -> DeviceValidation: + """Require a non-empty expected set that exactly matches stable observations.""" + reasons: list[str] = [] + + expected_keys = {device.key for device in expected} + observed_keys = {device.key for device in observed} + + if not expected_keys or expected_keys - observed_keys: + reasons.append(Reason.EXPECTED_DEVICE_MISSING) + if observed_keys - expected_keys: + reasons.append(Reason.UNEXPECTED_DEVICE) + # NOTE: a UUID must map 1:1 to a device key, or one physical GPU is counted twice. + if any(len(device.gpu_uuids) != 1 for device in observed): + reasons.append(Reason.GPU_UUID_CHANGED) + else: + uuids = [device.gpu_uuids[0] for device in observed] + if len(set(uuids)) != len(uuids): + reasons.append(Reason.GPU_UUID_CHANGED) + + _, role_conflicts = resolve_roles(expected) + _, group_conflicts = resolve_het_groups(expected) + reasons.extend(role_conflicts) + reasons.extend(group_conflicts) + + return DeviceValidation(valid=not reasons, reason_codes=dedupe(reasons)) diff --git a/src/srtctl/core/power/windows.py b/src/srtctl/core/power/windows.py new file mode 100644 index 000000000..8e92b2329 --- /dev/null +++ b/src/srtctl/core/power/windows.py @@ -0,0 +1,390 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Measurement-window coverage audit. + +Every regular ``windows/*.json`` file is scanned, so a stale, duplicate, +malformed, or unsafe artifact cannot be ignored. This is a structural audit +only: it never averages power or computes energy. +""" + +from __future__ import annotations + +import itertools +import json +import logging +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +from srtctl.core.power.contract import ( + CLOCK_SOURCE, + MAX_SAMPLE_GAP_SECONDS, + SCHEMA_VERSION, + WINDOWS_DIRNAME, + Reason, + atomic_write_json, + dedupe, + is_finite_number, + is_safe_relative_subpath, +) +from srtctl.core.power.manifest import ArtifactError, ExpectedWindow, WindowValidation +from srtctl.core.power.samples import ObservedDevice +from srtctl.core.power.topology import DeviceKey + +logger = logging.getLogger(__name__) + +WINDOW_STATUS_RUNNING = "running" +WINDOW_STATUS_COMPLETED = "completed" +WINDOW_STATUS_FAILED = "failed" +WINDOW_STATUS_INTERRUPTED = "interrupted" + +_ALLOWED_STATUSES = (WINDOW_STATUS_RUNNING, WINDOW_STATUS_COMPLETED, WINDOW_STATUS_FAILED, WINDOW_STATUS_INTERRUPTED) + +_CLOCK_TOLERANCE_SECONDS = 0.5 +_CLOCK_TOLERANCE_FRACTION = 0.01 + + +@dataclass(frozen=True) +class _ParsedWindow: + relative_path: str + benchmark_type: str + concurrency: int + result_path: str + status: str + start_unix: float + end_unix: float | None + duration: float | None + + +def validate_expected_windows( + *, + power_dir: Path, + result_root: Path, + expected_windows: Sequence[ExpectedWindow], + expected_device_keys: set[DeviceKey], + observed_devices: Sequence[ObservedDevice], + artifact_errors: list[ArtifactError], +) -> list[WindowValidation]: + """Emit exactly one validation row per expected window. + + A missing expected window must never pass vacuously, and every unexpected, + malformed, duplicate, or unsafe file lands in ``artifact_errors`` so the + expected key set can be compared against the valid observed key set. + """ + parsed, duplicates = _scan(power_dir / WINDOWS_DIRNAME, result_root, artifact_errors) + expected_keys = {window.key for window in expected_windows} + + for key, window in sorted(parsed.items()): + if key not in expected_keys: + artifact_errors.append( + ArtifactError(path=window.relative_path, reason_codes=(Reason.MEASUREMENT_WINDOW_UNEXPECTED,)) + ) + + return [ + _validate_one( + expected=expected, + window=parsed.get(expected.key), + duplicated=expected.key in duplicates, + result_root=result_root, + expected_device_keys=expected_device_keys, + observed_devices=observed_devices, + ) + for expected in expected_windows + ] + + +def convert_running_windows(windows_dir: Path, *, reason: str) -> int: + """Turn every still-``running`` window into ``interrupted``. + + Only safe after the benchmark child was proved reaped: a surviving child + could otherwise overwrite the file with ``completed``. The orchestrator + never invents a request-completion boundary, so end and duration stay null. + """ + if not windows_dir.is_dir(): + return 0 + if not _stays_below(windows_dir.parent, windows_dir.name): + return 0 + + try: + paths = sorted(windows_dir.glob("*.json")) + except OSError: + return 0 + + converted = 0 + for path in paths: + if path.is_symlink() or not path.is_file(): + continue + try: + payload = json.loads(path.read_text()) + except (OSError, ValueError): + continue + if not isinstance(payload, dict) or payload.get("status") != WINDOW_STATUS_RUNNING: + continue + payload["status"] = WINDOW_STATUS_INTERRUPTED + payload["benchmark_end_time_unix"] = None + payload["duration"] = None + payload["reason"] = reason + atomic_write_json(path, payload) + converted += 1 + return converted + + +def _scan( + windows_dir: Path, + result_root: Path, + artifact_errors: list[ArtifactError], +) -> tuple[dict[tuple[str, int], _ParsedWindow], set[tuple[str, int]]]: + """Parse every regular window file, recording anything unusable.""" + parsed: dict[tuple[str, int], _ParsedWindow] = {} + duplicates: set[tuple[str, int]] = set() + if not windows_dir.is_dir(): + return parsed, duplicates + + # NOTE: rejecting only child symlinks still lets the whole directory be one. + if not _stays_below(windows_dir.parent, WINDOWS_DIRNAME): + artifact_errors.append( + ArtifactError(path=WINDOWS_DIRNAME, reason_codes=(Reason.MEASUREMENT_WINDOW_ARTIFACT_PATH_INVALID,)) + ) + return parsed, duplicates + + try: + paths = sorted(windows_dir.iterdir()) + except OSError: + artifact_errors.append( + ArtifactError(path=WINDOWS_DIRNAME, reason_codes=(Reason.MEASUREMENT_WINDOW_ARTIFACT_PATH_INVALID,)) + ) + return parsed, duplicates + + for path in paths: + relative = f"{WINDOWS_DIRNAME}/{path.name}" + if path.is_symlink() or not path.is_file() or path.suffix != ".json": + artifact_errors.append( + ArtifactError(path=relative, reason_codes=(Reason.MEASUREMENT_WINDOW_ARTIFACT_PATH_INVALID,)) + ) + continue + + window, reasons = _parse(path, relative, result_root) + if window is None: + artifact_errors.append(ArtifactError(path=relative, reason_codes=reasons)) + continue + + key = (window.benchmark_type, window.concurrency) + if key in duplicates: + artifact_errors.append(ArtifactError(path=relative, reason_codes=(Reason.MEASUREMENT_WINDOW_DUPLICATE,))) + continue + + previous = parsed.pop(key, None) + if previous is not None: + duplicates.add(key) + artifact_errors.append( + ArtifactError(path=previous.relative_path, reason_codes=(Reason.MEASUREMENT_WINDOW_DUPLICATE,)) + ) + artifact_errors.append(ArtifactError(path=relative, reason_codes=(Reason.MEASUREMENT_WINDOW_DUPLICATE,))) + continue + parsed[key] = window + + return parsed, duplicates + + +def _is_strict_int(value) -> bool: + """``bool`` is an ``int`` subclass; JSON ``true`` must not key as concurrency 1.""" + return isinstance(value, int) and not isinstance(value, bool) + + +def _parse(path: Path, relative: str, result_root: Path) -> tuple[_ParsedWindow | None, tuple[str, ...]]: + try: + payload = json.loads(path.read_text()) + except (OSError, ValueError): + return None, (Reason.MEASUREMENT_WINDOW_MALFORMED,) + if not isinstance(payload, dict): + return None, (Reason.MEASUREMENT_WINDOW_MALFORMED,) + + status = payload.get("status") + concurrency = payload.get("concurrency") + benchmark_type = payload.get("benchmark_type") + start = payload.get("benchmark_start_time_unix") + if ( + not _is_strict_int(payload.get("schema_version")) + or payload.get("schema_version") != SCHEMA_VERSION + or payload.get("clock_source") != CLOCK_SOURCE + or status not in _ALLOWED_STATUSES + or not isinstance(benchmark_type, str) + or not _is_strict_int(concurrency) + or not is_finite_number(start) + ): + return None, (Reason.MEASUREMENT_WINDOW_MALFORMED,) + + result_path = payload.get("result_path") + if not isinstance(result_path, str) or not is_safe_relative_subpath(result_path): + return None, (Reason.MEASUREMENT_WINDOW_RESULT_PATH_INVALID,) + if not _stays_below(result_root, result_path) or Path(result_path).stem != path.stem: + return None, (Reason.MEASUREMENT_WINDOW_RESULT_PATH_INVALID,) + + end = payload.get("benchmark_end_time_unix") + duration = payload.get("duration") + if not _status_invariants_hold(status, end, duration, payload.get("reason")): + return None, (Reason.MEASUREMENT_WINDOW_MALFORMED,) + + return ( + _ParsedWindow( + relative_path=relative, + benchmark_type=benchmark_type, + concurrency=concurrency, + result_path=result_path, + status=status, + start_unix=float(start), + end_unix=float(end) if end is not None else None, + duration=float(duration) if duration is not None else None, + ), + (), + ) + + +def _status_invariants_hold(status: str, end, duration, reason) -> bool: + """End, duration, and reason nullability are all fixed by the status. + + ``running`` is written before anything can have gone wrong, so it carries no + reason; ``interrupted`` is only ever produced by the orchestrator, which + always records why. + """ + if status == WINDOW_STATUS_RUNNING: + return end is None and duration is None and reason is None + if status == WINDOW_STATUS_INTERRUPTED: + return end is None and duration is None and isinstance(reason, str) and bool(reason) + if not is_finite_number(end) or not is_finite_number(duration) or duration <= 0: + return False + if status == WINDOW_STATUS_COMPLETED: + return reason is None + return isinstance(reason, str) and bool(reason) + + +def _validate_one( + *, + expected: ExpectedWindow, + window: _ParsedWindow | None, + duplicated: bool, + result_root: Path, + expected_device_keys: set[DeviceKey], + observed_devices: Sequence[ObservedDevice], +) -> WindowValidation: + if window is None: + reasons = [Reason.MEASUREMENT_WINDOW_DUPLICATE] if duplicated else [Reason.MEASUREMENT_WINDOW_MISSING] + return WindowValidation( + benchmark_type=expected.benchmark_type, + concurrency=expected.concurrency, + window_file=None, + power_coverage_valid=False, + reason_codes=tuple(reasons), + ) + + reasons: list[str] = [] + if window.status != WINDOW_STATUS_COMPLETED: + reasons.append(Reason.MEASUREMENT_WINDOW_INCOMPLETE) + else: + reasons.extend(_check_result(window, result_root)) + + gaps: dict[str, float] = {} + if not reasons: + if window.end_unix is None: + reasons.append(Reason.MEASUREMENT_WINDOW_MALFORMED) + else: + gaps, coverage_reasons = _check_coverage( + window.start_unix, window.end_unix, expected_device_keys, observed_devices + ) + reasons.extend(coverage_reasons) + + return WindowValidation( + benchmark_type=expected.benchmark_type, + concurrency=expected.concurrency, + window_file=window.relative_path, + power_coverage_valid=not reasons, + reason_codes=dedupe(reasons), + per_device_max_sample_gap_seconds=gaps, + ) + + +def _check_result(window: _ParsedWindow, result_root: Path) -> list[str]: + """A completed window must match the result it points at, on both clocks.""" + result_file = result_root / window.result_path + if not result_file.is_file(): + return [Reason.MEASUREMENT_WINDOW_RESULT_MISSING] + try: + result = json.loads(result_file.read_text()) + except (OSError, ValueError): + return [Reason.MEASUREMENT_WINDOW_RESULT_MISSING] + if not isinstance(result, dict): + return [Reason.MEASUREMENT_WINDOW_RESULT_MISMATCH] + + if window.end_unix is None or window.duration is None: + return [Reason.MEASUREMENT_WINDOW_MALFORMED] + result_timings = ( + result.get("benchmark_start_time_unix"), + result.get("benchmark_end_time_unix"), + result.get("duration"), + ) + if not all(is_finite_number(value) for value in result_timings) or result_timings != ( + window.start_unix, + window.end_unix, + window.duration, + ): + return [Reason.MEASUREMENT_WINDOW_RESULT_MISMATCH] + + wall = window.end_unix - window.start_unix + if wall <= 0: + return [Reason.MEASUREMENT_WINDOW_CLOCK_MISMATCH] + tolerance = max(_CLOCK_TOLERANCE_SECONDS, _CLOCK_TOLERANCE_FRACTION * window.duration) + if abs(wall - window.duration) > tolerance: + return [Reason.MEASUREMENT_WINDOW_CLOCK_MISMATCH] + return [] + + +def _check_coverage( + start: float, + end: float, + expected_device_keys: set[DeviceKey], + observed_devices: Sequence[ObservedDevice], +) -> tuple[dict[str, float], list[str]]: + """Every expected device must bracket the window with small enough gaps.""" + by_key = {device.key: device for device in observed_devices} + gaps: dict[str, float] = {} + reasons: list[str] = [] + + for key in sorted(expected_device_keys): + device = by_key.get(key) + if device is None: + reasons.append(Reason.MEASUREMENT_WINDOW_NOT_BRACKETED) + continue + # NOTE: a changed UUID cannot be attributed to one GPU, so this window's coverage is unusable. + if len(device.gpu_uuids) != 1: + reasons.append(Reason.GPU_UUID_CHANGED) + continue + sequence = _bracketing_sequence(device.sample_times, start, end) + if sequence is None: + reasons.append(Reason.MEASUREMENT_WINDOW_NOT_BRACKETED) + continue + largest = max((later - earlier for earlier, later in itertools.pairwise(sequence)), default=0.0) + gaps[f"{device.hostname}/{device.gpu_uuids[0]}"] = largest + if largest > MAX_SAMPLE_GAP_SECONDS: + reasons.append(Reason.SAMPLE_GAP_EXCEEDED) + + return gaps, reasons + + +def _bracketing_sequence(times: Sequence[float], start: float, end: float) -> list[float] | None: + """The last sample at or before start, every in-window sample, the first at or after end.""" + ordered = sorted(times) + before = [value for value in ordered if value <= start] + after = [value for value in ordered if value >= end] + if not before or not after: + return None + inside = [value for value in ordered if start < value < end] + return [before[-1], *inside, after[0]] + + +def _stays_below(root: Path, relative: str) -> bool: + try: + (root / relative).resolve().relative_to(root.resolve()) + except (OSError, RuntimeError, ValueError): + return False + return True diff --git a/tests/test_power_artifacts.py b/tests/test_power_artifacts.py new file mode 100644 index 000000000..db55e273b --- /dev/null +++ b/tests/test_power_artifacts.py @@ -0,0 +1,858 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Raw artifact contract for the dcgm-power telemetry provider.""" + +import csv +import json +import os +from contextlib import suppress +from pathlib import Path + +import pytest + +import srtctl.core.power.parser as power_parser +from srtctl.core.power.contract import ( + MANIFEST_FILENAME, + SAMPLES_FILENAME, + SAMPLES_HEADER, + SCHEMA_VERSION, + WINDOWS_DIRNAME, + Reason, + atomic_write_json, + is_safe_relative_subpath, +) +from srtctl.core.power.manifest import ( + DcgmExporterIdentity, + ExpectedWindow, + PowerManifest, +) +from srtctl.core.power.parser import parse_power_scrape +from srtctl.core.power.samples import ( + ObservedDevice, + SampleRow, + SampleWriter, + derive_observed_devices, + read_samples, +) +from srtctl.core.power.topology import ( + ExpectedDevice, + build_expected_devices, + resolve_het_groups, + resolve_roles, + validate_devices, +) +from srtctl.core.power.windows import convert_running_windows, validate_expected_windows +from srtctl.core.topology import Process + + +def _process(node, gpus, mode, index=0, node_rank=0, het_group=None): + return Process( + node=node, + gpu_indices=frozenset(gpus), + sys_port=8080, + http_port=30000, + endpoint_mode=mode, + endpoint_index=index, + node_rank=node_rank, + het_group=het_group, + ) + + +def _metric(gpu, uuid, value, **labels): + label_text = ",".join( + [f'gpu="{gpu}"', f'UUID="{uuid}"', *[f'{k}="{v}"' for k, v in labels.items()]], + ) + return f"DCGM_FI_DEV_POWER_USAGE{{{label_text}}} {value}" + + +PREAMBLE = "# HELP DCGM_FI_DEV_POWER_USAGE Power draw (in W).\n# TYPE DCGM_FI_DEV_POWER_USAGE gauge\n" + + +def _scrape(*metrics, preamble=PREAMBLE): + return preamble + "\n".join(metrics) + "\n" + + +class TestDcgmParser: + """Strict DCGM power parsing.""" + + def test_parses_power_and_ignores_unrelated_metrics(self): + text = _scrape( + _metric(0, "GPU-aaa", 412.5, device="nvidia0", modelName="NVIDIA GB200"), + _metric(1, "GPU-bbb", 98.25, device="nvidia1", modelName="NVIDIA GB200"), + ) + text += '# TYPE DCGM_FI_DEV_GPU_TEMP gauge\nDCGM_FI_DEV_GPU_TEMP{gpu="0",UUID="GPU-aaa"} 42\n' + + scrape = parse_power_scrape(text) + + assert [(r.gpu_index, r.gpu_uuid, r.power_w) for r in scrape.readings] == [ + (0, "GPU-aaa", 412.5), + (1, "GPU-bbb", 98.25), + ] + assert scrape.reason_codes == () + + def test_label_order_and_escapes_do_not_change_identity(self): + text = ( + PREAMBLE + 'DCGM_FI_DEV_POWER_USAGE{modelName="NVIDIA GB200 \\"Grace\\"",UUID="GPU-aaa",' + 'Hostname="",gpu="3",device="nvidia3"} 100.0\n' + ) + + scrape = parse_power_scrape(text) + + assert [(r.gpu_index, r.gpu_uuid) for r in scrape.readings] == [(3, "GPU-aaa")] + + def test_devices_are_ordered_by_index_regardless_of_emission_order(self): + text = _scrape( + _metric(2, "GPU-ccc", 102.0), + _metric(0, "GPU-aaa", 100.0), + _metric(1, "GPU-bbb", 101.0), + ) + + scrape = parse_power_scrape(text) + + assert [r.gpu_index for r in scrape.readings] == [0, 1, 2] + assert [r.gpu_uuid for r in scrape.readings] == ["GPU-aaa", "GPU-bbb", "GPU-ccc"] + + def test_duplicate_device_metric_omits_row_and_reports(self): + text = _scrape( + _metric(0, "GPU-aaa", 100.0), + _metric(0, "GPU-aaa", 101.0), + _metric(1, "GPU-bbb", 102.0), + ) + + scrape = parse_power_scrape(text) + + assert [r.gpu_index for r in scrape.readings] == [1] + assert Reason.DUPLICATE_POWER_METRIC in scrape.reason_codes + + def test_mig_instance_is_rejected_without_marking_metric_missing(self): + text = _scrape(_metric(0, "GPU-aaa", 100.0, GPU_I_ID="3", GPU_I_PROFILE="1g.10gb")) + + scrape = parse_power_scrape(text) + + assert scrape.readings == () + assert scrape.reason_codes == (Reason.MIG_INSTANCE_UNSUPPORTED,) + + @pytest.mark.parametrize( + ("value", "reason"), + [ + ("NaN", Reason.INVALID_POWER_VALUE), + ("-1.0", Reason.INVALID_POWER_VALUE), + ], + ) + def test_invalid_power_values_are_rejected(self, value, reason): + scrape = parse_power_scrape(_scrape(_metric(0, "GPU-aaa", value))) + + assert scrape.readings == () + assert reason in scrape.reason_codes + + def test_missing_identity_labels_are_rejected(self): + text = PREAMBLE + 'DCGM_FI_DEV_POWER_USAGE{UUID="GPU-aaa"} 100.0\n' + 'DCGM_FI_DEV_POWER_USAGE{gpu="1"} 100.0\n' + + scrape = parse_power_scrape(text) + + assert scrape.readings == () + assert Reason.GPU_INDEX_MISSING in scrape.reason_codes + assert Reason.GPU_UUID_MISSING in scrape.reason_codes + + def test_absent_power_family_reports_missing_metric(self): + scrape = parse_power_scrape('# TYPE DCGM_FI_DEV_GPU_TEMP gauge\nDCGM_FI_DEV_GPU_TEMP{gpu="0"} 42\n') + + assert scrape.readings == () + assert Reason.POWER_METRIC_MISSING in scrape.reason_codes + + @pytest.mark.parametrize( + "error", + [ + ValueError("malformed exposition"), + KeyError("histogram field"), + IndexError("short directive in prometheus-client 0.20"), + ], + ) + def test_exporter_parse_failures_are_classified_without_escaping(self, monkeypatch, error): + def fail(_text): + raise error + + monkeypatch.setattr(power_parser, "text_string_to_metric_families", fail) + + scrape = parse_power_scrape("malformed") + + assert scrape.readings == () + assert scrape.reason_codes == ("endpoint_parse_error",) + + def test_parser_control_flow_exceptions_are_not_hidden(self, monkeypatch): + def fail(_text): + raise KeyboardInterrupt + + monkeypatch.setattr(power_parser, "text_string_to_metric_families", fail) + + with pytest.raises(KeyboardInterrupt): + parse_power_scrape("malformed") + + +class TestExpectedTopology: + """Expected devices derived from srt-slurm backend processes.""" + + def test_empty_device_assignments_are_rejected(self): + with pytest.raises(ValueError, match="at least one assignment"): + ExpectedDevice(hostname="node-a", gpu_index=0, assignments=()) + + def test_aggregated_topology(self): + processes = [_process("node-a", range(4), "agg")] + + devices = build_expected_devices(processes) + + assert [(d.hostname, d.gpu_index) for d in devices] == [("node-a", i) for i in range(4)] + assert {a.worker_role for d in devices for a in d.assignments} == {"agg"} + assert resolve_roles(devices) == ({("node-a", i): "agg" for i in range(4)}, ()) + assert resolve_het_groups(devices) == ({"node-a": None}, ()) + + def test_disaggregated_1p1d_topology(self): + processes = [ + _process("node-a", range(4), "prefill", het_group=0), + _process("node-b", range(4), "decode", het_group=1), + ] + + devices = build_expected_devices(processes) + + roles, role_conflicts = resolve_roles(devices) + groups, group_conflicts = resolve_het_groups(devices) + + assert len(devices) == 8 + assert sorted(roles.values()).count("prefill") == 4 + assert sorted(roles.values()).count("decode") == 4 + assert groups == {"node-a": 0, "node-b": 1} + assert role_conflicts == () and group_conflicts == () + + def test_repeated_same_role_assignments_are_allowed(self): + processes = [ + _process("node-a", [0, 1], "decode", index=0), + _process("node-a", [0, 1], "decode", index=1), + ] + + devices = build_expected_devices(processes) + + assert len(devices) == 2 + assert len(devices[0].assignments) == 2 + assert resolve_roles(devices)[1] == () + + def test_conflicting_roles_on_one_device_are_invalid(self): + processes = [ + _process("node-a", [0], "prefill"), + _process("node-a", [0], "decode"), + ] + + roles, conflicts = resolve_roles(build_expected_devices(processes)) + + assert roles == {} + assert Reason.CONFLICTING_WORKER_ROLES in conflicts + + def test_conflicting_het_groups_on_one_node_are_invalid(self): + processes = [ + _process("node-a", [0], "prefill", het_group=0), + _process("node-a", [1], "decode", het_group=1), + ] + + groups, conflicts = resolve_het_groups(build_expected_devices(processes)) + + assert groups == {} + assert Reason.CONFLICTING_HET_GROUPS in conflicts + + +class TestSampleArtifact: + """samples.csv round trip.""" + + def test_header_constant_is_pinned(self): + """Writer and reader both consume the constant, so pin it literally.""" + assert SAMPLES_HEADER == ( + "schema_version", + "timestamp_unix", + "scrape_seq", + "hostname", + "gpu_index", + "gpu_uuid", + "power_w", + ) + + def test_round_trip_preserves_rows_and_derives_devices(self, tmp_path): + path = tmp_path / SAMPLES_FILENAME + writer = SampleWriter(path) + writer.append( + [ + SampleRow(1000.0, 0, "node-a", 0, "GPU-aaa", 400.0), + SampleRow(1000.1, 0, "node-b", 0, "GPU-bbb", 401.0), + ] + ) + writer.append([SampleRow(1001.0, 1, "node-a", 0, "GPU-aaa", 402.0)]) + writer.close() + + rows, reasons = read_samples(path) + + with open(path, newline="") as handle: + assert next(csv.reader(handle)) == list(SAMPLES_HEADER) + assert reasons == () + assert writer.row_count == 3 + assert [row.schema_version for row in rows] == [SCHEMA_VERSION] * 3 + observed = derive_observed_devices(rows) + assert [(d.hostname, d.gpu_index, d.gpu_uuids) for d in observed] == [ + ("node-a", 0, ("GPU-aaa",)), + ("node-b", 0, ("GPU-bbb",)), + ] + assert observed[0].first_sample_time_unix == 1000.0 + assert observed[0].last_sample_time_unix == 1001.0 + + def test_writer_closes_handle_when_header_write_fails(self, tmp_path, monkeypatch): + opened = [] + + def tracking_open(*args, **kwargs): + handle = open(*args, **kwargs) # noqa: SIM115 - retain the handle to assert explicit cleanup + opened.append(handle) + return handle + + class FailingWriter: + def writerow(self, _row): + raise OSError("disk full") + + monkeypatch.setattr("srtctl.core.power.samples.open", tracking_open, raising=False) + monkeypatch.setattr("srtctl.core.power.samples.csv.writer", lambda _handle: FailingWriter()) + + with pytest.raises(OSError, match="disk full"): + SampleWriter(tmp_path / SAMPLES_FILENAME) + + assert len(opened) == 1 + assert opened[0].closed is True + + def test_uuid_change_is_retained_and_flagged(self, tmp_path): + path = tmp_path / SAMPLES_FILENAME + writer = SampleWriter(path) + writer.append( + [ + SampleRow(1000.0, 0, "node-a", 0, "GPU-aaa", 400.0), + SampleRow(1001.0, 1, "node-a", 0, "GPU-zzz", 400.0), + ] + ) + writer.close() + + observed = derive_observed_devices(read_samples(path)[0]) + + assert observed[0].gpu_uuids == ("GPU-aaa", "GPU-zzz") + + @pytest.mark.parametrize( + ("bad_row", "reason"), + [ + ("1,1000.0,0,node-a,0,GPU-aaa,not-a-number", Reason.SAMPLES_CSV_MALFORMED), + ("1,1000.0,0,node-a,0,GPU-aaa,NaN", Reason.SAMPLES_CSV_MALFORMED), + ("1,1000.0,0,node-a,0,GPU-aaa,-5", Reason.SAMPLES_CSV_MALFORMED), + ("1,1000.0,0,node-a,0,GPU-aaa", Reason.SAMPLES_CSV_MALFORMED), + ], + ) + def test_malformed_rows_are_reported(self, tmp_path, bad_row, reason): + path = tmp_path / SAMPLES_FILENAME + path.write_text(",".join(SAMPLES_HEADER) + "\n" + bad_row + "\n") + + rows, reasons = read_samples(path) + + assert rows == () + assert reason in reasons + + def test_duplicate_row_key_is_reported(self, tmp_path): + path = tmp_path / SAMPLES_FILENAME + path.write_text( + ",".join(SAMPLES_HEADER) + "\n1,1000.0,0,node-a,0,GPU-aaa,400.0\n1,1000.5,0,node-a,0,GPU-aaa,401.0\n" + ) + + _, reasons = read_samples(path) + + assert Reason.DUPLICATE_SAMPLE_ROW in reasons + + def test_non_monotonic_device_timestamps_are_reported(self, tmp_path): + path = tmp_path / SAMPLES_FILENAME + path.write_text( + ",".join(SAMPLES_HEADER) + "\n1,1001.0,0,node-a,0,GPU-aaa,400.0\n1,1000.0,1,node-a,0,GPU-aaa,401.0\n" + ) + + _, reasons = read_samples(path) + + assert Reason.TIMESTAMP_NON_MONOTONIC in reasons + + def test_equal_device_timestamps_are_allowed(self, tmp_path): + path = tmp_path / SAMPLES_FILENAME + path.write_text( + ",".join(SAMPLES_HEADER) + "\n1,1000.0,0,node-a,0,GPU-aaa,400.0\n1,1000.0,1,node-a,0,GPU-aaa,401.0\n" + ) + + _, reasons = read_samples(path) + + assert Reason.TIMESTAMP_NON_MONOTONIC not in reasons + + def test_invalid_utf8_bytes_are_reported(self, tmp_path): + path = tmp_path / SAMPLES_FILENAME + path.write_bytes(",".join(SAMPLES_HEADER).encode() + b"\n1,1000.0,0,node-\xff\xfe,0,GPU-aaa,400.0\n") + + rows, reasons = read_samples(path) + + assert rows == () + assert Reason.SAMPLES_CSV_MALFORMED in reasons + + def test_oversized_field_is_reported(self, tmp_path): + path = tmp_path / SAMPLES_FILENAME + giant = "x" * (csv.field_size_limit() + 1) + path.write_text(",".join(SAMPLES_HEADER) + f"\n1,1000.0,0,{giant},0,GPU-aaa,400.0\n") + + rows, reasons = read_samples(path) + + assert rows == () + assert Reason.SAMPLES_CSV_MALFORMED in reasons + + def test_header_mismatch_and_missing_file_are_reported(self, tmp_path): + wrong = tmp_path / "wrong.csv" + wrong.write_text("timestamp,power\n") + + assert Reason.SAMPLES_CSV_HEADER_MISMATCH in read_samples(wrong)[1] + assert Reason.SAMPLES_CSV_MISSING in read_samples(tmp_path / "nope.csv")[1] + + +class TestDeviceValidation: + """Identity and topology gates.""" + + def _expected(self): + return build_expected_devices( + [ + _process("node-a", [0, 1], "prefill", het_group=0), + _process("node-b", [0, 1], "decode", het_group=1), + ] + ) + + def _rows(self): + return [ + SampleRow(1000.0, 0, "node-a", 0, "GPU-a0", 400.0), + SampleRow(1000.0, 0, "node-a", 1, "GPU-a1", 400.0), + SampleRow(1000.0, 0, "node-b", 0, "GPU-b0", 400.0), + SampleRow(1000.0, 0, "node-b", 1, "GPU-b1", 400.0), + ] + + def test_matching_topology_is_valid(self): + result = validate_devices(self._expected(), derive_observed_devices(self._rows())) + + assert result.valid is True + assert result.reason_codes == () + + def test_missing_device_invalidates(self): + observed = derive_observed_devices(self._rows()[:3]) + + result = validate_devices(self._expected(), observed) + + assert result.valid is False + assert Reason.EXPECTED_DEVICE_MISSING in result.reason_codes + + def test_unexpected_device_invalidates(self): + rows = [*self._rows(), SampleRow(1000.0, 0, "node-c", 0, "GPU-c0", 400.0)] + + result = validate_devices(self._expected(), derive_observed_devices(rows)) + + assert result.valid is False + assert Reason.UNEXPECTED_DEVICE in result.reason_codes + + def test_changed_uuid_invalidates(self): + rows = [*self._rows(), SampleRow(1001.0, 1, "node-a", 0, "GPU-swapped", 400.0)] + + result = validate_devices(self._expected(), derive_observed_devices(rows)) + + assert result.valid is False + assert Reason.GPU_UUID_CHANGED in result.reason_codes + + def test_one_uuid_under_two_device_keys_invalidates(self): + """An endpoint misroute must not let one physical GPU be counted twice.""" + rows = self._rows()[:3] + rows.append(SampleRow(1000.0, 0, "node-b", 1, "GPU-a0", 400.0)) + + result = validate_devices(self._expected(), derive_observed_devices(rows)) + + assert result.valid is False + assert Reason.GPU_UUID_CHANGED in result.reason_codes + + def test_conflicting_role_invalidates(self): + expected = build_expected_devices([_process("node-a", [0], "prefill"), _process("node-a", [0], "decode")]) + observed = derive_observed_devices([SampleRow(1000.0, 0, "node-a", 0, "GPU-a0", 400.0)]) + + result = validate_devices(expected, observed) + + assert result.valid is False + assert Reason.CONFLICTING_WORKER_ROLES in result.reason_codes + + def test_empty_expected_set_is_invalid(self): + result = validate_devices([], []) + + assert result.valid is False + assert Reason.EXPECTED_DEVICE_MISSING in result.reason_codes + + +class TestManifest: + """manifest.json shape.""" + + def _manifest(self): + return PowerManifest( + job_id="12345", + run_name="recipe_12345", + sample_interval_seconds=1.0, + request_timeout_seconds=2.0, + required=True, + started_at_unix=1785168000.0, + producer_git_commit="abcdef0123456789abcdef0123456789abcdef01", + dcgm_exporter=DcgmExporterIdentity( + container_image_resolved="/containers/dcgm-exporter.sqsh", + container_image_sha256="0" * 64, + port=9401, + command="dcgm-exporter --collect-interval=100 --address :9401", + ), + expected_devices=build_expected_devices([_process("node-a", [0], "agg")]), + expected_windows=[ExpectedWindow(benchmark_type="sa-bench", concurrency=4)], + ) + + def test_starting_manifest_shape(self, tmp_path): + payload = self._manifest().to_dict() + + assert payload["schema_version"] == SCHEMA_VERSION + assert payload["producer"] == "srt-slurm.dcgm-power" + assert payload["source_metric"] == "DCGM_FI_DEV_POWER_USAGE" + assert payload["unit"] == "W" + assert payload["timestamp_source"] == "head_node_unix_clock" + assert payload["status"] == "starting" + assert payload["publication_valid"] is None + assert payload["stopped_at_unix"] is None + assert payload["max_scrape_duration_seconds"] is None + assert payload["scrape_count"] == 0 + assert payload["sample_row_count"] == 0 + assert payload["expected_windows"] == [{"benchmark_type": "sa-bench", "concurrency": 4}] + assert payload["expected_devices"] == [ + { + "hostname": "node-a", + "gpu_index": 0, + "assignments": [{"worker_role": "agg", "worker_index": 0, "worker_process": 0, "het_group": None}], + } + ] + + atomic_write_json(tmp_path / MANIFEST_FILENAME, payload) + assert json.loads((tmp_path / MANIFEST_FILENAME).read_text()) == payload + + def test_failed_startup_manifest_is_never_publishable(self): + manifest = self._manifest() + manifest.mark_terminal(status="failed", stopped_at_unix=1785168010.0, publication_valid=True) + + payload = manifest.to_dict() + + assert payload["status"] == "failed" + assert payload["publication_valid"] is False + + def test_terminal_manifest_cannot_be_changed_by_a_second_call(self): + manifest = self._manifest() + manifest.mark_terminal(status="complete", stopped_at_unix=1785168010.0, publication_valid=True) + first_terminal_state = (manifest.status, manifest.stopped_at_unix, manifest.publication_valid) + + with pytest.raises(RuntimeError, match="already terminal"): + manifest.mark_terminal(status="failed", stopped_at_unix=1785168020.0, publication_valid=False) + + assert (manifest.status, manifest.stopped_at_unix, manifest.publication_valid) == first_terminal_state + + def test_terminal_guard_survives_direct_status_reassignment(self): + manifest = self._manifest() + manifest.mark_terminal(status="complete", stopped_at_unix=1785168010.0, publication_valid=True) + first_terminal_evidence = (manifest.stopped_at_unix, manifest.publication_valid) + manifest.status = "running" + + with pytest.raises(RuntimeError, match="already terminal"): + manifest.mark_terminal(status="failed", stopped_at_unix=1785168020.0, publication_valid=False) + + assert (manifest.stopped_at_unix, manifest.publication_valid) == first_terminal_evidence + + def test_atomic_write_leaves_no_partial_file(self, tmp_path): + target = tmp_path / MANIFEST_FILENAME + atomic_write_json(target, {"a": 1}) + atomic_write_json(target, {"a": 2}) + + assert json.loads(target.read_text()) == {"a": 2} + assert sorted(p.name for p in tmp_path.iterdir()) == [MANIFEST_FILENAME] + + def test_atomic_write_closes_raw_fd_when_fdopen_fails(self, tmp_path, monkeypatch): + captured = {} + + def fail(fd, *_args, **_kwargs): + captured["fd"] = fd + raise OSError("too many open files") + + monkeypatch.setattr("srtctl.core.power.contract.os.fdopen", fail) + try: + with pytest.raises(OSError, match="too many open files"): + atomic_write_json(tmp_path / MANIFEST_FILENAME, {"a": 1}) + + with pytest.raises(OSError): + os.fstat(captured["fd"]) + assert list(tmp_path.iterdir()) == [] + finally: + with suppress(OSError): + os.close(captured["fd"]) + + +class TestMeasurementWindowArtifacts: + @staticmethod + def _write_completed_window(tmp_path, *, start, end, duration): + windows_dir = tmp_path / WINDOWS_DIRNAME + windows_dir.mkdir(exist_ok=True) + payload = { + "schema_version": SCHEMA_VERSION, + "benchmark_type": "sa-bench", + "result_path": "result.json", + "concurrency": 4, + "benchmark_start_time_unix": start, + "benchmark_end_time_unix": end, + "duration": duration, + "clock_source": "head_node_unix_clock", + "status": "completed", + "reason": None, + } + (windows_dir / "result.json").write_text(json.dumps(payload)) + (tmp_path / "result.json").write_text(json.dumps(payload)) + + @staticmethod + def _validate(tmp_path, *, expected_device_keys=None, observed_devices=()): + errors = [] + rows = validate_expected_windows( + power_dir=tmp_path, + result_root=tmp_path, + expected_windows=[ExpectedWindow("sa-bench", 4)], + expected_device_keys=set() if expected_device_keys is None else expected_device_keys, + observed_devices=observed_devices, + artifact_errors=errors, + ) + return rows[0], errors + + def test_unreadable_windows_directory_is_reported(self, tmp_path, monkeypatch): + windows_dir = tmp_path / WINDOWS_DIRNAME + windows_dir.mkdir() + original_iterdir = Path.iterdir + + def fail_for_windows(path): + if path == windows_dir: + raise PermissionError("permission denied") + return original_iterdir(path) + + monkeypatch.setattr(Path, "iterdir", fail_for_windows) + row, errors = self._validate(tmp_path) + + assert row.reason_codes == (Reason.MEASUREMENT_WINDOW_MISSING,) + assert [(error.path, error.reason_codes) for error in errors] == [ + (WINDOWS_DIRNAME, (Reason.MEASUREMENT_WINDOW_ARTIFACT_PATH_INVALID,)) + ] + + def test_unreadable_windows_directory_does_not_break_interruption_cleanup(self, tmp_path, monkeypatch): + windows_dir = tmp_path / WINDOWS_DIRNAME + windows_dir.mkdir() + original_glob = Path.glob + + def fail_for_windows(path, pattern): + if path == windows_dir: + raise PermissionError("permission denied") + return original_glob(path, pattern) + + monkeypatch.setattr(Path, "glob", fail_for_windows) + + assert convert_running_windows(windows_dir, reason="interrupted") == 0 + + def test_interruption_cleanup_rejects_windows_directory_symlink_escape(self, tmp_path): + power_dir = tmp_path / "power" + power_dir.mkdir() + external_dir = tmp_path / "external" + external_dir.mkdir() + victim = external_dir / "victim.json" + victim.write_text(json.dumps({"status": "running", "sentinel": "unchanged"})) + windows_dir = power_dir / WINDOWS_DIRNAME + windows_dir.symlink_to(external_dir, target_is_directory=True) + + original = victim.read_text() + converted = convert_running_windows(windows_dir, reason="interrupted") + + assert converted == 0 + assert victim.read_text() == original + + def test_interruption_cleanup_converts_running_window(self, tmp_path): + windows_dir = tmp_path / WINDOWS_DIRNAME + windows_dir.mkdir() + window_path = windows_dir / "result.json" + window_path.write_text( + json.dumps( + { + "status": "running", + "benchmark_end_time_unix": 1002.0, + "duration": 2.0, + "reason": None, + } + ) + ) + + converted = convert_running_windows(windows_dir, reason="benchmark child terminated") + + payload = json.loads(window_path.read_text()) + assert converted == 1 + assert payload["status"] == "interrupted" + assert payload["benchmark_end_time_unix"] is None + assert payload["duration"] is None + assert payload["reason"] == "benchmark child terminated" + + def test_unresolvable_windows_directory_is_reported(self, tmp_path, monkeypatch): + windows_dir = tmp_path / WINDOWS_DIRNAME + windows_dir.mkdir() + original_resolve = Path.resolve + + def fail_for_windows(path, *args, **kwargs): + if path == windows_dir: + raise RuntimeError("symlink loop") + return original_resolve(path, *args, **kwargs) + + monkeypatch.setattr(Path, "resolve", fail_for_windows) + row, errors = self._validate(tmp_path) + + assert row.reason_codes == (Reason.MEASUREMENT_WINDOW_MISSING,) + assert [(error.path, error.reason_codes) for error in errors] == [ + (WINDOWS_DIRNAME, (Reason.MEASUREMENT_WINDOW_ARTIFACT_PATH_INVALID,)) + ] + + def test_non_json_window_file_is_rejected(self, tmp_path): + self._write_completed_window(tmp_path, start=1000.0, end=1001.0, duration=1.0) + window_path = tmp_path / WINDOWS_DIRNAME / "result.json" + window_path.rename(window_path.with_suffix(".txt")) + + row, errors = self._validate(tmp_path) + + assert row.reason_codes == (Reason.MEASUREMENT_WINDOW_MISSING,) + assert [(error.path, error.reason_codes) for error in errors] == [ + (f"{WINDOWS_DIRNAME}/result.txt", (Reason.MEASUREMENT_WINDOW_ARTIFACT_PATH_INVALID,)) + ] + + @pytest.mark.parametrize( + ("start", "end"), + [ + (1000.1, 1000.0), + (1000.0, 1000.0), + ], + ) + def test_non_positive_wall_clock_interval_is_rejected(self, tmp_path, start, end): + self._write_completed_window(tmp_path, start=start, end=end, duration=0.1) + row, _ = self._validate(tmp_path) + + assert row.power_coverage_valid is False + assert row.reason_codes == (Reason.MEASUREMENT_WINDOW_CLOCK_MISMATCH,) + + @pytest.mark.parametrize( + ("field", "boolean_value"), + [ + ("benchmark_start_time_unix", False), + ("benchmark_end_time_unix", True), + ("duration", True), + ], + ) + def test_result_timing_fields_reject_boolean_numbers(self, tmp_path, field, boolean_value): + self._write_completed_window(tmp_path, start=0.0, end=1.0, duration=1.0) + result_path = tmp_path / "result.json" + result = json.loads(result_path.read_text()) + result[field] = boolean_value + result_path.write_text(json.dumps(result)) + + row, _ = self._validate(tmp_path) + + assert row.power_coverage_valid is False + assert row.reason_codes == (Reason.MEASUREMENT_WINDOW_RESULT_MISMATCH,) + + def test_computed_device_gaps_are_retained_when_coverage_is_invalid(self, tmp_path): + self._write_completed_window(tmp_path, start=1000.0, end=1004.0, duration=4.0) + observed = derive_observed_devices( + [ + SampleRow(999.0, 0, "node-a", 0, "GPU-a", 400.0), + SampleRow(1005.0, 1, "node-a", 0, "GPU-a", 400.0), + ] + ) + + row, _ = self._validate( + tmp_path, + expected_device_keys={("node-a", 0), ("node-b", 0)}, + observed_devices=observed, + ) + + assert row.power_coverage_valid is False + assert row.reason_codes == ( + Reason.SAMPLE_GAP_EXCEEDED, + Reason.MEASUREMENT_WINDOW_NOT_BRACKETED, + ) + assert row.per_device_max_sample_gap_seconds == {"node-a/GPU-a": 6.0} + + def test_window_coverage_does_not_depend_on_sample_time_order(self, tmp_path): + self._write_completed_window(tmp_path, start=1000.0, end=1002.0, duration=2.0) + observed = [ + ObservedDevice( + hostname="node-a", + gpu_index=0, + gpu_uuids=("GPU-a",), + first_sample_time_unix=999.0, + last_sample_time_unix=1003.0, + sample_times=(1000.0, 1003.0, 999.0, 1002.0), + ) + ] + + row, _ = self._validate( + tmp_path, + expected_device_keys={("node-a", 0)}, + observed_devices=observed, + ) + + assert row.power_coverage_valid is True + assert row.reason_codes == () + assert row.per_device_max_sample_gap_seconds == {"node-a/GPU-a": 2.0} + + def test_three_duplicate_windows_are_each_recorded_once(self, tmp_path): + windows_dir = tmp_path / WINDOWS_DIRNAME + windows_dir.mkdir() + for index in range(3): + stem = f"result-{index}" + (windows_dir / f"{stem}.json").write_text( + json.dumps( + { + "schema_version": 1, + "benchmark_type": "sa-bench", + "result_path": f"{stem}.json", + "concurrency": 4, + "benchmark_start_time_unix": 1000.0, + "benchmark_end_time_unix": 1020.0, + "duration": 20.0, + "clock_source": "head_node_unix_clock", + "status": "completed", + "reason": None, + } + ) + ) + errors = [] + + rows = validate_expected_windows( + power_dir=tmp_path, + result_root=tmp_path, + expected_windows=[ExpectedWindow("sa-bench", 4)], + expected_device_keys=set(), + observed_devices=[], + artifact_errors=errors, + ) + + assert rows[0].reason_codes == (Reason.MEASUREMENT_WINDOW_DUPLICATE,) + assert [error.path for error in errors] == [ + f"{WINDOWS_DIRNAME}/result-0.json", + f"{WINDOWS_DIRNAME}/result-1.json", + f"{WINDOWS_DIRNAME}/result-2.json", + ] + assert all(error.reason_codes == (Reason.MEASUREMENT_WINDOW_DUPLICATE,) for error in errors) + + +class TestStorageSubdirSafety: + @pytest.mark.parametrize("value", ["power", "telemetry/power", "a/b/c"]) + def test_safe_paths_accepted(self, value): + assert is_safe_relative_subpath(value) is True + + @pytest.mark.parametrize("value", ["", "/power", "../power", "power/../..", "a/../../b", "~/power"]) + def test_unsafe_paths_rejected(self, value): + assert is_safe_relative_subpath(value) is False