Capture vllm server log into test_vllm_inference output

cvs/lib/inference/unittests/test_vllm_job_server_reuse.py

@@ -60,6 +60,23 @@ def exec_on_head(self, cmd, *a, **k):
         return {}
 
 
+class FakeOrchMultiHost:
+    """Two-host fake orch that records which hosts each exec() call targeted."""
+
+    hosts = ["10.0.0.1", "10.0.0.2"]
+
+    def __init__(self):
+        self.exec_calls = []  # list of (cmd, hosts) actually issued
+
+    def exec(self, cmd, hosts=None, detailed=False):
+        self.exec_calls.append((cmd, hosts))
+        host = hosts[0]
+        return {host: f"content for {host}"}
+
+    def exec_on_head(self, cmd, *a, **k):
+        return {}
+
+
 def _make_job_for_check(tail_output="", grep_exit=1):
     """Construct a VllmJob suitable for testing _check_early_failure."""
     variant = mock.MagicMock()
@@ -261,6 +278,52 @@ def test_raises_on_cli_parse_error(self):
             job._check_early_failure()
 
 
+class TestDumpServerLog(unittest.TestCase):
+    def test_logs_full_content_per_rank(self):
+        job = _make_job_for_check(tail_output="line one\nline two")
+        with mock.patch("cvs.lib.inference.vllm_job.log") as mock_log:
+            job.dump_server_log()
+        logged_lines = [call.args[3] for call in mock_log.info.call_args_list if len(call.args) >= 4]
+        self.assertIn("line one", logged_lines)
+        self.assertIn("line two", logged_lines)
+
+    def test_mp_multinode_dumps_every_rank(self):
+        """mp backend: every rank runs its own vllm serve, so every rank is dumped."""
+        orch = FakeOrchMultiHost()
+        job = VllmJob(
+            orch=orch,
+            variant=_variant({"distributed-executor-backend": "mp"}),
+            hf_token="tok",
+            isl="1024",
+            osl="1024",
+            concurrency=8,
+            num_prompts="640",
+        )
+        with mock.patch("cvs.lib.inference.vllm_job.log") as mock_log:
+            job.dump_server_log()
+        ranks_dumped = {call.args[2] for call in mock_log.info.call_args_list if len(call.args) >= 4}
+        self.assertEqual(ranks_dumped, {0, 1})
+        self.assertEqual(len(orch.exec_calls), 2, "one cat per rank")
+
+    def test_ray_multinode_skips_worker_ranks(self):
+        """Ray multinode: only rank 0 runs vllm serve, so only rank 0 is dumped."""
+        orch = FakeOrchMultiHost()
+        job = VllmJob(
+            orch=orch,
+            variant=_variant({"distributed-executor-backend": "ray"}),
+            hf_token="tok",
+            isl="1024",
+            osl="1024",
+            concurrency=8,
+            num_prompts="640",
+        )
+        with mock.patch("cvs.lib.inference.vllm_job.log") as mock_log:
+            job.dump_server_log()
+        ranks_dumped = {call.args[2] for call in mock_log.info.call_args_list if len(call.args) >= 4}
+        self.assertEqual(ranks_dumped, {0}, "worker rank 1 has no server log under ray and must be skipped")
+        self.assertEqual(len(orch.exec_calls), 1, "only rank 0's cat should be issued")
+
+
 class TestRoleServerLogLevelValidator(unittest.TestCase):
     def test_invalid_log_level_rejected(self):
         with self.assertRaises(pydantic.ValidationError) as ctx:

cvs/lib/inference/vllm_job.py

@@ -613,6 +613,25 @@ def dump_client_log(self):
             for line in (text or "").splitlines():
                 log.info("[%s client.log] %s", host, line)
 
+    def dump_server_log(self):
+        """Emit each rank's full server log to the captured section once.
+
+        Mirrors dump_client_log(), but per-rank since every node (other than
+        ray-backend headless workers, which never run vllm serve) has its own
+        server log. Call this after the benchmark finishes -- success or
+        failure -- so a mid-run server crash (e.g. EngineDeadError) is
+        preserved in the captured output even though _check_early_failure()
+        stops tailing this log once startup is confirmed.
+        """
+        for rank, host in enumerate(self.orch.hosts):
+            if self._is_ray_backend and int(self.nnodes) > 1 and rank > 0:
+                continue
+            rank_log = self._rank_log(rank)
+            out = self.orch.exec(f"cat {shlex.quote(rank_log)}", hosts=[host])
+            for h, text in (out or {}).items():
+                for line in (text or "").splitlines():
+                    log.info("[%s rank%d server.log] %s", h, rank, line)
+
     def parse_results(self):
         """Fetch and parse the results artifact from the HEAD node via exec_on_head."""
         artifact = f"{self.out_dir}/results"

cvs/tests/inference/vllm/vllm.py

@@ -323,6 +323,10 @@ def test_vllm_inference(orch, variant_config, hf_token, seq_combo, concurrency,
         else:
             job.stop_server()
             job.build_server_cmd()
+            # Attribute the server-log dump target to this job as soon as it
+            # owns the (about-to-be-)running server, so a start_server/wait_ready
+            # failure below still dumps the right log via the except handler.
+            lifecycle.live_server_job = job
             pre_snap = _gpu_snap(orch)
             t = time.monotonic()
             job.start_server()
@@ -367,6 +371,15 @@ def test_vllm_inference(orch, variant_config, hf_token, seq_combo, concurrency,
         # A failed cell may have left the server in a bad state; force the next
         # cell to do a clean bringup rather than reuse a possibly-dead server.
         lifecycle.live_server_sig = None
+        # Dump from whichever job actually owns the running server -- on the
+        # reuse path (this cell only differs by concurrency) that's an earlier
+        # cell's job, not this one, since only the job that called
+        # build_server_cmd()/start_server() has a server_log path the server
+        # process is actually writing to. Only dumped on failure: the server
+        # log is per-server (not per-cell), so a success-path dump would
+        # re-emit the same growing log after every cell sharing a reused
+        # server.
+        getattr(lifecycle, "live_server_job", job).dump_server_log()
         raise
 
     agg = agg_readings(poll_readings)