[ci] Surface worker process logs on perf hard regressions

.buildkite/scripts/lanes/performance.sh

@@ -41,6 +41,7 @@ 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
+cp -rf fastvideo/tests/performance/results/worker_logs "$PERF_REPORTS_DIR/" 2>/dev/null || true
 
 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

@@ -297,6 +297,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,
@@ -482,6 +483,19 @@ 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 — and the performance lane
+(`.buildkite/scripts/lanes/performance.sh`) copies `results/worker_logs/` into
+`$PERF_REPORTS_DIR` next to the raw JSON results, so the CI host relays it as
+a build artifact even when the comparison phase never runs. 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 @@
         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 @@
         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__)),
@@ -693,6 +695,12 @@ def main() -> int:
         print("Performance regression check failed:")
         for item in all_failures:
             print(f"  - {item}")
+        for raw, fixed in zip(current_results, static_threshold_failures, strict=True):
+            if fixed:
+                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

@@ -1936,3 +1936,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):
+    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")
+
+    monkeypatch.setenv("PERF_RUN_SOURCE", "scheduled_main")
+    monkeypatch.delenv("PERF_PYTEST_RC", raising=False)
+    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):
+    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)})
+
+    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):
+    _run_main_with_static_breach(monkeypatch, tmp_path, {})
+
+    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

@@ -19,6 +19,10 @@
 
 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,
@@ -292,6 +296,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)
@@ -461,6 +473,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")
@@ -498,6 +511,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":
@@ -548,10 +563,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)
@@ -575,7 +599,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 copies results/worker_logs/ into PERF_REPORTS_DIR
+        # for artifact upload.
         _shutdown_executor(generator)
+        capture.close()
 
     avg_time = sum(times) / len(times)
     max_peak_memory = max(peak_memories)
@@ -594,12 +623,23 @@ 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)
+    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 = results["avg_generation_time_s"]
+    max_peak_memory = results["max_peak_memory_mb"]
     max_time = thresholds["max_generation_time_s"]
     max_mem = thresholds["max_peak_memory_mb"]
 

fastvideo/tests/performance/test_inference_performance_result_schema.py

@@ -107,9 +107,11 @@ def test_build_result_record_emits_v2_wan_shape(monkeypatch):
         },
         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"
@@ -136,6 +138,28 @@ def test_build_result_record_emits_v2_wan_shape(monkeypatch):
     assert record["vae_decode_time_s"] == 3.2
 
 
+def test_build_result_record_defaults_worker_log_path_to_none(monkeypatch):
+    monkeypatch.setenv("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)