[1/4] feat(power): add the DCGM power artifact data layer

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",

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.
+"""

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

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)),
+        }