From b30bf9500f8a4eedfcddac5626ab3d400e6f12b0 Mon Sep 17 00:00:00 2001 From: Rashid Kaleem <230885705+arekay-nv@users.noreply.github.com> Date: Mon, 2 Nov 2026 21:59:52 -0600 Subject: [PATCH] test: wait for the performance phase before ^C in sigint tests test_sigint_grace_expiry_abandons_wedged_drain waited for metrics/.ready and then slept a fixed 3 s before SIGSTOPping the aggregator. .ready precedes HTTP worker startup and the session's STARTED, and the aggregator publishes no live snapshot until STARTED. When worker startup outlasted the sleep, the wedged-then-SIGKILLed aggregator left neither final_snapshot.json nor a live frame, so no result_summary.json was written. Wait for the "Starting phase: performance" log line (logged right after STARTED) before the sleep, in this test and in test_group_sigint_exits_gracefully, which shares the pattern. Capture the run log in the grace-expiry test and print its tail if the summary is missing. Reuse the new _wait_for_log helper in the drain-window test, and note in BenchmarkSession._run_phase that tests key on this log line. Closes #558 Co-Authored-By: Claude Opus 5.5 --- .../load_generator/session.py | 4 ++ tests/integration/commands/test_sigint.py | 56 ++++++++++++++----- 2 files changed, 46 insertions(+), 14 deletions(-) diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py index 896646f81..51b3d21a1 100644 --- a/src/inference_endpoint/load_generator/session.py +++ b/src/inference_endpoint/load_generator/session.py @@ -500,6 +500,10 @@ async def run( async def _run_phase(self, phase: PhaseConfig) -> PhaseResult | None: """Run a single phase. Returns PhaseResult or None for warmup.""" + # Tests key on this log line: tests/integration/commands/test_sigint.py + # waits for it as the "workers up, STARTED published" signal, and + # tests/integration/commands/test_benchmark_command.py asserts on it. + # Update them if the wording changes. logger.info("Starting phase: %s (%s)", phase.name, phase.phase_type.value) phase_start = time.monotonic_ns() # Per-phase stop flag is scoped to this phase; clear any cap left set by diff --git a/tests/integration/commands/test_sigint.py b/tests/integration/commands/test_sigint.py index 1197744dc..24caab057 100644 --- a/tests/integration/commands/test_sigint.py +++ b/tests/integration/commands/test_sigint.py @@ -136,8 +136,11 @@ def _benchmark_proc( def _wait_services_ready( proc: subprocess.Popen, report_dir: Path, timeout_s: float = 60.0 ) -> None: - """Block until the aggregator touches metrics/.ready (handlers installed); - the session starts issuing right after service readiness.""" + """Block until the aggregator touches metrics/.ready (handlers installed). + + Readiness precedes HTTP worker startup and the session's STARTED, so the + run may not be measuring yet; wait for ``_PERF_PHASE_STARTED`` for that. + """ ready = report_dir / "metrics" / ".ready" deadline = time.monotonic() + timeout_s while not ready.exists(): @@ -146,6 +149,23 @@ def _wait_services_ready( time.sleep(0.1) +# Logged by BenchmarkSession._run_phase right after the session publishes +# STARTED; from here the aggregator is LIVE and publishes a snapshot every +# 0.25 s (its default --publish-interval). +_PERF_PHASE_STARTED = "Starting phase: performance" + + +def _wait_for_log( + proc: subprocess.Popen, log_file: Path, needle: str, timeout_s: float = 60.0 +) -> None: + """Block until the benchmark's captured output contains ``needle``.""" + deadline = time.monotonic() + timeout_s + while needle not in log_file.read_text(errors="replace"): + assert proc.poll() is None, f"benchmark exited before logging {needle!r}" + assert time.monotonic() < deadline, f"benchmark never logged {needle!r}" + time.sleep(0.05) + + def _iter_proc_cmdlines() -> Iterator[tuple[int, bytes]]: """(pid, cmdline) for every live process (Linux); racing exits skipped.""" for pid_dir in Path("/proc").iterdir(): @@ -221,7 +241,8 @@ def test_group_sigint_exits_gracefully(mock_http_echo_server, tmp_path): log_file, ) as proc: _wait_services_ready(proc, report_dir) - time.sleep(3.0) + _wait_for_log(proc, log_file, _PERF_PHASE_STARTED) + time.sleep(3.0) # well inside the ~120 s performance phase os.killpg(proc.pid, signal.SIGINT) rc = proc.wait(timeout=60.0) @@ -283,13 +304,18 @@ def test_sigint_grace_expiry_abandons_wedged_drain(mock_http_echo_server, tmp_pa interrupted_teardown_grace_s=3.0, ) + log_file = tmp_path / "run.log" agg_pid: int | None = None try: with _benchmark_proc( - [_cli(), "benchmark", "from-config", "-c", str(config_path)] + [_cli(), "benchmark", "from-config", "-c", str(config_path)], + log_file=log_file, ) as proc: _wait_services_ready(proc, report_dir) - time.sleep(3.0) # comfortably inside the ~120 s performance phase + # Without a live snapshot to fall back on, the SIGKILLed + # aggregator leaves nothing to build result_summary.json from. + _wait_for_log(proc, log_file, _PERF_PHASE_STARTED) + time.sleep(3.0) # many publish ticks past STARTED agg_pid = _pid_of_child("metrics_aggregator", str(report_dir)) assert agg_pid is not None, "aggregator child not found" @@ -314,9 +340,12 @@ def test_sigint_grace_expiry_abandons_wedged_drain(mock_http_echo_server, tmp_pa # The SIGKILLed aggregator never wrote a terminal snapshot, so the report # was built from the last live pub/sub frame — the split-brain guard must # still land the summary as interrupted, never state:"live". - summary = json.loads( - (report_dir / "performance" / "result_summary.json").read_text() + summary_path = report_dir / "performance" / "result_summary.json" + assert summary_path.exists(), ( + "no result_summary.json; run log tail:\n" + + log_file.read_text(errors="replace")[-4000:] ) + summary = json.loads(summary_path.read_text()) assert summary["state"] == "interrupted" assert summary["complete"] is False _assert_no_leftover_children(report_dir, "grace-expired abort") @@ -410,13 +439,12 @@ def test_sigint_in_drain_window_keeps_summary_authoritative( ) as proc: # The pipeline logs this line once the session has ENDED and the # aggregator drain wait begins — the divergence window. - deadline = time.monotonic() + 120.0 - while "Waiting for services to finish processing" not in log_file.read_text( - errors="replace" - ): - assert proc.poll() is None, "benchmark exited before the drain" - assert time.monotonic() < deadline, "drain window never reached" - time.sleep(0.05) + _wait_for_log( + proc, + log_file, + "Waiting for services to finish processing", + timeout_s=120.0, + ) os.kill(proc.pid, signal.SIGINT) # main process only: aggregator unsignaled rc = proc.wait(timeout=120.0)