DRAFT: feat(power): vendor-agnostic power/energy monitoring

docs/POWER_MONITORING.md

@@ -0,0 +1,132 @@
+# Power Monitoring
+
+First-class, **vendor-neutral** power/energy telemetry collected _during_ a
+benchmark run, windowed to the performance phase, and written to a sibling
+`power.json` (plus a section in `report.txt`). The `Report` /
+`result_summary.json` are never mutated — power is a separate artifact, exactly
+like profiling.
+
+## Design: a general framework + pluggable sources
+
+The core (`settings.power`) knows nothing about GPUs, Prometheus, or any vendor.
+A **source** is a small registered builder that turns the config into a sidecar
+command + a parse spec. `nvidia_smi` is just the **first reference source**;
+everything vendor-specific lives inside its builder, not in the schema.
+
+```
+setup → [start power sidecar] → warmup → PERFORMANCE phase → teardown → [stop sidecar]
+                                              └── trace sliced to this window ──┘
+```
+
+The sidecar is best-effort: a broken collector writes `status:"failed"` and the
+benchmark completes normally. It can never fail or perturb the run.
+
+## Config
+
+The vendor-neutral core is tiny; per-source settings go in `options` (each
+source documents its keys):
+
+```yaml
+settings:
+  power:
+    source: nvidia_smi # nvidia_smi | prometheus | command | <custom plugin>
+    interval_s: 1.0 # sampling interval (default 1.0)
+    options: {} # source-specific settings
+    # advanced (hidden, sane defaults): value_kind, env
+```
+
+When `source` is unset the whole block is omitted from the serialized config —
+disabling power leaves zero footprint.
+
+### Built-in sources
+
+```yaml
+# NVIDIA GPUs on the load-gen host (the reference source)
+power:
+  source: nvidia_smi
+  options: { gpu_indices: [0, 1, 2, 3] }   # optional; default = all GPUs
+
+# Remote GPUs / cluster: scrape a server-side exporter (DCGM, etc.)
+power:
+  source: prometheus
+  options:
+    url: "http://gpu-node:9400"
+    query: "DCGM_FI_DEV_POWER_USAGE"       # watts; one series per GPU/instance
+
+# Anything else, zero Python: a program that prints canonical JSONL
+power:
+  source: command
+  options:
+    argv: ["ssh", "gpu-node", "my-power-reader", "--interval", "1"]
+```
+
+The `command` argv must print **one JSONL sample per line**:
+`{"ts": <epoch_s>, "value": <float>, "label": "<series>"}`.
+
+## Adding your own source (plugin)
+
+No core edits, no entry-points — register a builder and pass it settings via
+`options`:
+
+```python
+from inference_endpoint.power import power_source, ResolvedSource
+
+@power_source("redfish")
+def _build(cfg):
+    return ResolvedSource(
+        argv=["redfish-power", cfg.options["bmc"], "--interval", str(cfg.interval_s)],
+        fmt="jsonl", value_kind="power_w",
+        ts_field="ts", value_field="value", label_field="label", csv_header=False,
+    )
+```
+
+```yaml
+power: { source: redfish, options: { bmc: "10.0.0.5" } }
+```
+
+(For non-Python sources, prefer the built-in `command` source — the process
+boundary is the plugin API.)
+
+## Output: `power.json`
+
+```json
+{
+  "schema_version": "1.0",
+  "status": "ok",
+  "window": {"start_epoch_s": ..., "end_epoch_s": ..., "duration_s": 305.0, "basis": "performance_phase"},
+  "totals": {
+    "energy_j": 1045212.2,
+    "mean_power_w": 3426.9,
+    "output_tokens": 4800915,
+    "energy_per_output_token_j": 0.2177,
+    "consistent_with_window": true
+  },
+  "sources": [
+    {"label": "gpu0", "value_kind": "power_w", "energy_j": 130651.5,
+     "power_w": {"mean": 428.4, "p50": 410.0, "p95": 658.0, "max": 879.1}, "sample_count": 305}
+  ],
+  "provenance": {"command": [...], "interval_s": 1.0, "samples_in_window": 305, "samples_dropped": 0}
+}
+```
+
+### Energy-per-output-token consistency
+
+`energy_per_output_token_j` is emitted **only** when the token count and the
+energy share the same window (single tracked performance phase). Otherwise it is
+`null` with an explanatory note — preventing a windowed-energy ÷ global-tokens
+mismatch.
+
+## Remote GPUs / multi-node
+
+The sidecar runs **where its command runs** (the load-gen host). For remote
+GPUs, either point `prometheus` at a server-side exporter (recommended) or use a
+`command` source. For an ssh-wrapped command, use `ssh -tt` so the remote
+sampler receives the stop signal. Clocks are assumed NTP-synced.
+
+## Caveats / scope
+
+- `power_w` energy uses a trapezoid integral with no edge interpolation
+  (≤ one sample-interval bias at each window edge — negligible at 1 s / 300 s+).
+- The window is the full performance phase (includes drain).
+- v1 collects a single source; multi-source totals / scope-aware double-count
+  guards are future work.

examples/11_Power_Monitoring/online_with_power.yaml

@@ -0,0 +1,47 @@
+# Online benchmark with vendor-agnostic power monitoring.
+# See docs/POWER_MONITORING.md. Power telemetry is collected during the
+# performance phase and written to <report_dir>/power.json + a section in
+# report.txt. A broken collector never fails the benchmark.
+name: "online-power-example"
+version: "1.0"
+type: "online"
+
+model_params:
+  name: "meta-llama/Llama-3.1-8B-Instruct"
+  temperature: 0.0
+  max_new_tokens: 128
+
+datasets:
+  - name: cnn_dailymail::llama3_8b
+    type: "performance"
+    samples: 2000
+    parser:
+      input: prompt
+
+settings:
+  runtime:
+    min_duration_ms: 300000
+    max_duration_ms: 1800000
+
+  load_pattern:
+    type: "concurrency"
+    target_concurrency: 64
+
+  client:
+    num_workers: 4
+
+  # --- Power monitoring -----------------------------------------------------
+  # Built-in nvidia-smi sidecar on the load-generator host. For a remote GPU
+  # node, switch to the `prometheus` source (DCGM exporter) or an ssh-wrapped
+  # `command` source — see docs/POWER_MONITORING.md.
+  power:
+    source: nvidia_smi
+    interval_s: 1.0
+    # options: { gpu_indices: [0, 1, 2, 3] }   # default: all visible GPUs
+
+endpoint_config:
+  endpoints:
+    - "http://localhost:8000"
+  api_key: null
+
+report_dir: logs/power_example

src/inference_endpoint/commands/benchmark/execute.py

@@ -100,6 +100,13 @@
     SessionResult,
 )
 from inference_endpoint.metrics.report import Report
+from inference_endpoint.power import (
+    PowerCollector,
+    build_power_report,
+    write_power_section,
+)
+from inference_endpoint.power.window import disabled_report
+from inference_endpoint.utils import monotime_to_datetime
 
 transformers_logging.set_verbosity_error()
 
@@ -148,6 +155,9 @@ class BenchmarkResult:
     # settings.profiling.engine is set; None otherwise. Rendered into
     # report.txt and a sibling profiling.json by finalize_benchmark.
     profiling: dict[str, Any] | None = None
+    # Power-telemetry sidecar when settings.power.source is set; None otherwise.
+    # finalize_benchmark windows its trace and writes a sibling power.json.
+    power: PowerCollector | None = None
 
 
 @dataclass
@@ -896,6 +906,15 @@ def _on_phase_start(phase: PhaseConfig) -> None:
                     )
                 profile_starts.append(rec)
 
+        # Power telemetry sidecar (best-effort; never fails the run). Started
+        # before session.run so it is warm; finalize_benchmark windows its
+        # trace to the performance phase.
+        power_cfg = config.settings.power
+        power_collector: PowerCollector | None = None
+        if power_cfg.source is not None:
+            power_collector = PowerCollector(power_cfg, ctx.report_dir / "power")
+            power_collector.start()
+
         loop.add_signal_handler(signal.SIGINT, session.stop)
         try:
             result = await session.run(phases, on_phase_start=_on_phase_start)
@@ -926,6 +945,9 @@ def _on_phase_start(phase: PhaseConfig) -> None:
                             rec["error"] or rec["status"],
                         )
                     profile_stops.append(rec)
+            # Stop the power sidecar (best-effort; bounded SIGTERM->SIGKILL).
+            if power_collector is not None:
+                power_collector.stop()
             logger.info("Cleaning up...")
             try:
                 if http_client:
@@ -1014,6 +1036,7 @@ def _on_phase_start(phase: PhaseConfig) -> None:
         report=report,
         tmpfs_dir=tmpfs_dir,
         profiling=profiling_payload,
+        power=power_collector,
     )
 
 
@@ -1066,13 +1089,60 @@ def _salvage_tmpfs(report_dir: Path, tmpfs_dir: Path) -> None:
         logger.debug(f"Copied {src_events} -> {dst_events}")
 
 
+def _build_power_payload(
+    ctx: BenchmarkContext, bench: BenchmarkResult
+) -> dict[str, Any] | None:
+    """Window the power trace to the performance phase and build the JSON payload.
+
+    Best-effort: any failure yields a ``status:"failed"`` payload, never an
+    exception that would abort finalization.
+    """
+    pc = bench.power
+    if pc is None:
+        return None
+    try:
+        if pc.resolved is None:
+            return disabled_report(pc.error or "power collector failed to start")
+        perf = bench.session.perf_results
+        if not perf:
+            return disabled_report("no performance phase to window power against")
+        # Phase windows are monotonic_ns; convert to wall-clock epoch (the clock
+        # power samples are stamped in) via the shared monotonic->wall anchor.
+        start_epoch = monotime_to_datetime(perf[0].start_time_ns).timestamp()
+        end_epoch = monotime_to_datetime(perf[-1].end_time_ns).timestamp()
+        output_tokens: int | None = None
+        if bench.report is not None:
+            osl = bench.report.output_sequence_lengths or {}
+            total = osl.get("total")
+            output_tokens = int(total) if total else None
+        return build_power_report(
+            resolved=pc.resolved,
+            trace_path=pc.trace_path,
+            window_start_epoch_s=start_epoch,
+            window_end_epoch_s=end_epoch,
+            output_tokens=output_tokens,
+            token_window_basis="performance_phase_tracked",
+            # Report tokens are scoped to the (single) tracked perf phase, so
+            # they share the energy window. Multiple perf phases would span gaps.
+            consistent_with_window=len(perf) == 1,
+            collector_status=pc.status,
+            collector_error=pc.error,
+            interval_s=pc.cfg.interval_s,
+        )
+    except Exception as e:  # noqa: BLE001 — power must never break finalization.
+        logger.warning("power: failed to build report: %s", e)
+        return disabled_report(f"power report build failed: {e}")
+
+
 def finalize_benchmark(ctx: BenchmarkContext, bench: BenchmarkResult) -> None:
     """Score accuracy, aggregate results, write JSON."""
     config = ctx.config
     result = bench.session
     collector = bench.collector
     report = bench.report
 
+    power_payload = _build_power_payload(ctx, bench)
+
     # Display report if available (from MetricsAggregator pub/sub snapshot).
     # result_summary.json is the self-complete machine-readable report (carries
     # qps/tps/seeds via Report.to_json); report.txt is the full human-readable
@@ -1086,6 +1156,8 @@ def finalize_benchmark(ctx: BenchmarkContext, bench: BenchmarkResult) -> None:
             report.display(fn=lambda s: print(s, file=f))
             if bench.profiling is not None:
                 _write_profiling_section(f, bench.profiling)
+            if power_payload is not None:
+                write_power_section(f, power_payload)
         logger.info("Report written to %s", report_txt)
 
     # Sibling profiling.json — kept separate so Report stays a pure
@@ -1095,6 +1167,10 @@ def finalize_benchmark(ctx: BenchmarkContext, bench: BenchmarkResult) -> None:
             json.dumps(bench.profiling, indent=2)
         )
 
+    # Sibling power.json — same rationale: Report stays snapshot-pure.
+    if power_payload is not None:
+        (ctx.report_dir / "power.json").write_text(json.dumps(power_payload, indent=2))
+
     # Write scoring artifacts + copy event log from tmpfs to disk
     _write_scoring_artifacts(ctx, result, bench.tmpfs_dir)
 

src/inference_endpoint/config/schema.py

@@ -702,6 +702,70 @@ def _validate_url_scheme(cls, v: list[str] | None) -> list[str] | None:
         return v
 
 
+@cyclopts.Parameter(name="*")
+class PowerConfig(BaseModel):
+    """Vendor-agnostic power/energy monitoring.
+
+    When ``source`` is set, a sidecar process streams power samples to a trace
+    file during the performance phase; at finalization the trace is sliced to
+    the measurement window, integrated into energy, and written to a sibling
+    ``power.json``. A broken collector never fails or skews the benchmark.
+
+    The sidecar runs *where the command runs* — i.e. the load-generator host.
+    For remote GPUs either point ``prometheus`` at a server-side DCGM exporter
+    or supply an ssh-wrapped ``command``; clocks are assumed NTP-synced.
+    """
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    # Vendor-neutral core. Source-specific settings live in `options`, owned and
+    # validated by the registered source builder (see power/sources.py) — the
+    # schema knows nothing about GPUs, Prometheus, etc.
+    source: Annotated[
+        str | None,
+        cyclopts.Parameter(
+            alias="--power",
+            help="Power source: nvidia_smi | prometheus | command | <custom plugin>",
+        ),
+    ] = Field(
+        None,
+        description="Power telemetry source name; None disables monitoring. "
+        "Built-ins: nvidia_smi, prometheus, command. Register custom sources "
+        "with @power_source('name').",
+    )
+    interval_s: Annotated[
+        float,
+        cyclopts.Parameter(alias="--power-interval"),
+    ] = Field(1.0, gt=0, description="Sampling interval in seconds")
+    options: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Source-specific settings (each source documents its keys). "
+        "e.g. nvidia_smi: {gpu_indices}; prometheus: {url, query}; "
+        "command: {argv}. command argv must print one JSONL sample per line: "
+        '{"ts": <epoch_s>, "value": <float>, "label": "<series>"}.',
+    )
+
+    # --- Advanced (hidden from CLI; rarely needed, sane defaults) ------------
+    value_kind: Annotated[
+        Literal["power_w", "energy_j"], cyclopts.Parameter(show=False)
+    ] = Field(
+        "power_w",
+        description="Sample semantics: instantaneous watts or a cumulative joule counter",
+    )
+    env: Annotated[dict[str, str], cyclopts.Parameter(show=False)] = Field(
+        default_factory=dict, description="Extra environment for the sidecar process"
+    )
+
+    @model_serializer(mode="wrap")
+    def _serialize(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
+        # When disabled (source=None), collapse to an empty mapping so a
+        # turned-off feature leaves no footprint in the serialized config.
+        data = handler(self)
+        if self.source is None:
+            return {}
+        return data
+
+
 @cyclopts.Parameter(name="*")
 class Settings(BaseModel):
     """Test settings."""
@@ -717,6 +781,7 @@ class Settings(BaseModel):
     )
     warmup: WarmupConfig = Field(default_factory=WarmupConfig)
     profiling: ProfilingConfig = Field(default_factory=ProfilingConfig)
+    power: PowerConfig = Field(default_factory=PowerConfig)
 
 
 class OfflineSettings(Settings):