From 0610a6539d4da26100fd471ba125a85c26b98807 Mon Sep 17 00:00:00 2001 From: Liangsheng Yin Date: Thu, 3 Sep 2026 15:46:33 -0700 Subject: [PATCH] [CI] Add Lark notifications for CUDA CI status, runner health, and queue time (#37881) --- .github/workflows/ci-lark-notify.yml | 125 ++++++ scripts/ci_monitor/README.md | 8 +- scripts/ci_monitor/lark_notify.py | 635 +++++++++++++++++++++++++++ 3 files changed, 767 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/ci-lark-notify.yml create mode 100755 scripts/ci_monitor/lark_notify.py diff --git a/.github/workflows/ci-lark-notify.yml b/.github/workflows/ci-lark-notify.yml new file mode 100644 index 000000000..ea38feda0 --- /dev/null +++ b/.github/workflows/ci-lark-notify.yml @@ -0,0 +1,125 @@ +name: CI Lark Notify + +# Lark cards for CUDA CI health; see scripts/ci_monitor/README.md. + +on: + workflow_run: + workflows: ["Nightly Test (Nvidia)", "Weekly Test (Nvidia)", "PR Test Base"] + types: [completed] + branches: [main] + schedule: + - cron: '*/15 * * * *' # runner-health + - cron: '30 */6 * * *' # queue-digest + workflow_dispatch: + inputs: + task: + description: 'Which report to run' + required: true + type: choice + options: [ci-status, runner-health, queue-digest] + run_id: + description: 'Workflow run id (ci-status only)' + required: false + type: string + dry_run: + description: 'Print the card instead of posting it' + required: false + type: boolean + default: false + +permissions: + contents: read + actions: read + +env: + PYTHONUNBUFFERED: 1 + DRY_RUN_FLAG: ${{ inputs.dry_run && '--dry-run' || '' }} + +jobs: + ci-status: + if: >- + github.repository == 'sgl-project/sglang' && ( + (github.event_name == 'workflow_run' && github.event.workflow_run.event == 'schedule') || + (github.event_name == 'workflow_dispatch' && inputs.task == 'ci-status') + ) + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + with: + sparse-checkout: scripts/ci_monitor + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - name: Post CI status card + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + LARK_WEBHOOK: ${{ secrets.LARK_WEBHOOK }} + RUN_ID: ${{ github.event.workflow_run.id || inputs.run_id }} + run: | + python scripts/ci_monitor/lark_notify.py $DRY_RUN_FLAG ci-status \ + --run-id "$RUN_ID" \ + ${{ github.event_name == 'workflow_dispatch' && '--any-event' || '' }} + + runner-health: + if: >- + github.repository == 'sgl-project/sglang' && ( + (github.event_name == 'schedule' && github.event.schedule == '*/15 * * * *') || + (github.event_name == 'workflow_dispatch' && inputs.task == 'runner-health') + ) + runs-on: ubuntu-latest + timeout-minutes: 15 + # Serialize so the state file is never written by two runs at once. + concurrency: + group: ci-lark-notify-runner-health + cancel-in-progress: false + steps: + - uses: actions/checkout@v4 + with: + sparse-checkout: scripts/ci_monitor + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + # Key never matches (run_id is unique) so restore-keys picks the most + # recent state and the save step writes a fresh entry every run. + - name: Restore runner-health state + uses: actions/cache/restore@v4 + with: + path: runner-health-state.json + key: lark-runner-health-${{ github.run_id }} + restore-keys: lark-runner-health- + - name: Check runner pools + env: + GITHUB_TOKEN: ${{ secrets.GH_PAT_FOR_RUNNER_ADMIN }} + LARK_WEBHOOK: ${{ secrets.LARK_WEBHOOK }} + run: | + python scripts/ci_monitor/lark_notify.py $DRY_RUN_FLAG runner-health \ + --state-file runner-health-state.json + - name: Save runner-health state + if: ${{ !inputs.dry_run }} + uses: actions/cache/save@v4 + with: + path: runner-health-state.json + key: lark-runner-health-${{ github.run_id }} + + queue-digest: + if: >- + github.repository == 'sgl-project/sglang' && ( + (github.event_name == 'schedule' && github.event.schedule == '30 */6 * * *') || + (github.event_name == 'workflow_dispatch' && inputs.task == 'queue-digest') + ) + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + with: + sparse-checkout: scripts/ci_monitor + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - name: Post queue digest + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + LARK_WEBHOOK: ${{ secrets.LARK_WEBHOOK }} + run: | + python scripts/ci_monitor/lark_notify.py $DRY_RUN_FLAG queue-digest --hours 6 diff --git a/scripts/ci_monitor/README.md b/scripts/ci_monitor/README.md index 22be6fb48..eb1ef1733 100644 --- a/scripts/ci_monitor/README.md +++ b/scripts/ci_monitor/README.md @@ -1,10 +1,16 @@ # SGLang CI failure monitoring -Scripts used by [.github/workflows/ci-failure-monitor.yml](../../.github/workflows/ci-failure-monitor.yml): scheduled failure analysis. +Scripts used by [.github/workflows/ci-failure-monitor.yml](../../.github/workflows/ci-failure-monitor.yml) (scheduled failure analysis) and [.github/workflows/ci-lark-notify.yml](../../.github/workflows/ci-lark-notify.yml) (Lark notifications). ## Tools 1. **Failures Analyzer** (`ci_failures_analysis.py`): Tracks consecutive failures, identifies flaky jobs, and monitors runner health across PR Test / Nightly workflows (Nvidia, AMD, Intel, XPU, NPU). +2. **Lark Notifier** (`lark_notify.py`): Posts CUDA CI health cards to a Lark group through an incoming webhook (`LARK_WEBHOOK` secret). Stdlib only. Three subcommands: + - `ci-status --run-id N`: one card per finished scheduled run of the Nvidia nightly / weekly / scheduled pr-test. The first attempt lists its failed jobs; a rerun (attempt N > 1) is compared with attempt N-1 of the same run (fixed by rerun / still failing). Triggered by `workflow_run`. + - `runner-health --state-file F`: per-pool online / offline counts for the primary CUDA labels (`N-gpu-h100|h200|h20|5090|b200|b300|gb200|gb300|a10`). Posts only on degraded / recovered transitions plus an hourly reminder while degraded; state is carried between runs via `actions/cache`. Needs an admin PAT to list runners. + - `queue-digest --hours 6`: per-pool queue time p50 / p90 / max over the window, plus currently queued jobs. + + All subcommands accept `--dry-run` to print the card JSON instead of posting. ## Installation diff --git a/scripts/ci_monitor/lark_notify.py b/scripts/ci_monitor/lark_notify.py new file mode 100755 index 000000000..596f9bc3a --- /dev/null +++ b/scripts/ci_monitor/lark_notify.py @@ -0,0 +1,635 @@ +#!/usr/bin/env python3 +""" +Post CUDA CI health cards to a Lark group via an incoming webhook. + +Used by .github/workflows/ci-lark-notify.yml; see scripts/ci_monitor/README.md. +Needs GITHUB_TOKEN (an admin PAT for runner-health) and LARK_WEBHOOK, or +--dry-run to print the card JSON instead of posting. +""" + +import argparse +import json +import os +import re +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +DEFAULT_REPO = "sgl-project/sglang" +GITHUB_API = "https://api.github.com" + +# Primary pool labels only; aliases (1-gpu-runner, 8-gpu-h200-deepep, ...) are +# excluded so every runner is counted under exactly one label. +CUDA_LABEL_RE = re.compile(r"^\d+-gpu-(h100|h200|h20|5090|b200|b300|gb200|gb300|a10)$") + +# Workflows whose jobs run on the CUDA pools; used for queue-digest. +CUDA_WORKFLOW_FILES = [ + "pr-test.yml", + "pr-test-extra.yml", + "nightly-test-nvidia.yml", + "weekly-test-nvidia.yml", +] + +FAILED_CONCLUSIONS = {"failure", "timed_out", "startup_failure", "action_required"} +# Aggregator jobs fail whenever any other job fails; listing them is noise. +AGGREGATOR_JOB_RE = re.compile(r"^(check-all-jobs|pr-test-finish)$") +MAX_LISTED_JOBS = 15 + + +# -------------------------------------------------------------------------- +# GitHub API +# -------------------------------------------------------------------------- + + +class GitHub: + def __init__(self, token: str, repo: str): + self.token = token + self.repo = repo + + def get(self, path: str, params: Optional[dict] = None, retries: int = 5) -> Any: + url = f"{GITHUB_API}/{path}" + if params: + url += "?" + urllib.parse.urlencode(params) + req = urllib.request.Request(url) + req.add_header("Authorization", f"Bearer {self.token}") + req.add_header("Accept", "application/vnd.github+json") + req.add_header("X-GitHub-Api-Version", "2022-11-28") + for attempt in range(retries): + try: + with urllib.request.urlopen(req, timeout=60) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + transient = e.code in (429, 502, 503, 504) or ( + e.code == 403 and "rate limit" in body.lower() + ) + if not transient or attempt == retries - 1: + raise RuntimeError(f"GET {url} -> {e.code}: {body[:300]}") from e + except urllib.error.URLError as e: + if attempt == retries - 1: + raise RuntimeError(f"GET {url} failed: {e}") from e + time.sleep(2**attempt) + raise RuntimeError("unreachable") + + def paginate( + self, path: str, key: str, params: Optional[dict] = None, max_pages: int = 30 + ) -> list: + params = dict(params or {}) + params.setdefault("per_page", 100) + items: list = [] + for page in range(1, max_pages + 1): + params["page"] = page + data = self.get(path, params) + chunk = data.get(key, []) + items.extend(chunk) + if len(chunk) < params["per_page"]: + break + return items + + def run(self, run_id: int) -> dict: + return self.get(f"repos/{self.repo}/actions/runs/{run_id}") + + def run_jobs(self, run_id: int) -> list: + return self.paginate( + f"repos/{self.repo}/actions/runs/{run_id}/jobs", + "jobs", + # latest attempt per job; jobs not rerun keep their earlier result + {"filter": "latest"}, + ) + + def run_attempt_jobs(self, run_id: int, attempt: int) -> list: + return self.paginate( + f"repos/{self.repo}/actions/runs/{run_id}/attempts/{attempt}/jobs", "jobs" + ) + + def workflow_runs( + self, workflow_id: Any, params: dict, max_pages: int = 30 + ) -> list: + return self.paginate( + f"repos/{self.repo}/actions/workflows/{workflow_id}/runs", + "workflow_runs", + params, + max_pages=max_pages, + ) + + def runners(self) -> list: + return self.paginate(f"repos/{self.repo}/actions/runners", "runners") + + +# -------------------------------------------------------------------------- +# Lark +# -------------------------------------------------------------------------- + + +def build_card(title: str, color: str, body_md: str, buttons: list) -> dict: + elements: list = [{"tag": "div", "text": {"tag": "lark_md", "content": body_md}}] + if buttons: + elements.append( + { + "tag": "action", + "actions": [ + { + "tag": "button", + "text": {"tag": "plain_text", "content": text}, + "url": url, + "type": "default", + } + for text, url in buttons + ], + } + ) + return { + "msg_type": "interactive", + "card": { + "config": {"wide_screen_mode": True}, + "header": { + "title": {"tag": "plain_text", "content": title}, + "template": color, + }, + "elements": elements, + }, + } + + +def post_card(card: dict, webhook: str, dry_run: bool) -> None: + if dry_run: + print(json.dumps(card, indent=2)) + return + req = urllib.request.Request( + webhook, + data=json.dumps(card).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=30) as resp: + body = json.loads(resp.read().decode("utf-8")) + if body.get("code", body.get("StatusCode")) not in (0, None): + raise RuntimeError(f"Lark webhook rejected message: {body}") + print(f"posted: {card['card']['header']['title']['content']}") + + +# -------------------------------------------------------------------------- +# helpers +# -------------------------------------------------------------------------- + + +def parse_time(s: Optional[str]) -> Optional[datetime]: + if not s: + return None + return datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + + +def fmt_duration(seconds: Optional[float]) -> str: + if seconds is None: + return "-" + seconds = int(seconds) + if seconds < 60: + return f"{seconds}s" + if seconds < 3600: + return f"{seconds // 60}m{seconds % 60:02d}s" + return f"{seconds // 3600}h{(seconds % 3600) // 60:02d}m" + + +def percentile(values: list, p: float) -> Optional[float]: + if not values: + return None + ordered = sorted(values) + idx = int(round((len(ordered) - 1) * p)) + return ordered[idx] + + +def primary_cuda_label(labels: list) -> Optional[str]: + for label in labels: + if CUDA_LABEL_RE.match(label): + return label + return None + + +def list_jobs_md(jobs: list, limit: int = MAX_LISTED_JOBS) -> str: + lines = [f" - [{j['name']}]({j['html_url']})" for j in jobs[:limit]] + if len(jobs) > limit: + lines.append(f" - ... and {len(jobs) - limit} more") + return "\n".join(lines) + + +def compress_runner_names(names: list) -> str: + groups: dict = {} + for n in sorted(names): + m = re.match(r"^(.*?)(\d+)$", n) + if m: + groups.setdefault(m.group(1), []).append(int(m.group(2))) + else: + groups.setdefault(n, []) + parts = [] + for prefix, nums in groups.items(): + if not nums: + parts.append(prefix) + elif len(nums) == 1: + parts.append(f"{prefix}{nums[0]}") + else: + parts.append(prefix + "{" + ",".join(str(x) for x in sorted(nums)) + "}") + return ", ".join(parts) + + +# -------------------------------------------------------------------------- +# ci-status +# -------------------------------------------------------------------------- + + +def is_reportable_job(job: dict) -> bool: + return job.get("conclusion") not in ( + None, + "skipped", + ) and not AGGREGATOR_JOB_RE.match(job["name"]) + + +def failed_job_names(jobs: list) -> dict: + return { + j["name"]: j + for j in jobs + if is_reportable_job(j) and j.get("conclusion") in FAILED_CONCLUSIONS + } + + +def diff_attempts(current: dict, previous: dict) -> dict: + return { + "fixed": [j for n, j in previous.items() if n not in current], + "still": [j for n, j in current.items() if n in previous], + "new": [j for n, j in current.items() if n not in previous], + } + + +def render_ci_status(run: dict, jobs: list, prev_failed: Optional[dict]) -> dict: + name = run["name"] + attempt = run.get("run_attempt", 1) + conclusion = run.get("conclusion") or "unknown" + counted = [j for j in jobs if is_reportable_job(j)] + failed = failed_job_names(jobs) + cancelled = [j for j in jobs if j.get("conclusion") == "cancelled"] + started = parse_time(run.get("run_started_at")) + updated = parse_time(run.get("updated_at")) + duration = ( + fmt_duration((updated - started).total_seconds()) + if started and updated + else "-" + ) + + sha = run["head_sha"][:9] + commit_msg = (run.get("head_commit") or {}).get("message", "").splitlines() + commit_line = f"`{sha}` {commit_msg[0] if commit_msg else ''}".strip() + rerun_prefix = f"Rerun #{attempt} - " if attempt > 1 else "" + + if conclusion == "cancelled": + title = f"{rerun_prefix}{name}: CANCELLED ({duration})" + color = "grey" + elif failed: + title = f"{rerun_prefix}{name}: FAILED ({len(failed)} failed / {len(counted)} jobs, {duration})" + color = "red" + else: + title = f"{rerun_prefix}{name}: PASSED ({len(counted)} jobs, {duration})" + color = "green" + + lines = [commit_line] + # None: first attempt, nothing to compare against + if prev_failed is None: + if failed: + lines.append(f"**Failed jobs ({len(failed)})**") + lines.append(list_jobs_md(list(failed.values()))) + else: + diff = diff_attempts(failed, prev_failed) + if diff["fixed"]: + lines.append(f"**Fixed by rerun ({len(diff['fixed'])})**") + lines.append(list_jobs_md(diff["fixed"])) + if diff["still"]: + lines.append(f"**Still failing ({len(diff['still'])})**") + lines.append(list_jobs_md(diff["still"])) + if diff["new"]: + lines.append(f"**New failures ({len(diff['new'])})**") + lines.append(list_jobs_md(diff["new"])) + if cancelled: + lines.append(f"Cancelled jobs: {len(cancelled)}") + + buttons = [("Run", run["html_url"])] + if attempt > 1: + buttons.append( + (f"Attempt {attempt - 1}", f"{run['html_url']}/attempts/{attempt - 1}") + ) + return build_card(title, color, "\n".join(lines), buttons) + + +def cmd_ci_status(args: argparse.Namespace, gh: GitHub) -> None: + run = gh.run(args.run_id) + if run["event"] != "schedule" and not args.any_event: + print( + f"run {args.run_id} event={run['event']} is not a scheduled run; skipping" + ) + return + if run.get("status") != "completed": + print( + f"run {args.run_id} status={run.get('status')} is not completed; skipping" + ) + return + jobs = gh.run_jobs(run["id"]) + attempt = run.get("run_attempt", 1) + prev_failed = None + if attempt > 1: + prev_failed = failed_job_names(gh.run_attempt_jobs(run["id"], attempt - 1)) + post_card(render_ci_status(run, jobs, prev_failed), args.webhook, args.dry_run) + + +# -------------------------------------------------------------------------- +# runner-health +# -------------------------------------------------------------------------- + + +def summarize_pools(runners: list) -> dict: + pools: dict = {} + for r in runners: + label = primary_cuda_label([l["name"] for l in r.get("labels", [])]) + if label is None: + continue + pool = pools.setdefault( + label, + {"total": 0, "online": 0, "offline": 0, "busy": 0, "offline_names": []}, + ) + pool["total"] += 1 + if r.get("status") == "online": + pool["online"] += 1 + if r.get("busy"): + pool["busy"] += 1 + else: + pool["offline"] += 1 + pool["offline_names"].append(r["name"]) + return pools + + +def is_degraded(pool: dict, threshold: float) -> bool: + if pool["total"] < 2: + return pool["offline"] == pool["total"] and pool["total"] > 0 + return pool["offline"] / pool["total"] >= threshold + + +def plan_health_events( + pools: dict, state: dict, now: datetime, threshold: float, remind_hours: float +) -> tuple: + events = [] # (kind, label, pool, since) + new_state: dict = {} + for label, pool in sorted(pools.items()): + prev = state.get(label) + degraded = is_degraded(pool, threshold) + if degraded: + since = parse_time(prev["degraded_since"]) if prev else now + last = parse_time(prev["last_notified"]) if prev else None + if prev is None: + events.append(("degraded", label, pool, since)) + last = now + elif last is None or (now - last) >= timedelta(hours=remind_hours): + events.append(("still_degraded", label, pool, since)) + last = now + new_state[label] = { + "degraded_since": since.strftime("%Y-%m-%dT%H:%M:%SZ"), + "last_notified": last.strftime("%Y-%m-%dT%H:%M:%SZ"), + "offline": pool["offline"], + } + elif prev is not None: + events.append( + ("recovered", label, pool, parse_time(prev["degraded_since"])) + ) + return events, new_state + + +def render_health_event( + kind: str, label: str, pool: dict, since: datetime, now: datetime, repo: str +) -> dict: + counts = ( + f"online {pool['online']} / offline {pool['offline']} / total {pool['total']}, " + f"busy {pool['busy']}" + ) + runners_url = f"https://github.com/{repo}/settings/actions/runners" + since_s = since.strftime("%Y-%m-%d %H:%M UTC") + elapsed = fmt_duration((now - since).total_seconds()) + if kind == "recovered": + title = f"Runner pool recovered: {label}" + body = f"{counts}\nDegraded for {elapsed} (since {since_s})." + return build_card(title, "green", body, [("Runners", runners_url)]) + all_down = pool["offline"] == pool["total"] + state_word = "DOWN" if all_down else "degraded" + if kind == "degraded": + title = f"Runner pool {state_word}: {label}" + else: + title = f"Runner pool still {state_word}: {label} ({elapsed})" + body = "\n".join( + [ + counts, + f"Offline: {compress_runner_names(pool['offline_names'])}", + f"Since {since_s}", + ] + ) + return build_card( + title, "red" if all_down else "orange", body, [("Runners", runners_url)] + ) + + +def cmd_runner_health(args: argparse.Namespace, gh: GitHub) -> None: + now = datetime.now(timezone.utc).replace(microsecond=0) + state = {} + if os.path.exists(args.state_file): + with open(args.state_file) as f: + state = json.load(f) + pools = summarize_pools(gh.runners()) + for label, pool in sorted(pools.items()): + print( + f"{label}: online {pool['online']} offline {pool['offline']} busy {pool['busy']}" + ) + events, new_state = plan_health_events( + pools, state, now, args.threshold, args.remind_hours + ) + for kind, label, pool, since in events: + post_card( + render_health_event(kind, label, pool, since, now, gh.repo), + args.webhook, + args.dry_run, + ) + if not events: + print("no runner-health transitions") + if not args.dry_run: + with open(args.state_file, "w") as f: + json.dump(new_state, f, indent=2) + + +# -------------------------------------------------------------------------- +# queue-digest +# -------------------------------------------------------------------------- + + +def job_queue_seconds(job: dict, now: datetime) -> Optional[float]: + created = parse_time(job.get("created_at")) + if created is None: + return None + if job.get("status") == "queued": + return (now - created).total_seconds() + started = parse_time(job.get("started_at")) + if started is None or started < created: + return None + return (started - created).total_seconds() + + +def summarize_queue(jobs: list, now: datetime) -> dict: + per_label: dict = {} + for job in jobs: + label = primary_cuda_label(job.get("labels") or []) + if label is None: + continue + q = job_queue_seconds(job, now) + if q is None: + continue + entry = per_label.setdefault( + label, {"waits": [], "queued_now": [], "started": 0} + ) + if job.get("status") == "queued": + entry["queued_now"].append(q) + else: + entry["waits"].append(q) + entry["started"] += 1 + result = {} + for label, e in per_label.items(): + result[label] = { + "n": e["started"], + "p50": percentile(e["waits"], 0.5), + "p90": percentile(e["waits"], 0.9), + "max": max(e["waits"]) if e["waits"] else None, + "queued_now": len(e["queued_now"]), + "oldest_queued": max(e["queued_now"]) if e["queued_now"] else None, + } + return result + + +def render_queue_digest( + stats: dict, hours: float, slow_minutes: float, repo: str +) -> dict: + slow = slow_minutes * 60 + ordered = sorted(stats.items(), key=lambda kv: -(kv[1]["p90"] or 0)) + lines = [] + for label, s in ordered: + flag = " (!)" if (s["p90"] or 0) >= slow else "" + queued = ( + f", queued now {s['queued_now']} (oldest {fmt_duration(s['oldest_queued'])})" + if s["queued_now"] + else "" + ) + lines.append( + f"**{label}**{flag}: {s['n']} jobs, p50 {fmt_duration(s['p50'])}, " + f"p90 {fmt_duration(s['p90'])}, max {fmt_duration(s['max'])}{queued}" + ) + if not lines: + lines.append("_No CUDA jobs in this window._") + any_slow = any((s["p90"] or 0) >= slow for s in stats.values()) + title = f"CUDA queue time, last {int(hours)}h" + if any_slow: + title += f" - p90 over {int(slow_minutes)}m on some pools" + return build_card( + title, + "orange" if any_slow else "blue", + "\n".join(lines), + [("Actions", f"https://github.com/{repo}/actions")], + ) + + +def fetch_window_jobs( + gh: GitHub, hours: float, workflow_files: list, workers: int +) -> list: + since = datetime.now(timezone.utc) - timedelta(hours=hours) + runs: list = [] + for wf in workflow_files: + runs.extend( + gh.workflow_runs( + wf, + {"created": ">=" + since.strftime("%Y-%m-%dT%H:%M:%SZ")}, + max_pages=10, + ) + ) + print(f"{len(runs)} runs in window across {len(workflow_files)} workflows") + with ThreadPoolExecutor(max_workers=workers) as pool: + job_lists = list(pool.map(lambda r: gh.run_jobs(r["id"]), runs)) + jobs = [j for jl in job_lists for j in jl] + print(f"{len(jobs)} jobs fetched") + return jobs + + +def cmd_queue_digest(args: argparse.Namespace, gh: GitHub) -> None: + now = datetime.now(timezone.utc) + jobs = fetch_window_jobs(gh, args.hours, args.workflows.split(","), args.workers) + stats = summarize_queue(jobs, now) + post_card( + render_queue_digest(stats, args.hours, args.slow_minutes, gh.repo), + args.webhook, + args.dry_run, + ) + + +# -------------------------------------------------------------------------- +# main +# -------------------------------------------------------------------------- + + +def main() -> int: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--repo", default=DEFAULT_REPO) + parser.add_argument("--token", default=os.environ.get("GITHUB_TOKEN")) + parser.add_argument("--webhook", default=os.environ.get("LARK_WEBHOOK")) + parser.add_argument( + "--dry-run", action="store_true", help="print card JSON instead of posting" + ) + sub = parser.add_subparsers(dest="command", required=True) + + p = sub.add_parser("ci-status", help="summarize a finished scheduled run") + p.add_argument("--run-id", type=int, required=True) + p.add_argument( + "--any-event", action="store_true", help="also report non-schedule runs" + ) + + p = sub.add_parser("runner-health", help="per-label online/offline transitions") + p.add_argument("--state-file", required=True) + p.add_argument( + "--threshold", + type=float, + default=0.5, + help="offline ratio that counts as degraded", + ) + p.add_argument("--remind-hours", type=float, default=1.0) + + p = sub.add_parser("queue-digest", help="per-label queue time percentiles") + p.add_argument("--hours", type=float, default=6.0) + p.add_argument( + "--slow-minutes", type=float, default=30.0, help="p90 above this is flagged" + ) + p.add_argument("--workflows", default=",".join(CUDA_WORKFLOW_FILES)) + p.add_argument("--workers", type=int, default=8) + + args = parser.parse_args() + if not args.token: + print("GITHUB_TOKEN (or --token) is required", file=sys.stderr) + return 2 + if not args.webhook and not args.dry_run: + print( + "LARK_WEBHOOK (or --webhook) is required unless --dry-run", file=sys.stderr + ) + return 2 + + gh = GitHub(args.token, args.repo) + { + "ci-status": cmd_ci_status, + "runner-health": cmd_runner_health, + "queue-digest": cmd_queue_digest, + }[args.command](args, gh) + return 0 + + +if __name__ == "__main__": + sys.exit(main())