From d311f311bc1a7e90ed0986227e68770c9af6dd09 Mon Sep 17 00:00:00 2001 From: Liangsheng Yin Date: Thu, 14 May 2026 00:36:25 -0700 Subject: [PATCH] ci: read est_time from sglang-ci-stats instead of scraping CI logs (#25255) --- .github/workflows/weekly-update-est-time.yml | 7 +- scripts/ci/update_est_time.py | 346 ++++++------------- 2 files changed, 101 insertions(+), 252 deletions(-) diff --git a/.github/workflows/weekly-update-est-time.yml b/.github/workflows/weekly-update-est-time.yml index 301b5af2f..b61a66921 100644 --- a/.github/workflows/weekly-update-est-time.yml +++ b/.github/workflows/weekly-update-est-time.yml @@ -25,9 +25,10 @@ jobs: - name: Update est_time values env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + MODEL_URL: "https://raw.githubusercontent.com/sgl-project/sglang-ci-stats/main/model.json" run: | python scripts/ci/update_est_time.py \ + --model-url "$MODEL_URL" \ --summary-file /tmp/est_time_summary.md - name: Check for changes @@ -61,9 +62,9 @@ jobs: { echo "## Summary" echo - echo "Updates \`est_time\` values in CI test registration calls based on the 90th percentile of the last 15 successful executions from scheduled PR Test runs on main." + echo "Refreshes \`est_time\` literals from [\`sgl-project/sglang-ci-stats\`](https://github.com/sgl-project/sglang-ci-stats)'s \`model.json\` (per-(suite, file) p90 over recent successful CI runs on \`main\`)." echo - echo "This keeps the LPT load-balancing algorithm accurate for partitioning tests across parallel CI jobs." + echo "This keeps the LPT load-balancing algorithm accurate for partitioning tests across parallel CI jobs, and serves as the static fallback when \`compute_partitions\` cannot fetch the live model at PR time." echo if [ -f /tmp/est_time_summary.md ]; then cat /tmp/est_time_summary.md diff --git a/scripts/ci/update_est_time.py b/scripts/ci/update_est_time.py index ce19e7113..de9b9bd40 100755 --- a/scripts/ci/update_est_time.py +++ b/scripts/ci/update_est_time.py @@ -1,36 +1,26 @@ #!/usr/bin/env python3 -"""Update est_time values in CI test files based on actual execution times. - -Fetches logs from recent scheduled PR Test workflow runs on main, -parses per-file elapsed times from successful jobs, computes the 90th -percentile, and updates the est_time literals in test registration calls. +"""Refresh est_time literals from sglang-ci-stats/model.json. Usage: - python scripts/ci/update_est_time.py [--dry-run] [--repo OWNER/REPO] + python scripts/ci/update_est_time.py [--dry-run] \\ + [--model-url URL] [--summary-file PATH] """ import argparse import json import re -import statistics import subprocess +import sys from collections import defaultdict from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent.parent - -# Regex to extract per-file elapsed time from CI logs. -# Matches lines like: -# filename='/actions-runner/_work/sglang/sglang/test/registered/core/test_x.py', elapsed=120, ... -# filename='/actions-runner/_work/sglang/sglang/python/sglang/jit_kernel/tests/test_x.py', ... -LOG_PATTERN = re.compile( - r"filename='[^']*?/sglang/((?:test|python)/[^']+\.py)', elapsed=(\d+)," +DEFAULT_MODEL_URL = ( + "https://raw.githubusercontent.com/sgl-project/sglang-ci-stats/main/model.json" ) -WORKFLOW_NAME = "PR Test" -MIN_DATA_POINTS = 3 -TARGET_DATA_POINTS = 15 -MAX_RUNS = 25 +# AMD / NPU live in separate workflows and are not scraped by sglang-ci-stats. +BACKENDS = ("cuda", "cpu") # A change is "significant" if |delta| >= this many seconds AND the relative # change is at least SIGNIFICANT_REL_DELTA. Dual threshold filters out both @@ -40,252 +30,110 @@ SIGNIFICANT_ABS_DELTA = 30 SIGNIFICANT_REL_DELTA = 0.3 -def gh_api(endpoint, paginate=False): - """Call gh api and return parsed JSON.""" - cmd = ["gh", "api", endpoint] - if paginate: - cmd.append("--paginate") - result = subprocess.run(cmd, capture_output=True, text=True, check=True) - return json.loads(result.stdout) - - -def gh_api_raw(endpoint): - """Call gh api and return raw bytes (for log downloads).""" - cmd = ["gh", "api", endpoint] - result = subprocess.run(cmd, capture_output=True, check=True) - return result.stdout - - -def get_workflow_id(repo): - """Find the workflow ID for the PR Test workflow.""" - data = gh_api(f"/repos/{repo}/actions/workflows") - for wf in data["workflows"]: - if wf["name"] == WORKFLOW_NAME: - return wf["id"] - raise RuntimeError(f"Workflow '{WORKFLOW_NAME}' not found in {repo}") - - -def get_scheduled_runs(repo, workflow_id): - """Get completed scheduled runs on main, newest first.""" - data = gh_api( - f"/repos/{repo}/actions/workflows/{workflow_id}/runs" - f"?branch=main&status=completed&event=schedule&per_page=100" +def fetch_model(url): + """Curl model.json. Fail loudly on network or parse errors -- the + weekly workflow will surface the failure rather than silently making + a no-op PR.""" + out = subprocess.run( + ["curl", "--fail", "--silent", "--show-error", "--max-time", "30", url], + capture_output=True, + text=True, + check=True, ) - return data["workflow_runs"] + return json.loads(out.stdout) -def get_successful_jobs(repo, run_id): - """Get successful jobs for a given run.""" - data = gh_api(f"/repos/{repo}/actions/runs/{run_id}/jobs?per_page=100") - return [j for j in data["jobs"] if j["conclusion"] == "success"] +def make_patterns(suite): + """Yield regex objects that match `register_{backend}_ci(est_time=N, ...)` + for the given suite, covering both registration styles: - -def job_name_to_suite(job_name): - """Extract the suite name from a job name. - - Job names look like "stage-c-test-4-gpu-h100 (2)" or "stage-a-test-cpu". - Strip the partition suffix " (N)" to get the suite name. + legacy: register_X_ci(est_time=N, suite="") + new: register_X_ci(est_time=N, stage="", runner_config="") """ - return re.sub(r"\s*\(\d+\)$", "", job_name) - - -def determine_backend(job_name): - """Determine backend from job name.""" - name = job_name.lower() - for backend in ["cpu", "amd", "npu"]: - if backend in name: - return backend - return "cuda" - - -def parse_job_logs(repo, job_id): - """Download and parse a job's logs for elapsed times. - - Returns list of (relative_path, elapsed_seconds) tuples. - """ - try: - raw = gh_api_raw(f"/repos/{repo}/actions/jobs/{job_id}/logs") - text = raw.decode("utf-8", errors="replace") - except subprocess.CalledProcessError: - return [] - - results = [] - for match in LOG_PATTERN.finditer(text): - rel_path = match.group(1) - elapsed = int(match.group(2)) - results.append((rel_path, elapsed)) - return results - - -def collect_timings(repo): - """Collect per-file elapsed times from recent scheduled CI runs. - - Returns dict mapping (relative_path, suite, backend) -> list of elapsed - times (newest first). - """ - workflow_id = get_workflow_id(repo) - print(f"Found workflow '{WORKFLOW_NAME}' (id={workflow_id})") - - runs = get_scheduled_runs(repo, workflow_id) - print(f"Found {len(runs)} completed scheduled runs on main") - - # timings[(rel_path, suite, backend)] = [elapsed1, elapsed2, ...] - timings = defaultdict(list) - runs_processed = 0 - - for run in runs: - run_id = run["id"] - jobs = get_successful_jobs(repo, run_id) - if not jobs: - continue - - runs_processed += 1 - test_jobs = [ - j - for j in jobs - if j["name"] != "check-changes" and "health" not in j["name"].lower() - ] - print( - f" Run {run_id} ({run['conclusion']}): " - f"{len(test_jobs)} successful test jobs" + stage_rc = None + if "-test-" in suite: + stage, _, rc = suite.partition("-test-") + stage_rc = (stage, rc) + for backend in BACKENDS: + yield re.compile( + rf"(register_{backend}_ci\(est_time=)(\d+)" + rf'(,\s*suite="{re.escape(suite)}")' ) - - for job in test_jobs: - suite = job_name_to_suite(job["name"]) - backend = determine_backend(job["name"]) - entries = parse_job_logs(repo, job["id"]) - for rel_path, elapsed in entries: - key = (rel_path, suite, backend) - timings[key].append(elapsed) - - if runs_processed >= MAX_RUNS: - print(f" Reached max {MAX_RUNS} runs, stopping collection") - break - - print( - f"\nProcessed {runs_processed} runs, " - f"collected timings for {len(timings)} (file, suite, backend) pairs" - ) - return timings + if stage_rc is not None: + stage, rc = stage_rc + yield re.compile( + rf"(register_{backend}_ci\(est_time=)(\d+)" + rf'(,\s*stage="{re.escape(stage)}",\s*runner_config="{re.escape(rc)}")' + ) -def compute_p90(timings): - """Compute 90th percentile of last TARGET_DATA_POINTS timings for each entry. +def update_files(model, dry_run=False): + """Walk `model.est`, apply each p90 to the matching register call. - Returns dict mapping (rel_path, suite, backend) -> p90 (int). - Only includes entries with >= MIN_DATA_POINTS data points. + Returns list of (relpath, suite, old, new) for every changed entry. """ - p90s = {} - for key, values in timings.items(): - recent = values[:TARGET_DATA_POINTS] - if len(recent) < MIN_DATA_POINTS: - continue - p90s[key] = round(statistics.quantiles(recent, n=10, method="inclusive")[8]) - return p90s - - -def update_est_times(p90s, dry_run=False): - """Update est_time values in source files. - - Each registration call is matched by both the function name and suite, - so files with multiple registrations for different suites get the correct - per-suite p90. - - Returns (updated_count, skipped_count, changes) where changes is a list - of (rel_path, suite, backend, old_val, new_val) for each modified entry. - """ - updated = 0 - skipped = 0 - changes = [] - - # Group p90s by file: {rel_path: [(suite, backend, p90), ...]} by_file = defaultdict(list) - for (rel_path, suite, backend), p90 in p90s.items(): - by_file[rel_path].append((suite, backend, p90)) + for suite, files in model.get("est", {}).items(): + for relpath, p90 in files.items(): + by_file[relpath].append((suite, p90)) - for rel_path, entries in sorted(by_file.items()): - filepath = REPO_ROOT / rel_path + changes = [] + for relpath, entries in sorted(by_file.items()): + filepath = REPO_ROOT / relpath if not filepath.exists(): - print(f" SKIP {rel_path}: file not found") - skipped += 1 continue - content = filepath.read_text() new_content = content - for suite, backend, p90 in entries: - # Match registration calls with this specific backend and suite. - # Two styles: - # legacy: register_X_ci(est_time=N, suite="stage-Y-test-Z") - # new: register_X_ci(est_time=N, stage="stage-Y", runner_config="Z") - # New-style files all use the canonical `stage=` then `runner_config=` order. - legacy_pattern = re.compile( - rf"(register_{backend}_ci\(est_time=)(\d+)" - rf'(,\s*suite="{re.escape(suite)}")' - ) - pattern = legacy_pattern if legacy_pattern.search(new_content) else None - if pattern is None and "-test-" in suite: - stage, _, rc = suite.partition("-test-") - new_style_pattern = re.compile( - rf"(register_{backend}_ci\(est_time=)(\d+)" - rf'(,\s*stage="{re.escape(stage)}",\s*runner_config="{re.escape(rc)}")' - ) - if new_style_pattern.search(new_content): - pattern = new_style_pattern - if pattern is None: - continue - match = pattern.search(new_content) - if not match: - continue + for suite, p90 in entries: + for pattern in make_patterns(suite): + match = pattern.search(new_content) + if match is None: + continue + old_val = int(match.group(2)) + if old_val != p90: + new_content = pattern.sub(rf"\g<1>{p90}\3", new_content) + changes.append((relpath, suite, old_val, p90)) + print( + f" {relpath}: suite={suite!r} " f"est_time {old_val} -> {p90}", + file=sys.stderr, + ) + break # one (file, suite) -> at most one register call - old_val = int(match.group(2)) - if old_val == p90: - continue + if new_content != content and not dry_run: + filepath.write_text(new_content) - new_content = pattern.sub(rf"\g<1>{p90}\3", new_content) - changes.append((rel_path, suite, backend, old_val, p90)) - print( - f" {rel_path}: register_{backend}_ci " - f'suite="{suite}" est_time={old_val} -> {p90}' - ) - - if new_content != content: - if not dry_run: - filepath.write_text(new_content) - updated += 1 - else: - skipped += 1 - - return updated, skipped, changes + return changes -def is_significant(old_val, new_val): - """Return True if the change meets both absolute and relative thresholds.""" - delta = abs(new_val - old_val) - return delta >= SIGNIFICANT_ABS_DELTA and delta / old_val >= SIGNIFICANT_REL_DELTA +def is_significant(old, new): + delta = abs(new - old) + return ( + delta >= SIGNIFICANT_ABS_DELTA and delta / max(old, 1) >= SIGNIFICANT_REL_DELTA + ) def write_summary(changes, summary_file): """Write a markdown summary of significant est_time changes.""" - significant = [c for c in changes if is_significant(c[3], c[4])] - significant.sort(key=lambda c: abs(c[4] - c[3]), reverse=True) + sig = [c for c in changes if is_significant(c[2], c[3])] + sig.sort(key=lambda c: abs(c[3] - c[2]), reverse=True) lines = [] - if significant: + if sig: lines.append( f"### Significant est_time changes " - f"({len(significant)} of {len(changes)} updates)" + f"({len(sig)} of {len(changes)} updates)" ) lines.append("") lines.append("| File | Suite | Old (s) | New (s) | Δ |") lines.append("| --- | --- | ---: | ---: | ---: |") - for rel_path, suite, _backend, old_val, new_val in significant: - delta = new_val - old_val + for relpath, suite, old, new in sig: + delta = new - old sign = "+" if delta > 0 else "" - pct = round(delta / old_val * 100) + pct = round(delta / max(old, 1) * 100) lines.append( - f"| `{Path(rel_path).name}` | `{suite}` | " - f"{old_val} | {new_val} | {sign}{delta} ({sign}{pct}%) |" + f"| `{Path(relpath).name}` | `{suite}` | " + f"{old} | {new} | {sign}{delta} ({sign}{pct}%) |" ) else: lines.append( @@ -298,19 +146,17 @@ def write_summary(changes, summary_file): def main(): - parser = argparse.ArgumentParser( - description="Update est_time values from CI run data" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--model-url", + default=DEFAULT_MODEL_URL, + help="URL of model.json from sglang-ci-stats (file:// is OK for testing)", ) parser.add_argument( "--dry-run", action="store_true", help="Print changes without modifying files", ) - parser.add_argument( - "--repo", - default="sgl-project/sglang", - help="GitHub repository (default: sgl-project/sglang)", - ) parser.add_argument( "--summary-file", default=None, @@ -318,25 +164,27 @@ def main(): ) args = parser.parse_args() - print("Collecting timings from CI logs...") - timings = collect_timings(args.repo) + print(f"Fetching {args.model_url}", file=sys.stderr) + model = fetch_model(args.model_url) + print( + f" model data_as_of={model.get('data_as_of')} " + f"n_runs={model.get('n_runs')} " + f"n_suites={len(model.get('est', {}))}", + file=sys.stderr, + ) - print("\nComputing 90th percentiles...") - p90s = compute_p90(timings) - print(f"Computed p90 for {len(p90s)} (file, suite, backend) entries") - - print("\nUpdating est_time values...") - updated, skipped, changes = update_est_times(p90s, dry_run=args.dry_run) + changes = update_files(model, dry_run=args.dry_run) + n_files = len({c[0] for c in changes}) action = "Would update" if args.dry_run else "Updated" - print(f"\n{action} {updated} files, skipped {skipped} files") + print( + f"\n{action} {len(changes)} est_time entries across {n_files} files", + file=sys.stderr, + ) if args.summary_file: write_summary(changes, args.summary_file) - print(f"Wrote summary to {args.summary_file}") - - if args.dry_run: - print("(dry-run mode, no files modified)") + print(f"Wrote summary to {args.summary_file}", file=sys.stderr) if __name__ == "__main__":