diff --git a/llmdbenchmark/analysis/benchmark_report/native_to_br0_1.py b/llmdbenchmark/analysis/benchmark_report/native_to_br0_1.py index 74a1aae84..a8ee982d8 100644 --- a/llmdbenchmark/analysis/benchmark_report/native_to_br0_1.py +++ b/llmdbenchmark/analysis/benchmark_report/native_to_br0_1.py @@ -2543,6 +2543,10 @@ def _import_categories(cat_list: list[dict[str, Any]]) -> list[dict[str, Any]]: "units": Units.S, "value": requester_info["dual_label_timestamp"], } + ri["container_start_timestamp"] = { + "units": Units.S, + "value": requester_info.get("container_start_timestamp", 0.0), + } info["requester_info"] = ri info["actuation_condition"] = launcher_info["actuation_condition"] @@ -2554,6 +2558,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["timing_source"] = launcher_info.get( + "timing_source", "kube_pod_create" + ) info["dpc_timing_available"] = launcher_info.get( "dpc_timing_available", False ) diff --git a/llmdbenchmark/analysis/scripts/nop-analyze_results.py b/llmdbenchmark/analysis/scripts/nop-analyze_results.py index d12f534cf..a1727f009 100755 --- a/llmdbenchmark/analysis/scripts/nop-analyze_results.py +++ b/llmdbenchmark/analysis/scripts/nop-analyze_results.py @@ -350,9 +350,24 @@ 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" - ) + # Three-way timing source so a degraded pod-create baseline is + # distinguishable from a container-start Kube fallback at a glance. + # Fall back to the legacy dpc_timing_available flag for older reports + # that predate the timing_source field. + source_map = { + "dpc": "DPC", + "kube_container_start": "Kube (container-start)", + "kube_pod_create": "Kube (pod-create)", + } + timing_source = launcher_info.get("timing_source") + if timing_source in source_map: + source = source_map[timing_source] + else: + source = ( + "DPC" + if launcher_info.get("dpc_timing_available", False) + else "Kube" + ) pandas_datas.append( { diff --git a/tests/test_dpc_log_parser.py b/tests/test_dpc_log_parser.py index d3109ed8a..653fc0b85 100644 --- a/tests/test_dpc_log_parser.py +++ b/tests/test_dpc_log_parser.py @@ -8,15 +8,16 @@ ) -# Representative klog lines from a DPC running with PR #522 at V(2) +# Representative klog lines from a DPC running with centralized "HTTP call done" line at V(5) +# The DPC now emits a single klog line per HTTP call with purpose="" field. 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:29:01.200000 1 inference-server.go:2074] "HTTP call done" purpose="wake" method="POST" url="http://10.0.0.1:8005/wake_up" requesterName="fma-req-1-1717424983-abcde" httpCallStartTime="2026-06-03T15:29:01.150000000Z" latencySecs="0.51" statusCode="200" +I0603 15:29:03.500000 1 inference-server.go:2074] "HTTP call done" purpose="relay_ready" method="POST" url="http://10.0.0.2:8888/v1/become-ready" requesterName="fma-req-1-1717424983-abcde" httpCallStartTime="2026-06-03T15:29:03.480000000Z" latencySecs="0.02" statusCode="200" +I0603 15:30:10.000000 1 inference-server.go:2074] "HTTP call done" purpose="create_instance" method="POST" url="http://10.0.0.3:8888/v1/models" requesterName="fma-req-2-1717424983-fghij" httpCallStartTime="2026-06-03T15:30:09.900000000Z" latencySecs="0.10" statusCode="200" +I0603 15:30:50.000000 1 inference-server.go:2074] "HTTP call done" purpose="relay_ready" method="POST" url="http://10.0.0.3:8888/v1/become-ready" requesterName="fma-req-2-1717424983-fghij" httpCallStartTime="2026-06-03T15:30:49.950000000Z" latencySecs="0.05" statusCode="200" 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" +I0603 15:32:05.000000 1 inference-server.go:2074] "HTTP call done" purpose="relay_ready" method="POST" url="http://10.0.0.4:8888/v1/become-ready" requesterName="fma-req-3-1717424983-klmno" httpCallStartTime="2026-06-03T15:32:04.900000000Z" latencySecs="0.10" statusCode="200" """ @@ -95,9 +96,13 @@ def test_indicator_message_past_256kb_still_parsed(self, tmp_path): (flag dumps, leader election, reconcile churn) before the first relay/wake/create message, so the indicator can sit well past that boundary. """ + # Filler must be a NON-indicator line (contains neither "HTTP call done" + # nor "Created launcher-based server-providing pod"), so the FIRST indicator + # is the real anchor from SAMPLE_LOG, sitting past the 256KB boundary. This + # keeps the regression bite: if _is_dpc_log_file ever sniffs only the first + # 256KB again, it would fail to recognize this file as a DPC log. filler_line = ( - "I0603 15:00:00.000000 1 inference-server.go:145] " - '"Reconciling server" serverUID="noise" requesterName="noise-pod"\n' + 'I0603 15:00:00.000000 1 flags.go:100] "Flag" name="foo" value="bar"\n' ) # Prepend well over 256KB of benign filler, then the real sample block. filler = filler_line * (300 * 1024 // len(filler_line) + 1) @@ -111,9 +116,11 @@ def test_indicator_message_past_256kb_still_parsed(self, tmp_path): def test_file_without_relay_but_with_wake_still_parsed(self, tmp_path): """DPC log where requester crashed before relay should still be found.""" + # New format with full fields for a wake call, no relay follows (requester crashed) 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' + 'I0603 15:29:01.000000 1 x.go:2074] "HTTP call done" purpose="wake" ' + 'method="POST" url="http://10.0.0.1:8005/wake_up" requesterName="req-crashed" ' + 'httpCallStartTime="2026-06-03T15:29:00.900Z" latencySecs="0.1" statusCode="200"\n' ) (tmp_path / "dpctlr--manager.log").write_text(partial_log) records = parse_dpc_log_file(str(tmp_path)) @@ -128,13 +135,13 @@ class TestEdgeCases: 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" ' + 'I0603 15:29:01.000000 1 x.go:2074] "HTTP call done" purpose="wake" ' '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" ' + 'I0603 15:29:03.000000 1 x.go:2074] "HTTP call done" purpose="relay_ready" ' + 'requesterName="req-retry" ' '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" ' + 'I0603 15:29:05.000000 1 x.go:2074] "HTTP call done" purpose="relay_ready" ' + 'requesterName="req-retry" ' 'httpCallStartTime="2026-06-03T15:29:04.800Z"', ] records = parse_dpc_log(lines) @@ -144,12 +151,12 @@ def test_multiple_relay_readiness_uses_last(self): 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.""" + """purpose='relay_unready' lines should not set relay_readiness_time.""" lines = [ - 'I0603 15:29:01.000000 1 x.go:1] "Woke inference server" ' + 'I0603 15:29:01.000000 1 x.go:2074] "HTTP call done" purpose="wake" ' '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" ' + 'I0603 15:29:03.000000 1 x.go:2074] "HTTP call done" purpose="relay_unready" ' + 'requesterName="req-unready" ' 'httpCallStartTime="2026-06-03T15:29:02.500Z"', ] records = parse_dpc_log(lines) @@ -160,10 +167,10 @@ def test_unready_relay_ignored(self): 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" ' + 'I0603 15:29:01.000000 1 x.go:2074] "HTTP call done" purpose="wake" ' '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" ' + 'I0603 15:29:03.000000 1 x.go:2074] "HTTP call done" purpose="relay_ready" ' + 'requesterName="req-bad" ' 'httpCallStartTime="2026-06-03T15:29:02.500Z"', ] records = parse_dpc_log(lines) @@ -174,15 +181,15 @@ def test_malformed_timestamp_skipped(self): def test_multiple_requesters_independent(self): """Different requesters get independent records.""" lines = [ - 'I0603 15:29:01.000000 1 x.go:1] "Woke inference server" ' + 'I0603 15:29:01.000000 1 x.go:2074] "HTTP call done" purpose="wake" ' 'requesterName="req-a" httpCallStartTime="2026-06-03T15:29:00.000Z"', - 'I0603 15:30:01.000000 1 x.go:1] "Created vLLM instance" ' + 'I0603 15:30:01.000000 1 x.go:2074] "HTTP call done" purpose="create_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" ' + 'I0603 15:29:05.000000 1 x.go:2074] "HTTP call done" purpose="relay_ready" ' + 'requesterName="req-a" ' '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" ' + 'I0603 15:31:05.000000 1 x.go:2074] "HTTP call done" purpose="relay_ready" ' + 'requesterName="req-b" ' 'httpCallStartTime="2026-06-03T15:31:04.000Z"', ] records = parse_dpc_log(lines) diff --git a/tests/test_fma_functions.py b/tests/test_fma_functions.py new file mode 100644 index 000000000..bf05c957a --- /dev/null +++ b/tests/test_fma_functions.py @@ -0,0 +1,266 @@ +"""Tests for FMA harness helpers: container-start baseline and timing_source. + +The Kube-timestamp fallback must anchor on the requester ``inference-server`` +container ``state.running.started_at`` (matching the dual-pods controller's +actuation baseline), and any revert to pod ``creation_timestamp`` must be +explicit (``timing_source == "kube_pod_create"``) and logged, never silent. +""" + +import logging +import os +import sys +from datetime import datetime, timezone +from types import SimpleNamespace + +# fma_functions.py uses bare imports (``from dpc_log_parser import ...``) that +# assume ``workload/harnesses`` is on the path, which is how it runs in the +# harness container. Mirror that here so the module imports as a plain module. +_HARNESS_DIR = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "workload", "harnesses") +) +if _HARNESS_DIR not in sys.path: + sys.path.insert(0, _HARNESS_DIR) + +import fma_functions as f # noqa: E402 + + +def _fake_pod_with_container_start(started_at, name="inference-server"): + """Build a minimal pod-like object with one running container status.""" + running = SimpleNamespace(started_at=started_at) + state = SimpleNamespace(running=running) + cs = SimpleNamespace(name=name, state=state) + status = SimpleNamespace(container_statuses=[cs]) + return SimpleNamespace(status=status) + + +class TestGetContainerStartTimestamp: + """get_container_start_timestamp mirrors the controller's actuation-baseline read.""" + + def test_returns_epoch_for_running_inference_server(self): + started = datetime(2026, 7, 1, 10, 39, 35, tzinfo=timezone.utc) + pod = _fake_pod_with_container_start(started) + ts = f.get_container_start_timestamp(pod) + assert ts == started.timestamp() + assert ts > 0.0 + + def test_zero_when_container_missing(self): + pod = _fake_pod_with_container_start( + datetime.now(timezone.utc), name="some-sidecar" + ) + assert f.get_container_start_timestamp(pod) == 0.0 + + def test_zero_when_not_running(self): + state = SimpleNamespace(running=None) + cs = SimpleNamespace(name="inference-server", state=state) + status = SimpleNamespace(container_statuses=[cs]) + pod = SimpleNamespace(status=status) + assert f.get_container_start_timestamp(pod) == 0.0 + + def test_zero_when_container_statuses_none(self): + status = SimpleNamespace(container_statuses=None) + pod = SimpleNamespace(status=status) + assert f.get_container_start_timestamp(pod) == 0.0 + + def test_zero_when_started_at_none(self): + pod = _fake_pod_with_container_start(None) + assert f.get_container_start_timestamp(pod) == 0.0 + + +class TestSelectKubeFallbackBaseline: + """The Kube fallback picks container-start when available, else pod-create.""" + + def test_uses_container_start_when_present(self): + ri = f.FMARequesterInfo( + name="fma-req-1", + creation_timestamp=1000.0, + ready_timestamp=1005.0, + container_start_timestamp=1002.0, + ) + baseline, source = f.select_kube_fallback_baseline(ri) + assert baseline == 1002.0 + assert source == "kube_container_start" + + def test_reverts_to_pod_create_selection(self, caplog): + ri = f.FMARequesterInfo( + name="fma-req-degraded", + creation_timestamp=1000.0, + ready_timestamp=1005.0, + container_start_timestamp=0.0, + ) + # Selection is pure: it picks the pod-create baseline but does NOT warn, + # because DPC refinement may still override the source to "dpc" later. + with caplog.at_level(logging.WARNING): + baseline, source = f.select_kube_fallback_baseline(ri) + assert baseline == 1000.0 + assert source == "kube_pod_create" + assert not any(rec.levelno == logging.WARNING for rec in caplog.records) + + +class TestWarnOnPodCreateBaseline: + """The deferred warning names the requester and is emitted at WARNING level.""" + + def test_warns_and_names_requester(self, caplog): + with caplog.at_level(logging.WARNING): + f.warn_on_pod_create_baseline("fma-req-degraded") + assert any( + rec.levelno == logging.WARNING and "fma-req-degraded" in rec.getMessage() + for rec in caplog.records + ) + + +def _emit_deferred_pod_create_warnings(launcher_infos): + """Mirror the production deferred-warning pass over final timing_source.""" + for li in launcher_infos: + if li.timing_source == "kube_pod_create": + f.warn_on_pod_create_baseline(li.requester_info.name) + + +class TestDeferredWarningRespectsFinalSource: + """Warning fires only for FINAL kube_pod_create, not DPC-overridden ones.""" + + def test_dpc_override_suppresses_warning(self, caplog): + # Requester started with container_start unavailable (tentative + # kube_pod_create) but DPC refinement overrode timing_source to "dpc". + refined = f.FMALauncherInfo( + requester_info=f.FMARequesterInfo( + name="fma-req-refined", creation_timestamp=1000.0 + ), + timing_source="dpc", + ) + with caplog.at_level(logging.WARNING): + _emit_deferred_pod_create_warnings([refined]) + # No spurious "may be overstated" warning for a DPC-timed requester. + assert not any("fma-req-refined" in rec.getMessage() for rec in caplog.records) + + def test_genuine_pod_create_still_warns(self, caplog): + degraded = f.FMALauncherInfo( + requester_info=f.FMARequesterInfo( + name="fma-req-degraded", creation_timestamp=1000.0 + ), + timing_source="kube_pod_create", + ) + with caplog.at_level(logging.WARNING): + _emit_deferred_pod_create_warnings([degraded]) + assert any( + rec.levelno == logging.WARNING and "fma-req-degraded" in rec.getMessage() + for rec in caplog.records + ) + + def test_container_start_source_does_not_warn(self, caplog): + cs = f.FMALauncherInfo( + requester_info=f.FMARequesterInfo(name="fma-req-cs"), + timing_source="kube_container_start", + ) + with caplog.at_level(logging.WARNING): + _emit_deferred_pod_create_warnings([cs]) + assert not any("fma-req-cs" in rec.getMessage() for rec in caplog.records) + + +class TestTimingSourceField: + """FMALauncherInfo tracks timing_source and derives dpc_timing_available.""" + + def test_default_timing_source(self): + li = f.FMALauncherInfo() + assert li.timing_source == "kube_pod_create" + assert li.dpc_timing_available is False + + def test_dpc_timing_available_derived_from_dpc(self): + li = f.FMALauncherInfo(timing_source="dpc") + assert li.dpc_timing_available is True + + def test_dpc_timing_available_false_for_kube_container_start(self): + li = f.FMALauncherInfo(timing_source="kube_container_start") + assert li.dpc_timing_available is False + + +class TestRequesterInfoContainerStartField: + """FMARequesterInfo carries container_start_timestamp and dumps it.""" + + def test_container_start_in_dump(self): + ri = f.FMARequesterInfo(name="r", container_start_timestamp=42.0) + dumped = ri.dump() + assert dumped["container_start_timestamp"] == 42.0 + + +def _fma_nop_results(timing_source, container_start_timestamp): + """Minimal nop results dict carrying one FMA launcher_info.""" + return { + "scenario": { + "model": {"name": "m"}, + "deploy_methods": "fma", + "load_format": "auto", + "sleep_mode": "1", + "gpus": 1, + "platform": { + "engines": [ + {"name": "vllm", "version": "0.1", "args": {}, "image": "img:tag"} + ] + }, + }, + "time": {"duration": 1.0, "start": 0.0, "stop": 1.0}, + "vllm_metrics": [], + "extra_metrics": [ + { + "name": "fma", + "iterations": [ + { + "iteration": 0, + "hot_hit_rate": 1.0, + "warm_hit_rate": 0.0, + "cold_launcher_hit_rate": 0.0, + "launcher_infos": [ + { + "name": "l1", + "requester_info": { + "name": "r1", + "creation_timestamp": 100.0, + "ready_timestamp": 105.0, + "dual_label_timestamp": 101.0, + "container_start_timestamp": ( + container_start_timestamp + ), + }, + "actuation_condition": "T_hot", + "launcher_endpoint": "", + "vllm_endpoint": "", + "ttft": 0.5, + "launcher_creation_timestamp": 0.0, + "launcher_node": "n1", + "timing_source": timing_source, + "dpc_timing_available": timing_source == "dpc", + "t_wake": 3.0, + } + ], + } + ], + } + ], + } + + +class TestTimingSourceSurvivesNativeToBr01: + """timing_source + container_start_timestamp survive native->br0.1 import.""" + + def _import(self, tmp_path, results): + import yaml + from llmdbenchmark.analysis.benchmark_report.native_to_br0_1 import import_nop + + path = tmp_path / "results.yaml" + path.write_text(yaml.safe_dump(results)) + br = import_nop(str(path)) + bd = br.model_dump() + md = next(m for m in bd["metrics"]["metadata"] if m["name"] == "extra_metrics") + return md["value"][0]["iterations"][0]["launcher_infos"][0] + + def test_kube_container_start_survives(self, tmp_path): + li = self._import(tmp_path, _fma_nop_results("kube_container_start", 102.0)) + assert li["timing_source"] == "kube_container_start" + assert li["requester_info"]["container_start_timestamp"]["value"] == 102.0 + + def test_kube_pod_create_survives(self, tmp_path): + li = self._import(tmp_path, _fma_nop_results("kube_pod_create", 0.0)) + assert li["timing_source"] == "kube_pod_create" + + def test_dpc_survives(self, tmp_path): + li = self._import(tmp_path, _fma_nop_results("dpc", 102.0)) + assert li["timing_source"] == "dpc" diff --git a/workload/harnesses/dpc_log_parser.py b/workload/harnesses/dpc_log_parser.py index cf04f2ff3..464e6ad99 100644 --- a/workload/harnesses/dpc_log_parser.py +++ b/workload/harnesses/dpc_log_parser.py @@ -1,8 +1,16 @@ """Parse DPC (dual-pods-controller) klog output to extract per-request timing. -The DPC emits structured klog lines at V(2) with httpCallStartTime/k8sCallStartTime -fields (added in llm-d-fast-model-actuation PR #522). This module extracts those -timestamps and correlates them by requesterName to produce per-request timing records. +The DPC emits two kinds of structured klog lines: +1. "HTTP call done" lines at V(5) with a purpose="" field identifying the + HTTP call type (purpose tokens: "wake", "create_instance", "relay_ready", etc.). + Each line includes httpCallStartTime field. +2. "Created launcher-based server-providing pod" lines at V(2) for cold-launcher + starts, with k8sCallStartTime field. + +This module correlates these timestamps by requesterName to produce per-request +timing records. Only specific purposes are tracked: "wake" (hot-start anchor), +"create_instance" (warm-start anchor), "relay_ready" (readiness end time). +Other purposes and relay_unready are ignored. """ import glob @@ -18,16 +26,12 @@ # klog structured field patterns (key="value" pairs) _FIELD_RE = re.compile(r'(\w+)="([^"]*)"') -# Log messages we care about -_RELAY_READINESS_MSG = "Successfully relayed the readiness" -_WAKE_MSG = "Woke inference server" -_CREATE_INSTANCE_MSG = "Created vLLM instance" +# Log messages we care about (DPC indicators) +_HTTP_CALL_DONE_MSG = "HTTP call done" _CREATE_LAUNCHER_POD_MSG = "Created launcher-based server-providing pod" _DPC_INDICATOR_MSGS = ( - _RELAY_READINESS_MSG, - _WAKE_MSG, - _CREATE_INSTANCE_MSG, + _HTTP_CALL_DONE_MSG, _CREATE_LAUNCHER_POD_MSG, ) @@ -109,21 +113,23 @@ def parse_dpc_log(lines: Iterable[str]) -> dict[str, DPCTimingRecord]: records[requester_name] = DPCTimingRecord(requester_name=requester_name) rec = records[requester_name] - if _RELAY_READINESS_MSG in line and fields.get("readiness") == "ready": + # Handle centralized "HTTP call done" line with purpose-based dispatch + if _HTTP_CALL_DONE_MSG in line: + purpose = fields.get("purpose", "") ts = _parse_rfc3339_nano(fields.get("httpCallStartTime", "")) - if ts is not None: - rec.relay_readiness_time = ts - - elif _WAKE_MSG in line: - ts = _parse_rfc3339_nano(fields.get("httpCallStartTime", "")) - if ts is not None: - rec.wake_start_time = ts - - elif _CREATE_INSTANCE_MSG in line: - ts = _parse_rfc3339_nano(fields.get("httpCallStartTime", "")) - if ts is not None: - rec.instance_create_start_time = ts + if purpose == "wake": + if ts is not None: + rec.wake_start_time = ts + elif purpose == "create_instance": + if ts is not None: + rec.instance_create_start_time = ts + elif purpose == "relay_ready": + if ts is not None: + rec.relay_readiness_time = ts + # else: ignore relay_unready, sleep, query_sleeping, etc. + + # Handle cold-launcher start elif _CREATE_LAUNCHER_POD_MSG in line: ts = _parse_rfc3339_nano(fields.get("k8sCallStartTime", "")) if ts is not None: diff --git a/workload/harnesses/fma_functions.py b/workload/harnesses/fma_functions.py index e8c3dfed0..67f85d8fd 100755 --- a/workload/harnesses/fma_functions.py +++ b/workload/harnesses/fma_functions.py @@ -38,6 +38,10 @@ DUAL_LABEL = "dual-pods.llm-d.ai/dual" FMA_TIMEOUT = 10.0 * 60.0 # time (seconds) to wait +# Name of the requester container whose start time is the actuation baseline +# used by the dual-pods controller. Mirrors FMA pkg/api InferenceServerContainerName. +INFERENCE_SERVER_CONTAINER_NAME = "inference-server" + @dataclass class FMARequesterInfo: @@ -47,6 +51,7 @@ class FMARequesterInfo: creation_timestamp: float = 0.0 ready_timestamp: float = 0.0 dual_label_timestamp: float = 0.0 + container_start_timestamp: float = 0.0 pod: Any | None = None def dump(self) -> dict[str, Any]: @@ -84,7 +89,19 @@ class FMALauncherInfo: # pylint: disable=too-many-instance-attributes t_wake: float | None = None t_instance_create: float | None = None t_cold_launcher: float | None = None - dpc_timing_available: bool = False + # Which baseline produced this iteration's actuation timing, one of: + # "dpc" -- DPC "HTTP call done" log (highest fidelity) + # "kube_container_start"-- Kube fallback, requester inference-server + # container state.running.started_at (matches the + # dual-pods controller's actuation baseline) + # "kube_pod_create" -- Kube fallback, container-start unavailable, + # reverted to requester pod creation_timestamp + timing_source: str = "kube_pod_create" + + @property + def dpc_timing_available(self) -> bool: + """Derived convenience: True only when timing came from the DPC log.""" + return self.timing_source == "dpc" def dump(self) -> dict[str, Any]: """Convert FMALauncherInfo to dict. @@ -103,6 +120,10 @@ def dump(self) -> dict[str, Any]: else value ) + # dpc_timing_available is a derived property (not a dataclass field), so + # include it explicitly for backward-compatible downstream readers. + dump_dict["dpc_timing_available"] = self.dpc_timing_available + return dump_dict @@ -325,6 +346,96 @@ def get_dual_label_timestamp(pod: Any) -> float: return 0.0 +def get_container_start_timestamp(pod: Any) -> float: + """Return the requester inference-server container start time as an epoch. + + Mirrors the dual-pods controller's actuation baseline, which reads + ``getContainerStatus(requestingPod, "inference-server").State.Running.StartedAt``. + + Returns the ``state.running.started_at`` epoch (float) for the container + named ``inference-server``, or 0.0 when the container is missing, not + running, or its start time is unavailable. + """ + container_statuses = pod.status.container_statuses + if not container_statuses: + return 0.0 + for cs in container_statuses: + if cs.name != INFERENCE_SERVER_CONTAINER_NAME: + continue + state = getattr(cs, "state", None) + running = getattr(state, "running", None) if state is not None else None + started_at = getattr(running, "started_at", None) if running else None + if started_at is None: + return 0.0 + return started_at.astimezone(timezone.utc).timestamp() + return 0.0 + + +def select_kube_fallback_baseline( + requester_info: "FMARequesterInfo", +) -> tuple[float, str]: + """Choose the Kube-timestamp fallback baseline for an actuation. + + This is the fallback used only when DPC-log HTTP timing is unavailable. The + three ``timing_source`` values measure three DIFFERENT intervals -- they are + not one interval at three fidelities -- because each subtracts from a + different start point (all end at the requester's readiness): + + - ``dpc`` (set elsewhere, by DPC-log refinement): duration measured inside + the DPC log as ``relay_readiness - httpCallStartTime`` of the wake / + instance-create call. Its baseline is the HTTP call start, which occurs + *after* the container is already up, so it is the tightest interval + (excludes scheduling + container startup). + - ``kube_container_start`` (this function, preferred): duration measured as + ``requester ready - container state.running.startedAt``. This baseline + matches the baseline the dual-pods controller uses for its own actuation + metric -- but NOT the ``dpc`` log baseline above; it starts earlier (at + container start) and so is a coarser upper bound than ``dpc``. + - ``kube_pod_create`` (this function, reverted): duration measured from the + requester pod ``creation_timestamp`` -- an even earlier start point, so + the coarsest and most degraded of the three. (This was the harness's + original baseline before it was aligned to container start.) + + Because the sources measure different intervals, a ``dpc`` number and a + ``kube_*`` fallback number for the "same" actuation are not directly + comparable; per-iteration ``timing_source`` exists precisely so consumers + do not mix them blindly. + + This function is pure (no logging): the tentative ``timing_source`` it + returns may be overridden later by DPC-log refinement, so any warning about + a ``kube_pod_create`` reversion is deferred until the final source is known + (see :func:`warn_on_pod_create_baseline`). + + Returns a ``(baseline_epoch, timing_source)`` tuple where ``timing_source`` + is ``"kube_container_start"`` or ``"kube_pod_create"``. + """ + if requester_info.container_start_timestamp > 0.0: + return requester_info.container_start_timestamp, "kube_container_start" + + return requester_info.creation_timestamp, "kube_pod_create" + + +def warn_on_pod_create_baseline( + requester_name: str, + logger_: logging.Logger = logger, +) -> None: + """Warn that an actuation's FINAL baseline reverted to pod creation time. + + Emitted only for iterations whose final ``timing_source`` is + ``"kube_pod_create"`` (i.e. neither a container-start Kube fallback nor a + DPC-log override), so a genuine reversion is never silent while a reversion + that DPC refinement later moots produces no spurious warning. + """ + logger_.warning( + "Requester '%s': inference-server container start time unavailable; " + "reverted Kube-fallback baseline to pod creation_timestamp " + "(timing_source=kube_pod_create). This iteration's actuation number is " + "measured from an earlier baseline than the dual-pods controller's own " + "actuation metric and may be overstated.", + requester_name, + ) + + def wait_for_requester_pods( # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-branches,too-many-statements v1: client.CoreV1Api, namespace: str, @@ -358,6 +469,7 @@ def wait_for_requester_pods( # pylint: disable=too-many-arguments,too-many-posi ).timestamp() requester_info.ready_timestamp = get_ready_timestamp(p) requester_info.dual_label_timestamp = get_dual_label_timestamp(p) + requester_info.container_start_timestamp = get_container_start_timestamp(p) requester_info.pod = p all_requester_pods[p.metadata.name] = requester_info @@ -409,6 +521,9 @@ def wait_for_requester_pods( # pylint: disable=too-many-arguments,too-many-posi ).timestamp() ) requester_info.ready_timestamp = get_ready_timestamp(pod) + requester_info.container_start_timestamp = ( + get_container_start_timestamp(pod) + ) # only calculate if it wasn't already calculated if requester_info.dual_label_timestamp == 0.0: requester_info.dual_label_timestamp = get_dual_label_timestamp( @@ -860,21 +975,31 @@ def benchmark_fma( # pylint: disable=too-many-arguments,too-many-positional-arg FMAActuationCondition.T_COLD_LAUNCHER ) - # Compute per-path timing (upper bound via Kube timestamps) + # Compute per-path timing (upper bound via Kube timestamps). + # Anchor hot/warm actuation on the requester + # inference-server container start (matches the dual-pods + # controller's actuation baseline); revert to pod + # creation_timestamp only when the container start is + # unavailable, and record which + # baseline was used via timing_source. ready_ts = launcher_info.requester_info.ready_timestamp - creation_ts = launcher_info.requester_info.creation_timestamp + actuation_baseline, launcher_info.timing_source = ( + select_kube_fallback_baseline(launcher_info.requester_info) + ) if ( launcher_info.actuation_condition == FMAActuationCondition.T_HOT and ready_ts > 0.0 ): - launcher_info.t_wake = ready_ts - creation_ts + launcher_info.t_wake = ready_ts - actuation_baseline elif ( launcher_info.actuation_condition == FMAActuationCondition.T_WARM and ready_ts > 0.0 ): - launcher_info.t_instance_create = ready_ts - creation_ts + launcher_info.t_instance_create = ( + ready_ts - actuation_baseline + ) elif ( launcher_info.actuation_condition == FMAActuationCondition.T_COLD_LAUNCHER @@ -956,7 +1081,7 @@ def benchmark_fma( # pylint: disable=too-many-arguments,too-many-positional-arg refined, ) launcher_info.t_wake = refined - launcher_info.dpc_timing_available = True + launcher_info.timing_source = "dpc" elif ( launcher_info.actuation_condition == FMAActuationCondition.T_WARM @@ -970,7 +1095,7 @@ def benchmark_fma( # pylint: disable=too-many-arguments,too-many-positional-arg refined, ) launcher_info.t_instance_create = refined - launcher_info.dpc_timing_available = True + launcher_info.timing_source = "dpc" elif ( launcher_info.actuation_condition == FMAActuationCondition.T_COLD_LAUNCHER @@ -984,8 +1109,16 @@ def benchmark_fma( # pylint: disable=too-many-arguments,too-many-positional-arg refined, ) launcher_info.t_cold_launcher = refined - launcher_info.dpc_timing_available = True + launcher_info.timing_source = "dpc" else: logger.info( "No DPC timing records found; using Kube-timestamp upper bounds." ) + + # Now that DPC refinement has run, warn only for iterations whose FINAL + # timing_source is kube_pod_create -- so a reversion that DPC refinement + # overrode to "dpc" does not emit a spurious "may be overstated" warning. + for fma_iter in fma_metrics.iterations: + for launcher_info in fma_iter.launcher_infos: + if launcher_info.timing_source == "kube_pod_create": + warn_on_pod_create_baseline(launcher_info.requester_info.name)