Add NVLink profiling fallback

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.
+        if canonical.get(NVLINK_AGGREGATE_FIELD) is None:
+            tx = raw.fields.get(NVLINK_TX_FIELD)
+            rx = raw.fields.get(NVLINK_RX_FIELD)
+            if tx is not None or rx is not None:
+                canonical[NVLINK_AGGREGATE_FIELD] = (
+                    float(tx or 0.0) + float(rx or 0.0)
+                )
+
         # 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  110000000000 120000000000
+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/unit/test_dcgmi_reader.py

@@ -25,11 +25,21 @@ 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_gpu0 = records[2]
+    assert warm_gpu0.fields["DCGM_FI_DEV_NVLINK_BANDWIDTH_TOTAL"] is None
+    assert warm_gpu0.fields["DCGM_FI_PROF_NVLINK_TX_BYTES"] == 110000000000.0
+    assert warm_gpu0.fields["DCGM_FI_PROF_NVLINK_RX_BYTES"] == 120000000000.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,18 @@ 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():
+    _, recs = _translate_all()
+    g0t2 = recs[2]  # aggregate field is N/A, profiling TX/RX are valid
+    assert g0t2.gpu_nvlink_aggregate_throughput_bytes_per_second == 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 test_framebuffer_total_derived():
     _, recs = _translate_all()
     g1t2 = recs[3]