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

ci: add per-host utilization view to runner-utilization report (#24102)

Browse Source
This commit is contained in:
Alison Shao
2026-04-30 10:05:16 -07:00
committed by GitHub
parent 651af06a0b
commit 7bb7f6049a
1 changed files with 284 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/ci/utils/runner_utilization_report.py
+284 -182
View File
@@ -9,7 +9,9 @@ Reports idle time, active time, and utilization percentage per runner label.
import argparse import argparse
import json import json
import os import os
import random
import subprocess import subprocess
import time
from collections import defaultdict from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta, timezone 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"} GITHUB_HOSTED_LABELS = {"ubuntu-latest", "ubuntu-22.04", "ubuntu-24.04"}
def run_gh_command(args: list[str]) -> dict: def run_gh_command(args: list[str], max_retries: int = 10) -> dict:
"""Run gh CLI command and return JSON result.""" """Run gh CLI command and return JSON result.
result = subprocess.run(
["gh", "api"] + args, Retries on transient failures (5xx, secondary rate limits, network
capture_output=True, blips) with exponential backoff. The previous fail-fast behavior
text=True, combined with `except Exception: return None` in the threadpool
) callers caused entire workflow runs to be silently dropped from
if result.returncode != 0: the utilization numerator whenever GH API hiccuped, severely
raise Exception(f"gh api failed: {result.stderr}") undercounting busy time on busy days.
return json.loads(result.stdout) """
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]: 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: if len(page_runs) < 100:
break break
page += 1 page += 1
if page > 20: # Safety limit if page > 50: # Safety limit (5000 runs)
break break
return runs return runs
def get_jobs_for_run(repo: str, run_id: int) -> list[dict]: 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 = [] jobs = []
page = 1 page = 1
while True: while True:
data = run_gh_command( 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", [])) jobs.extend(data.get("jobs", []))
if len(data.get("jobs", [])) < 100: if len(data.get("jobs", [])) < 100:
break break
page += 1 page += 1
if page > 5: # Safety limit if page > 20: # Safety limit (2000 jobs per run)
break break
return jobs return jobs
@@ -112,33 +157,13 @@ def parse_time(time_str: str) -> datetime:
return datetime.fromisoformat(time_str.replace("Z", "+00:00")) 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( def calculate_concurrency_metrics(
jobs: list[dict], jobs: list,
window_start: datetime, window_start: datetime,
window_end: datetime, window_end: datetime,
num_runners: int, num_runners: int,
) -> dict: ) -> dict:
""" """Sweep-line algorithm: peak/avg concurrent, saturation time, peak queue."""
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
"""
if not jobs: if not jobs:
return { return {
"peak_concurrent": 0, "peak_concurrent": 0,
@@ -147,7 +172,6 @@ def calculate_concurrency_metrics(
"saturation_pct": 0.0, "saturation_pct": 0.0,
"peak_queue": 0, "peak_queue": 0,
} }
window_seconds = (window_end - window_start).total_seconds() window_seconds = (window_end - window_start).total_seconds()
if window_seconds <= 0: if window_seconds <= 0:
return { return {
@@ -157,75 +181,49 @@ def calculate_concurrency_metrics(
"saturation_pct": 0.0, "saturation_pct": 0.0,
"peak_queue": 0, "peak_queue": 0,
} }
# Create events for running jobs: +1 at start, -1 at end
running_events = [] running_events = []
for job in jobs: for job in jobs:
start = job["start"] start, end = job["start"], job["end"]
end = job["end"]
# Clamp to window
if end < window_start or start > window_end: if end < window_start or start > window_end:
continue continue
clamped_start = max(start, window_start) running_events.append((max(start, window_start), 1))
clamped_end = min(end, window_end) running_events.append((min(end, window_end), -1))
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)
queue_events = [] queue_events = []
for job in jobs: for job in jobs:
created_at = job.get("created_at") created_at = job.get("created_at")
started_at = job["start"] started_at = job["start"]
if created_at and created_at < started_at: if created_at and created_at < started_at:
# Clamp to window
if started_at < window_start or created_at > window_end: if started_at < window_start or created_at > window_end:
continue continue
clamped_created = max(created_at, window_start) queue_events.append((max(created_at, window_start), 1))
clamped_started = min(started_at, window_end) queue_events.append((min(started_at, window_end), -1))
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
running_events.sort(key=lambda e: (e[0], e[1] == 1)) running_events.sort(key=lambda e: (e[0], e[1] == 1))
# Process running events to get concurrency metrics
current_running = 0 current_running = 0
peak_running = 0 peak_running = 0
prev_time = window_start prev_time = window_start
total_running_seconds = 0.0 total_running_seconds = 0.0
saturation_seconds = 0.0 saturation_seconds = 0.0
for event_time, delta in running_events:
for event_time, delta, _ in running_events: td = (event_time - prev_time).total_seconds()
# Accumulate time at previous concurrency level if td > 0:
time_delta = (event_time - prev_time).total_seconds() total_running_seconds += current_running * td
if time_delta > 0:
total_running_seconds += current_running * time_delta
if current_running >= num_runners: if current_running >= num_runners:
saturation_seconds += time_delta saturation_seconds += td
# Update concurrency
current_running += delta current_running += delta
peak_running = max(peak_running, current_running) peak_running = max(peak_running, current_running)
prev_time = event_time prev_time = event_time
# Handle remaining time after last event
if prev_time < window_end: if prev_time < window_end:
time_delta = (window_end - prev_time).total_seconds() td = (window_end - prev_time).total_seconds()
total_running_seconds += current_running * time_delta total_running_seconds += current_running * td
if current_running >= num_runners: if current_running >= num_runners:
saturation_seconds += time_delta saturation_seconds += td
# Sort queue events and calculate peak queue depth
queue_events.sort(key=lambda e: (e[0], e[1] == 1)) queue_events.sort(key=lambda e: (e[0], e[1] == 1))
current_queued = 0 current_queued = 0
peak_queue = 0 peak_queue = 0
for _, delta in queue_events:
for _, delta, _ in queue_events:
current_queued += delta current_queued += delta
peak_queue = max(peak_queue, current_queued) peak_queue = max(peak_queue, current_queued)
avg_concurrent = total_running_seconds / window_seconds if window_seconds > 0 else 0 avg_concurrent = total_running_seconds / window_seconds if window_seconds > 0 else 0
return { return {
"peak_concurrent": peak_running, "peak_concurrent": peak_running,
"avg_concurrent": avg_concurrent, "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): def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None):
"""Calculate runner utilization metrics.""" """Calculate runner utilization metrics."""
print(f"Fetching workflow runs from last {hours} hours...") print(f"Fetching workflow runs from last {hours} hours...")
runs = get_workflow_runs(repo, hours) all_runs = get_workflow_runs(repo, hours)
print(f"Found {len(runs)} workflow runs") 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 # Try to get online runners from API
print("Fetching online runners...") 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) # Track runners seen in jobs (for labels not in API or when API unavailable)
job_label_runners = defaultdict(set) job_label_runners = defaultdict(set)
label_jobs = defaultdict(list) # label -> list of job_info 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) total_runs = len(runs)
print(f"Fetching jobs for {total_runs} runs in parallel...") print(f"Fetching jobs for {total_runs} runs in parallel...")
def fetch_jobs_for_run(run): 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: try:
return (run["id"], get_jobs_for_run(repo, run["id"])) return (run["id"], get_jobs_for_run(repo, run["id"]), None)
except Exception: except Exception as e:
return (run["id"], None) return (run["id"], None, str(e)[:200])
all_jobs = [] 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] futures = [executor.submit(fetch_jobs_for_run, run) for run in runs]
completed = 0 completed = 0
for future in as_completed(futures): for future in as_completed(futures):
completed += 1 completed += 1
if completed % 50 == 0: if completed % 100 == 0:
print(f"Fetched jobs for {completed}/{total_runs} runs...") print(
run_id, jobs = future.result() f"Fetched jobs for {completed}/{total_runs} runs "
if jobs: 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) all_jobs.extend(jobs)
print(f"Processing {len(all_jobs)} 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: for job in all_jobs:
runner_name = job.get("runner_name") 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, "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) # Use job labels directly (available in job data)
job_labels = job.get("labels", []) job_labels = job.get("labels", [])
for label in job_labels: for label in job_labels:
@@ -321,6 +392,7 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
continue continue
job_label_runners[label].add(runner_name) job_label_runners[label].add(runner_name)
label_jobs[label].append(job_info) label_jobs[label].append(job_info)
host_labels[runner_name].add(label)
# Merge API runners and job-observed runners # Merge API runners and job-observed runners
# Prefer API count (online runners) when available # 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)}") print(f"Tracking {len(all_labels)} runner labels: {sorted(all_labels)}")
# Calculate metrics per label
window_seconds = hours * 3600 window_seconds = hours * 3600
window_end = datetime.now(timezone.utc) window_end = datetime.now(timezone.utc)
window_start = window_end - timedelta(hours=hours) 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 = [] results = []
for label in sorted(all_labels): for label in sorted(all_labels):
# Use API runner count if available, otherwise use job-observed count # Hosts to attribute to this label = union of currently-online
if label in api_label_runners and api_label_runners[label]: # runners advertising the label PLUS hosts that actually ran a
num_runners = len(api_label_runners[label]) # job under it during the window. The union catches hosts that
elif label in job_label_runners: # went offline mid-window (their busy time is still real
num_runners = len(job_label_runners[label]) # capacity consumed) and hosts that came online late.
else: hosts = api_label_runners.get(label, set()) | job_label_runners.get(
num_runners = KNOWN_RUNNER_COUNTS.get(label, 1) label, set()
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
) )
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] 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 avg_queue = sum(queue_times) / len(queue_times) if queue_times else 0
max_queue_time = max(queue_times) if queue_times else 0 max_queue = max(queue_times) if queue_times else 0
# Calculate concurrency metrics # Concurrency / saturation / queue-depth metrics. Use observed
# First pass: get peak concurrent to determine effective capacity # peak as effective capacity if it's lower than the API count
concurrency_initial = calculate_concurrency_metrics( # (e.g. for autoscaling pools where most listeners sit idle).
conc_initial = calculate_concurrency_metrics(
jobs, window_start, window_end, num_runners jobs, window_start, window_end, num_runners
) )
effective_runners = (
# Use observed peak as effective capacity if lower than API count min(num_runners, conc_initial["peak_concurrent"]) or num_runners
# This handles cases where not all runners are active all the time )
effective_runners = min(num_runners, concurrency_initial["peak_concurrent"])
if effective_runners < num_runners and effective_runners > 0: if effective_runners < num_runners and effective_runners > 0:
# Recalculate with effective capacity for accurate saturation conc = calculate_concurrency_metrics(
concurrency = calculate_concurrency_metrics(
jobs, window_start, window_end, effective_runners jobs, window_start, window_end, effective_runners
) )
else: else:
concurrency = concurrency_initial conc = conc_initial
effective_runners = num_runners
results.append( results.append(
{ {
@@ -389,38 +475,76 @@ def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None)
"num_runners": num_runners, "num_runners": num_runners,
"effective_runners": effective_runners, "effective_runners": effective_runners,
"num_jobs": len(jobs), "num_jobs": len(jobs),
"total_active_hours": total_active_seconds / 3600, "total_active_hours": active_seconds / 3600,
"total_idle_hours": idle_seconds / 3600,
"total_capacity_hours": total_capacity_seconds / 3600,
"utilization_pct": utilization, "utilization_pct": utilization,
"avg_queue_min": avg_queue_time / 60, "avg_queue_min": avg_queue / 60,
"max_queue_min": max_queue_time / 60, "max_queue_min": max_queue / 60,
# Concurrency metrics "peak_concurrent": conc_initial["peak_concurrent"],
"peak_concurrent": concurrency_initial["peak_concurrent"], "avg_concurrent": conc["avg_concurrent"],
"avg_concurrent": concurrency["avg_concurrent"], "saturation_hours": conc["saturation_seconds"] / 3600,
"saturation_hours": concurrency["saturation_seconds"] / 3600, "saturation_pct": conc["saturation_pct"],
"saturation_pct": concurrency["saturation_pct"], "peak_queue": conc["peak_queue"],
"peak_queue": concurrency["peak_queue"],
} }
) )
return results return results, fetch_failure_pct
def format_report(results: list[dict], hours: int) -> str: def format_report(
"""Format results as markdown 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 = [ lines = [
"# Runner Utilization Report", "# 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')}", 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: for r in results:
effective = r["effective_runners"] effective = r["effective_runners"]
avg_pct = (r["avg_concurrent"] / effective * 100) if effective > 0 else 0 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 |" f"{r['peak_queue']} jobs |"
) )
# Add recommendations section # Recommendations
lines.extend(["", "## Recommendations", ""]) lines.extend(["", "## Recommendations", ""])
has_recommendations = False has_recs = False
for r in results: for r in results:
label = r["label"] label = r["label"]
saturation_pct = r["saturation_pct"] sat_pct = r["saturation_pct"]
peak_queue = r["peak_queue"] peak_q = r["peak_queue"]
effective = r["effective_runners"] effective = r["effective_runners"]
avg_pct = (r["avg_concurrent"] / effective * 100) if effective > 0 else 0 avg_pct = (r["avg_concurrent"] / effective * 100) if effective > 0 else 0
if sat_pct > 50 or peak_q > 5:
if saturation_pct > 50 or peak_queue > 5:
lines.append( lines.append(
f"⚠️ **{label}**: High saturation ({saturation_pct:.0f}%) " f"⚠️ **{label}**: High saturation ({sat_pct:.0f}%) "
f"with queue buildup ({peak_queue} jobs). Consider adding runners." f"with queue buildup ({peak_q} jobs). Consider adding runners."
) )
has_recommendations = True has_recs = True
elif saturation_pct > 20 or peak_queue > 0: elif sat_pct > 20 or peak_q > 0:
lines.append( lines.append(
f"📊 **{label}**: Moderate saturation ({saturation_pct:.0f}%), " f"📊 **{label}**: Moderate saturation ({sat_pct:.0f}%), "
f"peak queue {peak_queue} jobs. Monitor for trends." f"peak queue {peak_q} jobs. Monitor for trends."
) )
has_recommendations = True has_recs = True
elif avg_pct < 30 and r["num_jobs"] > 0: elif avg_pct < 30 and r["num_jobs"] > 0:
lines.append( lines.append(
f"💡 **{label}**: Low average utilization ({avg_pct:.0f}%). " f"💡 **{label}**: Low average utilization ({avg_pct:.0f}%). "
f"Runner pool may be oversized." f"Runner pool may be oversized."
) )
has_recommendations = True has_recs = True
else: else:
lines.append(f"✓ **{label}**: Healthy utilization with minimal queueing.") lines.append(f"✓ **{label}**: Healthy utilization with minimal queueing.")
if not has_recs and results:
if not has_recommendations and results:
lines.append("All runner pools have healthy utilization.") 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) return "\n".join(lines)
@@ -506,8 +606,10 @@ def main():
parser.add_argument("--output", type=str, help="Output file (default: stdout)") parser.add_argument("--output", type=str, help="Output file (default: stdout)")
args = parser.parse_args() args = parser.parse_args()
results = calculate_utilization(args.repo, args.hours, args.filter) results, fetch_failure_pct = calculate_utilization(
report = format_report(results, args.hours) args.repo, args.hours, args.filter
)
report = format_report(results, args.hours, fetch_failure_pct)
if args.output: if args.output:
with open(args.output, "w") as f: with open(args.output, "w") as f: