ci: read est_time from sglang-ci-stats instead of scraping CI logs (#25255)

This commit is contained in:
Liangsheng Yin
2026-05-14 00:36:25 -07:00
committed by GitHub
parent bc265c5f82
commit d311f311bc
2 changed files with 101 additions and 252 deletions
+4 -3
View File
@@ -25,9 +25,10 @@ jobs:
- name: Update est_time values - name: Update est_time values
env: env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} MODEL_URL: "https://raw.githubusercontent.com/sgl-project/sglang-ci-stats/main/model.json"
run: | run: |
python scripts/ci/update_est_time.py \ python scripts/ci/update_est_time.py \
--model-url "$MODEL_URL" \
--summary-file /tmp/est_time_summary.md --summary-file /tmp/est_time_summary.md
- name: Check for changes - name: Check for changes
@@ -61,9 +62,9 @@ jobs:
{ {
echo "## Summary" echo "## Summary"
echo 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
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 echo
if [ -f /tmp/est_time_summary.md ]; then if [ -f /tmp/est_time_summary.md ]; then
cat /tmp/est_time_summary.md cat /tmp/est_time_summary.md
+97 -249
View File
@@ -1,36 +1,26 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Update est_time values in CI test files based on actual execution times. """Refresh est_time literals from sglang-ci-stats/model.json.
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.
Usage: 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 argparse
import json import json
import re import re
import statistics
import subprocess import subprocess
import sys
from collections import defaultdict from collections import defaultdict
from pathlib import Path from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent.parent REPO_ROOT = Path(__file__).resolve().parent.parent.parent
DEFAULT_MODEL_URL = (
# Regex to extract per-file elapsed time from CI logs. "https://raw.githubusercontent.com/sgl-project/sglang-ci-stats/main/model.json"
# 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+),"
) )
WORKFLOW_NAME = "PR Test" # AMD / NPU live in separate workflows and are not scraped by sglang-ci-stats.
MIN_DATA_POINTS = 3 BACKENDS = ("cuda", "cpu")
TARGET_DATA_POINTS = 15
MAX_RUNS = 25
# A change is "significant" if |delta| >= this many seconds AND the relative # A change is "significant" if |delta| >= this many seconds AND the relative
# change is at least SIGNIFICANT_REL_DELTA. Dual threshold filters out both # 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 SIGNIFICANT_REL_DELTA = 0.3
def gh_api(endpoint, paginate=False): def fetch_model(url):
"""Call gh api and return parsed JSON.""" """Curl model.json. Fail loudly on network or parse errors -- the
cmd = ["gh", "api", endpoint] weekly workflow will surface the failure rather than silently making
if paginate: a no-op PR."""
cmd.append("--paginate") out = subprocess.run(
result = subprocess.run(cmd, capture_output=True, text=True, check=True) ["curl", "--fail", "--silent", "--show-error", "--max-time", "30", url],
return json.loads(result.stdout) capture_output=True,
text=True,
check=True,
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"
) )
return data["workflow_runs"] return json.loads(out.stdout)
def get_successful_jobs(repo, run_id): def make_patterns(suite):
"""Get successful jobs for a given run.""" """Yield regex objects that match `register_{backend}_ci(est_time=N, ...)`
data = gh_api(f"/repos/{repo}/actions/runs/{run_id}/jobs?per_page=100") for the given suite, covering both registration styles:
return [j for j in data["jobs"] if j["conclusion"] == "success"]
legacy: register_X_ci(est_time=N, suite="<full-suite>")
def job_name_to_suite(job_name): new: register_X_ci(est_time=N, stage="<stage>", runner_config="<rc>")
"""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.
""" """
return re.sub(r"\s*\(\d+\)$", "", job_name) stage_rc = None
if "-test-" in suite:
stage, _, rc = suite.partition("-test-")
def determine_backend(job_name): stage_rc = (stage, rc)
"""Determine backend from job name.""" for backend in BACKENDS:
name = job_name.lower() yield re.compile(
for backend in ["cpu", "amd", "npu"]: rf"(register_{backend}_ci\(est_time=)(\d+)"
if backend in name: rf'(,\s*suite="{re.escape(suite)}")'
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"
) )
if stage_rc is not None:
for job in test_jobs: stage, rc = stage_rc
suite = job_name_to_suite(job["name"]) yield re.compile(
backend = determine_backend(job["name"]) rf"(register_{backend}_ci\(est_time=)(\d+)"
entries = parse_job_logs(repo, job["id"]) rf'(,\s*stage="{re.escape(stage)}",\s*runner_config="{re.escape(rc)}")'
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
def compute_p90(timings): def update_files(model, dry_run=False):
"""Compute 90th percentile of last TARGET_DATA_POINTS timings for each entry. """Walk `model.est`, apply each p90 to the matching register call.
Returns dict mapping (rel_path, suite, backend) -> p90 (int). Returns list of (relpath, suite, old, new) for every changed entry.
Only includes entries with >= MIN_DATA_POINTS data points.
""" """
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) by_file = defaultdict(list)
for (rel_path, suite, backend), p90 in p90s.items(): for suite, files in model.get("est", {}).items():
by_file[rel_path].append((suite, backend, p90)) for relpath, p90 in files.items():
by_file[relpath].append((suite, p90))
for rel_path, entries in sorted(by_file.items()): changes = []
filepath = REPO_ROOT / rel_path for relpath, entries in sorted(by_file.items()):
filepath = REPO_ROOT / relpath
if not filepath.exists(): if not filepath.exists():
print(f" SKIP {rel_path}: file not found")
skipped += 1
continue continue
content = filepath.read_text() content = filepath.read_text()
new_content = content new_content = content
for suite, backend, p90 in entries: for suite, p90 in entries:
# Match registration calls with this specific backend and suite. for pattern in make_patterns(suite):
# Two styles: match = pattern.search(new_content)
# legacy: register_X_ci(est_time=N, suite="stage-Y-test-Z") if match is None:
# new: register_X_ci(est_time=N, stage="stage-Y", runner_config="Z") continue
# New-style files all use the canonical `stage=` then `runner_config=` order. old_val = int(match.group(2))
legacy_pattern = re.compile( if old_val != p90:
rf"(register_{backend}_ci\(est_time=)(\d+)" new_content = pattern.sub(rf"\g<1>{p90}\3", new_content)
rf'(,\s*suite="{re.escape(suite)}")' changes.append((relpath, suite, old_val, p90))
) print(
pattern = legacy_pattern if legacy_pattern.search(new_content) else None f" {relpath}: suite={suite!r} " f"est_time {old_val} -> {p90}",
if pattern is None and "-test-" in suite: file=sys.stderr,
stage, _, rc = suite.partition("-test-") )
new_style_pattern = re.compile( break # one (file, suite) -> at most one register call
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
old_val = int(match.group(2)) if new_content != content and not dry_run:
if old_val == p90: filepath.write_text(new_content)
continue
new_content = pattern.sub(rf"\g<1>{p90}\3", new_content) return changes
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
def is_significant(old_val, new_val): def is_significant(old, new):
"""Return True if the change meets both absolute and relative thresholds.""" delta = abs(new - old)
delta = abs(new_val - old_val) return (
return delta >= SIGNIFICANT_ABS_DELTA and delta / old_val >= SIGNIFICANT_REL_DELTA delta >= SIGNIFICANT_ABS_DELTA and delta / max(old, 1) >= SIGNIFICANT_REL_DELTA
)
def write_summary(changes, summary_file): def write_summary(changes, summary_file):
"""Write a markdown summary of significant est_time changes.""" """Write a markdown summary of significant est_time changes."""
significant = [c for c in changes if is_significant(c[3], c[4])] sig = [c for c in changes if is_significant(c[2], c[3])]
significant.sort(key=lambda c: abs(c[4] - c[3]), reverse=True) sig.sort(key=lambda c: abs(c[3] - c[2]), reverse=True)
lines = [] lines = []
if significant: if sig:
lines.append( lines.append(
f"### Significant est_time changes " f"### Significant est_time changes "
f"({len(significant)} of {len(changes)} updates)" f"({len(sig)} of {len(changes)} updates)"
) )
lines.append("") lines.append("")
lines.append("| File | Suite | Old (s) | New (s) | Δ |") lines.append("| File | Suite | Old (s) | New (s) | Δ |")
lines.append("| --- | --- | ---: | ---: | ---: |") lines.append("| --- | --- | ---: | ---: | ---: |")
for rel_path, suite, _backend, old_val, new_val in significant: for relpath, suite, old, new in sig:
delta = new_val - old_val delta = new - old
sign = "+" if delta > 0 else "" sign = "+" if delta > 0 else ""
pct = round(delta / old_val * 100) pct = round(delta / max(old, 1) * 100)
lines.append( lines.append(
f"| `{Path(rel_path).name}` | `{suite}` | " f"| `{Path(relpath).name}` | `{suite}` | "
f"{old_val} | {new_val} | {sign}{delta} ({sign}{pct}%) |" f"{old} | {new} | {sign}{delta} ({sign}{pct}%) |"
) )
else: else:
lines.append( lines.append(
@@ -298,19 +146,17 @@ def write_summary(changes, summary_file):
def main(): def main():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(description=__doc__)
description="Update est_time values from CI run data" 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( parser.add_argument(
"--dry-run", "--dry-run",
action="store_true", action="store_true",
help="Print changes without modifying files", help="Print changes without modifying files",
) )
parser.add_argument(
"--repo",
default="sgl-project/sglang",
help="GitHub repository (default: sgl-project/sglang)",
)
parser.add_argument( parser.add_argument(
"--summary-file", "--summary-file",
default=None, default=None,
@@ -318,25 +164,27 @@ def main():
) )
args = parser.parse_args() args = parser.parse_args()
print("Collecting timings from CI logs...") print(f"Fetching {args.model_url}", file=sys.stderr)
timings = collect_timings(args.repo) 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...") changes = update_files(model, dry_run=args.dry_run)
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)
n_files = len({c[0] for c in changes})
action = "Would update" if args.dry_run else "Updated" 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: if args.summary_file:
write_summary(changes, args.summary_file) write_summary(changes, args.summary_file)
print(f"Wrote summary to {args.summary_file}") print(f"Wrote summary to {args.summary_file}", file=sys.stderr)
if args.dry_run:
print("(dry-run mode, no files modified)")
if __name__ == "__main__": if __name__ == "__main__":