704 lines
27 KiB
Python
704 lines
27 KiB
Python
# 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())
|