minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

ci: fix runner utilization report undercounting busy time ~25x (#31396)

Browse Source
This commit is contained in:
Alison Shao
2026-07-15 22:02:38 -07:00
committed by GitHub
parent b871a509e6
commit 238448f4c8
3 changed files with 501 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/runner-utilization.yml
+4 -1
View File
@@ -56,7 +56,10 @@ jobs:
-p 'test_runner_utilization_report.py' -v
- name: Generate Utilization Report
timeout-minutes: 30
# Full-coverage fetch (created-filtered listing + 12h lookback) is
# ~2-3x the truncated 5000-run fetch this replaces; the old fetch
# took ~10 min, so give the full one comfortable headroom.
timeout-minutes: 55
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
scripts/ci/utils/runner_utilization_report.py
+267 -42
View File
@@ -84,35 +84,98 @@ def run_gh_command(args: list[str], max_retries: int = 10) -> dict:
raise Exception(f"gh api failed after {max_retries} attempts: {last_err[:300]}")
def get_workflow_runs(repo: str, hours: int = 24) -> list[dict]:
"""Get workflow runs from the last N hours."""
since = datetime.now(timezone.utc) - timedelta(hours=hours)
def get_workflow_runs(
repo: str, since: datetime, max_pages: int = 400
) -> tuple[list[dict], bool]:
"""Get workflow runs created after `since`.
Returns (runs, truncated). `truncated` is True when the safety cap cut
the listing short, i.e. the OLDEST part of the window is missing (runs
come back newest-first).
The `created` filter is applied server-side so pagination ends exactly
when the window is exhausted. The previous implementation listed ALL
runs newest-first and stopped at a hard 50-page cap (5000 runs); on
busy days the repo creates ~15-18k runs per 24h, so the cap silently
dropped the oldest ~2/3 of the window while the utilization denominator
still assumed full 24h coverage. Worse, under queue backlog job
execution lags run creation by hours, so the surviving newest slice
held mostly still-queued jobs — saturated pools (hosts busy 24h/24h)
reported ~4% utilization.
GitHub API gotcha: a `created`-filtered listing serves at most 1000
results per query (search-style cap; page 11 comes back empty even
though total_count is larger). We therefore walk a cursor: whenever a
range holds more than 1000 runs, re-query with the range's upper bound
moved down to the oldest created_at fetched so far, deduping the
boundary overlap by run id, until the range is exhausted.
"""
since_str = since.strftime("%Y-%m-%dT%H:%M:%SZ")
runs = []
seen_ids = set()
upper_str = None # walking upper bound (inclusive), None = now
pages_used = 0
truncated = False
while True:
created_q = f"{since_str}..{upper_str}" if upper_str else f">={since_str}"
chunk = []
chunk_total = None
page = 1
while True:
data = run_gh_command(
[
f"repos/{repo}/actions/runs?per_page=100&page={page}",
f"repos/{repo}/actions/runs"
f"?per_page=100&page={page}&created={created_q}",
]
)
pages_used += 1
if chunk_total is None:
chunk_total = data.get("total_count", 0)
page_runs = data.get("workflow_runs", [])
# Filter by time
for run in page_runs:
created_at = parse_time(run.get("created_at"))
if created_at and created_at >= since:
runs.append(run)
elif created_at and created_at < since:
# Runs are ordered by created_at desc, so we can stop
return runs
if len(page_runs) < 100:
chunk.extend(page_runs)
# Stop at a short page (range exhausted), the 1000-result cap
# (10 full pages — page 11 would be empty), or the global
# request budget.
if len(page_runs) < 100 or page >= 10 or pages_used >= max_pages:
break
page += 1
if page > 50: # Safety limit (5000 runs)
new_runs = []
for r in chunk:
rid = r.get("id")
if rid in seen_ids:
continue
seen_ids.add(rid)
new_runs.append(r)
runs.extend(new_runs)
if chunk_total is not None and len(chunk) >= chunk_total:
break # saw everything matching this range -> reached `since`
if pages_used >= max_pages:
truncated = True
break
return runs
# More results exist below the 1000-result cap: move the upper
# bound down to the oldest run fetched in this chunk and re-query.
chunk_created = [
parse_time(r.get("created_at")) for r in chunk if r.get("created_at")
]
if not new_runs or not chunk_created:
truncated = True # cursor can't advance; avoid looping forever
break
new_upper = min(chunk_created).strftime("%Y-%m-%dT%H:%M:%SZ")
if new_upper == upper_str:
truncated = True
break
upper_str = new_upper
if truncated:
print(
f"WARNING: run listing truncated at {len(runs)} runs "
f"(request budget {max_pages} pages exhausted or cursor "
f"stalled). The oldest part of the window is missing."
)
return runs, truncated
def get_jobs_for_run(repo: str, run_id: int) -> list[dict]:
@@ -173,6 +236,45 @@ def parse_time(time_str: str) -> datetime:
return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
def carried_over_fingerprint(job: dict):
"""Identity of one physical job execution, for deduping re-run carryover.
When a run is re-run, the completed jobs that were NOT re-run reappear
in the new attempt as new job records (new id, identical
name/runner/timestamps), and `filter=all` returns every attempt's
records. Two completed records agreeing on run, name, runner, and both
timestamps are one execution — a real runner cannot run two identical
jobs at the same instant. Returns None for non-completed jobs (their
records are attempt-specific, and e.g. still-queued jobs lack the
timestamps that make the fingerprint discriminating).
"""
if job.get("status") != "completed":
return None
return (
job.get("run_id"),
job.get("name"),
job.get("runner_name"),
job.get("started_at"),
job.get("completed_at"),
)
def union_seconds(intervals: list[tuple]) -> float:
"""Total length in seconds of the union of (start, end) intervals."""
busy = 0.0
cur_start = cur_end = None
for start, end in sorted(intervals):
if cur_end is None or start > cur_end:
if cur_end is not None:
busy += (cur_end - cur_start).total_seconds()
cur_start, cur_end = start, end
else:
cur_end = max(cur_end, end)
if cur_end is not None:
busy += (cur_end - cur_start).total_seconds()
return busy
def classify_job(job: dict, now: datetime):
"""Derive the queue-wait and busy interval for a single job.
@@ -351,16 +453,23 @@ _NON_GPU_WORKFLOW_HINTS = (
"stale",
"dependabot",
"codeql",
# Pure API-bookkeeping workflows that fire on every PR event. "PR
# States" alone is ~25% of all runs on a busy day; none of these ever
# dispatch to a self-hosted runner.
"pr states",
"slash command",
"cancel", # cancel-unfinished-pr-tests, cancel-pr-workflows-on-close
"model inventory",
)
def _likely_no_gpu_jobs(workflow_name: str) -> bool:
"""Heuristic: skip per-run job-fetch for workflows that don't dispatch
to self-hosted GPU runners. The GH API rate limit (~5000 req/hr per
token) is the bottleneck on busy 24h windows where ~4000 workflow
runs fire — but only a fraction of those (pr-test, nightly-test,
pr-test-*kernel, etc.) actually run on GPU runners. Skipping the
docs/lint/release runs cuts the API call budget by 2-4x.
to self-hosted GPU runners. The GH API rate limit is the bottleneck on
busy 24h windows where ~18k workflow runs fire — but only a fraction
of those (pr-test, nightly-test, pr-test-*kernel, etc.) actually run
on GPU runners. Skipping the bookkeeping/docs/lint/release runs cuts
the API call budget roughly in half.
"""
if not workflow_name:
return False
@@ -368,16 +477,81 @@ def _likely_no_gpu_jobs(workflow_name: str) -> bool:
return any(h in n for h in _NON_GPU_WORKFLOW_HINTS)
def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None):
"""Calculate runner utilization metrics."""
def calculate_utilization(
repo: str,
hours: float = 24,
runner_filter: str = None,
lookback_hours: float = None,
):
"""Calculate runner utilization metrics.
print(f"Fetching workflow runs from last {hours} hours...")
all_runs = get_workflow_runs(repo, hours)
runs = [r for r in all_runs if not _likely_no_gpu_jobs(r.get("name", ""))]
skipped = len(all_runs) - len(runs)
`lookback_hours` extends the run *listing* (not the analysis window)
back before the window start. Under queue backlog, jobs execute hours
after their run is created, so the busy time observed inside the window
largely belongs to runs created before it — measured at ~33h of
in-window busy time from pre-window runs on one saturated 4-host pool.
Busy intervals are clamped to the window, so the lookback only restores
missing numerator; it never inflates it.
"""
fetch_start = datetime.now(timezone.utc)
if lookback_hours is None:
lookback_hours = min(12.0, hours / 2)
window_start_precheck = fetch_start - timedelta(hours=hours)
since = fetch_start - timedelta(hours=hours + lookback_hours)
print(
f"Fetching workflow runs from last {hours}h "
f"(+{lookback_hours:.1f}h lookback for long-queued jobs)..."
)
all_runs, truncated = get_workflow_runs(repo, since)
runs = []
skipped_non_gpu = 0
skipped_lookback = 0
for r in all_runs:
if _likely_no_gpu_jobs(r.get("name", "")):
skipped_non_gpu += 1
continue
created_at = parse_time(r.get("created_at"))
if created_at and created_at < window_start_precheck:
# Lookback region: only runs that were still active at the
# window start can contribute in-window busy time. A completed
# run whose last update predates the window finished before it
# — skip the (expensive) per-run job fetch. In-flight runs are
# always kept: their updated_at can be stale while a job is
# still running.
updated_at = parse_time(r.get("updated_at"))
if (
r.get("status") == "completed"
and updated_at
and updated_at < window_start_precheck
):
skipped_lookback += 1
continue
runs.append(r)
print(
f"Found {len(all_runs)} workflow runs "
f"({skipped} skipped as non-GPU: docs/lint/release/etc.)"
f"({skipped_non_gpu} skipped as non-GPU: docs/lint/release/etc.; "
f"{skipped_lookback} lookback runs skipped as finished pre-window)"
)
# If the run listing was truncated (newest-first), the oldest part of
# the window has no data. Shrink the analysis window to what the data
# actually covers so busy time isn't divided by capacity-hours we never
# observed — that's exactly the bug that made saturated pools report
# single-digit utilization.
coverage_hours = hours
if truncated and all_runs:
oldest_created = min(
parse_time(r["created_at"]) for r in all_runs if r.get("created_at")
)
coverage_hours = min(
hours,
(datetime.now(timezone.utc) - oldest_created).total_seconds() / 3600,
)
print(
f"Shrinking analysis window: {coverage_hours:.1f}h covered of "
f"{hours}h requested."
)
# Try to get online runners from API
@@ -462,11 +636,37 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
# captured once so every in-flight job is measured against a single
# reference (matches window_end below to within processing time).
now = datetime.now(timezone.utc)
window_seconds = coverage_hours * 3600
window_end = now
window_start = window_end - timedelta(hours=coverage_hours)
all_job_infos = [] # one entry per job (deduped across labels) for detail views
seen_fingerprints = set()
for job in all_jobs:
# Re-run attempts carry over the completed jobs they did NOT re-run
# as brand-new job records: a different job id but identical
# name/runner/timestamps. `filter=all` returns every attempt, so on
# rerun-heavy days each carried-over job was double-counted —
# inflating busy time to ~110% on saturated hosts and padding the
# pass counts. Job ids differ, so dedup by execution fingerprint.
fp = carried_over_fingerprint(job)
if fp is not None:
if fp in seen_fingerprints:
continue
seen_fingerprints.add(fp)
job_info = classify_job(job, now)
if job_info is None:
continue
# Lookback runs bring in jobs that finished before the window even
# started. They can't contribute clamped busy time and would only
# pollute job counts / queue stats — drop them. A job is entirely
# pre-window when both its busy interval and its queue wait ended
# before window_start.
latest_activity = max(
t for t in (job_info["end"], job_info["queue_end"]) if t is not None
)
if latest_activity < window_start:
continue
all_job_infos.append(job_info)
runner_name = job_info["runner_name"]
@@ -497,21 +697,22 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
print(f"Tracking {len(all_labels)} runner labels: {sorted(all_labels)}")
window_seconds = hours * 3600
window_end = datetime.now(timezone.utc)
window_start = window_end - timedelta(hours=hours)
# Per-host window-clamped busy time (each physical machine counted once).
# This is the source of truth for how loaded each host actually is.
# Busy time is the length of the UNION of the host's job intervals, not
# their sum: a named runner executes one job at a time, so overlapping
# records (e.g. an orphaned job stuck reporting in_progress while the
# host serves new jobs) are API artifacts. Summing them pushed saturated
# hosts to ~110% utilization; the union caps every host at 100%.
host_busy_seconds = {}
for host, jobs in host_jobs.items():
busy = 0.0
intervals = []
for j in jobs:
cs = max(j["start"], window_start)
ce = min(j["end"], window_end)
if ce > cs:
busy += (ce - cs).total_seconds()
host_busy_seconds[host] = busy
intervals.append((cs, ce))
host_busy_seconds[host] = union_seconds(intervals)
results = []
for label in sorted(all_labels):
@@ -586,15 +787,16 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
# Per-job detail (deduped across labels), longest waits first, for the
# links + status section of the report.
longest_waits = sorted(all_job_infos, key=lambda j: j["queue_time"], reverse=True)
return results, fetch_failure_pct, longest_waits
return results, fetch_failure_pct, longest_waits, coverage_hours
def format_report(
results: list[dict],
hours: int,
hours: float,
fetch_failure_pct: float = 0.0,
longest_waits: list = None,
top_n: int = 20,
coverage_hours: float = None,
) -> str:
"""One compact summary table — original schema, fixed columns.
@@ -606,13 +808,23 @@ def format_report(
physical hosts, so their utilization now reflects real hardware
saturation instead of being divided across labels.
"""
if coverage_hours is None:
coverage_hours = hours
lines = [
"# Runner Utilization Report",
"",
f"**Time window:** Last {hours} hours · "
f"**Time window:** Last {coverage_hours:.1f} hours · "
f"**Generated:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}",
"",
]
if coverage_hours < hours - 0.05:
lines.append(
f"⚠️ **Coverage warning**: the run listing hit the API safety "
f"cap, so only the most recent {coverage_hours:.1f}h of the "
f"requested {hours:.0f}h window is covered. All metrics below "
f"are computed over the covered {coverage_hours:.1f}h."
)
lines.append("")
if fetch_failure_pct > 1.0:
lines.append(
f"⚠️ **Data completeness warning**: {fetch_failure_pct:.0f}% of "
@@ -732,17 +944,30 @@ def main():
parser.add_argument(
"--hours", type=float, default=24, help="Time window in hours (fractional ok)"
)
parser.add_argument(
"--lookback-hours",
type=float,
default=None,
help=(
"How far before the window to list runs whose long-queued jobs "
"may still execute inside it (default: min(12, hours/2))"
),
)
parser.add_argument(
"--filter", type=str, help="Filter runner labels (e.g., '5090', 'h200')"
)
parser.add_argument("--output", type=str, help="Output file (default: stdout)")
args = parser.parse_args()
results, fetch_failure_pct, longest_waits = calculate_utilization(
args.repo, args.hours, args.filter
results, fetch_failure_pct, longest_waits, coverage_hours = calculate_utilization(
args.repo, args.hours, args.filter, lookback_hours=args.lookback_hours
)
report = format_report(
results, args.hours, fetch_failure_pct, longest_waits=longest_waits
results,
args.hours,
fetch_failure_pct,
longest_waits=longest_waits,
coverage_hours=coverage_hours,
)
if args.output:
scripts/ci/utils/test_runner_utilization_report.py
+222
View File
@@ -191,5 +191,227 @@ class TestStatusAndFormatting(unittest.TestCase):
self.assertIn("⏳", report)
class TestGetWorkflowRuns(unittest.TestCase):
"""Regression guard for the run-listing truncation bugs.
Two distinct failure modes are covered:
- the original silent 50-page cap (5000 runs) on an UNfiltered listing
while ~18k runs fire per busy 24h — fixed by server-side `created`
filtering, with honest truncation reporting when a budget is hit;
- the GitHub API's 1000-result cap on any `created`-FILTERED listing
(page 11 comes back empty despite a larger total_count) — fixed by
walking the range's upper bound down past each cap.
"""
def _fake_gh_api(self, all_runs):
"""Emulate the runs-listing API over `all_runs` (newest-first),
including the 1000-result-per-filtered-query cap."""
calls = []
def fake(args):
path = args[0]
calls.append(path)
page = int(path.split("&page=")[1].split("&")[0])
created = path.split("&created=")[1]
if created.startswith(">="):
lo, hi = created[2:], None
else:
lo, hi = created.split("..")
matching = [
r
for r in all_runs
if r["created_at"] >= lo and (hi is None or r["created_at"] <= hi)
]
offset = (page - 1) * 100
# The API serves at most 1000 results per filtered query even
# though total_count reports the full match count.
page_runs = matching[offset : offset + 100] if offset < 1000 else []
return {"total_count": len(matching), "workflow_runs": page_runs}
return fake, calls
def _make_runs(self, n):
"""n runs, newest-first, one per minute."""
return [
{
"id": i,
"created_at": _iso(NOW - timedelta(minutes=i + 1)),
}
for i in range(n)
]
def _run(self, all_runs, **kw):
fake, calls = self._fake_gh_api(all_runs)
orig = rur.run_gh_command
rur.run_gh_command = fake
try:
result = rur.get_workflow_runs("o/r", since=NOW - timedelta(hours=24), **kw)
finally:
rur.run_gh_command = orig
return result, calls
def test_uses_created_filter_and_stops_when_exhausted(self):
(runs, truncated), calls = self._run(self._make_runs(130))
self.assertEqual(len(runs), 130)
self.assertFalse(truncated)
self.assertEqual(len(calls), 2)
self.assertIn("created=>=", calls[0])
def test_walks_past_the_1000_result_cap(self):
# 1327 runs in-window (the exact live-validation failure): a single
# filtered query serves only the newest 1000; the cursor walkdown
# must fetch the remaining 327 via a narrowed created range.
(runs, truncated), calls = self._run(self._make_runs(1327))
self.assertEqual(len(runs), 1327)
self.assertEqual(len({r["id"] for r in runs}), 1327) # deduped
self.assertFalse(truncated)
# Second sweep queries a bounded range, not the open-ended filter.
self.assertTrue(any(".." in c.split("&created=")[1] for c in calls))
def test_reports_truncation_at_request_budget(self):
(runs, truncated), _ = self._run(self._make_runs(1000), max_pages=3)
self.assertEqual(len(runs), 300)
self.assertTrue(truncated)
class TestPreWindowJobFiltering(unittest.TestCase):
"""Lookback runs bring in jobs that finished before the window started.
Those jobs must be droppable by comparing their latest activity
(busy end or queue end) against window_start — kept only when they
touch the window. This mirrors the inline filter in
calculate_utilization.
"""
def _latest_activity(self, info):
return max(t for t in (info["end"], info["queue_end"]) if t is not None)
def test_finished_pre_window_job_is_droppable(self):
window_start = NOW - timedelta(hours=24)
info = rur.classify_job(
_job(
created_at=_iso(NOW - timedelta(hours=30)),
started_at=_iso(NOW - timedelta(hours=29)),
completed_at=_iso(NOW - timedelta(hours=26)),
),
NOW,
)
self.assertLess(self._latest_activity(info), window_start)
def test_job_spanning_window_start_is_kept(self):
window_start = NOW - timedelta(hours=24)
info = rur.classify_job(
_job(
created_at=_iso(NOW - timedelta(hours=30)),
started_at=_iso(NOW - timedelta(hours=26)),
completed_at=_iso(NOW - timedelta(hours=20)),
),
NOW,
)
self.assertGreaterEqual(self._latest_activity(info), window_start)
def test_still_queued_pre_window_job_is_kept(self):
# queue_end anchors to `now` for still-queued jobs, so an old run's
# still-waiting job always touches the window.
window_start = NOW - timedelta(hours=24)
info = rur.classify_job(
_job(
status="queued",
runner_name="",
created_at=_iso(NOW - timedelta(hours=30)),
started_at=_iso(NOW - timedelta(hours=30)),
completed_at=None,
),
NOW,
)
self.assertGreaterEqual(self._latest_activity(info), window_start)
class TestCarriedOverFingerprint(unittest.TestCase):
"""Re-run attempts duplicate the completed jobs they did NOT re-run as
new records (new id, identical name/runner/timestamps). Verified live:
a 2-attempt run listed the same successful job twice under different
ids, double-counting its busy time and pass count."""
def _record(self, **kw):
base = _job(
started_at=_iso(CREATED + timedelta(minutes=5)),
completed_at=_iso(CREATED + timedelta(minutes=25)),
)
base.update({"run_id": 42, "id": 1})
base.update(kw)
return base
def test_carried_over_copy_same_fingerprint(self):
a = self._record(id=100, run_attempt=1)
b = self._record(id=200, run_attempt=2) # copy with a new job id
self.assertIsNotNone(rur.carried_over_fingerprint(a))
self.assertEqual(
rur.carried_over_fingerprint(a), rur.carried_over_fingerprint(b)
)
def test_genuine_retry_differs(self):
# An actually re-run job has its own execution timestamps.
a = self._record(id=100, run_attempt=1)
b = self._record(
id=200,
run_attempt=2,
started_at=_iso(CREATED + timedelta(hours=2)),
completed_at=_iso(CREATED + timedelta(hours=2, minutes=20)),
)
self.assertNotEqual(
rur.carried_over_fingerprint(a), rur.carried_over_fingerprint(b)
)
def test_non_completed_jobs_not_fingerprinted(self):
queued = self._record(status="queued", runner_name="", completed_at=None)
self.assertIsNone(rur.carried_over_fingerprint(queued))
class TestUnionSeconds(unittest.TestCase):
"""Per-host busy time must be the union of job intervals, not the sum.
A named runner executes one job at a time; overlapping records (e.g.
an orphaned job stuck in_progress while the host serves new jobs) are
API artifacts. Summing them reported saturated hosts at ~110%
utilization.
"""
def _iv(self, s_min, e_min):
return (NOW + timedelta(minutes=s_min), NOW + timedelta(minutes=e_min))
def test_disjoint_intervals_sum(self):
got = rur.union_seconds([self._iv(0, 10), self._iv(20, 30)])
self.assertAlmostEqual(got, 20 * 60)
def test_overlap_counted_once(self):
# 0..60 ghost overlapping three real 15-min jobs -> union is 60min,
# not 105min.
got = rur.union_seconds(
[self._iv(0, 60), self._iv(0, 15), self._iv(20, 35), self._iv(40, 55)]
)
self.assertAlmostEqual(got, 60 * 60)
def test_unsorted_input_and_touching_edges(self):
got = rur.union_seconds([self._iv(10, 20), self._iv(0, 10)])
self.assertAlmostEqual(got, 20 * 60)
def test_empty(self):
self.assertEqual(rur.union_seconds([]), 0.0)
class TestCoverageWarning(unittest.TestCase):
def test_banner_when_coverage_below_requested(self):
report = rur.format_report([], 24, 0.0, coverage_hours=9.3)
self.assertIn("Coverage warning", report)
self.assertIn("9.3h", report)
def test_no_banner_at_full_coverage(self):
report = rur.format_report([], 24, 0.0, coverage_hours=24)
self.assertNotIn("Coverage warning", report)
report = rur.format_report([], 24, 0.0) # default: full coverage
self.assertNotIn("Coverage warning", report)
if __name__ == "__main__":
unittest.main()