[ci] Surface worker process logs on perf hard regressions (#1701)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: SolitaryThinker <wlsaidhi@gmail.com>

.buildkite/scripts/lanes/performance.sh

@@ -41,6 +41,18 @@ if [ "$pytest_rc" -eq 0 ] || [ "$PERF_UPLOAD_POLICY" = always ]; then
 fi
 python ./fastvideo/tests/performance/dashboard.py || true
 cp -f fastvideo/tests/performance/results/*.json "$PERF_REPORTS_DIR/" 2>/dev/null || true
+# The trusted host relays only .md/.html/.json/.csv from PERF_REPORTS_DIR, so
+# mirror each captured worker log with an allowlisted extension.
+for worker_log in fastvideo/tests/performance/results/worker_logs/*.log; do
+  [ -f "$worker_log" ] || continue
+  base=$(basename "${worker_log%.log}")
+  # WorkerLogCapture keeps a .log.1 backup after rollover, and read_log_tail
+  # includes it; mirror that retained history too so the artifact is complete.
+  if [ -f "$worker_log.1" ]; then
+    cp -f "$worker_log.1" "$PERF_REPORTS_DIR/${base}.1.md" 2>/dev/null || true
+  fi
+  cp -f "$worker_log" "$PERF_REPORTS_DIR/${base}.md" 2>/dev/null || true
+done
 
 echo "--- GPU telemetry (clocks.sm vs clocks.max.sm reveals capped hosts) ---"
 cat "$PERF_REPORTS_DIR/gpu_telemetry.csv" || true

docs/contributing/performance_benchmarks.md

@@ -357,6 +357,7 @@ Written by `test_inference_performance.py`. One file per benchmark run.
     "max_dit_time_s": 10.0,
     "max_vae_decode_time_s": 10.0
   },
+  "worker_log_path": "<container-local path to the captured worker log, or null>",
   "regression_thresholds": {
     "latency": {
       "threshold_percent": 0.10,
@@ -542,6 +543,22 @@ record is checked against its own static thresholds: a measured breach reports
 nonzero pytest exit with no attributable static-threshold breach reports
 `INFRA_ERROR`. Failed records have `success=false` and are excluded from future
 rolling baselines. The dashboard still runs best-effort for observability.
+
+Each benchmark run also captures worker-process logs into
+`results/worker_logs/worker_<benchmark_id>_<ts>.log` (raw records point to it
+via `worker_log_path`; the field is null/absent in older and HF-synced
+records). On a hard regression, a 200-line tail of that log is printed in the
+failure output — by the pytest assertion on PR runs and by
+`compare_baseline.py` on scheduled runs. The performance lane
+(`.buildkite/scripts/lanes/performance.sh`) also mirrors each captured log into
+`$PERF_REPORTS_DIR` as `worker_<benchmark_id>_<ts>.md` next to the raw JSON
+results, so the log is retrievable as a build artifact even when the
+comparison phase never runs. The mirror uses an allowlisted extension because
+the trusted host relays only `.md`, `.html`, `.json`, and `.csv` files from
+that directory; the `.log` files themselves are never uploaded. Coverage
+caveat: only the `fastvideo` logger is captured (no torch/NCCL or raw stderr
+output), and ranks > 0 suppress `logger.info` by default, so the file contains
+rank-0 INFO plus WARNING/ERROR from all ranks.
 When the rolling-baseline phase runs, it emits:
 
 * **Markdown summary** — appended to `$GITHUB_STEP_SUMMARY` when that variable

fastvideo/tests/performance/compare_baseline.py

@@ -35,6 +35,7 @@ try:
         resolve_metric_policies,
         serialize_metric_thresholds,
     )
+    from fastvideo.tests.performance.worker_log_capture import format_worker_log_tail
 except ImportError:
     repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
     if repo_root not in sys.path:
@@ -52,6 +53,7 @@ except ImportError:
         resolve_metric_policies,
         serialize_metric_thresholds,
     )
+    from fastvideo.tests.performance.worker_log_capture import format_worker_log_tail
 
 RESULTS_DIR = os.path.join(
     os.path.dirname(os.path.abspath(__file__)),
@@ -602,6 +604,7 @@ def main() -> int:
     unattributed_pytest_failure = performance_pytest_failed and not any(static_threshold_failures)
 
     all_failures: list[str] = []
+    regression_records: list[dict[str, Any]] = []
     summary_rows: list[dict[str, Any]] = []
 
     if persist_tracking:
@@ -670,6 +673,8 @@ def main() -> int:
         record["baseline_eligible"] = (identity_filters is not None and _is_baseline_eligible(
             record["run_source"], record["success"], comparison_status))
         all_failures.extend(failures)
+        if comparison_status == STATUS_REGRESSION:
+            regression_records.append(raw)
 
         print(f"{record['model_id']} comparison status: "
               f"{comparison_status} - {comparison_status_reason}")
@@ -693,6 +698,11 @@ def main() -> int:
         print("Performance regression check failed:")
         for item in all_failures:
             print(f"  - {item}")
+        for raw in regression_records:
+            print(format_worker_log_tail(
+                raw.get("benchmark_id", "unknown"),
+                raw.get("worker_log_path"),
+            ))
         return 1
 
     print("Performance baseline comparison passed")

fastvideo/tests/performance/test_compare_baseline_policy.py

@@ -1942,3 +1942,50 @@ def test_upload_policy_pass_allows_calibration_record(monkeypatch):
         "success": True,
         "comparison_status": compare_baseline.STATUS_CALIBRATION_NEEDED,
     }) is True
+
+
+def _run_main_with_static_breach(monkeypatch, tmp_path, raw_overrides, env_overrides):
+    results_dir = tmp_path / "results"
+    reports_dir = tmp_path / "reports"
+    tracking_root = tmp_path / "tracking"
+    results_dir.mkdir()
+
+    raw_result = _v2_raw_result(
+        avg_generation_time_s=18.252,
+        thresholds={"max_generation_time_s": 10.0},
+        **raw_overrides,
+    )
+    (results_dir / "perf_breach.json").write_text(json.dumps(raw_result), encoding="utf-8")
+
+    env_overrides.enter_context(envs.override_external("PERF_RUN_SOURCE", "scheduled_main"))
+    env_overrides.enter_context(envs.override_external("PERF_PYTEST_RC", None))
+    monkeypatch.setattr(compare_baseline, "RESULTS_DIR", str(results_dir))
+    monkeypatch.setattr(compare_baseline, "PERF_REPORTS_DIR", str(reports_dir))
+    monkeypatch.setattr(compare_baseline, "TRACKING_ROOT", str(tracking_root))
+    monkeypatch.setattr(compare_baseline, "UPLOAD_POLICY", "never")
+    monkeypatch.setattr(compare_baseline, "sync_from_hf", lambda local_dir, strict=False: local_dir)
+
+    assert compare_baseline.main() == 1
+
+
+def test_static_threshold_failure_prints_worker_log_tail(monkeypatch, tmp_path, capsys, env_overrides):
+    worker_log = tmp_path / "worker_wan-t2v-1.3b-2gpu.log"
+    worker_log.write_text("attention backend fell back to slow path\n", encoding="utf-8")
+
+    _run_main_with_static_breach(
+        monkeypatch, tmp_path, {"worker_log_path": str(worker_log)}, env_overrides)
+
+    output = capsys.readouterr().out
+    assert "exceeded fixed threshold" in output
+    assert "Worker log tail for wan-t2v-1.3b-2gpu" in output
+    assert "attention backend fell back to slow path" in output
+
+
+def test_static_threshold_failure_without_worker_log_degrades_gracefully(
+        monkeypatch, tmp_path, capsys, env_overrides):
+    _run_main_with_static_breach(monkeypatch, tmp_path, {}, env_overrides)
+
+    output = capsys.readouterr().out
+    assert "exceeded fixed threshold" in output
+    assert "no log file recorded" in output
+    assert "worker log unavailable" in output

fastvideo/tests/performance/test_inference_performance.py

@@ -20,6 +20,10 @@ import pytest
 import fastvideo.envs as envs
 from fastvideo import VideoGenerator
 from fastvideo.logger import init_logger
+from fastvideo.tests.performance.worker_log_capture import (
+    WorkerLogCapture,
+    format_worker_log_tail,
+)
 from fastvideo.tests.performance.identity import (
     benchmark_identity_from_config,
     build_recipe_from_benchmark_config,
@@ -315,6 +319,14 @@ def _write_results(results):
     logger.info("Performance results written to %s", filepath)
 
 
+_WORKER_LOG_DIRNAME = "worker_logs"
+
+
+def _worker_log_dir():
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    return os.path.join(script_dir, "results", _WORKER_LOG_DIRNAME)
+
+
 def _backend_name(value) -> str:
     if hasattr(value, "name"):
         return str(value.name)
@@ -484,6 +496,7 @@ def _build_result_record(
     runtime_identity: Mapping[str, Any],
     device_name: str,
     timestamp: str | None = None,
+    worker_log_path: str | None = None,
 ) -> dict[str, Any]:
     if not times or not peak_memories:
         raise ValueError("Cannot build a performance result record without measurement runs")
@@ -521,6 +534,8 @@ def _build_result_record(
         "individual_peak_memories_mb": [round(m, 1) for m in peak_memories],
         "thresholds":
         dict(thresholds),
+        "worker_log_path":
+        worker_log_path,
         "regression_thresholds":
         cfg.get("regression_thresholds", {}),
         "commit":
@@ -594,10 +609,19 @@ def _run_benchmark(cfg):
     os.makedirs(output_dir, exist_ok=True)
     gen_kwargs["output_path"] = output_dir
 
+    capture = WorkerLogCapture(
+        _worker_log_dir(),
+        cfg["benchmark_id"],
+        datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"),
+    )
     generator = None
     try:
+        # log_queue only goes to from_pretrained: workers keep the handler for
+        # their lifetime, covering model load + warmups + measured runs.
+        # Passing it to generate_video would detach it after the first call.
         generator = VideoGenerator.from_pretrained(
             model_path=model_info["model_path"],
+            log_queue=capture.log_queue,
             **init_kwargs,
         )
         runtime_identity = _runtime_identity_from_generator(generator)
@@ -621,7 +645,12 @@ def _run_benchmark(cfg):
             peak_memories.append(peak_mb)
             all_component_times.append(component_times)
     finally:
+        # Shutdown stops workers producing; close() then drains the queue so
+        # the log file is complete before any assertion reads it back. The
+        # performance CI lane mirrors results/worker_logs/*.log into
+        # PERF_REPORTS_DIR as allowlisted .md files for artifact upload.
         _shutdown_executor(generator)
+        capture.close()
 
     avg_time = sum(times) / len(times)
     max_peak_memory = max(peak_memories)
@@ -640,12 +669,22 @@ def _run_benchmark(cfg):
         prompt=prompt,
         runtime_identity=runtime_identity,
         device_name=device_name,
+        worker_log_path=capture.log_path,
     )
 
     logger.info("Performance results: avg_time=%.2fs, "
                 "max_peak_memory=%.0fMB", avg_time, max_peak_memory)
     _write_results(results)
 
+    try:
+        _assert_thresholds(results, thresholds, device_name, avg_time=avg_time, max_peak_memory=max_peak_memory)
+    except AssertionError:
+        print(format_worker_log_tail(cfg["benchmark_id"], capture.log_path), flush=True)
+        raise
+
+
+def _assert_thresholds(results, thresholds, device_name, avg_time, max_peak_memory):
+    """Gate on the raw measurements: the emitted record stores rounded copies."""
     max_time = thresholds["max_generation_time_s"]
     max_mem = thresholds["max_peak_memory_mb"]
 

fastvideo/tests/performance/test_inference_performance_result_schema.py

@@ -108,9 +108,11 @@ def test_build_result_record_emits_v2_wan_shape(monkeypatch, env_overrides):
         },
         device_name="NVIDIA L40S",
         timestamp="2026-07-05T00:00:00+00:00",
+        worker_log_path="/tmp/worker_logs/worker_wan-t2v-1.3b-2gpu.log",
     )
 
     assert record["result_schema_version"] == perf_test.RESULT_SCHEMA_VERSION
+    assert record["worker_log_path"] == "/tmp/worker_logs/worker_wan-t2v-1.3b-2gpu.log"
     assert record["benchmark_id"] == "wan-t2v-1.3b-2gpu"
     assert record["workload_id"] == "wan-t2v"
     assert record["variant_id"] == "1.3b-sp2"
@@ -137,6 +139,28 @@ def test_build_result_record_emits_v2_wan_shape(monkeypatch, env_overrides):
     assert record["vae_decode_time_s"] == 3.2
 
 
+def test_build_result_record_defaults_worker_log_path_to_none(monkeypatch, env_overrides):
+    env_overrides.enter_context(envs.override_external("PERF_RUN_SOURCE", "scheduled_main"))
+    record = perf_test._build_result_record(
+        cfg={"benchmark_id": "wan-t2v-1.3b-2gpu"},
+        model_info={},
+        init_kwargs={},
+        gen_kwargs={},
+        num_warmup=1,
+        num_measure=1,
+        thresholds={},
+        times=[10.0],
+        peak_memories=[10000.0],
+        all_component_times=[],
+        prompt="A cinematic video.",
+        runtime_identity={},
+        device_name="NVIDIA L40S",
+        timestamp="2026-07-05T00:00:00+00:00",
+    )
+
+    assert record["worker_log_path"] is None
+
+
 def test_validate_run_counts_rejects_zero_measurement_runs():
     with pytest.raises(ValueError, match="num_measurement_runs"):
         perf_test._validate_run_counts({

fastvideo/tests/performance/test_worker_log_capture.py

@@ -0,0 +1,93 @@
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+
+from fastvideo.tests.performance.worker_log_capture import (
+    DEFAULT_TAIL_LINES,
+    LOG_DELIMITER,
+    WorkerLogCapture,
+    format_worker_log_tail,
+    read_log_tail,
+)
+
+
+def _make_record(msg):
+    return logging.LogRecord(
+        name="fastvideo.worker",
+        level=logging.INFO,
+        pathname=__file__,
+        lineno=1,
+        msg=msg,
+        args=(),
+        exc_info=None,
+    )
+
+
+def test_capture_round_trips_queue_and_parent_logger(tmp_path):
+    capture = WorkerLogCapture(str(tmp_path), "bench", "20260807T000000Z")
+    try:
+        capture.log_queue.put(_make_record("from worker queue"))
+        logging.getLogger("fastvideo").warning("from parent process")
+    finally:
+        capture.close()
+
+    content = open(capture.log_path, encoding="utf-8").read()
+    assert "from worker queue" in content
+    assert "from parent process" in content
+
+
+def test_close_drains_pending_queue_records(tmp_path):
+    capture = WorkerLogCapture(str(tmp_path), "bench", "20260807T000000Z")
+    for i in range(50):
+        capture.log_queue.put(_make_record(f"pending record {i}"))
+    capture.close()
+
+    content = open(capture.log_path, encoding="utf-8").read()
+    for i in range(50):
+        assert f"pending record {i}" in content
+
+
+def test_read_log_tail_truncates_to_last_lines(tmp_path):
+    log_path = tmp_path / "worker_bench.log"
+    total = DEFAULT_TAIL_LINES + 50
+    log_path.write_text("".join(f"line {i}\n" for i in range(total)), encoding="utf-8")
+
+    tail = read_log_tail(str(log_path))
+    assert tail is not None
+    assert f"showing last {DEFAULT_TAIL_LINES} of {total} lines" in tail
+    assert f"line {total - 1}" in tail
+    assert "line 0\n" not in tail
+
+
+def test_read_log_tail_concatenates_rotated_backup(tmp_path):
+    log_path = tmp_path / "worker_bench.log"
+    (tmp_path / "worker_bench.log.1").write_text("older rotated line\n", encoding="utf-8")
+    log_path.write_text("newer live line\n", encoding="utf-8")
+
+    tail = read_log_tail(str(log_path))
+    assert tail == "older rotated line\nnewer live line\n"
+
+
+def test_read_log_tail_handles_missing_paths():
+    assert read_log_tail(None) is None
+    assert read_log_tail("/nonexistent/worker.log") is None
+
+
+def test_format_worker_log_tail_renders_unavailable_block():
+    block = format_worker_log_tail("bench", None)
+    assert LOG_DELIMITER in block
+    assert "no log file recorded" in block
+    assert "worker log unavailable" in block
+
+    block = format_worker_log_tail("bench", "/nonexistent/worker.log")
+    assert "worker log unavailable" in block
+
+
+def test_format_worker_log_tail_renders_content(tmp_path):
+    log_path = tmp_path / "worker_bench.log"
+    log_path.write_text("dit slowdown detected\n", encoding="utf-8")
+
+    block = format_worker_log_tail("bench", str(log_path))
+    assert "Worker log tail for bench" in block
+    assert "dit slowdown detected" in block
+    assert block.endswith(LOG_DELIMITER)

fastvideo/tests/performance/worker_log_capture.py

@@ -0,0 +1,97 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Capture worker-process logs during performance benchmark runs.
+
+Workers spawned by MultiprocExecutor forward records from the "fastvideo"
+logger into a multiprocessing queue when one is passed to
+``VideoGenerator.from_pretrained(..., log_queue=...)``. This module consumes
+that queue into a size-capped per-benchmark log file so the log can be
+attached to CI failures and uploaded as a build artifact.
+
+Coverage caveats: only the "fastvideo" logger is forwarded (no torch/NCCL or
+raw stderr output), and ranks > 0 suppress ``logger.info`` by default
+(``local_main_process_only=True``), so the file contains rank-0 INFO plus
+WARNING/ERROR from all ranks.
+"""
+
+import logging
+import logging.handlers
+import multiprocessing
+import os
+
+LOG_MAX_BYTES = 10 * 1024 * 1024
+DEFAULT_TAIL_LINES = 200
+LOG_DELIMITER = "=" * 78
+
+
+class WorkerLogCapture:
+    """Capture "fastvideo" logger output from the test process and all worker
+    subprocesses into a size-capped per-benchmark log file."""
+
+    def __init__(self, log_dir: str, benchmark_id: str, timestamp: str):
+        os.makedirs(log_dir, exist_ok=True)
+        self.log_path = os.path.join(log_dir, f"worker_{benchmark_id}_{timestamp}.log")
+        # Manager queue required: with the spawn start method a plain
+        # mp.Queue() cannot be shipped to workers through the executor RPC.
+        self._manager = multiprocessing.Manager()
+        self.log_queue = self._manager.Queue()
+        self._file_handler = logging.handlers.RotatingFileHandler(
+            self.log_path, maxBytes=LOG_MAX_BYTES, backupCount=1, encoding="utf-8")
+        self._file_handler.setFormatter(
+            logging.Formatter("%(asctime)s [%(levelname)s] %(processName)s %(name)s: %(message)s"))
+        self._listener = logging.handlers.QueueListener(
+            self.log_queue, self._file_handler, respect_handler_level=True)
+        self._listener.start()
+        # Also persist parent-process logs (run markers, load orchestration)
+        # so the file reads as a single timeline.
+        self._parent_logger = logging.getLogger("fastvideo")
+        self._parent_logger.addHandler(self._file_handler)
+
+    def close(self) -> None:
+        """Drain the queue and release resources.
+
+        Call after the executor is shut down (workers stop producing) and
+        before any threshold assertion reads the log back.
+        """
+        try:
+            self._parent_logger.removeHandler(self._file_handler)
+            self._listener.stop()
+            self._file_handler.close()
+        finally:
+            self._manager.shutdown()
+
+
+def read_log_tail(log_path: str | None, max_lines: int = DEFAULT_TAIL_LINES) -> str | None:
+    """Return the last ``max_lines`` of the capture, or None if unavailable.
+
+    Concatenates the rotated ``.log.1`` backup (older) with the live file
+    (newer) so the tail spans a rollover boundary.
+    """
+    if not log_path:
+        return None
+    lines: list[str] = []
+    for path in (f"{log_path}.1", log_path):
+        try:
+            if os.path.isfile(path):
+                with open(path, encoding="utf-8", errors="replace") as f:
+                    lines.extend(f.readlines())
+        except OSError:
+            continue
+    if not lines:
+        return None
+    tail = lines[-max_lines:]
+    if len(lines) > max_lines:
+        tail.insert(0, f"... truncated, showing last {max_lines} of {len(lines)} lines ...\n")
+    return "".join(tail)
+
+
+def format_worker_log_tail(benchmark_id: str, log_path: str | None) -> str:
+    """Render a delimited worker-log tail block for CI failure output."""
+    tail = read_log_tail(log_path)
+    header = (f"{LOG_DELIMITER}\n"
+              f"Worker log tail for {benchmark_id} ({log_path or 'no log file recorded'})\n"
+              f"{LOG_DELIMITER}")
+    if tail is None:
+        return f"{header}\n<worker log unavailable: file missing or unreadable>\n{LOG_DELIMITER}"
+    if not tail.endswith("\n"):
+        tail += "\n"
+    return f"{header}\n{tail}{LOG_DELIMITER}"