Add runner-fleet-metrics workflow (registered runner counts -> ClickHouse) (#8403)

Track how many self-hosted runners are registered with the `pytorch` org
over time, bucketed by label (macOS + B200), for capacity trending in
Grafana.

Every 30 min the workflow counts `total`/`online`/`busy` per label,
uploads gzipped JSONEachRow to `s3://gha-artifacts/runner_fleet_count/`
(assuming the shared `arn:aws:iam::308535385114:role/arc` via OIDC —
same role `_linux-build.yml` uses), and `clickhouse-replicator-s3`
ingests it into `misc.runner_fleet_count`. No ClickHouse creds in CI.

**Changes:** table schema, collector script, cron workflow (protected
env), and `runner_fleet_count_adapter` +
`SUPPORTED_PATHS`/`OBJECT_CONVERTER` entries in the replicator (needs
redeploy).

**Requires:** companion AWS wiring meta-pytorch/pytorch-gha-infra#1364;
a `runner-fleet-metrics` Environment (restricted to `main`) with secret
`RUNNER_LIST_GH_TOKEN` (org self-hosted-runners:read).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: huydhn

.github/workflows/runner-fleet-metrics.yml

@@ -0,0 +1,79 @@
+name: Runner fleet metrics
+
+# Every 30 minutes, count the self-hosted runners registered with the pytorch
+# org (macOS + B200 by default), then upload the counts to S3. The
+# clickhouse-replicator-s3 lambda ingests them into misc.runner_fleet_count,
+# which is graphed in Grafana via the ClickHouse datasource.
+#
+# No ClickHouse credentials live here: the workflow only authenticates to AWS
+# (OIDC) to write the object; the replicator holds the CH creds.
+
+on:
+  schedule:
+    - cron: "*/30 * * * *"
+  workflow_dispatch:
+    inputs:
+      dry_run:
+        description: "Collect and print counts without uploading to S3"
+        type: boolean
+        default: false
+
+concurrency:
+  group: runner-fleet-metrics
+  cancel-in-progress: false
+
+permissions:
+  contents: read
+  id-token: write  # required for AWS OIDC role assumption
+
+jobs:
+  collect:
+    runs-on: ubuntu-latest
+    # Protected environment restricted to the main branch; holds the runner-list
+    # token.
+    environment: runner-fleet-metrics
+    env:
+      RUNNER_ORG: pytorch
+      S3_BUCKET: gha-artifacts
+      S3_PREFIX: runner_fleet_count
+      AWS_REGION: us-east-1
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - run: python -m pip install --no-cache-dir requests
+
+      - name: Collect runner counts
+        env:
+          # GitHub reserves the "GITHUB_" prefix for secret *names*, so the
+          # runner-list token is stored as RUNNER_LIST_GH_TOKEN and exposed to
+          # the script as GITHUB_TOKEN. Needs org self-hosted-runners:read.
+          GITHUB_TOKEN: ${{ secrets.RUNNER_LIST_GH_TOKEN }}
+          TRACKED_LABEL_PREFIXES: "macos,linux.dgx.b200"
+          OUTPUT_FILE: runner_fleet_count.jsonl
+        run: python tools/runner_metrics/collect_runner_fleet.py
+
+      - name: Configure AWS credentials
+        if: ${{ !inputs.dry_run }}
+        # Shared ARC role: trusts repo:pytorch/test-infra:* and can write
+        # gha-artifacts (same role _linux-build.yml uses for artifact upload).
+        uses: aws-actions/configure-aws-credentials@v4
+        with:
+          role-to-assume: arn:aws:iam::308535385114:role/arc
+          aws-region: ${{ env.AWS_REGION }}
+
+      - name: Upload to S3 for ClickHouse ingestion
+        if: ${{ !inputs.dry_run }}
+        run: |
+          set -euo pipefail
+          ts="$(date -u +%Y%m%dT%H%M%SZ)"
+          gzip -c runner_fleet_count.jsonl > runner_fleet_count.json.gz
+          # NB: do NOT set --content-encoding gzip. ClickHouse's s3() reader
+          # decompresses via the explicit compression arg in the replicator; a
+          # gzip Content-Encoding header makes it mis-read the object and ingest
+          # 0 rows silently (no error). Store plain gzip bytes instead.
+          aws s3 cp runner_fleet_count.json.gz \
+            "s3://${S3_BUCKET}/${S3_PREFIX}/${RUNNER_ORG}/${ts}.json.gz"

aws/lambda/clickhouse-replicator-s3/lambda_function.py

@@ -709,6 +709,24 @@ def autorevert_advisor_verdicts_adapter(table, bucket, key):
     general_adapter(table, bucket, key, schema, ["none"], "JSONEachRow")
 
 
+def runner_fleet_count_adapter(table, bucket, key):
+    # Column order must match clickhouse_db_schema/misc.runner_fleet_count/
+    # schema.sql (minus `_meta`, which general_adapter appends via `SELECT *,
+    # (bucket, key)`). JSONEachRow maps by field name.
+    # NB: no timezone literal here -- general_adapter wraps this schema in single
+    # quotes, so an inner 'UTC' would break the s3() SQL. Timestamps are UTC
+    # wall-clock strings; the table column carries the UTC tz.
+    schema = """
+    `time_stamp` DateTime64(0),
+    `org` String,
+    `label` String,
+    `total_count` UInt32,
+    `online_count` UInt32,
+    `busy_count` UInt32
+    """
+    general_adapter(table, bucket, key, schema, ["gzip", "none"], "JSONEachRow")
+
+
 SUPPORTED_PATHS = {
     "merges": "default.merges",
     "queue_times_historical": "default.queue_times_historical",
@@ -734,6 +752,7 @@ def autorevert_advisor_verdicts_adapter(table, bucket, key):
     # fbossci-cloudwatch-metrics bucket
     "ghci-related": "infra_metrics.cloudwatch_metrics",
     "test_jsons_while_running": "tests.all_test_runs",
+    "runner_fleet_count": "misc.runner_fleet_count",
 }
 
 OBJECT_CONVERTER = {
@@ -760,6 +779,7 @@ def autorevert_advisor_verdicts_adapter(table, bucket, key):
     "misc.claude_code_usage": claude_code_usage_adapter,
     "misc.autorevert_advisor_verdicts": autorevert_advisor_verdicts_adapter,
     "infra_metrics.cloudwatch_metrics": cloudwatch_metrics_adapter,
+    "misc.runner_fleet_count": runner_fleet_count_adapter,
 }
 
 

clickhouse_db_schema/misc.runner_fleet_count/schema.sql

@@ -0,0 +1,26 @@
+-- Point-in-time counts of self-hosted runners registered with a GitHub org,
+-- bucketed by runner label. One row per (sample, label). Populated by the S3
+-- replicator (clickhouse-replicator-s3): the runner-fleet-metrics workflow
+-- uploads gzipped JSONEachRow to s3://gha-artifacts/runner_fleet_count/..., the
+-- replicator does `INSERT INTO misc.runner_fleet_count SELECT *, (bucket, key)
+-- AS _meta FROM s3(...)`. Graphed in Grafana via the ClickHouse datasource.
+--
+-- IMPORTANT: the general_adapter insert is positional (`SELECT *, _meta`), so
+-- the column order here MUST match the adapter's schema string exactly, with
+-- `_meta` last. Do not add DEFAULT/normal columns in between -- they would
+-- shift the positional mapping and break ingestion.
+CREATE TABLE misc.runner_fleet_count
+(
+    `time_stamp` DateTime64(0, 'UTC'),   -- when the fleet was sampled (UTC)
+    `org` String,                        -- e.g. pytorch
+    `label` String,                      -- e.g. linux.dgx.b200, macos-m1-stable
+    `total_count` UInt32,                -- runners carrying this label
+    `online_count` UInt32,               -- subset with status = online
+    `busy_count` UInt32,                 -- subset currently running a job
+    `_meta` Tuple(bucket String, key String)  -- S3 provenance added by replicator
+)
+ENGINE = SharedMergeTree('/clickhouse/tables/{uuid}/{shard}', '{replica}')
+PARTITION BY toYYYYMM(time_stamp)
+ORDER BY (label, org, time_stamp)
+TTL toDate(time_stamp) + toIntervalYear(2)
+SETTINGS index_granularity = 8192

tools/runner_metrics/collect_runner_fleet.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""Sample how many self-hosted runners are registered with a GitHub org,
+bucketed by label, and write the counts as JSONEachRow for S3 -> ClickHouse
+ingestion (misc.runner_fleet_count via clickhouse-replicator-s3).
+
+A runner is counted under every tracked label it carries, so a host with both
+``macos-m1-14`` and ``macos-m1-stable`` contributes to both series -- these are
+distinct scheduling labels, which is what we want to trend.
+
+The output is newline-delimited JSON (one object per line); each object's keys
+match the ClickHouse column names exactly (JSONEachRow maps by name). The
+workflow gzips this file and uploads it under the runner_fleet_count/ prefix.
+
+Environment:
+  GITHUB_TOKEN            token with org "self-hosted runners: read"
+                          (fine-grained) or classic ``admin:org`` read scope.
+  RUNNER_ORG              org to query (default: pytorch).
+  TRACKED_LABEL_PREFIXES  comma-separated label prefixes to record
+                          (default: "macos,linux.dgx.b200").
+  OUTPUT_FILE             path to write JSONEachRow to (default:
+                          runner_fleet_count.jsonl).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from collections import defaultdict
+from datetime import datetime, timezone
+
+import requests
+
+
+GITHUB_API = "https://api.github.com"
+
+
+def list_org_runners(org: str, token: str) -> list[dict]:
+    """Return every self-hosted runner registered with ``org`` (all pages)."""
+    session = requests.Session()
+    session.headers.update(
+        {
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/vnd.github+json",
+            "X-GitHub-Api-Version": "2022-11-28",
+        }
+    )
+    runners: list[dict] = []
+    page = 1
+    while True:
+        resp = session.get(
+            f"{GITHUB_API}/orgs/{org}/actions/runners",
+            params={"per_page": 100, "page": page},
+            timeout=30,
+        )
+        resp.raise_for_status()
+        payload = resp.json()
+        batch = payload.get("runners", [])
+        runners.extend(batch)
+        if not batch or len(runners) >= payload.get("total_count", 0):
+            break
+        page += 1
+    return runners
+
+
+def bucket_by_label(
+    runners: list[dict], prefixes: list[str]
+) -> dict[str, dict[str, int]]:
+    counts: dict[str, dict[str, int]] = defaultdict(
+        lambda: {"total": 0, "online": 0, "busy": 0}
+    )
+    for runner in runners:
+        is_online = str(runner.get("status", "")).lower() == "online"
+        is_busy = bool(runner.get("busy"))
+        for name in {label.get("name", "") for label in runner.get("labels", [])}:
+            if not any(name.startswith(p) for p in prefixes):
+                continue
+            counts[name]["total"] += 1
+            counts[name]["online"] += int(is_online)
+            counts[name]["busy"] += int(is_busy)
+    return counts
+
+
+def main() -> int:
+    token = os.environ.get("GITHUB_TOKEN")
+    if not token:
+        print("GITHUB_TOKEN is required", file=sys.stderr)
+        return 1
+    org = os.environ.get("RUNNER_ORG", "pytorch")
+    prefixes = [
+        p.strip()
+        for p in os.environ.get("TRACKED_LABEL_PREFIXES", "macos,linux.dgx.b200").split(
+            ","
+        )
+        if p.strip()
+    ]
+    output_file = os.environ.get("OUTPUT_FILE", "runner_fleet_count.jsonl")
+
+    runners = list_org_runners(org, token)
+    counts = bucket_by_label(runners, prefixes)
+
+    # ClickHouse DateTime64(0, 'UTC') parses this "YYYY-MM-DD HH:MM:SS" form.
+    sample_ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+    lines = []
+    for label, c in sorted(counts.items()):
+        row = {
+            "time_stamp": sample_ts,
+            "org": org,
+            "label": label,
+            "total_count": c["total"],
+            "online_count": c["online"],
+            "busy_count": c["busy"],
+        }
+        lines.append(json.dumps(row))
+        print(
+            f"{org} {label}: total={c['total']} online={c['online']} busy={c['busy']}"
+        )
+
+    if not lines:
+        # An empty match set is a real (all-zero) observation, but we cannot
+        # synthesize per-label zero rows without knowing every expected label.
+        # Fail loudly rather than silently uploading an empty object.
+        print(
+            f"ERROR: no runners matched prefixes {prefixes} in org {org}.",
+            file=sys.stderr,
+        )
+        return 2
+
+    with open(output_file, "w") as f:
+        f.write("\n".join(lines) + "\n")
+    print(f"Wrote {len(lines)} rows to {output_file}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())