Skip to content

Consume centralized DPC "HTTP call done" log line; align actuation baseline #1583

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
maugustosilva merged 6 commits into from
Jul 6, 2026
Merged

Consume centralized DPC "HTTP call done" log line; align actuation baseline #1583

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3b592ee
feat: Parse centralized "HTTP call done" DPC line
rubambiza Jul 1, 2026
3ab58a8
test: keep large-file + partial-log coverage for new format
rubambiza Jul 1, 2026
fd408c8
test: use non-indicator filler to preserve 256KB regression
rubambiza Jul 1, 2026
ce751ee
feat: align actuation baseline to container start (#599)
rubambiza Jul 1, 2026
98ba62e
fix: only warn on final kube_pod_create timing_source
rubambiza Jul 1, 2026
f5fbc11
docs: clarify timing_source baselines, drop bare issue refs
rubambiza Jul 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions llmdbenchmark/analysis/benchmark_report/native_to_br0_1.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand All @@ -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
)
Expand Down
21 changes: 18 additions & 3 deletions llmdbenchmark/analysis/scripts/nop-analyze_results.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
{
Expand Down
63 changes: 35 additions & 28 deletions tests/test_dpc_log_parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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="<token>" 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"
"""


Expand Down Expand Up @@ -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)
Expand All @@ -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))
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand Down
Loading
Loading