[TRTLLM-13409][test] fail fast + surface server logs when a perf-sanity server dies or never becomes healthy

tests/integration/defs/perf/test_perf_sanity.py

@@ -28,7 +28,7 @@
 import pytest
 import yaml
 from test_common.error_utils import report_error
-from test_common.http_utils import wait_for_endpoint_ready
+from test_common.http_utils import fail_if_proc_died, wait_for_endpoint_ready
 
 from defs.trt_test_alternative import print_info
 from tensorrt_llm._utils import get_free_port
@@ -96,6 +96,46 @@ def ensure_bench_serving_repo() -> str:
 
 
 DEFAULT_TIMEOUT = 10800
+# Defaults for the server *ready* wait, separate from the whole-test timeout:
+# a server that is not healthy after this long is not going to be, and failing
+# here (with server-log tails, see wait_for_endpoint_ready) instead of at the
+# per-test pytest kill both saves GPU-hours and leaves a classifiable failure
+# in the CI log. The disagg bound is larger because its /health only answers
+# once EVERY ctx/gen worker has finished model load + autotune + warmup.
+AGG_SERVER_READY_TIMEOUT = 1800
+DISAGG_SERVER_READY_TIMEOUT = 3600
+
+
+def server_ready_timeout(default: int, mode: str) -> int:
+    """Ready-wait bound for one serving mode ("AGG" or "DISAGG").
+
+    Agg and disagg servers have very different init times (disagg's /health
+    answers only after every ctx/gen worker is up), so each mode has its own
+    override var, with the generic one as a shared fallback:
+    TRTLLM_TEST_<mode>_SERVER_READY_TIMEOUT > TRTLLM_TEST_SERVER_READY_TIMEOUT
+    > the built-in per-mode default.
+
+    Read at call time (not import time) so the env vars can be adjusted per
+    invocation, and parsed defensively so a malformed value cannot break
+    pytest collection of this module.
+    """
+    for var in (
+        f"TRTLLM_TEST_{mode.upper()}_SERVER_READY_TIMEOUT",
+        "TRTLLM_TEST_SERVER_READY_TIMEOUT",
+    ):
+        raw = os.environ.get(var)
+        if not raw:
+            continue
+        try:
+            timeout = int(raw)
+        except ValueError:
+            timeout = 0
+        if timeout > 0:
+            return timeout
+        print_info(f"Invalid {var}={raw!r}; ignoring it")
+    return default
+
+
 AGG_CONFIG_FOLDER = os.environ.get("AGG_CONFIG_FOLDER", "tests/scripts/perf-sanity/aggregated")
 DISAGG_CONFIG_FOLDER = os.environ.get(
     "DISAGG_CONFIG_FOLDER", "tests/scripts/perf-sanity/disaggregated"
@@ -1095,7 +1135,9 @@ def run_cmd(self, server_idx: int) -> List[str]:
 
                 wait_for_endpoint_ready(
                     f"http://{server_hostname}:{server_port}/health",
-                    timeout=self.timeout,
+                    timeout=min(
+                        self.timeout, server_ready_timeout(AGG_SERVER_READY_TIMEOUT, "AGG")
+                    ),
                     check_files=[server_file_path],
                     server_proc=server_proc,
                 )
@@ -1270,15 +1312,37 @@ def _get_disagg_server_hostname_and_port(self, server_idx: int) -> Tuple[str, in
             server_config = yaml.safe_load(f)
         return server_config["hostname"], server_config["port"]
 
-    def wait_for_benchmark_ready(self, benchmark_status_file: str):
-        """Wait for benchmark to complete."""
+    def wait_for_benchmark_ready(
+        self,
+        benchmark_status_file: str,
+        server_proc: subprocess.Popen | None = None,
+        server_log: str | None = None,
+    ):
+        """Wait for benchmark to complete, failing fast if our server dies.
+
+        The liveness check is event-driven (process exit), not a timeout: a
+        ctx/gen/disagg server that dies here raises within one loop iteration
+        with its log tail in the CI log, and the rank exits nonzero. Teardown
+        of the rest of the stage then follows from the launcher
+        (``srun --kill-on-bad-exit=1`` kills this rank's step) plus the
+        benchmark rank's bounded ready-wait failing fast on the dead endpoint
+        -- instead of every rank sitting in this loop for the full timeout.
+
+        The benchmark-done check runs FIRST so a server exiting just after a
+        completed benchmark cannot fail an otherwise-passing test.
+        """
         start_time = time.time()
         while True:
             if os.path.exists(benchmark_status_file):
                 print_info(
                     f"Benchmark status file found, terminating server {self.disagg_serving_type}"
                 )
                 break
+            fail_if_proc_died(
+                server_proc,
+                f"{self.disagg_serving_type} server",
+                [server_log] if server_log else None,
+            )
             elapsed_time = time.time() - start_time
             print_info(f"Waiting for benchmark status file, elapsed time: {elapsed_time}s")
             if elapsed_time > self.timeout:
@@ -1359,7 +1423,11 @@ def run_cmd(self, server_idx: int) -> List[str]:
                         stdout=server_ctx,
                         stderr=subprocess.STDOUT,
                     )
-                    self.wait_for_benchmark_ready(benchmark_status_file)
+                    self.wait_for_benchmark_ready(
+                        benchmark_status_file,
+                        server_proc=server_proc,
+                        server_log=server_file_path,
+                    )
             finally:
                 print_info(f"Server {self.disagg_serving_type} stopped")
                 server_proc.terminate()
@@ -1384,7 +1452,11 @@ def run_cmd(self, server_idx: int) -> List[str]:
                         stdout=disagg_server_ctx,
                         stderr=subprocess.STDOUT,
                     )
-                    self.wait_for_benchmark_ready(benchmark_status_file)
+                    self.wait_for_benchmark_ready(
+                        benchmark_status_file,
+                        server_proc=disagg_server_proc,
+                        server_log=disagg_server_file_path,
+                    )
             finally:
                 print_info(f"Disagg server {self.disagg_serving_type} stopped")
                 disagg_server_proc.terminate()
@@ -1398,7 +1470,9 @@ def run_cmd(self, server_idx: int) -> List[str]:
 
                 wait_for_endpoint_ready(
                     f"http://{disagg_server_hostname}:{disagg_server_port}/health",
-                    timeout=self.timeout,
+                    timeout=min(
+                        self.timeout, server_ready_timeout(DISAGG_SERVER_READY_TIMEOUT, "DISAGG")
+                    ),
                     check_files=self.get_server_logs(server_idx),
                 )
 

tests/integration/test_lists/test-db/l0_cpu_arm.yml

@@ -14,3 +14,4 @@ l0_cpu_arm:
       orchestrator: mpi
   tests:
   - unittest/executor/test_rpc.py
+  - unittest/others/test_http_utils_fail_fast.py

tests/integration/test_lists/test-db/l0_cpu_x86.yml

@@ -14,4 +14,5 @@ l0_cpu_x86:
       orchestrator: mpi
   tests:
   - unittest/executor/test_rpc.py
+  - unittest/others/test_http_utils_fail_fast.py
   - unittest/executor/test_multi_frontend_routing.py

tests/test_common/error_utils.py

@@ -12,6 +12,22 @@
     "PMI2_Init failed to initialize",
     "OSError",
 ]
+# Autotuner warmup intentionally probes tactics that can OOM and logs e.g.
+# "[Autotuner] Single-pair run failed ... CUDA out of memory ...". Only that
+# specific marker+OOM combination is benign: an autotuner-prefixed line with
+# any other error keyword (e.g. "[Autotuner] RuntimeError: ...") is a real
+# failure and must still be reported.
+AUTOTUNER_MARKER = "[Autotuner]"
+AUTOTUNER_BENIGN_TEXTS = [
+    "out of memory",
+]
+
+
+def is_benign_line(line: str) -> bool:
+    """True for lines expected during a HEALTHY run despite an error keyword."""
+    return AUTOTUNER_MARKER in line and any(text in line for text in AUTOTUNER_BENIGN_TEXTS)
+
+
 SLURM_LOG_TAIL_LINES = 200  # Number of lines to print from slurm job logs
 ERROR_CONTEXT_LINES = 100  # Number of lines to print before and after error line
 
@@ -23,6 +39,8 @@ def check_error(file_path: str) -> list[tuple[int, str]]:
     error_lines = []
     with open(file_path, "r", errors="replace") as f:
         for line_idx, line in enumerate(f, start=1):
+            if is_benign_line(line):
+                continue
             for keyword in ERROR_KEYWORDS:
                 if keyword in line:
                     error_lines.append((line_idx, line.strip()))
@@ -45,17 +63,22 @@ def report_error(
     for log_file in log_files:
         if not os.path.exists(log_file):
             messages.append(f"Failed to read {log_file}: Path doesn't exist")
+            continue
 
         all_lines = None
         error_lines = []
         try:
             with open(log_file, "r", errors="replace") as f:
                 all_lines = f.readlines()
-                for line_idx, line in enumerate(f, start=1):
-                    for keyword in ERROR_KEYWORDS:
-                        if keyword in line:
-                            error_lines.append((line_idx, line.strip()))
-                            break
+            # Scan the buffered lines (iterating the exhausted file handle
+            # after readlines() would yield nothing).
+            for line_idx, line in enumerate(all_lines, start=1):
+                if is_benign_line(line):
+                    continue
+                for keyword in ERROR_KEYWORDS:
+                    if keyword in line:
+                        error_lines.append((line_idx, line.strip()))
+                        break
         except Exception as e:
             all_lines = None
             error_lines = []
@@ -74,7 +97,10 @@ def report_error(
                 end_idx = min(len(all_lines), first_idx + ERROR_CONTEXT_LINES)
                 context_lines = all_lines[start_idx:end_idx]
                 messages.append("".join(context_lines))
-        else:
+        if all_lines is not None:
+            # ALWAYS include the end of the log, even when a keyword matched
+            # above: the first keyword hit may be benign noise while the
+            # actual fatal error sits at the end of the file.
             tail_content = "".join(all_lines[-tail_lines:]) if all_lines else "(empty)"
             messages.append(f"--- {log_file} [last {tail_lines} lines] ---")
             messages.append(tail_content)

tests/test_common/http_utils.py

@@ -1,38 +1,92 @@
+import os
 import subprocess
 import time
 
 import requests
 
-from test_common.error_utils import check_error
+from test_common.error_utils import check_error, report_error
+
+
+def _fail_with_server_logs(message: str, check_files: list[str] | None):
+    """Raise RuntimeError(message), appending the tail of each server log.
+
+    A server that never becomes healthy often prints its story only to its own
+    log file; without dumping it here the CI log shows nothing but the client
+    poll loop, making the failure unclassifiable.
+    """
+    if check_files:
+        # report_error raises RuntimeError(message + error context / log tails).
+        report_error(message, check_files)
+    raise RuntimeError(message)
+
+
+def fail_if_proc_died(
+    proc: subprocess.Popen | None,
+    what: str,
+    check_files: list[str] | None = None,
+):
+    """Event-driven fail-fast: raise (with server-log tails) if ``proc`` exited.
+
+    Process death is an event, not a timeout: a babysitter that checks its
+    child converts a dead server into an immediate, diagnosable failure
+    instead of burning GPUs until a timeout. The resulting nonzero rank exit
+    ends that rank's SLURM step (``srun --kill-on-bad-exit=1`` in the perf
+    scripts); the remaining steps then fail fast on the dead endpoint via the
+    bounded ready-wait.
+    """
+    if proc is None:
+        return
+    exit_code = proc.poll()
+    if exit_code is not None:
+        _fail_with_server_logs(
+            f"{what} exited unexpectedly with code {exit_code} while it was still needed.",
+            check_files,
+        )
 
 
 def wait_for_endpoint_ready(
     url: str,
     timeout: int = 300,
     check_files: list[str] | None = None,
     server_proc: subprocess.Popen | None = None,
+    check_interval: float = 30.0,
 ):
+    """Poll ``url`` until it returns 200, failing fast and loudly otherwise.
+
+    Fail-fast paths (all of which dump the tails of ``check_files`` so the
+    server-side story lands in the CI log):
+      - ``server_proc`` exited -> no point polling a dead server for the
+        remaining timeout;
+      - an error keyword appears in a ``check_files`` log (scanned every
+        ``check_interval`` seconds; time-based rather than iteration-based);
+      - ``timeout`` elapses without the endpoint becoming ready.
+    """
     start = time.monotonic()
-    iteration = 0
+    next_file_check = start + check_interval
+    missing_warned: set[str] = set()
     while time.monotonic() - start < timeout:
         # Check server_proc if provided (singular)
-        if server_proc is not None:
-            exit_code = server_proc.poll()
-            if exit_code is not None:
-                raise RuntimeError(
-                    f"Server process exited with code {exit_code} before becoming ready."
-                )
-
-        iteration += 1
-        if check_files and iteration % 300 == 0:
+        fail_if_proc_died(server_proc, "Server process (before becoming ready)", check_files)
+
+        if check_files and time.monotonic() >= next_file_check:
+            next_file_check = time.monotonic() + check_interval
             for check_file in check_files:
+                if not os.path.exists(check_file):
+                    if check_file not in missing_warned:
+                        missing_warned.add(check_file)
+                        print(
+                            f"[WARNING] server log {check_file} does not exist "
+                            "(yet?); cannot scan it for errors"
+                        )
+                    continue
                 error_lines = check_error(check_file)
                 if error_lines:
                     error_lines_str = ", ".join(
                         [f"line {line_idx}: {line_str}" for line_idx, line_str in error_lines]
                     )
-                    raise RuntimeError(
-                        f"Found error in server file {check_file}: {error_lines_str}"
+                    _fail_with_server_logs(
+                        f"Found error in server file {check_file}: {error_lines_str}",
+                        check_files,
                     )
         try:
             time.sleep(1)
@@ -41,7 +95,9 @@ def wait_for_endpoint_ready(
                 return
         except Exception as err:
             print(f"endpoint {url} is not ready, with exception: {err}")
-    raise RuntimeError(f"Endpoint {url} did not become ready within {timeout} seconds")
+    _fail_with_server_logs(
+        f"Endpoint {url} did not become ready within {timeout} seconds", check_files
+    )
 
 
 def wait_for_endpoint_down(url: str, timeout: int = 300):