From 67a4af9846fb7a5ece6a95c253faa2153ee96785 Mon Sep 17 00:00:00 2001 From: naindev <193035803+Nain9Dev@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:36:09 +0200 Subject: [PATCH] feat(report): add a runner watchdog that toasts an offline runner or a long queue The weekly report says how the queue behaved; nothing said that it was wrong now. runner_watch.py reads the runners of the organisation and of each private repository plus the queued runs, and reports a runner seen offline for 10 minutes (the first sighting is remembered, GitHub does not say since when) or a run queued for 15, once per finding, through the desktop's notify.ps1. An ignored runner or an unreadable answer raises nothing. --- README.md | 10 ++ report/runner_watch.py | 257 +++++++++++++++++++++++++++++++++++++ tests/test_runner_watch.py | 239 ++++++++++++++++++++++++++++++++++ 3 files changed, 506 insertions(+) create mode 100644 report/runner_watch.py create mode 100644 tests/test_runner_watch.py diff --git a/README.md b/README.md index 1eeba40..f990679 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,16 @@ Status: in progress. The design is in `specs/next-ci-reliability` of merged outside the train since `--train-since`, and `--train-dir D:\Development\Tools\ci-train` adds a section on the installed train: age of its log (silent after 90 minutes) and whether its script differs from the one in `ci-shared` main. + +- `report/runner_watch.py` (available): the runner watchdog. The report says how the queue behaved + over a period; this says that it is wrong now. Run from a scheduled task every few minutes + (`py -3 report/runner_watch.py --org naindev-contrast3d --notify `), it reads + the runners of the organisation and of each private repository, plus the queued runs, and raises one + toast per finding: a runner offline for `--offline-min` minutes (default 10; GitHub does not say + since when, so the first sighting is remembered in a state file) or a run queued for `--queue-min` + (default 15). A finding is reported once and again only after it has cleared; `--ignore NAME` mutes + a runner that is offline on purpose; an unreadable answer is skipped, never reported. Exit code 1 + when it reported something. `--until 2026-10-14T11:00` ends the period in the past, so a baseline can be reproduced after later changes; `--limit` raises the number of runs read per repository (default 300). diff --git a/report/runner_watch.py b/report/runner_watch.py new file mode 100644 index 0000000..eafcc28 --- /dev/null +++ b/report/runner_watch.py @@ -0,0 +1,257 @@ +"""Runner watchdog: a toast when a self-hosted runner is offline or a run waits too long in the queue. + + py -3 runner_watch.py --org naindev-contrast3d --notify D:\\...\\notify.ps1 + +Run it from a scheduled task every few minutes. Each cycle reads, over `gh`, the runners of the +organisation and of every private repository in it, plus the runs still queued. It reports a runner +that has been seen offline for `--offline-min` minutes (GitHub does not say since when, so the first +sighting is remembered in the state file) and a run queued for `--queue-min` minutes. Every finding +is reported once, and again only after it has gone away and come back. An unreadable answer is +skipped, never reported: a watchdog that cries wolf on a network blip gets muted. + +The report in this folder (`ci_report.py`) says how the queue behaved over a period; this says that +it is wrong now. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import subprocess +import sys +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import ci_report # same folder; its Github class does the `gh` reads + +STAMP = "%Y-%m-%dT%H:%M:%SZ" +OFFLINE_MIN = 10 +QUEUE_MIN = 15 + + +@dataclass +class Alert: + key: str + title: str + body: str + + +class Github(ci_report.Github): + """The reads of the report plus the two the watchdog needs at organisation scope.""" + + def org_runners(self, org: str) -> list[dict]: + out = self._run(["gh", "api", f"orgs/{org}/actions/runners?per_page=100"]) + return json.loads(out or "{}").get("runners", []) + + def org_repos(self, org: str) -> list[dict]: + out = self._run( + [ + "gh", + "repo", + "list", + org, + "--limit", + "100", + "--json", + "name,visibility,isArchived", + ] + ) + return json.loads(out or "[]") + + +def collect(gh, org: str) -> dict: + """What GitHub says now: (scope, runner name, status) and (repository, run id, created at).""" + runners: list[tuple[str, str, str]] = [] + queued: list[tuple[str, int, str]] = [] + try: + for runner in gh.org_runners(org): + runners.append((org, runner["name"], runner["status"])) + except subprocess.CalledProcessError: + pass + try: + repos = gh.org_repos(org) + except subprocess.CalledProcessError: + repos = [] + for repo in repos: + if repo.get("isArchived") or repo.get("visibility") != "PRIVATE": + continue + scope = f"{org}/{repo['name']}" + try: + for runner in gh.runners(scope): + runners.append((scope, runner["name"], runner["status"])) + except subprocess.CalledProcessError: + pass + try: + for item in gh.queued_runs(scope): + queued.append((scope, item["databaseId"], item["createdAt"])) + except subprocess.CalledProcessError: + pass + return {"runners": runners, "queued": queued} + + +def _moment(text: str) -> dt.datetime: + return dt.datetime.fromisoformat(text.replace("Z", "+00:00")) + + +def evaluate( + snapshot: dict, + state: dict, + now: dt.datetime, + offline_min: int = OFFLINE_MIN, + queue_min: int = QUEUE_MIN, + ignore=frozenset(), +) -> tuple[list[Alert], dict]: + """The alerts that are new now, and the state to keep for the next cycle.""" + alerted = set(state.get("alerted", [])) + seen_offline = dict(state.get("offline_since", {})) + new_offline: dict[str, str] = {} + active: set[str] = set() + alerts: list[Alert] = [] + + for scope, name, status in snapshot["runners"]: + if name in ignore or status == "online": + continue + key = f"{scope}/{name}" + since = seen_offline.get(key) or now.strftime(STAMP) + new_offline[key] = since + minutes = int((now - _moment(since)).total_seconds() // 60) + if minutes < offline_min: + continue + alert_key = f"offline:{key}" + active.add(alert_key) + if alert_key not in alerted: + alerts.append( + Alert( + alert_key, + "Runner offline", + f"{name} ({scope}) has been offline for {minutes} min.", + ) + ) + + for scope, run_id, created in snapshot["queued"]: + minutes = int((now - _moment(created)).total_seconds() // 60) + if minutes < queue_min: + continue + alert_key = f"queue:{scope}/{run_id}" + active.add(alert_key) + if alert_key not in alerted: + alerts.append( + Alert( + alert_key, + "CI run waiting for a runner", + f"A run of {scope} has been queued for {minutes} min.", + ) + ) + + kept = sorted((alerted & active) | {a.key for a in alerts}) + return alerts, {"offline_since": new_offline, "alerted": kept} + + +def notify( + alerts: list[Alert], notify_path: str, runner: Callable[[list[str]], object] +) -> None: + """One toast per alert through the desktop's own `notify.ps1`; a failing toast never raises.""" + for alert in alerts: + command = [ + "powershell.exe", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + notify_path, + "-Title", + alert.title, + "-Body", + alert.body, + ] + try: + runner(command) + except (OSError, subprocess.SubprocessError): + pass + + +def load_state(path: Path) -> dict: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + +def save_state(path: Path, state: dict) -> None: + scratch = path.with_name(path.name + ".tmp") + scratch.write_text( + json.dumps(state, indent=1, sort_keys=True), encoding="utf-8", newline="\n" + ) + scratch.replace(path) + + +def _run_hidden(command: list[str]) -> None: + no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0) + subprocess.run(command, check=False, capture_output=True, creationflags=no_window) + + +def run( + gh, + org: str, + state_path: Path, + now: dt.datetime, + *, + notify_path: str | None, + offline_min: int = OFFLINE_MIN, + queue_min: int = QUEUE_MIN, + ignore=frozenset(), + runner: Callable[[list[str]], object] = _run_hidden, +) -> list[Alert]: + alerts, state = evaluate( + collect(gh, org), load_state(state_path), now, offline_min, queue_min, ignore + ) + if notify_path: + notify(alerts, notify_path, runner) + save_state(state_path, state) + return alerts + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--org", required=True, help="the organisation to watch") + parser.add_argument( + "--state", help="JSON remembered between cycles (default: beside this file)" + ) + parser.add_argument( + "--notify", help="path of notify.ps1: one toast per new finding" + ) + parser.add_argument("--offline-min", type=int, default=OFFLINE_MIN) + parser.add_argument("--queue-min", type=int, default=QUEUE_MIN) + parser.add_argument( + "--ignore", + action="append", + default=[], + help="a runner name never reported; repeatable", + ) + args = parser.parse_args(argv) + state = ( + Path(args.state) + if args.state + else Path(__file__).resolve().with_name("runner-watch.json") + ) + alerts = run( + Github(), + args.org, + state, + dt.datetime.now(dt.UTC), + notify_path=args.notify, + offline_min=args.offline_min, + queue_min=args.queue_min, + ignore=frozenset(args.ignore), + ) + for alert in alerts: + print(f"{alert.title}: {alert.body}") + return 1 if alerts else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_runner_watch.py b/tests/test_runner_watch.py new file mode 100644 index 0000000..37aa3d5 --- /dev/null +++ b/tests/test_runner_watch.py @@ -0,0 +1,239 @@ +"""Tests for report/runner_watch.py: when a runner or a queue deserves a toast, and only once.""" + +from __future__ import annotations + +import datetime as dt +import json +import subprocess +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "report")) + +import runner_watch as watch + +NOW = dt.datetime(2026, 10, 15, 12, 0, 0, tzinfo=dt.UTC) +ORG = "acme" + + +def ago(minutes: float) -> str: + return (NOW - dt.timedelta(minutes=minutes)).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def snapshot(runners=(), queued=()): + return {"runners": list(runners), "queued": list(queued)} + + +class Offline(unittest.TestCase): + def test_an_online_runner_raises_nothing(self): + alerts, state = watch.evaluate( + snapshot([("acme/a", "a-01", "online")]), {}, NOW + ) + self.assertEqual(alerts, []) + self.assertEqual(state["offline_since"], {}) + + def test_a_runner_just_seen_offline_is_remembered_not_reported(self): + alerts, state = watch.evaluate( + snapshot([("acme/a", "a-01", "offline")]), {}, NOW + ) + self.assertEqual(alerts, []) + self.assertEqual(state["offline_since"], {"acme/a/a-01": ago(0)}) + + def test_it_is_reported_once_it_has_been_offline_long_enough(self): + state = {"offline_since": {"acme/a/a-01": ago(11)}} + alerts, state = watch.evaluate( + snapshot([("acme/a", "a-01", "offline")]), state, NOW + ) + self.assertEqual([a.key for a in alerts], ["offline:acme/a/a-01"]) + self.assertIn("a-01", alerts[0].body) + self.assertIn("11 min", alerts[0].body) + + def test_a_runner_offline_for_less_than_the_threshold_is_not_reported(self): + state = {"offline_since": {"acme/a/a-01": ago(9)}} + alerts, _ = watch.evaluate( + snapshot([("acme/a", "a-01", "offline")]), state, NOW + ) + self.assertEqual(alerts, []) + + def test_the_same_runner_is_not_reported_twice(self): + state = {"offline_since": {"acme/a/a-01": ago(30)}} + alerts, state = watch.evaluate( + snapshot([("acme/a", "a-01", "offline")]), state, NOW + ) + self.assertEqual(len(alerts), 1) + again, _ = watch.evaluate(snapshot([("acme/a", "a-01", "offline")]), state, NOW) + self.assertEqual(again, []) + + def test_a_runner_that_comes_back_is_forgotten_and_can_be_reported_again(self): + state = { + "offline_since": {"acme/a/a-01": ago(30)}, + "alerted": ["offline:acme/a/a-01"], + } + alerts, state = watch.evaluate( + snapshot([("acme/a", "a-01", "online")]), state, NOW + ) + self.assertEqual(alerts, []) + self.assertEqual(state["offline_since"], {}) + self.assertEqual(state["alerted"], []) + + def test_an_ignored_runner_is_never_reported(self): + state = {"offline_since": {"acme/a/a-01": ago(300)}} + alerts, state = watch.evaluate( + snapshot([("acme/a", "a-01", "offline")]), state, NOW, ignore={"a-01"} + ) + self.assertEqual(alerts, []) + self.assertEqual(state["offline_since"], {}) + + def test_a_runner_that_vanished_from_the_list_is_forgotten(self): + state = { + "offline_since": {"acme/a/gone": ago(300)}, + "alerted": ["offline:acme/a/gone"], + } + _, state = watch.evaluate(snapshot([]), state, NOW) + self.assertEqual(state["offline_since"], {}) + self.assertEqual(state["alerted"], []) + + +class Queue(unittest.TestCase): + def test_a_young_queued_run_is_not_reported(self): + alerts, _ = watch.evaluate(snapshot(queued=[("acme/a", 7, ago(14))]), {}, NOW) + self.assertEqual(alerts, []) + + def test_an_old_queued_run_is_reported_once(self): + snap = snapshot(queued=[("acme/a", 7, ago(16))]) + alerts, state = watch.evaluate(snap, {}, NOW) + self.assertEqual([a.key for a in alerts], ["queue:acme/a/7"]) + self.assertIn("16 min", alerts[0].body) + again, _ = watch.evaluate(snap, state, NOW) + self.assertEqual(again, []) + + def test_a_run_that_started_is_forgotten(self): + state = {"alerted": ["queue:acme/a/7"]} + _, state = watch.evaluate(snapshot(queued=[]), state, NOW) + self.assertEqual(state["alerted"], []) + + def test_the_thresholds_are_parameters(self): + snap = snapshot(queued=[("acme/a", 7, ago(5))]) + alerts, _ = watch.evaluate(snap, {}, NOW, queue_min=3) + self.assertEqual(len(alerts), 1) + + +class Collecting(unittest.TestCase): + def test_it_reads_the_organisation_and_the_private_repositories(self): + class Gh: + def org_repos(self, org): + return [ + {"name": "a", "visibility": "PRIVATE", "isArchived": False}, + {"name": "old", "visibility": "PRIVATE", "isArchived": True}, + {"name": "open", "visibility": "PUBLIC", "isArchived": False}, + ] + + def org_runners(self, org): + return [{"name": "pool-01", "status": "online"}] + + def runners(self, repo): + return [{"name": f"{repo}-01", "status": "offline"}] + + def queued_runs(self, repo): + return [{"databaseId": 5, "createdAt": ago(40)}] + + snap = watch.collect(Gh(), ORG) + self.assertIn(("acme", "pool-01", "online"), snap["runners"]) + self.assertIn(("acme/a", "acme/a-01", "offline"), snap["runners"]) + self.assertEqual(snap["queued"], [("acme/a", 5, ago(40))]) + self.assertEqual( + len(snap["runners"]), 2, "archived and public repositories are skipped" + ) + + def test_an_unreadable_part_does_not_stop_the_rest(self): + class Gh: + def org_repos(self, org): + return [{"name": "a", "visibility": "PRIVATE", "isArchived": False}] + + def org_runners(self, org): + raise subprocess.CalledProcessError(1, ["gh"], stderr="HTTP 403") + + def runners(self, repo): + return [{"name": "a-01", "status": "online"}] + + def queued_runs(self, repo): + raise subprocess.CalledProcessError(1, ["gh"], stderr="HTTP 502") + + snap = watch.collect(Gh(), ORG) + self.assertEqual(snap["runners"], [("acme/a", "a-01", "online")]) + self.assertEqual(snap["queued"], []) + + +class RealCommands(unittest.TestCase): + def setUp(self): + self.calls = [] + + def fake(self, command): + self.calls.append(command) + return json.dumps({"runners": [{"name": "pool-01", "status": "online"}]}) + + def test_org_runners_asks_the_organisation_endpoint(self): + runners = watch.Github(self.fake).org_runners(ORG) + self.assertEqual(runners[0]["name"], "pool-01") + self.assertIn("orgs/acme/actions/runners", " ".join(self.calls[0])) + + def test_org_repos_asks_for_visibility_and_archived(self): + watch.Github(lambda c: self.calls.append(c) or "[]").org_repos(ORG) + joined = " ".join(self.calls[0]) + self.assertIn("repo list acme", joined) + self.assertIn("name,visibility,isArchived", joined) + + +class Notifying(unittest.TestCase): + def test_each_alert_becomes_one_toast_command(self): + sent = [] + alerts = [watch.Alert("k1", "T1", "B1"), watch.Alert("k2", "T2", "B2")] + watch.notify(alerts, r"C:\x\notify.ps1", lambda c: sent.append(c)) + self.assertEqual(len(sent), 2) + self.assertIn("-File", sent[0]) + self.assertEqual(sent[0][sent[0].index("-Title") + 1], "T1") + self.assertEqual(sent[1][sent[1].index("-Body") + 1], "B2") + + def test_a_failing_toast_never_raises(self): + def boom(command): + raise OSError("no powershell") + + watch.notify([watch.Alert("k", "T", "B")], "notify.ps1", boom) + + +class Main(unittest.TestCase): + def test_state_is_saved_and_a_second_run_stays_quiet(self): + import tempfile + + class Gh: + def org_repos(self, org): + return [] + + def org_runners(self, org): + return [{"name": "pool-01", "status": "offline"}] + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "state.json" + state = {"offline_since": {"acme/pool-01": ago(40)}} + path.write_text(json.dumps(state), encoding="utf-8") + first = watch.run(Gh(), ORG, path, NOW, notify_path=None) + second = watch.run(Gh(), ORG, path, NOW, notify_path=None) + self.assertEqual(len(first), 1) + self.assertEqual(second, []) + self.assertIn( + "offline:acme/pool-01", + json.loads(path.read_text(encoding="utf-8"))["alerted"], + ) + + def test_a_corrupt_state_file_is_a_clean_slate(self): + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "state.json" + path.write_text("{not json", encoding="utf-8") + self.assertEqual(watch.load_state(path), {}) + + +if __name__ == "__main__": + unittest.main()