diff --git a/.buildkite/scripts/lanes/performance.sh b/.buildkite/scripts/lanes/performance.sh index ab2d95d6b7..2662bfc554 100755 --- a/.buildkite/scripts/lanes/performance.sh +++ b/.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 diff --git a/docs/contributing/performance_benchmarks.md b/docs/contributing/performance_benchmarks.md index b2405eb421..a77936b1e7 100644 --- a/docs/contributing/performance_benchmarks.md +++ b/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": "", "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__.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 diff --git a/fastvideo/tests/performance/compare_baseline.py b/fastvideo/tests/performance/compare_baseline.py index a6fa9b37b7..2e0bf6054a 100644 --- a/fastvideo/tests/performance/compare_baseline.py +++ b/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") diff --git a/fastvideo/tests/performance/test_compare_baseline_policy.py b/fastvideo/tests/performance/test_compare_baseline_policy.py index 5de5b7316e..9b31f3bc42 100644 --- a/fastvideo/tests/performance/test_compare_baseline_policy.py +++ b/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 diff --git a/fastvideo/tests/performance/test_inference_performance.py b/fastvideo/tests/performance/test_inference_performance.py index 99ceea70e0..abe578b703 100644 --- a/fastvideo/tests/performance/test_inference_performance.py +++ b/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"] diff --git a/fastvideo/tests/performance/test_inference_performance_result_schema.py b/fastvideo/tests/performance/test_inference_performance_result_schema.py index 30c6191a81..da3b138d90 100644 --- a/fastvideo/tests/performance/test_inference_performance_result_schema.py +++ b/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({ diff --git a/fastvideo/tests/performance/test_worker_log_capture.py b/fastvideo/tests/performance/test_worker_log_capture.py new file mode 100644 index 0000000000..abe1f74c9b --- /dev/null +++ b/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) diff --git a/fastvideo/tests/performance/worker_log_capture.py b/fastvideo/tests/performance/worker_log_capture.py new file mode 100644 index 0000000000..1416ccf0c1 --- /dev/null +++ b/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\n{LOG_DELIMITER}" + if not tail.endswith("\n"): + tail += "\n" + return f"{header}\n{tail}{LOG_DELIMITER}"