From 7bb7f6049a19026bb537f7159ab6297a46035d20 Mon Sep 17 00:00:00 2001 From: Alison Shao <54658187+alisonshao@users.noreply.github.com> Date: Thu, 30 Apr 2026 10:05:16 -0700 Subject: [PATCH] ci: add per-host utilization view to runner-utilization report (#24102) --- scripts/ci/utils/runner_utilization_report.py | 466 +++++++++++------- 1 file changed, 284 insertions(+), 182 deletions(-) diff --git a/scripts/ci/utils/runner_utilization_report.py b/scripts/ci/utils/runner_utilization_report.py index eed6e67f0..2792ea2a5 100755 --- a/scripts/ci/utils/runner_utilization_report.py +++ b/scripts/ci/utils/runner_utilization_report.py @@ -9,7 +9,9 @@ Reports idle time, active time, and utilization percentage per runner label. import argparse import json import os +import random import subprocess +import time from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta, timezone @@ -19,16 +21,51 @@ DEFAULT_LABELS_TO_IGNORE = {"self-hosted", "Linux", "X64", "ARM64"} GITHUB_HOSTED_LABELS = {"ubuntu-latest", "ubuntu-22.04", "ubuntu-24.04"} -def run_gh_command(args: list[str]) -> dict: - """Run gh CLI command and return JSON result.""" - result = subprocess.run( - ["gh", "api"] + args, - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise Exception(f"gh api failed: {result.stderr}") - return json.loads(result.stdout) +def run_gh_command(args: list[str], max_retries: int = 10) -> dict: + """Run gh CLI command and return JSON result. + + Retries on transient failures (5xx, secondary rate limits, network + blips) with exponential backoff. The previous fail-fast behavior + combined with `except Exception: return None` in the threadpool + callers caused entire workflow runs to be silently dropped from + the utilization numerator whenever GH API hiccuped, severely + undercounting busy time on busy days. + """ + last_err = "" + for attempt in range(max_retries): + result = subprocess.run( + ["gh", "api"] + args, + capture_output=True, + text=True, + ) + if result.returncode == 0: + return json.loads(result.stdout) + last_err = result.stderr or "(no stderr)" + # Detect retryable conditions: HTTP 5xx, secondary rate limit, abuse + # detection, network resets. 4xx other than 429 are non-retryable. + retryable = any( + s in last_err + for s in ( + "rate limit", + "abuse", + "Internal Server Error", + "502", + "503", + "504", + "Bad Gateway", + "Gateway Time-out", + "connection reset", + "Connection reset", + "EOF", + "timeout", + ) + ) + if not retryable: + break + # Exponential backoff with jitter, capped at 60s. + delay = min(60, (2**attempt) + random.uniform(0, 1)) + time.sleep(delay) + raise Exception(f"gh api failed after {max_retries} attempts: {last_err[:300]}") def get_workflow_runs(repo: str, hours: int = 24) -> list[dict]: @@ -57,26 +94,34 @@ def get_workflow_runs(repo: str, hours: int = 24) -> list[dict]: if len(page_runs) < 100: break page += 1 - if page > 20: # Safety limit + if page > 50: # Safety limit (5000 runs) break return runs def get_jobs_for_run(repo: str, run_id: int) -> list[dict]: - """Get all jobs for a workflow run.""" + """Get all jobs for a workflow run, including all retry attempts. + + `filter=all` is required so that re-run attempts of the same job + appear separately. Each attempt consumed host time on the runner + pool, so for utilization we want them all summed in. The default + (`filter=latest`) only returns the most recent attempt and silently + hides time spent on prior retries. + """ jobs = [] page = 1 while True: data = run_gh_command( [ - f"repos/{repo}/actions/runs/{run_id}/jobs?per_page=100&page={page}", + f"repos/{repo}/actions/runs/{run_id}/jobs" + f"?per_page=100&page={page}&filter=all", ] ) jobs.extend(data.get("jobs", [])) if len(data.get("jobs", [])) < 100: break page += 1 - if page > 5: # Safety limit + if page > 20: # Safety limit (2000 jobs per run) break return jobs @@ -112,33 +157,13 @@ def parse_time(time_str: str) -> datetime: return datetime.fromisoformat(time_str.replace("Z", "+00:00")) -# Known runner counts per label (fallback when API unavailable) -KNOWN_RUNNER_COUNTS = { - "1-gpu-5090": 16, - "h200": 8, - "h20": 4, - "b200": 4, - "amd": 8, - "github-hosted": 20, # GitHub hosted runners (variable) - "other": 10, -} - - def calculate_concurrency_metrics( - jobs: list[dict], + jobs: list, window_start: datetime, window_end: datetime, num_runners: int, ) -> dict: - """ - Calculate concurrency metrics using a sweep line algorithm. - - Tracks: - - Peak concurrent runners in use - - Average concurrent runners over time - - Time at saturation (all runners busy) - - Queue depth when runners are saturated - """ + """Sweep-line algorithm: peak/avg concurrent, saturation time, peak queue.""" if not jobs: return { "peak_concurrent": 0, @@ -147,7 +172,6 @@ def calculate_concurrency_metrics( "saturation_pct": 0.0, "peak_queue": 0, } - window_seconds = (window_end - window_start).total_seconds() if window_seconds <= 0: return { @@ -157,75 +181,49 @@ def calculate_concurrency_metrics( "saturation_pct": 0.0, "peak_queue": 0, } - - # Create events for running jobs: +1 at start, -1 at end running_events = [] for job in jobs: - start = job["start"] - end = job["end"] - # Clamp to window + start, end = job["start"], job["end"] if end < window_start or start > window_end: continue - clamped_start = max(start, window_start) - clamped_end = min(end, window_end) - running_events.append((clamped_start, 1, "start")) # +1 for start - running_events.append((clamped_end, -1, "end")) # -1 for end - - # Create events for queue tracking (jobs created but not started) + running_events.append((max(start, window_start), 1)) + running_events.append((min(end, window_end), -1)) queue_events = [] for job in jobs: created_at = job.get("created_at") started_at = job["start"] if created_at and created_at < started_at: - # Clamp to window if started_at < window_start or created_at > window_end: continue - clamped_created = max(created_at, window_start) - clamped_started = min(started_at, window_end) - queue_events.append((clamped_created, 1, "queued")) - queue_events.append((clamped_started, -1, "dequeued")) - - # Sort running events: by time, then ends before starts at same time + queue_events.append((max(created_at, window_start), 1)) + queue_events.append((min(started_at, window_end), -1)) running_events.sort(key=lambda e: (e[0], e[1] == 1)) - - # Process running events to get concurrency metrics current_running = 0 peak_running = 0 prev_time = window_start total_running_seconds = 0.0 saturation_seconds = 0.0 - - for event_time, delta, _ in running_events: - # Accumulate time at previous concurrency level - time_delta = (event_time - prev_time).total_seconds() - if time_delta > 0: - total_running_seconds += current_running * time_delta + for event_time, delta in running_events: + td = (event_time - prev_time).total_seconds() + if td > 0: + total_running_seconds += current_running * td if current_running >= num_runners: - saturation_seconds += time_delta - - # Update concurrency + saturation_seconds += td current_running += delta peak_running = max(peak_running, current_running) prev_time = event_time - - # Handle remaining time after last event if prev_time < window_end: - time_delta = (window_end - prev_time).total_seconds() - total_running_seconds += current_running * time_delta + td = (window_end - prev_time).total_seconds() + total_running_seconds += current_running * td if current_running >= num_runners: - saturation_seconds += time_delta - - # Sort queue events and calculate peak queue depth + saturation_seconds += td queue_events.sort(key=lambda e: (e[0], e[1] == 1)) current_queued = 0 peak_queue = 0 - - for _, delta, _ in queue_events: + for _, delta in queue_events: current_queued += delta peak_queue = max(peak_queue, current_queued) - avg_concurrent = total_running_seconds / window_seconds if window_seconds > 0 else 0 - return { "peak_concurrent": peak_running, "avg_concurrent": avg_concurrent, @@ -237,12 +235,49 @@ def calculate_concurrency_metrics( } +_NON_GPU_WORKFLOW_HINTS = ( + "lint", + "deploy", + "release", + "publish", + "docs", + "doc", + "mintlify", + "runner utilization", # this very script + "tag-and-rerun", + "auto", # auto-merge etc. + "label", + "stale", + "dependabot", + "codeql", +) + + +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. + """ + if not workflow_name: + return False + n = workflow_name.lower() + 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.""" print(f"Fetching workflow runs from last {hours} hours...") - runs = get_workflow_runs(repo, hours) - print(f"Found {len(runs)} workflow runs") + 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) + print( + f"Found {len(all_runs)} workflow runs " + f"({skipped} skipped as non-GPU: docs/lint/release/etc.)" + ) # Try to get online runners from API print("Fetching online runners...") @@ -263,31 +298,64 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None) # Track runners seen in jobs (for labels not in API or when API unavailable) job_label_runners = defaultdict(set) label_jobs = defaultdict(list) # label -> list of job_info + # Per-host accumulation: each physical machine appears once regardless of + # how many overlapping labels it advertises. This is what we use for the + # "Per Host Utilization" section (the source-of-truth view). + host_jobs = defaultdict(list) # runner_name -> list of job_info + host_labels = defaultdict(set) # runner_name -> set of labels it ran jobs under - # Fetch jobs for all runs in parallel + # Fetch jobs for all runs in parallel. Cap concurrency lower than the + # GH API secondary rate-limit threshold to avoid bursts that silently + # drop runs even with retries. total_runs = len(runs) print(f"Fetching jobs for {total_runs} runs in parallel...") def fetch_jobs_for_run(run): - """Fetch jobs for a single run, returning (run_id, jobs) or (run_id, None) on error.""" + """Fetch jobs for a single run. + + Returns (run_id, jobs, error_msg). `error_msg` is None on success. + We surface failures rather than silently dropping the run so the + caller can report how many runs' jobs are missing — silently + dropping previously caused 4-gpu-b200 (and every other label) to + report wildly different numbers depending on transient API hiccups. + """ try: - return (run["id"], get_jobs_for_run(repo, run["id"])) - except Exception: - return (run["id"], None) + return (run["id"], get_jobs_for_run(repo, run["id"]), None) + except Exception as e: + return (run["id"], None, str(e)[:200]) all_jobs = [] - with ThreadPoolExecutor(max_workers=20) as executor: + failed_runs = [] + # Concurrency=4 with longer retry budget keeps us well below the GH + # API secondary rate-limit threshold (~10 req/s). On a 24h window + # with ~1500 GPU-relevant runs (post-filter), this completes in ~5 + # min and almost never hits the rate limit. + with ThreadPoolExecutor(max_workers=4) as executor: futures = [executor.submit(fetch_jobs_for_run, run) for run in runs] completed = 0 for future in as_completed(futures): completed += 1 - if completed % 50 == 0: - print(f"Fetched jobs for {completed}/{total_runs} runs...") - run_id, jobs = future.result() - if jobs: + if completed % 100 == 0: + print( + f"Fetched jobs for {completed}/{total_runs} runs " + f"({len(failed_runs)} failed so far)..." + ) + run_id, jobs, err = future.result() + if err: + failed_runs.append((run_id, err)) + elif jobs: all_jobs.extend(jobs) print(f"Processing {len(all_jobs)} jobs...") + if failed_runs: + print( + f"WARNING: {len(failed_runs)}/{total_runs} runs failed to fetch " + f"after retries. Utilization will be undercounted. " + f"First few errors:" + ) + for rid, err in failed_runs[:5]: + print(f" run {rid}: {err}") + fetch_failure_pct = len(failed_runs) / total_runs * 100 if total_runs > 0 else 0 for job in all_jobs: runner_name = job.get("runner_name") @@ -313,6 +381,9 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None) "runner_name": runner_name, } + # Per-host: every job on this physical machine, regardless of label. + host_jobs[runner_name].append(job_info) + # Use job labels directly (available in job data) job_labels = job.get("labels", []) for label in job_labels: @@ -321,6 +392,7 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None) continue job_label_runners[label].add(runner_name) label_jobs[label].append(job_info) + host_labels[runner_name].add(label) # Merge API runners and job-observed runners # Prefer API count (online runners) when available @@ -332,56 +404,70 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None) print(f"Tracking {len(all_labels)} runner labels: {sorted(all_labels)}") - # Calculate metrics per label 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. + host_busy_seconds = {} + for host, jobs in host_jobs.items(): + busy = 0.0 + 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 + results = [] - for label in sorted(all_labels): - # Use API runner count if available, otherwise use job-observed count - if label in api_label_runners and api_label_runners[label]: - num_runners = len(api_label_runners[label]) - elif label in job_label_runners: - num_runners = len(job_label_runners[label]) - else: - num_runners = KNOWN_RUNNER_COUNTS.get(label, 1) - - total_capacity_seconds = window_seconds * num_runners - - jobs = label_jobs.get(label, []) - total_active_seconds = sum(j["duration"] for j in jobs) - - utilization = ( - (total_active_seconds / total_capacity_seconds * 100) - if total_capacity_seconds > 0 - else 0 + # Hosts to attribute to this label = union of currently-online + # runners advertising the label PLUS hosts that actually ran a + # job under it during the window. The union catches hosts that + # went offline mid-window (their busy time is still real + # capacity consumed) and hosts that came online late. + hosts = api_label_runners.get(label, set()) | job_label_runners.get( + label, set() ) - idle_seconds = total_capacity_seconds - total_active_seconds + num_runners = len(hosts) if hosts else 1 - # Calculate queue time metrics + # Pool busy time: sum of busy time across the hosts that could + # serve this label, regardless of which sibling label actually + # dispatched the job. This is the right denominator/numerator for + # asking "how saturated is the underlying hardware that this + # label depends on?" — sibling labels (e.g. `4-gpu-b200` and + # `4-gpu-b200-low-disk`) compete for the same physical machines, + # so their busy time should not be double-counted into separate + # capacity buckets. + active_seconds = sum(host_busy_seconds.get(h, 0.0) for h in hosts) + capacity_seconds = num_runners * window_seconds + utilization = ( + (active_seconds / capacity_seconds * 100) if capacity_seconds > 0 else 0 + ) + + # Job count + queue stats stay label-specific (only jobs that + # were dispatched under THIS label). + jobs = label_jobs.get(label, []) queue_times = [j["queue_time"] for j in jobs if j["queue_time"] > 0] - avg_queue_time = sum(queue_times) / len(queue_times) if queue_times else 0 - max_queue_time = max(queue_times) if queue_times else 0 + avg_queue = sum(queue_times) / len(queue_times) if queue_times else 0 + max_queue = max(queue_times) if queue_times else 0 - # Calculate concurrency metrics - # First pass: get peak concurrent to determine effective capacity - concurrency_initial = calculate_concurrency_metrics( + # Concurrency / saturation / queue-depth metrics. Use observed + # peak as effective capacity if it's lower than the API count + # (e.g. for autoscaling pools where most listeners sit idle). + conc_initial = calculate_concurrency_metrics( jobs, window_start, window_end, num_runners ) - - # Use observed peak as effective capacity if lower than API count - # This handles cases where not all runners are active all the time - effective_runners = min(num_runners, concurrency_initial["peak_concurrent"]) + effective_runners = ( + min(num_runners, conc_initial["peak_concurrent"]) or num_runners + ) if effective_runners < num_runners and effective_runners > 0: - # Recalculate with effective capacity for accurate saturation - concurrency = calculate_concurrency_metrics( + conc = calculate_concurrency_metrics( jobs, window_start, window_end, effective_runners ) else: - concurrency = concurrency_initial - effective_runners = num_runners + conc = conc_initial results.append( { @@ -389,38 +475,76 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None) "num_runners": num_runners, "effective_runners": effective_runners, "num_jobs": len(jobs), - "total_active_hours": total_active_seconds / 3600, - "total_idle_hours": idle_seconds / 3600, - "total_capacity_hours": total_capacity_seconds / 3600, + "total_active_hours": active_seconds / 3600, "utilization_pct": utilization, - "avg_queue_min": avg_queue_time / 60, - "max_queue_min": max_queue_time / 60, - # Concurrency metrics - "peak_concurrent": concurrency_initial["peak_concurrent"], - "avg_concurrent": concurrency["avg_concurrent"], - "saturation_hours": concurrency["saturation_seconds"] / 3600, - "saturation_pct": concurrency["saturation_pct"], - "peak_queue": concurrency["peak_queue"], + "avg_queue_min": avg_queue / 60, + "max_queue_min": max_queue / 60, + "peak_concurrent": conc_initial["peak_concurrent"], + "avg_concurrent": conc["avg_concurrent"], + "saturation_hours": conc["saturation_seconds"] / 3600, + "saturation_pct": conc["saturation_pct"], + "peak_queue": conc["peak_queue"], } ) - return results + return results, fetch_failure_pct -def format_report(results: list[dict], hours: int) -> str: - """Format results as markdown report.""" +def format_report( + results: list[dict], hours: int, fetch_failure_pct: float = 0.0 +) -> str: + """One compact summary table — original schema, fixed columns. + + Active (hrs) and Utilization now reflect the actual host pool's + busy time (sum across all jobs on the hosts that advertise this + label, regardless of which sibling label dispatched them). This + makes the column meaningful for shared host pools — e.g. + `4-gpu-b200` and `4-gpu-b200-low-disk` both consume the same + physical hosts, so their utilization now reflects real hardware + saturation instead of being divided across labels. + """ lines = [ "# Runner Utilization Report", "", - f"**Time window:** Last {hours} hours", + f"**Time window:** Last {hours} hours · " f"**Generated:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}", "", - "## Concurrency Analysis", - "", - "| Label | Runners (API/Effective) | Peak Concurrent | Avg Concurrent | Saturation Time | Peak Queue |", - "|-------|-------------------------|-----------------|----------------|-----------------|------------|", ] + if fetch_failure_pct > 1.0: + lines.append( + f"⚠️ **Data completeness warning**: {fetch_failure_pct:.0f}% of " + f"GPU-relevant workflow runs failed to fetch jobs after retries " + f"(GH API rate limit). Active hours and utilization below are " + f"under-counted by approximately this fraction." + ) + lines.append("") + lines.extend( + [ + "| Label | Runners | Jobs | Active (hrs) | Utilization | Avg Queue | Max Queue |", + "|-------|---------|------|--------------|-------------|-----------|-----------|", + ] + ) + for r in results: + bar = "█" * int(r["utilization_pct"] / 10) + "░" * ( + 10 - int(r["utilization_pct"] / 10) + ) + lines.append( + f"| {r['label']} | {r['num_runners']} | {r['num_jobs']} | " + f"{r['total_active_hours']:.1f} | " + f"{r['utilization_pct']:.1f}% {bar} | " + f"{r['avg_queue_min']:.1f}m | {r['max_queue_min']:.1f}m |" + ) + # Concurrency Analysis section + lines.extend( + [ + "", + "## Concurrency Analysis", + "", + "| Label | Runners (API/Effective) | Peak Concurrent | Avg Concurrent | Saturation Time | Peak Queue |", + "|-------|-------------------------|-----------------|----------------|-----------------|------------|", + ] + ) for r in results: effective = r["effective_runners"] avg_pct = (r["avg_concurrent"] / effective * 100) if effective > 0 else 0 @@ -437,62 +561,38 @@ def format_report(results: list[dict], hours: int) -> str: f"{r['peak_queue']} jobs |" ) - # Add recommendations section + # Recommendations lines.extend(["", "## Recommendations", ""]) - has_recommendations = False + has_recs = False for r in results: label = r["label"] - saturation_pct = r["saturation_pct"] - peak_queue = r["peak_queue"] + sat_pct = r["saturation_pct"] + peak_q = r["peak_queue"] effective = r["effective_runners"] avg_pct = (r["avg_concurrent"] / effective * 100) if effective > 0 else 0 - - if saturation_pct > 50 or peak_queue > 5: + if sat_pct > 50 or peak_q > 5: lines.append( - f"⚠️ **{label}**: High saturation ({saturation_pct:.0f}%) " - f"with queue buildup ({peak_queue} jobs). Consider adding runners." + f"⚠️ **{label}**: High saturation ({sat_pct:.0f}%) " + f"with queue buildup ({peak_q} jobs). Consider adding runners." ) - has_recommendations = True - elif saturation_pct > 20 or peak_queue > 0: + has_recs = True + elif sat_pct > 20 or peak_q > 0: lines.append( - f"📊 **{label}**: Moderate saturation ({saturation_pct:.0f}%), " - f"peak queue {peak_queue} jobs. Monitor for trends." + f"📊 **{label}**: Moderate saturation ({sat_pct:.0f}%), " + f"peak queue {peak_q} jobs. Monitor for trends." ) - has_recommendations = True + has_recs = True elif avg_pct < 30 and r["num_jobs"] > 0: lines.append( f"💡 **{label}**: Low average utilization ({avg_pct:.0f}%). " f"Runner pool may be oversized." ) - has_recommendations = True + has_recs = True else: lines.append(f"✓ **{label}**: Healthy utilization with minimal queueing.") - - if not has_recommendations and results: + if not has_recs and results: lines.append("All runner pools have healthy utilization.") - # Add summary table - lines.extend( - [ - "", - "## Summary by Runner Label", - "", - "| Label | Runners | Jobs | Active (hrs) | Utilization | Avg Queue | Max Queue |", - "|-------|---------|------|--------------|-------------|-----------|-----------|", - ] - ) - - for r in results: - utilization_bar = "█" * int(r["utilization_pct"] / 10) + "░" * ( - 10 - int(r["utilization_pct"] / 10) - ) - lines.append( - f"| {r['label']} | {r['num_runners']} | {r['num_jobs']} | " - f"{r['total_active_hours']:.1f} | " - f"{r['utilization_pct']:.1f}% {utilization_bar} | " - f"{r['avg_queue_min']:.1f}m | {r['max_queue_min']:.1f}m |" - ) - return "\n".join(lines) @@ -506,8 +606,10 @@ def main(): parser.add_argument("--output", type=str, help="Output file (default: stdout)") args = parser.parse_args() - results = calculate_utilization(args.repo, args.hours, args.filter) - report = format_report(results, args.hours) + results, fetch_failure_pct = calculate_utilization( + args.repo, args.hours, args.filter + ) + report = format_report(results, args.hours, fetch_failure_pct) if args.output: with open(args.output, "w") as f: