Add NVLink profiling fallback

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **NVLink fallback for systems where the aggregate gauge is unavailable**:
+  `DCGM_FI_DEV_NVLINK_BANDWIDTH_TOTAL` (field 449) reports `N/A` on some systems
+  while the NVLink profiling fields `DCGM_FI_PROF_NVLINK_TX_BYTES` and
+  `DCGM_FI_PROF_NVLINK_RX_BYTES` (1011/1012) still carry usable per-direction
+  rates. The direct `dcgm` backend now collects those two fields, and the
+  canonical NVLink aggregate falls back to their sum when 449 is missing. A
+  real `0.0` from 449 means idle and does *not* trigger the fallback, and a
+  one-sided reading is reported as unknown rather than as a silently halved
+  total (#21)
+
 ## [0.5.0] - 2026-07-27
 
 This release replaces the single-file implementation with a layered package.

docs/guide/backends.md

@@ -32,6 +32,10 @@ HTTP endpoint (or reads a saved exposition-format file).
 - **Aggregation:** each sample is a **window** average over the exporter's
   collection interval.
 - **Requires:** a reachable exporter endpoint (`--source`).
+- **NVLink fallback:** the shipped `dcgm-exporter` counters CSVs omit
+  `DCGM_FI_PROF_NVLINK_TX_BYTES` and `DCGM_FI_PROF_NVLINK_RX_BYTES`. Add both
+  fields to the exporter's counters CSV to enable the TX+RX fallback when field
+  449 is unavailable; otherwise field 449 remains the only NVLink source.
 - **Best for:** fleet-level monitoring where an exporter is already deployed.
 
 ```bash

docs/metrics.md

@@ -72,11 +72,20 @@ Memory used % = `100 × FB_USED / total`.
 | PCIe RX | `DCGM_FI_PROF_PCIE_RX_BYTES` | PCIe receive throughput. ≥ 1 GB/s triggers I/O classification. | Classification, Fleet / Focus / Plot views |
 | PCIe TX | `DCGM_FI_PROF_PCIE_TX_BYTES` | PCIe transmit throughput. ≥ 1 GB/s triggers I/O classification. | Classification, Fleet / Focus / Plot views |
 
-## NVLink (bytes, monotonic counter converted to rate)
+## NVLink (throughput gauges)
 
 | Metric | DCGM Field | Description | Used In |
 |--------|-----------|-------------|---------|
-| NVLink BW Total | `DCGM_FI_DEV_NVLINK_BANDWIDTH_TOTAL` | Cumulative NVLink bytes. Dashboard computes Δ/s and displays GB/s. | Fleet / Focus views |
+| NVLink BW Total | `DCGM_FI_DEV_NVLINK_BANDWIDTH_TOTAL` (`449`) | Aggregate throughput gauge in MB/s. Converted directly to bytes/s, without differencing. | Fleet / Focus views |
+| NVLink TX | `DCGM_FI_PROF_NVLINK_TX_BYTES` (`1011`) | Transmit throughput in bytes/s. | Aggregate fallback |
+| NVLink RX | `DCGM_FI_PROF_NVLINK_RX_BYTES` (`1012`) | Receive throughput in bytes/s. | Aggregate fallback |
+
+Field 449 is preferred whenever it has a value, including `0.0` for an idle
+link. When it is unavailable, KempnerPulse sums fields 1011 and 1012 only if
+both directions are present; if either is missing, the canonical aggregate
+remains unknown. The direct `dcgm` backend requests all three fields. The
+Prometheus backend depends on which fields `dcgm-exporter` is configured to
+publish; see [Backends](guide/backends.md).
 
 ## Energy (monotonic counter)
 

docs/schema.md

@@ -97,7 +97,7 @@ local / non-Slurm runs still produce valid records.
 | `gpu_pcie_transmit_throughput_bytes_per_second` | bytes/s (differenced) |
 | `gpu_pcie_receive_throughput_bytes_per_second` | bytes/s (differenced) |
 | `gpu_pcie_replay_count` | count (cumulative) |
-| `gpu_nvlink_aggregate_throughput_bytes_per_second` | bytes/s (differenced) |
+| `gpu_nvlink_aggregate_throughput_bytes_per_second` | bytes/s (gauge; not differenced) |
 | `gpu_board_power_draw_watts` | watts |
 | `gpu_board_total_energy_joules` | joules (cumulative) |
 | `gpu_board_enforced_power_limit_watts` | watts |

src/kempnerpulse/reader/dcgmi.py

@@ -59,6 +59,8 @@
     (1008, "DCGM_FI_PROF_PIPE_FP16_ACTIVE"),
     (1009, "DCGM_FI_PROF_PCIE_TX_BYTES"),           # bytes/sec
     (1010, "DCGM_FI_PROF_PCIE_RX_BYTES"),           # bytes/sec
+    (1011, "DCGM_FI_PROF_NVLINK_TX_BYTES"),         # bytes/sec
+    (1012, "DCGM_FI_PROF_NVLINK_RX_BYTES"),         # bytes/sec
 )
 
 DCGM_DMON_FIELD_IDS = ",".join(str(fid) for fid, _ in DCGM_DMON_FIELDS)
@@ -198,7 +200,7 @@ def run_dmon_once(
 ) -> str:
     """Run one ``dcgmi dmon -c 2`` collection and return its raw stdout text.
 
-    Two samples are requested because profiling fields (1001-1010) return
+    Two samples are requested because profiling fields (1001-1012) return
     ``N/A`` on the first sample of a cold invocation; the valid second tick lets
     a downstream last-non-``None``-wins merge recover the real values.
     """

src/kempnerpulse/translate/translator.py

@@ -18,6 +18,10 @@
 from .mapping import GPU_UUID_LABELS, convert, map_field
 from .schema import SCHEMA_VERSION, CanonicalRecord
 
+NVLINK_AGGREGATE_FIELD = "gpu_nvlink_aggregate_throughput_bytes_per_second"
+NVLINK_TX_FIELD = "DCGM_FI_PROF_NVLINK_TX_BYTES"
+NVLINK_RX_FIELD = "DCGM_FI_PROF_NVLINK_RX_BYTES"
+
 # Static-metadata keys read from SourceContext.slurm_metadata -> record fields.
 _SLURM_FIELDS = (
     "record_slurm_job_id",
@@ -55,6 +59,17 @@ def translate(self, raw: RawRecord) -> Optional[CanonicalRecord]:
             canonical_name, unit_kind = mapped
             canonical[canonical_name] = convert(unit_kind, value)
 
+        # DCGM field 449 is the preferred aggregate NVLink gauge. On some
+        # systems it reports N/A while profiling fields 1011/1012 still expose
+        # usable TX/RX rates, so synthesize the same canonical aggregate.
+        # Both directions are required: summing a one-sided reading would
+        # silently under-report the total rather than admit it is unknown.
+        if canonical.get(NVLINK_AGGREGATE_FIELD) is None:
+            tx = convert("number", raw.fields.get(NVLINK_TX_FIELD))
+            rx = convert("number", raw.fields.get(NVLINK_RX_FIELD))
+            if tx is not None and rx is not None:
+                canonical[NVLINK_AGGREGATE_FIELD] = max(0.0, tx + rx)
+
         # Derived: total framebuffer = used + free + reserved (when all present).
         used = canonical.get("gpu_framebuffer_used_mebibytes")
         free = canonical.get("gpu_framebuffer_free_mebibytes")

tests/fixtures/dcgmi_dmon_2tick.txt

@@ -1,6 +1,6 @@
-#Entity  SMCLK  MMCLK  MTMP  GTMP  POWER  TOTEC      PCIRP  GPUTL  MCUTL  FBFRE  FBUSD  FBRSV  NVLBW  GRACT  SMACT  SMOCC  TENSO  DRACT  FP64A  FP32A  FP16A  PCITX       PCIRX
+#Entity  SMCLK  MMCLK  MTMP  GTMP  POWER  TOTEC      PCIRP  GPUTL  MCUTL  FBFRE  FBUSD  FBRSV  NVLBW  GRACT  SMACT  SMOCC  TENSO  DRACT  FP64A  FP32A  FP16A  PCITX       PCIRX       NVLTX         NVLRX
 ID
-GPU 0    1980   1593   45    38    350.5  123456789  0      85     30     70000  11000  512    25000  N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A         N/A
-GPU 1    1980   1593   46    39    352.1  223456789  0      90     35     69000  12000  512    25500  N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A         N/A
-GPU 0    1980   1593   46    39    351.0  123466789  0      88     32     70000  11050  512    N/A    0.910  0.950  0.620  0.480  0.300  0.000  0.120  0.880  1500000000  1200000000
-GPU 1    1980   1593   47    40    353.0  223466789  0      92     36     69000  12050  512    25600  0.930  0.970  0.640  0.500  0.320  0.000  0.130  0.900  1600000000  1300000000
+GPU 0    1980   1593   45    38    350.5  123456789  0      85     30     70000  11000  512    25000  N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A         N/A         N/A           N/A
+GPU 1    1980   1593   46    39    352.1  223456789  0      90     35     69000  12000  512    25500  N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A    N/A         N/A         N/A           N/A
+GPU 0    1980   1593   46    39    351.0  123466789  0      88     32     70000  11050  512    N/A    0.910  0.950  0.620  0.480  0.300  0.000  0.120  0.880  1500000000  1200000000  N/A           N/A
+GPU 1    1980   1593   47    40    353.0  223466789  0      92     36     69000  12050  512    25600  0.930  0.970  0.640  0.500  0.320  0.000  0.130  0.900  1600000000  1300000000  210000000000 220000000000

tests/integration/test_pipeline.py

@@ -63,14 +63,15 @@ def test_pipeline_yields_computed_records():
 
 def test_pipeline_csv_export_matches_header_width():
     last = _computed_ticks()[-1]
+    assert last[0].record.gpu_nvlink_aggregate_throughput_bytes_per_second is None
     cols = resolve_columns("all")
     header = csv_header(cols)
     assert "real_util_pct" in header and "sm_active_pct" in header and "nvlink_gbps" in header
     ts = last[0].record.record_timestamp_wallclock_unix_seconds
     row = csv_row(last[0], ts, cols)
     assert len(row) == len(header)
-    # N/A stays empty, never coerced to 0 (gpu0 NVLink was N/A in tick 2)
-    assert all(cell == "" or cell is not None for cell in row)
+    # N/A stays empty, never coerced to 0 (GPU 0 NVLink is unavailable in tick 2).
+    assert row[header.index("nvlink_gbps")] == ""
 
 
 def test_pipeline_dashboard_renders():

tests/unit/test_dcgmi_reader.py

@@ -25,11 +25,20 @@ def test_na_becomes_none_never_zero():
     cold_gpu0 = records[0]  # first tick: profiling fields are N/A
     assert cold_gpu0.fields["DCGM_FI_PROF_SM_ACTIVE"] is None
     assert cold_gpu0.fields["DCGM_FI_PROF_PIPE_TENSOR_ACTIVE"] is None
+    assert cold_gpu0.fields["DCGM_FI_PROF_NVLINK_TX_BYTES"] is None
+    assert cold_gpu0.fields["DCGM_FI_PROF_NVLINK_RX_BYTES"] is None
     # device fields in the same row are real values
     assert cold_gpu0.fields["DCGM_FI_DEV_POWER_USAGE"] == 350.5
     assert cold_gpu0.fields["DCGM_FI_DEV_NVLINK_BANDWIDTH_TOTAL"] == 25000.0
 
 
+def test_nvlink_profile_fields_are_parsed():
+    records = parse_dmon_block(_load())
+    warm_gpu1 = records[3]
+    assert warm_gpu1.fields["DCGM_FI_PROF_NVLINK_TX_BYTES"] == 210000000000.0
+    assert warm_gpu1.fields["DCGM_FI_PROF_NVLINK_RX_BYTES"] == 220000000000.0
+
+
 def test_skips_headers_and_blank_lines():
     text = "#Entity ...\nID\n\n   \nGPU 0 1\n"
     # Only one data row, one metric column present.

tests/unit/test_translator.py

@@ -36,6 +36,55 @@ def test_units_normalized_to_canonical():
     assert g1t2.gpu_pcie_transmit_throughput_bytes_per_second == 1600000000.0
 
 
+def test_nvlink_profile_fields_fallback_when_aggregate_missing():
+    assert _nvlink_raw(
+        DCGM_FI_PROF_NVLINK_TX_BYTES=110e9,
+        DCGM_FI_PROF_NVLINK_RX_BYTES=120e9,
+    ) == 230e9
+
+
+def test_nvlink_aggregate_field_takes_precedence_over_profile_fields():
+    _, recs = _translate_all()
+    g1t2 = recs[3]  # aggregate field is present, so TX/RX are ignored
+    assert g1t2.gpu_nvlink_aggregate_throughput_bytes_per_second == 25600 * 1e6
+
+
+def _nvlink_raw(**fields):
+    tr = make_translator(BackendKind.DCGMI, hostname="testnode")
+    raw = RawRecord(
+        timestamp=0.0, wallclock=0.0, entity_id=0,
+        fields=fields, source="dcgm", source_version="test",
+    )
+    return tr.translate(raw).gpu_nvlink_aggregate_throughput_bytes_per_second
+
+
+def test_nvlink_zero_aggregate_is_idle_not_missing():
+    """A real 0.0 from field 449 means idle, so the fallback must not fire."""
+    assert _nvlink_raw(
+        DCGM_FI_DEV_NVLINK_BANDWIDTH_TOTAL=0.0,
+        DCGM_FI_PROF_NVLINK_TX_BYTES=110e9,
+        DCGM_FI_PROF_NVLINK_RX_BYTES=120e9,
+    ) == 0.0
+
+
+def test_nvlink_one_sided_profile_reading_is_not_reported():
+    """Half a sum would under-report the total, so report nothing instead."""
+    assert _nvlink_raw(DCGM_FI_PROF_NVLINK_TX_BYTES=110e9) is None
+    assert _nvlink_raw(DCGM_FI_PROF_NVLINK_RX_BYTES=120e9) is None
+
+
+def test_nvlink_negative_profile_reading_is_clamped():
+    """Matches the clamp the mapped 449 path gets, so validation cannot fail."""
+    assert _nvlink_raw(
+        DCGM_FI_PROF_NVLINK_TX_BYTES=-5e9,
+        DCGM_FI_PROF_NVLINK_RX_BYTES=1e9,
+    ) == 0.0
+
+
+def test_nvlink_absent_everywhere_stays_none():
+    assert _nvlink_raw(DCGM_FI_DEV_POWER_USAGE=100.0) is None
+
+
 def test_framebuffer_total_derived():
     _, recs = _translate_all()
     g1t2 = recs[3]