# SPDX-License-Identifier: Apache-2.0 """Track performance results and compare against historical baseline. This script: 1) reads current benchmark results from fastvideo/tests/performance/results, 2) syncs the canonical baseline from the configured HF dataset repo, 3) compares each current record against the median of up to 5 prior baseline-eligible successful records in the same workload/variant/version, GPU, recipe, hardware, and software cohort, 4) writes normalized records back to the HF dataset repo according to PERF_UPLOAD_POLICY, 5) exits non-zero if any gated metric exceeds both its percent and absolute regression floors. """ import glob import json import os import statistics import sys from datetime import datetime, timezone from typing import Any try: from fastvideo.performance.hf_store import ( load_records_for_identity, safe_float, sanitize, sync_from_hf, upload_record, ) from fastvideo.performance.metric_policy import ( MetricPolicy, regression_delta, resolve_metric_policies, serialize_metric_thresholds, ) except ImportError: repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")) if repo_root not in sys.path: sys.path.insert(0, repo_root) from fastvideo.performance.hf_store import ( load_records_for_identity, safe_float, sanitize, sync_from_hf, upload_record, ) from fastvideo.performance.metric_policy import ( MetricPolicy, regression_delta, resolve_metric_policies, serialize_metric_thresholds, ) RESULTS_DIR = os.path.join( os.path.dirname(os.path.abspath(__file__)), "results", ) TRACKING_ROOT = os.environ.get( "PERFORMANCE_TRACKING_ROOT", "/tmp/perf-tracking", ) PERF_REPORTS_DIR = os.environ.get("PERF_REPORTS_DIR", "/root/data/perf_reports") UPLOAD_POLICY = os.environ.get("PERF_UPLOAD_POLICY", "never").strip().lower() VALID_UPLOAD_POLICIES = {"never", "pass", "always"} VALID_RUN_SOURCES = {"pr", "local", "scheduled_main", "unknown"} IDENTITY_KEYS = ( "result_schema_version", "workload_id", "variant_id", "benchmark_version", "recipe", "recipe_fingerprint", "hardware_profile", "hardware_profile_id", "software_profile", "software_profile_id", "environment_metadata", "environment_fingerprint", "quality_metadata", "variant_metadata", ) COMPARISON_IDENTITY_KEYS = ( "workload_id", "variant_id", "benchmark_version", "recipe_fingerprint", "hardware_profile_id", "software_profile_id", ) STATUS_PASS = "PASS" STATUS_REGRESSION = "REGRESSION" STATUS_CALIBRATION_NEEDED = "CALIBRATION_NEEDED" STATUS_RECIPE_MISMATCH = "RECIPE_MISMATCH" STATUS_INFRA_ERROR = "INFRA_ERROR" # Reserved for the promoted-baseline workflow; never emitted here. STATUS_QUALITY_BLOCKED = "QUALITY_BLOCKED" STATIC_THRESHOLD_FIELDS = ( ("avg_generation_time_s", "max_generation_time_s"), ("max_peak_memory_mb", "max_peak_memory_mb"), ("text_encoder_time_s", "max_text_encoder_time_s"), ("dit_time_s", "max_dit_time_s"), ("vae_decode_time_s", "max_vae_decode_time_s"), ) def _should_persist_tracking() -> bool: return _normalized_upload_policy() != "never" def _normalized_upload_policy() -> str: if UPLOAD_POLICY in VALID_UPLOAD_POLICIES: return UPLOAD_POLICY print(f"Invalid PERF_UPLOAD_POLICY={UPLOAD_POLICY!r}; using 'never'") return "never" def _truthy_pr_number(value: str | None) -> bool: return bool(value and value not in {"false", "0", "None", "none"}) def _detect_run_source() -> str: explicit = os.environ.get("PERF_RUN_SOURCE", "").strip().lower() if explicit in VALID_RUN_SOURCES: return explicit if explicit: print(f"Invalid PERF_RUN_SOURCE={explicit!r}; inferring run source") if _truthy_pr_number(os.environ.get("BUILDKITE_PULL_REQUEST")): return "pr" if os.environ.get("BUILDKITE_BRANCH") == "main" and os.environ.get("TEST_SCOPE") == "full": return "scheduled_main" if not os.environ.get("BUILDKITE_COMMIT"): return "local" return "unknown" def _is_baseline_eligible( run_source: str, success: bool, comparison_status: str | None = None, ) -> bool: if comparison_status is not None and comparison_status != STATUS_PASS: return False return run_source == "scheduled_main" and success def _upload_allowed(record: dict[str, Any]) -> bool: policy = _normalized_upload_policy() if policy == "always": return True if policy == "pass": return bool(record.get("success", True)) return False def _performance_pytest_failed() -> bool: value = os.environ.get("PERF_PYTEST_RC", "") if not value: return False try: return int(value) != 0 except ValueError: return False def _record_metadata(run_source: str, result: dict[str, Any]) -> dict[str, Any]: raw_run_source = str(result.get("run_source") or "").strip().lower() if raw_run_source in VALID_RUN_SOURCES: run_source = raw_run_source pr_number = result.get("pr_number") or os.environ.get("BUILDKITE_PULL_REQUEST", "") if not _truthy_pr_number(str(pr_number)): pr_number = "" return { "run_source": run_source, "baseline_eligible": False, "branch": result.get("branch") or os.environ.get("BUILDKITE_BRANCH", ""), "pr_number": pr_number, "test_scope": result.get("test_scope") or os.environ.get("TEST_SCOPE", ""), "build_url": result.get("build_url") or os.environ.get("BUILDKITE_BUILD_URL", ""), "build_id": result.get("build_id") or os.environ.get("BUILDKITE_BUILD_ID", ""), "job_id": result.get("job_id") or os.environ.get("BUILDKITE_JOB_ID", ""), } def _identity_metadata(result: dict[str, Any]) -> dict[str, Any]: metadata = {key: result[key] for key in IDENTITY_KEYS if key in result and result[key] is not None} recipe = result.get("recipe") benchmark = recipe.get("benchmark") if isinstance(recipe, dict) else None if isinstance(benchmark, dict): for key in ("workload_id", "variant_id", "benchmark_version"): if key not in metadata and benchmark.get(key) is not None: metadata[key] = benchmark[key] return metadata def _comparison_identity_filters(record: dict[str, Any]) -> dict[str, str]: missing = [ key for key in COMPARISON_IDENTITY_KEYS if key not in record or record[key] is None or (isinstance(record[key], str) and not record[key].strip()) ] if missing: raise ValueError("Performance record missing required comparison identity fields: " + ", ".join(missing)) return {key: str(record[key]) for key in COMPARISON_IDENTITY_KEYS} def _record_uses_v2_identity(record: dict[str, Any]) -> bool: schema_version = record.get("result_schema_version") return str(schema_version) == "2" or any(key in record for key in COMPARISON_IDENTITY_KEYS) def _recipe_cohort_filters(record: dict[str, Any]) -> dict[str, str]: identity = _comparison_identity_filters(record) return {key: identity[key] for key in ("workload_id", "variant_id", "benchmark_version")} def _comparison_identity_filters_or_none(record: dict[str, Any]) -> dict[str, str] | None: try: return _comparison_identity_filters(record) except ValueError as exc: if _record_uses_v2_identity(record): print(f"Invalid v2 comparison identity for {record.get('model_id', 'unknown')}: {exc}") else: print("Skipping rolling baseline comparison for " f"{record.get('model_id', 'unknown')} on " f"{record.get('gpu_type', 'unknown')}: {exc}. " "The normalized artifact will still be written, but this record " "will not be baseline eligible.") return None def _recipe_mismatch_records( record: dict[str, Any], cohort_records: list[dict[str, Any]], ) -> list[dict[str, Any]]: current_recipe = str(record.get("recipe_fingerprint")) return [ item for item in cohort_records if _record_can_authorize_recipe_mismatch(item) and item.get("recipe_fingerprint") is not None and str(item.get("recipe_fingerprint")).strip() and str(item.get("recipe_fingerprint")) != current_recipe ] def _record_can_authorize_recipe_mismatch(record: dict[str, Any]) -> bool: return record.get("run_source") == "scheduled_main" or record.get("baseline_eligible") is True def _format_identity_filters(filters: dict[str, str]) -> str: if not filters: return "" compact = ", ".join(f"{key}={value}" for key, value in filters.items()) return f" ({compact})" def _load_current_results() -> list[dict[str, Any]]: pattern = os.path.join(RESULTS_DIR, "perf_*.json") records: list[dict[str, Any]] = [] for path in sorted(glob.glob(pattern)): with open(path, encoding="utf-8") as f: records.append(json.load(f)) return records def normalize_performance_result(result: dict[str, Any]) -> dict[str, Any]: """Normalize a raw perf_*.json result into the HF tracking schema. The Buildkite artifact intentionally keeps the raw benchmark output from test_inference_performance.py. Baseline comparison, main-branch persistence, and manual baseline reseeds should all use this mapping so the stored HF records do not drift from the artifact schema. """ benchmark_id = result.get("benchmark_id", "unknown") model_id = benchmark_id timestamp = result.get("timestamp") if not timestamp: timestamp = datetime.now(timezone.utc).isoformat() commit_sha = result.get("commit") or os.environ.get("BUILDKITE_COMMIT", "") latency = safe_float(result.get("avg_generation_time_s")) throughput = safe_float(result.get("throughput_fps")) memory = safe_float(result.get("max_peak_memory_mb")) text_encoder_time = safe_float(result.get("text_encoder_time_s")) dit_time = safe_float(result.get("dit_time_s")) vae_decode_time = safe_float(result.get("vae_decode_time_s")) metric_policies = resolve_metric_policies(result.get("regression_thresholds")) record = { "model_id": model_id, "timestamp": timestamp, "commit_sha": commit_sha, "gpu_type": result.get("device", "unknown"), "latency": latency, "throughput": throughput, "memory": memory, "text_encoder_time_s": text_encoder_time, "dit_time_s": dit_time, "vae_decode_time_s": vae_decode_time, "regression_thresholds": serialize_metric_thresholds(metric_policies), "success": True, **_record_metadata(_detect_run_source(), result), } record.update(_identity_metadata(result)) return record def _normalize_record(result: dict[str, Any]) -> dict[str, Any]: return normalize_performance_result(result) def _write_tracking_record(record: dict[str, Any]) -> str: model_dir = os.path.join(TRACKING_ROOT, sanitize(record["model_id"])) os.makedirs(model_dir, exist_ok=True) timestamp = sanitize(record["timestamp"]) commit = sanitize(record["commit_sha"] or "unknown") out_path = os.path.join(model_dir, f"{timestamp}_{commit}.json") with open(out_path, "w", encoding="utf-8") as f: json.dump(record, f, indent=2) return out_path def _write_normalized_artifact(record: dict[str, Any]) -> None: try: results_dir = os.path.join(PERF_REPORTS_DIR, "results") os.makedirs(results_dir, exist_ok=True) timestamp = sanitize(record["timestamp"]) model_id = sanitize(record["model_id"]) commit = sanitize(record["commit_sha"] or "unknown") path = os.path.join( results_dir, f"normalized_perf_{model_id}_{timestamp}_{commit}.json", ) with open(path, "w", encoding="utf-8") as f: json.dump(record, f, indent=2) print(f"Normalized performance result written to {path}") except Exception as e: print(f"Failed to write normalized performance result artifact: {e}") def _baseline_metric(records: list[dict[str, Any]], key: str) -> float | None: values = [safe_float(r.get(key)) for r in records] values = [v for v in values if v is not None] if not values: return None return statistics.median(values) def _metric_policy_summary(policy: MetricPolicy) -> str: gated = "gated" if policy.gated else "info" return (f"{gated}, >{policy.threshold_percent * 100:.1f}% " f"and >{policy.threshold_absolute:.{policy.precision}f}") def _check_regressions( current: dict[str, Any], baseline_records: list[dict[str, Any]], metric_policies: tuple[MetricPolicy, ...], ) -> list[str]: failures: list[str] = [] for policy in metric_policies: baseline = _baseline_metric(baseline_records, policy.key) curr = safe_float(current.get(policy.key)) if baseline is None or curr is None: continue delta = regression_delta(policy, curr, baseline) if delta is None or not delta.regressed: continue failures.append(f"{current['model_id']} {policy.key} regressed by " f"{delta.percent * 100:.1f}% and " f"{delta.absolute:.{policy.precision}f} " f"(current={curr:.{policy.precision}f}, " f"baseline_median={baseline:.{policy.precision}f}, " f"threshold={_metric_policy_summary(policy)})") return failures def _fixed_threshold_failures(raw_result: dict[str, Any]) -> list[str]: thresholds = raw_result.get("thresholds") if not isinstance(thresholds, dict): return [] failures = [] for result_key, threshold_key in STATIC_THRESHOLD_FIELDS: current = safe_float(raw_result.get(result_key)) threshold = safe_float(thresholds.get(threshold_key)) if current is None or threshold is None or current <= threshold: continue failures.append(f"{raw_result.get('benchmark_id', 'unknown')} {result_key} exceeded fixed threshold " f"(current={current:.3f}, threshold={threshold:.3f})") return failures def _pytest_infra_failure(record: dict[str, Any]) -> str: return (f"{record['model_id']} performance pytest failed without an " "attributable static-threshold regression " f"(PERF_PYTEST_RC={os.environ.get('PERF_PYTEST_RC')})") def _recipe_mismatch_failure( record: dict[str, Any], recipe_mismatch_records: list[dict[str, Any]], ) -> str: seen = sorted({ str(item.get("recipe_fingerprint")) for item in recipe_mismatch_records if item.get("recipe_fingerprint") is not None }) suffix = f"; existing recipe_fingerprint values: {', '.join(seen)}" if seen else "" return (f"{record['model_id']} recipe_fingerprint={record.get('recipe_fingerprint')} " "does not match baseline records for the same workload, variant, " "and benchmark version across hardware and software profiles" f"{suffix}") def _evaluate_record_comparison( record: dict[str, Any], baseline_records: list[dict[str, Any]], recipe_mismatch_records: list[dict[str, Any]], metric_policies: tuple[MetricPolicy, ...], static_threshold_failures: list[str], unattributed_pytest_failure: bool, ) -> tuple[list[str], str, str]: identity_filters = None if _record_uses_v2_identity(record): try: identity_filters = _comparison_identity_filters(record) except ValueError as exc: failure = f"{record['model_id']} cannot compare v2 record: {exc}" return [failure], STATUS_INFRA_ERROR, failure if not baseline_records and recipe_mismatch_records: failure = _recipe_mismatch_failure(record, recipe_mismatch_records) return [failure], STATUS_RECIPE_MISMATCH, failure if static_threshold_failures: return static_threshold_failures, STATUS_REGRESSION, "; ".join(static_threshold_failures) if unattributed_pytest_failure: failure = _pytest_infra_failure(record) return [failure], STATUS_INFRA_ERROR, failure if not baseline_records: if identity_filters is not None: return [], STATUS_CALIBRATION_NEEDED, ("No baseline found for exact comparable identity" f"{_format_identity_filters(identity_filters)}") return [], STATUS_PASS, f"No legacy baseline for {record['model_id']} on {record['gpu_type']}" failures = _check_regressions(record, baseline_records, metric_policies) if failures: return failures, STATUS_REGRESSION, "; ".join(failures) return [], STATUS_PASS, "Comparable baseline found with no gated regressions" def _compact_value(value: float | None, precision: int = 3) -> str: if value is None: return "n/a" return f"{value:.{precision}f}" def _markdown_cell(value: Any) -> str: text = str(value) if value is not None else "" return text.replace("\n", " ").replace("|", "/") def _build_summary_row( record: dict[str, Any], baseline_records: list[dict[str, Any]], metric_policies: tuple[MetricPolicy, ...], has_failed: bool, ) -> dict[str, Any]: """Format a single benchmark result as a row for the Markdown table.""" metric_values: dict[str, dict[str, Any]] = {} regressions: list[float] = [] failing_metrics: list[str] = [] threshold_exceeded_metrics: list[str] = [] for policy in metric_policies: curr = safe_float(record.get(policy.key)) baseline = _baseline_metric(baseline_records, policy.key) delta = (regression_delta(policy, curr, baseline) if curr is not None and baseline is not None else None) regression = None if delta is None else delta.percent * 100.0 absolute_delta = None if delta is None else delta.absolute metric_values[policy.key] = { "curr": curr, "base": baseline, "regression_pct": regression, "absolute_delta": absolute_delta, "threshold_percent": policy.threshold_percent * 100.0, "threshold_absolute": policy.threshold_absolute, "gated": policy.gated, "threshold_exceeded": False if delta is None else delta.threshold_exceeded, "regressed": False if delta is None else delta.regressed, } if regression is not None: regressions.append(regression) if delta is not None and delta.threshold_exceeded: threshold_exceeded_metrics.append(policy.key) if delta is not None and delta.regressed: failing_metrics.append(policy.key) worst_regression_pct = max(regressions) if regressions else None return { "model_id": record["model_id"], "gpu_type": record["gpu_type"], "baseline_n": len(baseline_records), "metrics": metric_values, "worst_regression_pct": worst_regression_pct, "threshold_exceeded_metrics": threshold_exceeded_metrics, "failing_metrics": failing_metrics, "failed": has_failed, "status": record.get("comparison_status", STATUS_REGRESSION if has_failed else STATUS_PASS), "status_reason": record.get("comparison_status_reason", ""), } def _build_markdown_summary( summary_rows: list[dict[str, Any]], metric_policies: tuple[MetricPolicy, ...], ) -> str: lines = [ "## Performance Baseline Comparison", "", "Threshold: gated metrics fail only when both percent and absolute " "regression floors are exceeded.", "", ("| Model | GPU | Baseline N | Latency (curr/base) | " "Throughput (curr/base) | Memory (curr/base) | " "Text Enc (curr/base) | DiT (curr/base) | " "VAE Decode (curr/base) | Worst Regression | Exceeded Metrics | " "Failing Metrics | Status | Status Detail |"), "|---|---|---:|---|---|---|---|---|---|---:|---|---|---|---|", ] for row in summary_rows: metric_cells = [] for policy in metric_policies: values = row["metrics"][policy.key] metric_cells.append(f"{_compact_value(values['curr'], policy.precision)} / " f"{_compact_value(values['base'], policy.precision)}") worst_reg = ("n/a" if row["worst_regression_pct"] is None else f"{row['worst_regression_pct']:.1f}%") exceeded_metrics = (", ".join(row["threshold_exceeded_metrics"]) if row["threshold_exceeded_metrics"] else "none") failing_metrics = ", ".join(row["failing_metrics"]) if row["failing_metrics"] else "none" status = row["status"] status_detail = row["status_reason"] lines.append(f"| {row['model_id']} | {row['gpu_type']} | " f"{row['baseline_n']} | " f"{' | '.join(metric_cells)} | " f"{worst_reg} | {exceeded_metrics} | {failing_metrics} | " f"{_markdown_cell(status)} | {_markdown_cell(status_detail)} |") return "\n".join(lines) + "\n" def _emit_markdown_summary(markdown: str, commit_sha: str) -> None: print("\n" + markdown) # 1. Existing GitHub logic (safe to keep) summary_path = os.environ.get("GITHUB_STEP_SUMMARY") if summary_path: with open(summary_path, "a", encoding="utf-8") as f: f.write(markdown + "\n") # 2. Write to Modal volume for Buildkite to pick up in post-run hook try: os.makedirs(PERF_REPORTS_DIR, exist_ok=True) short_sha = commit_sha[:7] if commit_sha else "unknown" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") report_path = os.path.join(PERF_REPORTS_DIR, f"perf_{short_sha}_{timestamp}.md") with open(report_path, "w", encoding="utf-8") as f: f.write(markdown + "\n") print(f"Performance report written to {report_path}") except Exception as e: print(f"Failed to write performance report to Modal volume: {e}") def main() -> int: persist_tracking = _should_persist_tracking() upload_policy = _normalized_upload_policy() performance_pytest_failed = _performance_pytest_failed() # Strict on upload-enabled runs: silent sync failure would make comparison # and upload state ambiguous. sync_from_hf(TRACKING_ROOT, strict=persist_tracking) current_results = _load_current_results() if not current_results: print(f"No performance result files found in {RESULTS_DIR}") return 0 static_threshold_failures = [_fixed_threshold_failures(raw) for raw in current_results] unattributed_pytest_failure = performance_pytest_failed and not any(static_threshold_failures) all_failures: list[str] = [] summary_rows: list[dict[str, Any]] = [] if persist_tracking: print(f"Tracking persistence enabled: PERF_UPLOAD_POLICY={upload_policy}") else: print("Tracking persistence disabled: PERF_UPLOAD_POLICY=never") if performance_pytest_failed: if unattributed_pytest_failure: print("Performance pytest failed without a measured static-threshold regression: " f"PERF_PYTEST_RC={os.environ.get('PERF_PYTEST_RC')}") else: print("Performance pytest failure attributed to per-record static thresholds: " f"PERF_PYTEST_RC={os.environ.get('PERF_PYTEST_RC')}") for raw, fixed_threshold_failures in zip(current_results, static_threshold_failures, strict=True): record = _normalize_record(raw) metric_policies = resolve_metric_policies(record.get("regression_thresholds")) identity_filters = _comparison_identity_filters_or_none(record) baseline_records: list[dict[str, Any]] = [] recipe_mismatch_records: list[dict[str, Any]] = [] try: if identity_filters is None: record["baseline_status"] = ("invalid_identity" if _record_uses_v2_identity(record) else "skipped_missing_identity") else: baseline_records = load_records_for_identity( TRACKING_ROOT, identity_filters, last_n=5, successful_only=True, baseline_eligible_only=True, ) if baseline_records: record["baseline_status"] = "compared" else: recipe_cohort_records = load_records_for_identity( TRACKING_ROOT, _recipe_cohort_filters(record), successful_only=True, ) recipe_mismatch_records = _recipe_mismatch_records(record, recipe_cohort_records) record["baseline_status"] = ("recipe_mismatch" if recipe_mismatch_records else "initialized_new_cohort") failures, comparison_status, comparison_status_reason = _evaluate_record_comparison( record, baseline_records, recipe_mismatch_records, metric_policies, fixed_threshold_failures, unattributed_pytest_failure, ) except Exception as exc: comparison_status = STATUS_INFRA_ERROR comparison_status_reason = f"{record['model_id']} baseline comparison hit an infra error: {exc}" failures = [comparison_status_reason] baseline_records = [] record["baseline_status"] = "infra_error" record["comparison_status"] = comparison_status record["comparison_status_reason"] = comparison_status_reason record["success"] = not failures record["baseline_eligible"] = (identity_filters is not None and _is_baseline_eligible( record["run_source"], record["success"], comparison_status)) all_failures.extend(failures) print(f"{record['model_id']} comparison status: " f"{comparison_status} - {comparison_status_reason}") _write_normalized_artifact(record) if _upload_allowed(record): current_path = _write_tracking_record(record) upload_record(current_path, record, strict=True) else: print("Tracking upload skipped for " f"{record['model_id']} ({record['run_source']}, success={record['success']})") summary_rows.append(_build_summary_row(record, baseline_records, metric_policies, bool(failures))) commit_sha = os.environ.get("BUILDKITE_COMMIT", "unknown")[:7] markdown = _build_markdown_summary(summary_rows, resolve_metric_policies(None)) _emit_markdown_summary(markdown, commit_sha) if all_failures: print("Performance regression check failed:") for item in all_failures: print(f" - {item}") return 1 print("Performance baseline comparison passed") return 0 if __name__ == "__main__": sys.exit(main())