grm/scripts/cleanup_stale_runners.py

#!/usr/bin/env python3
"""Clean up stale runner registrations from Gitea.

A runner is considered stale if it hasn't been online for more than a
configurable threshold (default: 1 hour).  Stale runners accumulate when:
- A runner host is rebuilt or re-provisioned (old registration remains)
- A runner is re-registered (old entry remains alongside the new one)
- A runner process dies and the healthcheck can't auto-recover

This script queries the Gitea API for all runners, identifies stale ones,
and deletes them via ``DELETE /api/v1/admin/actions/runners/{id}``.

Usage::

    python3 scripts/cleanup_stale_runners.py --gitea-url https://git.example.com --token <admin-token>
    python3 scripts/cleanup_stale_runners.py --gitea-url https://git.example.com --token <admin-token> --dry-run
    python3 scripts/cleanup_stale_runners.py --gitea-url https://git.example.com --token <admin-token> \\
        --stale-threshold 3600
"""

from __future__ import annotations

import argparse
import json
import sys
import time
import urllib.error
import urllib.request  # noqa: PTH123  # nosec B404
from typing import Any


def _api_request(base_url: str, token: str, method: str, path: str) -> Any:
    url = f"{base_url.rstrip('/')}/api/v1{path}"
    req = urllib.request.Request(url, method=method)  # nosec B310
    req.add_header("Authorization", f"token {token}")
    req.add_header("Accept", "application/json")
    try:
        with urllib.request.urlopen(req) as resp:  # noqa: PTH123  # nosec B310
            if resp.status == 204:
                return None
            raw = resp.read()
            return json.loads(raw) if raw else None
    except urllib.error.HTTPError as e:
        detail = e.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"Gitea API error {e.code}: {detail}") from e


def list_runners(base_url: str, token: str) -> list[dict[str, Any]]:
    data = _api_request(base_url, token, "GET", "/admin/actions/runners")
    if data is None:
        return []
    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        return data.get("runners", [])
    return []


def delete_runner(base_url: str, token: str, runner_id: int) -> bool:
    try:
        _api_request(base_url, token, "DELETE", f"/admin/actions/runners/{runner_id}")
        return True
    except RuntimeError as e:
        print(f"  ERROR deleting runner {runner_id}: {e}", file=sys.stderr)
        return False


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Clean up stale Gitea runner registrations")
    parser.add_argument("--gitea-url", required=True, help="Gitea base URL")
    parser.add_argument("--token", required=True, help="Gitea admin API token")
    parser.add_argument(
        "--stale-threshold",
        type=int,
        default=3600,
        help="Seconds since last_online before a runner is considered stale (default: 3600 = 1h)",
    )
    parser.add_argument("--dry-run", action="store_true", help="List stale runners without deleting")
    args = parser.parse_args(argv)

    runners = list_runners(args.gitea_url, args.token)
    if not runners:
        print("No runners found.")
        return 0

    now = int(time.time())
    stale: list[dict[str, Any]] = []
    online: list[dict[str, Any]] = []

    for runner in runners:
        last_online = runner.get("last_online", 0) or 0
        seconds_since = now - last_online
        runner["seconds_since_online"] = seconds_since
        if seconds_since > args.stale_threshold:
            stale.append(runner)
        else:
            online.append(runner)

    print(f"Total runners: {len(runners)}")
    print(f"Online (within {args.stale_threshold}s): {len(online)}")
    print(f"Stale (>{args.stale_threshold}s): {len(stale)}")
    print()

    if not stale:
        print("No stale runners to clean up.")
        return 0

    print("Stale runners:")
    for r in stale:
        rid = r.get("id", "?")
        name = r.get("name", "?")
        uuid = r.get("uuid", "?")[:8]
        secs = r.get("seconds_since_online", 0)
        hours = secs / 3600
        print(f"  id={rid} name={name} uuid={uuid}... offline={hours:.1f}h ago")

    if args.dry_run:
        print("\n--dry-run: not deleting. Remove --dry-run to clean up.")
        return 0

    print(f"\nDeleting {len(stale)} stale runners...")
    deleted = 0
    for r in stale:
        rid = r.get("id")
        if rid is None:
            continue
        if delete_runner(args.gitea_url, args.token, rid):
            deleted += 1
            print(f"  Deleted runner id={rid} ({r.get('name', '?')})")

    print(f"\nDone: {deleted}/{len(stale)} stale runners deleted.")
    return 0 if deleted == len(stale) else 1


if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())