feat: Tighter FMA timing via DPC log parsing (#1504)

* feat: Add DPC log parser for per-request timing extraction

Signed-off-by: Gloire Rubambiza <gloire@ibm.com>

* feat: Integrate DPC log parsing to refine per-path timing

Signed-off-by: Gloire Rubambiza <gloire@ibm.com>

* feat: Display timing source (DPC/Kube) in analysis output

Signed-off-by: Gloire Rubambiza <gloire@ibm.com>

* fix: Make DPC log detection scan whole file

The DPC timing refinement silently fell back to Kube upper bounds on
real cluster runs. Two bugs, both surfaced during cluster validation:

1. _is_dpc_log_file() only sniffed the first 256KB of a .log for an
   indicator message. A real controller log was 1.08MB with its first
   relay/wake/create message at byte ~429K (after startup noise), so the
   log was discarded and the parser never ran. Now stream line-by-line
   with early-exit. Added a regression test with an indicator past 256KB.

2. native_to_br0_1 dropped the dpc_timing_available flag during
   conversion, so the analyzer's Source column always showed Kube even
   when timing was DPC-derived. Propagate the flag into launcher_infos.

Validated on cluster: dpc_timing_available true on all iterations,
hot-start t_wake ~0.53-0.59s (vs 8s Kube upper bound), Source=DPC.

Assisted-By: Claude Opus 4.6 <noreply@anthropic.com>

Signed-off-by: Gloire Rubambiza <gloire@ibm.com>

Author: rubambiza

llmdbenchmark/analysis/benchmark_report/native_to_br0_1.py

@@ -2554,6 +2554,9 @@ def _import_categories(cat_list: list[dict[str, Any]]) -> list[dict[str, Any]]:
                     "value": launcher_info.get("launcher_creation_timestamp", 0.0),
                 }
                 info["launcher_node"] = launcher_info.get("launcher_node", "")
+                info["dpc_timing_available"] = launcher_info.get(
+                    "dpc_timing_available", False
+                )
                 if launcher_info.get("t_wake") is not None:
                     info["t_wake"] = {
                         "units": Units.S,

llmdbenchmark/analysis/scripts/nop-analyze_results.py

@@ -300,9 +300,12 @@ def write_fma_metrics(  # pylint: disable=too-many-locals,too-many-statements
     file.write("  T_warm: existing launcher creates new vLLM instance\n")
     file.write("  T_hot: waking sleeping vLLM instance\n\n")
     file.write("T_actuation: Time for the Requester Pod to be ready\n")
-    file.write("T_hot: Hot-start timing (upper bound)\n")
-    file.write("T_warm: Warm-start timing (upper bound)\n")
-    file.write("T_cold_launcher: Cold-start-with-launcher timing (upper bound)\n")
+    file.write("T_hot: Hot-start timing\n")
+    file.write("T_warm: Warm-start timing\n")
+    file.write("T_cold_launcher: Cold-start-with-launcher timing\n")
+    file.write(
+        "Source: DPC = tighter timing from DPC logs, Kube = upper bound from Kube timestamps\n"
+    )
     file.write("T_first_token: Time for vLLM server to return first token\n")
     file.write("Each iteration scales ReplicaSet from 0 to 1 and then from 1 to 0\n")
 
@@ -344,6 +347,9 @@ def write_fma_metrics(  # pylint: disable=too-many-locals,too-many-statements
                 else None
             )
             node = launcher_info.get("launcher_node", "")
+            source = (
+                "DPC" if launcher_info.get("dpc_timing_available", False) else "Kube"
+            )
 
             pandas_datas.append(
                 {
@@ -356,6 +362,7 @@ def write_fma_metrics(  # pylint: disable=too-many-locals,too-many-statements
                     "T_warm(s)": t_warm_val,
                     "T_cold(s)": t_cold_val,
                     "T_first_token(s)": ttft,
+                    "Source": source,
                 }
             )
 

tests/test_dpc_log_parser.py

@@ -0,0 +1,192 @@
+"""Tests for DPC log parser that extracts per-request timing from klog output."""
+
+import pytest
+
+from workload.harnesses.dpc_log_parser import (
+    parse_dpc_log,
+    parse_dpc_log_file,
+)
+
+
+# Representative klog lines from a DPC running with PR #522 at V(2)
+SAMPLE_LOG = """\
+I0603 15:29:00.100000  1 inference-server.go:145] "Reconciling server" serverUID="abc-123" requesterName="fma-req-1-1717424983-abcde"
+I0603 15:29:01.200000  1 inference-server.go:1131] "Woke inference server" endpoint="http://10.0.0.1:8005/wake_up" description="discovered-bound" requesterName="fma-req-1-1717424983-abcde" httpCallStartTime="2026-06-03T15:29:01.150000000Z"
+I0603 15:29:03.500000  1 inference-server.go:513] "Successfully relayed the readiness" requesterName="fma-req-1-1717424983-abcde" name="launcher-27hdz" readiness="ready" url="http://10.0.0.2:8888/v1/become-ready" httpCallStartTime="2026-06-03T15:29:03.480000000Z"
+I0603 15:30:10.000000  1 inference-server.go:464] "Created vLLM instance" instance_id="inst-456" status="running" requesterName="fma-req-2-1717424983-fghij" httpCallStartTime="2026-06-03T15:30:09.900000000Z"
+I0603 15:30:50.000000  1 inference-server.go:513] "Successfully relayed the readiness" requesterName="fma-req-2-1717424983-fghij" name="launcher-27hdz" readiness="ready" url="http://10.0.0.3:8888/v1/become-ready" httpCallStartTime="2026-06-03T15:30:49.950000000Z"
+I0603 15:31:00.000000  1 inference-server.go:667] "Created launcher-based server-providing pod" name="launcher-xq8lg" gpus="GPU-abc" requesterName="fma-req-3-1717424983-klmno" k8sCallStartTime="2026-06-03T15:30:59.800000000Z"
+I0603 15:32:05.000000  1 inference-server.go:513] "Successfully relayed the readiness" requesterName="fma-req-3-1717424983-klmno" name="launcher-xq8lg" readiness="ready" url="http://10.0.0.4:8888/v1/become-ready" httpCallStartTime="2026-06-03T15:32:04.900000000Z"
+"""
+
+
+class TestParseDpcLog:
+    """Test parse_dpc_log extracts timing records per requester."""
+
+    def test_hot_start_timing(self):
+        records = parse_dpc_log(SAMPLE_LOG.splitlines())
+        rec = records["fma-req-1-1717424983-abcde"]
+        assert rec.relay_readiness_time is not None
+        assert rec.wake_start_time is not None
+        assert rec.instance_create_start_time is None
+        assert rec.launcher_create_start_time is None
+        t_hot = rec.t_hot()
+        assert t_hot is not None
+        assert 2.3 < t_hot < 2.4  # 15:29:03.48 - 15:29:01.15 = 2.33s
+
+    def test_warm_start_timing(self):
+        records = parse_dpc_log(SAMPLE_LOG.splitlines())
+        rec = records["fma-req-2-1717424983-fghij"]
+        assert rec.instance_create_start_time is not None
+        assert rec.relay_readiness_time is not None
+        assert rec.wake_start_time is None
+        t_warm = rec.t_instance_create()
+        assert t_warm is not None
+        assert 40.0 < t_warm < 40.1  # 15:30:49.95 - 15:30:09.90 = 40.05s
+
+    def test_cold_launcher_timing(self):
+        records = parse_dpc_log(SAMPLE_LOG.splitlines())
+        rec = records["fma-req-3-1717424983-klmno"]
+        assert rec.launcher_create_start_time is not None
+        assert rec.relay_readiness_time is not None
+        t_cold = rec.t_cold_launcher()
+        assert t_cold is not None
+        assert 65.0 < t_cold < 65.2  # 15:32:04.90 - 15:30:59.80 = 65.1s
+
+    def test_unknown_requester_not_present(self):
+        records = parse_dpc_log(SAMPLE_LOG.splitlines())
+        assert "nonexistent-pod" not in records
+
+    def test_empty_log(self):
+        records = parse_dpc_log([])
+        assert records == {}
+
+    def test_log_with_no_timing_fields(self):
+        records = parse_dpc_log(
+            ['I0603 15:29:00.100000  1 foo.go:1] "Something unrelated"']
+        )
+        assert records == {}
+
+
+class TestParseDpcLogFile:
+    """Test file-based entry point."""
+
+    def test_reads_log_file_and_parses(self, tmp_path):
+        log_file = tmp_path / "dpctlr-pod--manager.log"
+        log_file.write_text(SAMPLE_LOG)
+        records = parse_dpc_log_file(str(tmp_path))
+        assert "fma-req-1-1717424983-abcde" in records
+        assert records["fma-req-1-1717424983-abcde"].t_hot() is not None
+
+    def test_missing_directory_returns_empty(self, tmp_path):
+        records = parse_dpc_log_file(str(tmp_path / "nonexistent"))
+        assert records == {}
+
+    def test_no_matching_files_returns_empty(self, tmp_path):
+        (tmp_path / "unrelated.log").write_text("nothing relevant here")
+        records = parse_dpc_log_file(str(tmp_path))
+        assert records == {}
+
+    def test_indicator_message_past_256kb_still_parsed(self, tmp_path):
+        """A DPC log whose first indicator message lands past 256KB must still parse.
+
+        Regression for the original _is_dpc_log_file heuristic, which only sniffed
+        the first 256KB. Real controller logs prepend large amounts of startup noise
+        (flag dumps, leader election, reconcile churn) before the first
+        relay/wake/create message, so the indicator can sit well past that boundary.
+        """
+        filler_line = (
+            "I0603 15:00:00.000000  1 inference-server.go:145] "
+            '"Reconciling server" serverUID="noise" requesterName="noise-pod"\n'
+        )
+        # Prepend well over 256KB of benign filler, then the real sample block.
+        filler = filler_line * (300 * 1024 // len(filler_line) + 1)
+        assert len(filler) > 256 * 1024
+        log_file = tmp_path / "dpctlr-pod--manager.log"
+        log_file.write_text(filler + SAMPLE_LOG)
+
+        records = parse_dpc_log_file(str(tmp_path))
+        assert "fma-req-1-1717424983-abcde" in records
+        assert records["fma-req-1-1717424983-abcde"].t_hot() is not None
+
+    def test_file_without_relay_but_with_wake_still_parsed(self, tmp_path):
+        """DPC log where requester crashed before relay should still be found."""
+        partial_log = (
+            'I0603 15:29:01.000000  1 x.go:1] "Woke inference server" '
+            'requesterName="req-crashed" httpCallStartTime="2026-06-03T15:29:00.900Z"\n'
+        )
+        (tmp_path / "dpctlr--manager.log").write_text(partial_log)
+        records = parse_dpc_log_file(str(tmp_path))
+        assert "req-crashed" in records
+        assert records["req-crashed"].wake_start_time is not None
+        assert records["req-crashed"].t_hot() is None  # no relay = no duration
+
+
+class TestEdgeCases:
+    """Edge cases for DPC log parsing robustness."""
+
+    def test_multiple_relay_readiness_uses_last(self):
+        """If DPC retries readiness relay, use the last successful one."""
+        lines = [
+            'I0603 15:29:01.000000  1 x.go:1] "Woke inference server" '
+            'requesterName="req-retry" httpCallStartTime="2026-06-03T15:29:00.900Z"',
+            'I0603 15:29:03.000000  1 x.go:1] "Successfully relayed the readiness" '
+            'requesterName="req-retry" readiness="ready" '
+            'httpCallStartTime="2026-06-03T15:29:02.500Z"',
+            'I0603 15:29:05.000000  1 x.go:1] "Successfully relayed the readiness" '
+            'requesterName="req-retry" readiness="ready" '
+            'httpCallStartTime="2026-06-03T15:29:04.800Z"',
+        ]
+        records = parse_dpc_log(lines)
+        rec = records["req-retry"]
+        t_hot = rec.t_hot()
+        assert t_hot is not None
+        assert 3.8 < t_hot < 4.0  # 04.8 - 00.9 = 3.9s
+
+    def test_unready_relay_ignored(self):
+        """readiness='unready' lines should not set relay_readiness_time."""
+        lines = [
+            'I0603 15:29:01.000000  1 x.go:1] "Woke inference server" '
+            'requesterName="req-unready" httpCallStartTime="2026-06-03T15:29:00.900Z"',
+            'I0603 15:29:03.000000  1 x.go:1] "Successfully relayed the readiness" '
+            'requesterName="req-unready" readiness="unready" '
+            'httpCallStartTime="2026-06-03T15:29:02.500Z"',
+        ]
+        records = parse_dpc_log(lines)
+        rec = records["req-unready"]
+        assert rec.relay_readiness_time is None
+        assert rec.t_hot() is None
+
+    def test_malformed_timestamp_skipped(self):
+        """Malformed httpCallStartTime should not crash, just skip."""
+        lines = [
+            'I0603 15:29:01.000000  1 x.go:1] "Woke inference server" '
+            'requesterName="req-bad" httpCallStartTime="not-a-timestamp"',
+            'I0603 15:29:03.000000  1 x.go:1] "Successfully relayed the readiness" '
+            'requesterName="req-bad" readiness="ready" '
+            'httpCallStartTime="2026-06-03T15:29:02.500Z"',
+        ]
+        records = parse_dpc_log(lines)
+        rec = records["req-bad"]
+        assert rec.wake_start_time is None
+        assert rec.relay_readiness_time is not None
+
+    def test_multiple_requesters_independent(self):
+        """Different requesters get independent records."""
+        lines = [
+            'I0603 15:29:01.000000  1 x.go:1] "Woke inference server" '
+            'requesterName="req-a" httpCallStartTime="2026-06-03T15:29:00.000Z"',
+            'I0603 15:30:01.000000  1 x.go:1] "Created vLLM instance" '
+            'requesterName="req-b" httpCallStartTime="2026-06-03T15:30:00.000Z"',
+            'I0603 15:29:05.000000  1 x.go:1] "Successfully relayed the readiness" '
+            'requesterName="req-a" readiness="ready" '
+            'httpCallStartTime="2026-06-03T15:29:04.000Z"',
+            'I0603 15:31:05.000000  1 x.go:1] "Successfully relayed the readiness" '
+            'requesterName="req-b" readiness="ready" '
+            'httpCallStartTime="2026-06-03T15:31:04.000Z"',
+        ]
+        records = parse_dpc_log(lines)
+        assert records["req-a"].t_hot() == pytest.approx(4.0, abs=0.01)
+        assert records["req-b"].t_instance_create() == pytest.approx(64.0, abs=0.01)
+        assert records["req-a"].t_instance_create() is None
+        assert records["req-b"].t_hot() is None