diff --git a/.github/workflows/ci-lark-notify.yml b/.github/workflows/ci-lark-notify.yml index 20ddca2e9..c520ae765 100644 --- a/.github/workflows/ci-lark-notify.yml +++ b/.github/workflows/ci-lark-notify.yml @@ -9,14 +9,13 @@ on: branches: [main] schedule: - cron: '*/15 * * * *' # runner-health - - cron: '30 */8 * * *' # queue-digest workflow_dispatch: inputs: task: description: 'Which report to run' required: true type: choice - options: [ci-status, runner-health, queue-digest] + options: [ci-status, runner-health] run_id: description: 'Workflow run id (ci-status only)' required: false @@ -94,33 +93,11 @@ jobs: LARK_WEBHOOK: ${{ secrets.LARK_WEBHOOK }} run: | python scripts/ci_monitor/lark_notify.py $DRY_RUN_FLAG runner-health \ - --state-file runner-health-state.json + --state-file runner-health-state.json \ + --remind-hours 6 - 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 */8 * * *') || - (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: - # PAT: an 8h window is ~400-500 API calls, too close to GITHUB_TOKEN's 1000/h - GITHUB_TOKEN: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} - LARK_WEBHOOK: ${{ secrets.LARK_WEBHOOK }} - run: | - python scripts/ci_monitor/lark_notify.py $DRY_RUN_FLAG queue-digest --hours 8 --only-if-slow diff --git a/.github/workflows/runner-utilization.yml b/.github/workflows/runner-utilization.yml index d126ea40b..15147e33a 100644 --- a/.github/workflows/runner-utilization.yml +++ b/.github/workflows/runner-utilization.yml @@ -68,4 +68,18 @@ jobs: python scripts/ci/utils/runner_utilization_report.py \ --repo ${{ github.repository }} \ --hours ${{ (github.event_name == 'pull_request' && '0.34') || inputs.hours || '24' }} \ - ${{ inputs.filter && format('--filter {0}', inputs.filter) || '' }} + ${{ inputs.filter && format('--filter {0}', inputs.filter) || '' }} \ + --queue-series-out queue-series.json + + # The card rides on the scan above rather than running its own: a 24h + # window is ~5000 API calls, the whole hourly budget of ci-lark-notify's PAT. + - name: Post queue timeline card + if: >- + github.repository == 'sgl-project/sglang' && + github.event_name != 'pull_request' + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + LARK_WEBHOOK: ${{ secrets.LARK_WEBHOOK }} + run: | + python scripts/ci_monitor/lark_notify.py queue-timeline \ + --series-file queue-series.json diff --git a/scripts/ci/utils/runner_utilization_report.py b/scripts/ci/utils/runner_utilization_report.py index 45dfe7bd1..85bd8cc23 100755 --- a/scripts/ci/utils/runner_utilization_report.py +++ b/scripts/ci/utils/runner_utilization_report.py @@ -463,6 +463,85 @@ _NON_GPU_WORKFLOW_HINTS = ( ) +def percentile(values: list[float], p: float) -> float | None: + if not values: + return None + ordered = sorted(values) + return ordered[int(round((len(ordered) - 1) * p))] + + +def queue_timeline( + jobs: list[dict], + window_start: datetime, + window_end: datetime, + bucket_minutes: int = 60, +) -> list[dict]: + """Per-bucket queue backlog and wait p90, for the Lark timeline card. + + `backlog` is the sweep-line counter calculate_concurrency_metrics reduces + to a single peak, kept per bucket; `p90_wait_min` covers only the jobs + picked up inside the bucket, so a bucket can show a deep backlog and no + p90 at all. + """ + # Buckets start on the hour so the x-axis labels mean what they say. The + # partly-covered hour window_start lands in is dropped rather than reported + # short; on a 24h window it would also repeat the last bucket's label. + bucket = timedelta(minutes=bucket_minutes) + origin = window_start.replace(minute=0, second=0, microsecond=0) + if origin < window_start: + origin += bucket + starts = [] + t = origin + while t < window_end: + starts.append(t) + t += bucket + if not starts: + return [] + peaks = [0] * len(starts) + + def bucket_index(when: datetime) -> int: + """Bucket `when` falls in. Callers drop anything before origin.""" + offset = (when - origin).total_seconds() // (bucket_minutes * 60) + return min(len(starts) - 1, int(offset)) + + events = [] + for job in jobs: + created_at, queue_end = job.get("created_at"), job.get("queue_end") + if not created_at or not queue_end or created_at >= queue_end: + continue + if queue_end < window_start or created_at > window_end: + continue + events.append((max(created_at, window_start), 1)) + events.append((min(queue_end, window_end), -1)) + # Level between two events applies to every bucket the gap spans, so a + # backlog that persists without any event still shows in later buckets. + events.sort(key=lambda e: (e[0], e[1] == 1)) + current, prev_time = 0, window_start + for event_time, delta in events + [(window_end, 0)]: + if current > 0 and event_time > prev_time and event_time >= origin: + first = bucket_index(max(prev_time, origin)) + for i in range(first, bucket_index(event_time) + 1): + peaks[i] = max(peaks[i], current) + current += delta + prev_time = event_time + + waits: list[list[float]] = [[] for _ in starts] + for job in jobs: + start = job.get("start") + if start is None or start < origin or start > window_end: + continue + waits[bucket_index(start)].append(job["queue_time"]) + return [ + { + "start": starts[i].strftime("%Y-%m-%dT%H:%M:%SZ"), + "backlog": peaks[i], + "started": len(waits[i]), + "p90_wait_min": (percentile(waits[i], 0.9) or 0.0) / 60, + } + for i in range(len(starts)) + ] + + 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 is the bottleneck on @@ -781,6 +860,7 @@ def calculate_utilization( "saturation_pct": conc["saturation_pct"], "peak_queue": conc["peak_queue"], "status_counts": status_counts, + "queue_timeline": queue_timeline(jobs, window_start, window_end), } ) @@ -957,6 +1037,11 @@ def main(): "--filter", type=str, help="Filter runner labels (e.g., '5090', 'h200')" ) parser.add_argument("--output", type=str, help="Output file (default: stdout)") + parser.add_argument( + "--queue-series-out", + type=str, + help="Write the per-label hourly queue series as JSON (for the Lark card)", + ) args = parser.parse_args() results, fetch_failure_pct, longest_waits, coverage_hours = calculate_utilization( @@ -970,6 +1055,17 @@ def main(): coverage_hours=coverage_hours, ) + if args.queue_series_out: + # Every bucket carries its own start, so the window needs no separate + # field -- and cannot drift from the one calculate_utilization used. + series = { + "bucket_minutes": 60, + "labels": {r["label"]: r["queue_timeline"] for r in results}, + } + with open(args.queue_series_out, "w") as f: + json.dump(series, f) + print(f"Queue series written to {args.queue_series_out}") + if args.output: with open(args.output, "w") as f: f.write(report) diff --git a/scripts/ci_monitor/README.md b/scripts/ci_monitor/README.md index c4c4a0477..ec24056b9 100644 --- a/scripts/ci_monitor/README.md +++ b/scripts/ci_monitor/README.md @@ -7,8 +7,8 @@ Scripts used by [.github/workflows/ci-failure-monitor.yml](../../.github/workflo 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 8 --only-if-slow`: per-pool queue time p50 / p90 / max over the window plus currently queued jobs, posted only when some pool's p90 exceeds `--slow-minutes` (default 30). Links to the latest Runner Utilization Report 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`). Posts only on degraded / recovered transitions plus a reminder every `--remind-hours` (6h in CI) while degraded; state is carried between runs via `actions/cache`. Needs an admin PAT to list runners. + - `queue-timeline --series-file F`: one card a day charting the hourly queue backlog and wait p90 across the CUDA pools, naming the pool behind each peak. The series comes from `runner_utilization_report.py --queue-series-out`, so this runs inside runner-utilization.yml rather than here -- a 24h scan of its own would cost a PAT's whole hourly API budget. All subcommands accept `--dry-run` to print the card JSON instead of posting. diff --git a/scripts/ci_monitor/lark_notify.py b/scripts/ci_monitor/lark_notify.py index 194aecc7f..5eba30223 100755 --- a/scripts/ci_monitor/lark_notify.py +++ b/scripts/ci_monitor/lark_notify.py @@ -27,20 +27,17 @@ GITHUB_API = "https://api.github.com" UTILIZATION_WORKFLOW = "runner-utilization.yml" # 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", -] +# excluded so every runner is counted under exactly one label. a10 is left out +# as well: it serves no per-commit test, so its transitions are noise. +CUDA_LABEL_RE = re.compile(r"^\d+-gpu-(h100|h200|h20|5090|b200|b300|gb200|gb300)$") 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)$") +# Jobs from a called workflow arrive prefixed ("call-pr-test-extra / "), +# so the aggregator name is matched on the last segment. +AGGREGATOR_JOB_RE = re.compile( + r"^(?:.+ / )?(check-all-jobs|pr-test-finish|pr-test-extra-finish)$" +) MAX_LISTED_JOBS = 15 @@ -162,25 +159,6 @@ def kv_columns(pairs: list) -> dict: } -def table(columns: list, rows: list, page_size: int = 12) -> dict: - # columns: (key, display_name, data_type); rows: {key: value} - return { - "tag": "table", - "page_size": page_size, - "row_height": "low", - "header_style": { - "text_align": "left", - "bold": True, - "background_style": "grey", - }, - "columns": [ - {"name": k, "display_name": name, "data_type": dtype, "width": "auto"} - for k, name, dtype in columns - ], - "rows": rows, - } - - def button(text: str, url: str) -> dict: return { "tag": "button", @@ -190,6 +168,17 @@ def button(text: str, url: str) -> dict: } +def chart(spec: dict, aspect_ratio: str = "16:9") -> dict: + # The spec is VChart JSON rendered by the Lark client, so no image upload + # (hence no Lark app credentials) is involved. Needs Lark client 7.1+. + return { + "tag": "chart", + "aspect_ratio": aspect_ratio, + "color_theme": "brand", + "chart_spec": spec, + } + + HR = {"tag": "hr"} @@ -241,6 +230,13 @@ def fmt_local(dt: Optional[datetime]) -> str: return dt.astimezone(LOCAL_TZ).strftime("%Y-%m-%d %I:%M %p %Z") +def fmt_local_hour(dt: Optional[datetime]) -> str: + """Timeline bucket label: the local hour alone, e.g. "9am".""" + if dt is None: + return "-" + return dt.astimezone(LOCAL_TZ).strftime("%I%p").lstrip("0").lower() + + def plural(n: int, word: str) -> str: return f"{n} {word}" if n == 1 else f"{n} {word}s" @@ -256,14 +252,6 @@ def fmt_duration(seconds: Optional[float]) -> str: 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 name in labels: if CUDA_LABEL_RE.match(name): @@ -550,134 +538,135 @@ def cmd_runner_health(args: argparse.Namespace, gh: GitHub) -> None: # -------------------------------------------------------------------------- -# queue-digest +# queue-timeline # -------------------------------------------------------------------------- -def job_queue_seconds(job: dict, now: datetime) -> Optional[float]: - created = parse_time(job.get("created_at")) - if created is None: - return None - # a still-queued job reports a placeholder started_at; measure against now - 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 merge_timeline(series: dict) -> list: + """Fold the per-label series into one CUDA-wide series, bucket by bucket. - -def summarize_queue(jobs: list, now: datetime) -> dict: - per_label: dict = {} - for job in jobs: - pool_label = primary_cuda_label(job.get("labels") or []) - if pool_label is None: + Backlog sums across pools; the wait takes the max of the per-pool p90s, + since averaging would let idle pools mask the one pool that is stuck. The + merge costs the answer to "which pool?", so each bucket keeps the label + behind the deepest backlog and the longest wait -- routinely not the same. + """ + buckets: dict = {} + for label, rows in series.get("labels", {}).items(): + if not CUDA_LABEL_RE.match(label): continue - q = job_queue_seconds(job, now) - if q is None: - continue - entry = per_label.setdefault(pool_label, {"waits": [], "queued_now": []}) - if job.get("status") == "queued": - entry["queued_now"].append(q) - else: - entry["waits"].append(q) - result = {} - for pool_label, e in per_label.items(): - result[pool_label] = { - "n": len(e["waits"]), - "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 + for row in rows: + b = buckets.setdefault( + row["start"], + { + "backlog": 0, + "started": 0, + "p90": 0.0, + "p90_pool": "-", + "top_pool": "-", + "top_backlog": 0, + }, + ) + b["backlog"] += row["backlog"] + b["started"] += row["started"] + if row["p90_wait_min"] > b["p90"]: + b["p90"], b["p90_pool"] = row["p90_wait_min"], label + if row["backlog"] > b["top_backlog"]: + b["top_backlog"], b["top_pool"] = row["backlog"], label + return [dict(start=k, **v) for k, v in sorted(buckets.items())] -def slow_pools(stats: dict, slow_minutes: float) -> set: - return {k for k, s in stats.items() if (s["p90"] or 0) >= slow_minutes * 60} - - -QUEUE_COLUMNS = [ - ("pool", "Pool", "lark_md"), - ("jobs", "Jobs", "text"), - ("p50", "p50", "text"), - ("p90", "p90", "text"), - ("max", "Max", "text"), - ("queued", "Queued now", "text"), - ("oldest", "Oldest wait", "text"), -] - - -def render_queue_digest( - stats: dict, hours: float, slow_minutes: float, now: datetime, report_url: str -) -> dict: - slow = slow_pools(stats, slow_minutes) - ordered = sorted(stats.items(), key=lambda kv: -(kv[1]["p90"] or 0)) - rows = [] - for pool_label, s in ordered: - is_slow = pool_label in slow - rows.append( +def timeline_chart_spec(rows: list) -> dict: + hours = [fmt_local_hour(parse_time(r["start"])) for r in rows] + return { + "type": "common", + "data": [ { - "pool": f"**{pool_label}** (!)" if is_slow else pool_label, - "jobs": str(s["n"]), - "p50": fmt_duration(s["p50"]), - "p90": fmt_duration(s["p90"]), - "max": fmt_duration(s["max"]), - "queued": str(s["queued_now"]) if s["queued_now"] else "-", - "oldest": fmt_duration(s["oldest_queued"]) if s["queued_now"] else "-", - } - ) - title = f"CUDA queue time, last {int(hours)}h" - if slow: - title += f" - p90 over {int(slow_minutes)}m on some pools" - window = f"{fmt_local(now - timedelta(hours=hours))} to {fmt_local(now)}" + "id": "backlog", + "values": [ + {"hour": h, "value": r["backlog"]} for h, r in zip(hours, rows) + ], + }, + { + "id": "wait", + "values": [ + {"hour": h, "value": round(r["p90"], 1)} + for h, r in zip(hours, rows) + ], + }, + ], + "series": [ + { + "type": "bar", + "id": "backlog", + "dataIndex": 0, + "xField": "hour", + "yField": "value", + "name": "Jobs waiting (peak)", + }, + { + "type": "line", + "id": "wait", + "dataIndex": 1, + "xField": "hour", + "yField": "value", + "name": "p90 wait (min)", + }, + ], + "axes": [ + {"orient": "left", "seriesIndex": [0], "title": {"visible": False}}, + {"orient": "right", "seriesId": ["wait"], "grid": {"visible": False}}, + {"orient": "bottom", "type": "band", "label": {"visible": True}}, + ], + "legends": {"visible": True, "orient": "bottom"}, + } + + +def render_queue_timeline(rows: list, report_url: str) -> dict: + peak = max(rows, key=lambda r: r["backlog"]) + slowest = max(rows, key=lambda r: r["p90"]) + span = f"{fmt_local(parse_time(rows[0]['start']))} to {fmt_local(parse_time(rows[-1]['start']))}" + peak_hour = fmt_local_hour(parse_time(peak["start"])) + slowest_hour = fmt_local_hour(parse_time(slowest["start"])) elements = [ - md(f"{grey('Window')} {window}\n{grey('(!)')} p90 over {int(slow_minutes)}m"), - table(QUEUE_COLUMNS, rows) if rows else md("_No CUDA jobs in this window._"), + md(f"{grey('Window')} {span} {grey('(bucket: 1h)')}"), + kv_columns( + [ + ("Jobs started", str(sum(r["started"] for r in rows))), + ("Peak backlog", f"{peak['backlog']} jobs"), + ("Worst p90 wait", fmt_duration(slowest["p90"] * 60)), + ] + ), + chart(timeline_chart_spec(rows)), + md( + f"{grey('Peak backlog')} {peak_hour}, mostly " + f"**{peak['top_pool']}** ({peak['top_backlog']})\n" + f"{grey('Worst wait')} {slowest_hour}, **{slowest['p90_pool']}**" + ), ] return build_card( - title, - "orange" if slow else "blue", + "CUDA queue over the day", + "blue", elements, [("View utilization report", report_url)], ) -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) - if args.only_if_slow and not slow_pools(stats, args.slow_minutes): - print(f"no pool with p90 over {int(args.slow_minutes)}m; skipping") +def cmd_queue_timeline(args: argparse.Namespace, gh: GitHub) -> None: + with open(args.series_file) as f: + series = json.load(f) + rows = merge_timeline(series) + if not rows: + print("no CUDA buckets in the series; skipping") return - report_url = gh.latest_run_url(UTILIZATION_WORKFLOW) - post_card( - render_queue_digest(stats, args.hours, args.slow_minutes, now, report_url), - args.webhook, - args.dry_run, + # The card is built from THIS run's scan, so link to it rather than to the + # last successful one, which would be yesterday's report. + run_id = os.environ.get("GITHUB_RUN_ID") + report_url = ( + f"https://github.com/{gh.repo}/actions/runs/{run_id}" + if run_id + else gh.latest_run_url(UTILIZATION_WORKFLOW) ) + post_card(render_queue_timeline(rows, report_url), args.webhook, args.dry_run) # -------------------------------------------------------------------------- @@ -713,18 +702,12 @@ def main() -> int: ) 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=8.0) + p = sub.add_parser("queue-timeline", help="daily queue backlog / wait chart") p.add_argument( - "--slow-minutes", type=float, default=30.0, help="p90 above this is flagged" + "--series-file", + required=True, + help="JSON written by runner_utilization_report.py --queue-series-out", ) - p.add_argument( - "--only-if-slow", - action="store_true", - help="post only when some pool's p90 exceeds --slow-minutes", - ) - 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: @@ -740,7 +723,7 @@ def main() -> int: { "ci-status": cmd_ci_status, "runner-health": cmd_runner_health, - "queue-digest": cmd_queue_digest, + "queue-timeline": cmd_queue_timeline, }[args.command](args, gh) return 0