[ci] Surface worker process logs on perf hard regressions

When the performance CI failed on a fixed-threshold breach, the only
output was a one-line message like "wan-t2v-1.3b-2gpu dit_time_s
exceeded fixed threshold (current=18.252, threshold=10.000)" - the
worker-process logs that explain why were never persisted or attached.

Capture worker logs per benchmark run via the existing executor
log_queue mechanism into a size-capped per-benchmark file, record its
path in the raw result (worker_log_path), print a 200-line tail on
hard-regression failures in both the pytest path (PR runs) and the
compare_baseline path (scheduled runs), and upload the log files as
Buildkite artifacts.

Fixes #1604

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: yogyam

.buildkite/scripts/pr_test.sh

@@ -151,6 +151,19 @@ upload_performance_artifacts() {
         fi
     }
 
+    _upload_worker_logs() {
+        local found=0
+        while IFS= read -r -d '' target; do
+            found=1
+            log "Found worker log: $target. Uploading to Buildkite..."
+            buildkite-agent artifact upload "$target"
+        done < <(find "$LOCAL_DIR" -path "*/worker_logs/worker_*.log*" -print0)
+
+        if [ "$found" -eq 0 ]; then
+            log "No worker log artifacts found."
+        fi
+    }
+
     _cleanup_modal_volume() {
         log "Cleaning up perf_reports/ from Modal Volume..."
         if modal volume rm hf-model-weights "perf_reports/" --recursive; then
@@ -170,6 +183,7 @@ upload_performance_artifacts() {
     _upload_dashboard
     _upload_perf_summary
     _upload_normalized_perf_results
+    _upload_worker_logs
     _cleanup_modal_volume
     _cleanup_local
 }

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,
@@ -480,6 +481,18 @@ 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 log file is copied to
+`$PERF_REPORTS_DIR/worker_logs/` so `upload_performance_artifacts` uploads it
+as a Buildkite 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__)),
@@ -725,6 +727,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

@@ -9,6 +9,7 @@
 import glob
 import json
 import os
+import shutil
 import time
 from collections.abc import Mapping
 from datetime import datetime, timezone
@@ -19,6 +20,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,
@@ -304,6 +309,28 @@ 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 _copy_log_to_perf_reports(log_path):
+    """Best-effort copy so the log survives as a Buildkite artifact even when
+    compare_baseline.py never runs (PR hard-regression path). No-op locally."""
+    reports_dir = os.environ.get("PERF_REPORTS_DIR", "/root/data/perf_reports")
+    try:
+        dest_dir = os.path.join(reports_dir, _WORKER_LOG_DIRNAME)
+        os.makedirs(dest_dir, exist_ok=True)
+        for path in (f"{log_path}.1", log_path):
+            if os.path.isfile(path):
+                shutil.copy2(path, dest_dir)
+    except OSError as exc:
+        logger.warning("Could not copy worker log to %s: %s", reports_dir, exc)
+
+
 def _backend_name(value) -> str:
     if hasattr(value, "name"):
         return str(value.name)
@@ -480,6 +507,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")
@@ -512,6 +540,7 @@ def _build_result_record(
         "max_peak_memory_mb": round(max_peak_memory, 1),
         "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": os.environ.get("BUILDKITE_COMMIT", ""),
         **_ci_provenance_fields(),
@@ -558,10 +587,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)
@@ -585,7 +623,11 @@ 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.
         _shutdown_executor(generator)
+        capture.close()
+        _copy_log_to_perf_reports(capture.log_path)
 
     avg_time = sum(times) / len(times)
     max_peak_memory = max(peak_memories)
@@ -604,13 +646,24 @@ 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)