#!/usr/bin/env python3 """Collect CI job/step timings from the GitHub API and generate an HTML diff report. In CI, the script reads GITHUB_TOKEN, GITHUB_REPOSITORY, GITHUB_RUN_ID, and GITHUB_SHA from the environment to collect timings via the REST API. If a baseline JSON file (ci-timings-baseline.json by default) exists, the report includes a diff with per-job and per-step deltas, plus a gantt chart overlaying current vs baseline bars. Usage: # Collect from API (CI mode): python scripts/ci/timings_report.py # Regenerate HTML from saved JSON (testing): python scripts/ci/timings_report.py --from-json ci-timings.json """ from __future__ import annotations import argparse import glob import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from datetime import datetime from html import escape API_BASE = "https://api.github.com" # Retry policy for GitHub API calls. The repo-scoped GITHUB_TOKEN shares a # rate-limit budget across every concurrent workflow run; when several PRs # run CI at once, this report job (which makes dozens of paginated calls) # regularly hits 403 rate-limit responses. Those are transient — retry with # backoff, honoring Retry-After / X-RateLimit-Reset when present. _RETRY_STATUSES = {403, 429, 500, 502, 503, 504} _MAX_ATTEMPTS = 5 _MAX_RETRY_WAIT_S = 120.0 class TimingsUnavailable(Exception): """GitHub API data could not be collected (rate limit, outage, ...). This is a REPORT job — never a reason to fail the PR's checks. main() catches this and exits 0 with a degraded summary. """ def _retry_wait_s(headers, attempt: int) -> float: """Seconds to wait before the next attempt, honoring server hints.""" retry_after = (headers.get("Retry-After") or "").strip() if headers else "" if retry_after.isdigit(): return min(float(retry_after), _MAX_RETRY_WAIT_S) reset = (headers.get("X-RateLimit-Reset") or "").strip() if headers else "" remaining = (headers.get("X-RateLimit-Remaining") or "").strip() if headers else "" if remaining == "0" and reset.isdigit(): return min(max(float(reset) - time.time(), 1.0), _MAX_RETRY_WAIT_S) return min(2.0 ** attempt * 2.0, _MAX_RETRY_WAIT_S) # 4s, 8s, 16s, 32s def _urlopen_with_retry(req: urllib.request.Request): """urlopen with backoff on rate-limit/transient statuses. Returns (parsed_json, link_header). Raises TimingsUnavailable when attempts are exhausted — callers treat that as "no report this run", not a job failure. """ last_err: Exception | None = None for attempt in range(1, _MAX_ATTEMPTS + 1): try: with urllib.request.urlopen(req) as resp: return json.loads(resp.read()), resp.headers.get("Link", "") except urllib.error.HTTPError as e: last_err = e if e.code not in _RETRY_STATUSES or attempt == _MAX_ATTEMPTS: break wait = _retry_wait_s(e.headers, attempt) print(f"GitHub API {e.code} on {req.full_url} — " f"retry {attempt}/{_MAX_ATTEMPTS - 1} in {wait:.0f}s", file=sys.stderr) time.sleep(wait) except urllib.error.URLError as e: last_err = e if attempt == _MAX_ATTEMPTS: break wait = _retry_wait_s(None, attempt) print(f"GitHub API connection error on {req.full_url} ({e.reason}) — " f"retry {attempt}/{_MAX_ATTEMPTS - 1} in {wait:.0f}s", file=sys.stderr) time.sleep(wait) raise TimingsUnavailable( f"GitHub API unavailable after {_MAX_ATTEMPTS} attempts: {last_err}" ) # --------------------------------------------------------------------------- # GitHub API helpers # --------------------------------------------------------------------------- def api_get(path: str, token: str, params: dict | None = None, list_key: str | None = None) -> list | dict: """Authenticated GitHub API GET with automatic pagination. For list endpoints, pass list_key to extract items from the paginated wrapper response (e.g. list_key='jobs' for {'total_count': N, 'jobs': [...]}). When list_key is omitted, a non-list response is returned as-is (single object). Transient failures (403 rate limit, 429, 5xx, connection errors) are retried with backoff; exhausted retries raise TimingsUnavailable. """ url = f"{API_BASE}{path}" if params: url += "?" + urllib.parse.urlencode(params) results: list = [] while url: req = urllib.request.Request(url, headers={ "Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", "User-Agent": "ci-timings-report", }) data, link_header = _urlopen_with_retry(req) if list_key: results.extend(data.get(list_key, [])) elif isinstance(data, list): results.extend(data) else: return data next_url = None for part in link_header.split(","): part = part.strip() if 'rel="next"' in part: next_url = part[part.find("<") + 1:part.find(">")] break url = next_url return results def parse_ts(ts: str | None) -> datetime | None: if not ts: return None return datetime.fromisoformat(ts.replace("Z", "+00:00")) def dur_s(started: str | None, completed: str | None) -> float | None: s = parse_ts(started) e = parse_ts(completed) if not s or not e: return None return (e - s).total_seconds() def is_skipped(job: dict) -> bool: """A job is 'skipped' when GitHub didn't actually run it. Skipped jobs have conclusion == 'skipped' and typically have null or equal started_at/completed_at timestamps, yielding duration_s of None or 0. They should be excluded from delta comparisons, gantt bars, regression tables, and aggregate stats (wall/compute). """ return job.get("conclusion") == "skipped" # --------------------------------------------------------------------------- # Timings collection # --------------------------------------------------------------------------- def _normalize_job(raw: dict) -> dict: steps = [] for step in (raw.get("steps") or []): steps.append({ "name": step.get("name", ""), "number": step.get("number", 0), "status": step.get("status", ""), "conclusion": step.get("conclusion", ""), "started_at": step.get("started_at"), "completed_at": step.get("completed_at"), "duration_s": dur_s(step.get("started_at"), step.get("completed_at")), }) return { "name": raw.get("name", "unknown"), "workflow_name": raw.get("_workflow_name", ""), "job_id": raw.get("id"), "status": raw.get("status", ""), "conclusion": raw.get("conclusion", ""), "started_at": raw.get("started_at"), "completed_at": raw.get("completed_at"), "duration_s": dur_s(raw.get("started_at"), raw.get("completed_at")), "html_url": raw.get("html_url", ""), "steps": steps, } # Step categories. "Overhead" is everything a job pays before and after its # actual work: runner setup, checkout, dependency restore/install, teardown. # Naming is GitHub's, not ours — a `uses:` step is reported as # "Run /@" and its cleanup as "Post Run <...>", while a # `run:` step keeps whatever `name:` the workflow gave it. So the setup # patterns match action refs and the well-known implicit steps, and anything # unmatched is treated as work (better to under-report overhead than to # silently classify a real test step as setup). STEP_SETUP = "setup" STEP_WORK = "work" STEP_TEARDOWN = "teardown" _SETUP_PATTERNS = ( "set up job", "set up python", "set up node", "checkout", "actions/checkout", "actions/setup-", "actions/cache", "restore ", # "Restore uv cache", "Restore baseline cache", ... "install ", # "Install dependencies", "Install ruff + ty", ... "minimize uv cache", "uv-cache", "set up docker buildx", "log in to", "authenticate to", "mint read-only cache token", "pull ghcr.io/", "get-app-token", "determine base ref", ) _TEARDOWN_PATTERNS = ( "complete job", "stop containers", "upload", "export results", ) def classify_step(name: str) -> str: """Bucket a step into setup / work / teardown. Any ``Post ...`` step is teardown regardless of what it post-processes — that's GitHub's own cleanup phase for a ``uses:`` step. """ low = (name or "").strip().lower() if low.startswith("post "): return STEP_TEARDOWN for pat in _TEARDOWN_PATTERNS: if pat in low: return STEP_TEARDOWN for pat in _SETUP_PATTERNS: if pat in low: return STEP_SETUP return STEP_WORK def display_step_name(name: str) -> str: """Strip the pinned SHA from an action ref for display. GitHub names a ``uses:`` step after the full pinned ref, e.g. ``Run actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd``. The 40-char SHA crowds out the part a reader cares about and makes the overhead table unreadable, so drop it. Only the DISPLAY changes — the raw name stays the key for baseline comparison and aggregation, since two different pins are genuinely two different steps. """ if not name: return "" return re.sub(r"@[0-9a-f]{7,40}\b", "", name) def compute_overhead(timings: dict) -> dict: """Aggregate setup/work/teardown seconds across every non-skipped job. Returns totals plus the worst individual setup steps, so the report can answer "how much of CI is spent getting ready to work" with a number instead of an impression. """ totals = {STEP_SETUP: 0.0, STEP_WORK: 0.0, STEP_TEARDOWN: 0.0} by_step: dict[str, dict] = {} jobs_with_steps = 0 for j in timings.get("jobs", []): if is_skipped(j) or not j.get("steps"): continue jobs_with_steps += 1 for s in j["steps"]: dur = s.get("duration_s") if dur is None or dur < 0: continue cat = classify_step(s.get("name", "")) totals[cat] += dur if cat != STEP_WORK: agg = by_step.setdefault( s.get("name", ""), {"name": s.get("name", ""), "total_s": 0.0, "count": 0, "category": cat} ) agg["total_s"] += dur agg["count"] += 1 accounted = sum(totals.values()) overhead = totals[STEP_SETUP] + totals[STEP_TEARDOWN] return { "setup_s": totals[STEP_SETUP], "work_s": totals[STEP_WORK], "teardown_s": totals[STEP_TEARDOWN], "overhead_s": overhead, "accounted_s": accounted, "overhead_pct": (overhead / accounted * 100) if accounted > 0 else 0.0, "jobs_with_steps": jobs_with_steps, "top_overhead_steps": sorted( by_step.values(), key=lambda x: -x["total_s"] )[:8], } def _annotate_wait_times(jobs: list[dict]) -> None: """Annotate each job with ``wait_s`` — how long it sat idle before starting. Wait time = ``started_at - max(completed_at of all jobs that finished before this job started)``. Jobs with no predecessor (e.g. ``detect``) get ``wait_s = 0``. Skipped jobs get ``wait_s = None``. This is a timestamp heuristic, not a workflow-YAML dependency parse: it infers dependencies from temporal ordering rather than ``needs:`` declarations. It's accurate for pipeline-shaped CI where the critical path is linear at each stage (detect → parallel lanes → gate → report). """ for j in jobs: if is_skipped(j): j["wait_s"] = None continue started = parse_ts(j.get("started_at")) if started is None: j["wait_s"] = None continue latest_dep_end: datetime | None = None for other in jobs: if other is j or is_skipped(other): continue other_end = parse_ts(other.get("completed_at")) if other_end is None or other_end > started: continue if latest_dep_end is None or other_end > latest_dep_end: latest_dep_end = other_end j["wait_s"] = (started - latest_dep_end).total_seconds() if latest_dep_end else 0.0 def collect_timings(token: str, repo: str, run_id: str, head_sha: str) -> dict: """Collect job/step timings from the GitHub API. 1. Get orchestrator run's direct jobs (detect, all-checks-pass, etc.). Skip workflow-call placeholder jobs (step name starts with "Run ./.github/workflows/"). 2. Find sub-workflow runs via head_sha + event=workflow_call. 3. Get each sub-workflow run's jobs with full step timing. """ owner, repo_name = repo.split("/") # Orchestrator run info run_info = api_get(f"/repos/{owner}/{repo_name}/actions/runs/{run_id}", token) created_at = run_info.get("created_at", "") # Orchestrator direct jobs orch_jobs = api_get(f"/repos/{owner}/{repo_name}/actions/runs/{run_id}/jobs", token, list_key="jobs") direct = [] for job in orch_jobs: steps = job.get("steps") or [] if any(s.get("name", "").startswith("Run ./.github/workflows/") for s in steps): continue # workflow-call placeholder if job.get("status") in ("in_progress", "queued"): continue # skip self / unfinished direct.append(job) # Sub-workflow runs sub_runs = api_get(f"/repos/{owner}/{repo_name}/actions/runs", token, params={ "head_sha": head_sha, "event": "workflow_call", "per_page": 100, }, list_key="workflow_runs") sub_runs = [r for r in sub_runs if r.get("created_at", "") >= created_at] sub_jobs_raw = [] for sr in sub_runs: sr_id = sr["id"] sr_name = sr.get("name", "") sr_jobs = api_get(f"/repos/{owner}/{repo_name}/actions/runs/{sr_id}/jobs", token, list_key="jobs") for j in sr_jobs: j["_workflow_name"] = sr_name j["_workflow_run_id"] = sr_id sub_jobs_raw.append(j) # Normalize + sort all_jobs = [_normalize_job(j) for j in direct + sub_jobs_raw] all_jobs = [j for j in all_jobs if j["status"] not in ("in_progress", "queued")] all_jobs.sort(key=lambda j: j.get("started_at") or "") _annotate_wait_times(all_jobs) return { "run_id": run_id, "head_sha": head_sha, "created_at": created_at, "jobs": all_jobs, } # --------------------------------------------------------------------------- # Formatting helpers # --------------------------------------------------------------------------- def fmt_dur(seconds: float | None) -> str: if seconds is None: return "—" if seconds < 60: return f"{seconds:.1f}s" m = int(seconds // 60) s = seconds % 60 if s == 0: return f"{m}m" return f"{m}m{s:.0f}s" def fmt_delta(current: float | None, baseline: float | None) -> tuple[str, str]: """Return (text, css_class) for a delta.""" if current is None or baseline is None: return ("—", "neutral") delta = current - baseline if baseline == 0: pct_str = "new" if delta > 0 else "0%" else: pct = (delta / baseline) * 100 pct_str = f"{pct:+.1f}%" if abs(delta) < 1.0: cls = "neutral" elif delta > 0: cls = "slower" else: cls = "faster" sign = "+" if delta >= 0 else "" return (f"{sign}{delta:.1f}s ({pct_str})", cls) def nice_ticks(max_seconds: float, num_ticks: int = 8) -> list[int]: if max_seconds <= 0: return [0] raw = max_seconds / num_ticks for nice in [5, 10, 15, 30, 60, 120, 180, 300, 600, 900, 1800, 3600, 7200]: if nice >= raw: step = nice break else: step = max(int(raw), 3600) return list(range(0, int(max_seconds) + step + 1, step)) def fmt_tick(seconds: int) -> str: if seconds < 60: return f"{seconds}s" m, s = divmod(seconds, 60) if s == 0: return f"{m}m" return f"{m}m{s}s" # --------------------------------------------------------------------------- # Stats computation # --------------------------------------------------------------------------- def compute_stats(timings: dict, baseline: dict | None = None) -> dict: jobs_all = timings.get("jobs", []) jobs = [j for j in jobs_all if not is_skipped(j)] bl_jobs_all = (baseline or {}).get("jobs", []) bl_jobs = [j for j in bl_jobs_all if not is_skipped(j)] bl_map = {j["name"]: j for j in bl_jobs} # Wall time (skipped jobs have no real timestamps) starts = [s for s in (parse_ts(j.get("started_at")) for j in jobs) if s is not None] ends = [e for e in (parse_ts(j.get("completed_at")) for j in jobs) if e is not None] wall = (max(ends) - min(starts)).total_seconds() if starts and ends else 0 compute = sum(j.get("duration_s") or 0 for j in jobs) # Baseline wall/compute bl_wall = None bl_compute = None if baseline: bl_starts = [s for s in (parse_ts(j.get("started_at")) for j in bl_jobs) if s is not None] bl_ends = [e for e in (parse_ts(j.get("completed_at")) for j in bl_jobs) if e is not None] if bl_starts and bl_ends: bl_wall = (max(bl_ends) - min(bl_starts)).total_seconds() bl_compute = sum(j.get("duration_s") or 0 for j in bl_jobs) # Per-job deltas (skipped excluded) faster = 0 slower = 0 unchanged = 0 no_baseline = 0 for j in jobs: bl = bl_map.get(j["name"]) if not bl: no_baseline += 1 continue cur_d = j.get("duration_s") or 0 bl_d = bl.get("duration_s") or 0 if abs(cur_d - bl_d) < 1.0: unchanged += 1 elif cur_d > bl_d: slower += 1 else: faster += 1 skipped = sum(1 for j in jobs_all if is_skipped(j)) bl_skipped = sum(1 for j in bl_jobs_all if is_skipped(j)) total_wait = sum(j.get("wait_s") or 0 for j in jobs) bl_total_wait = sum(j.get("wait_s") or 0 for j in bl_jobs) return { "wall": wall, "compute": compute, "bl_wall": bl_wall, "bl_compute": bl_compute, "faster": faster, "slower": slower, "unchanged": unchanged, "no_baseline": no_baseline, "skipped": skipped, "bl_skipped": bl_skipped, "total_wait": total_wait, "bl_total_wait": bl_total_wait, "total_jobs": len(jobs_all), } # --------------------------------------------------------------------------- # Resource profile loading + bottleneck analysis # --------------------------------------------------------------------------- def load_resource_profiles(directory: str) -> dict[str, dict]: """Load all resource-profile-*/resource-profile.json artifacts. Returns {label: profile_dict}. Labels are derived from the artifact directory name (resource-profile-