xiaoxia-saas/scripts/ci/runner_monitor/runner_status.py

#!/usr/bin/env python3
"""
Runner 状态巡检 - 调 Gitea API 查 runner 列表 + 状态 + 队列积压

功能:
  - 获取所有 runner 的在线状态、忙闲状态
  - 检测离线/禁用 runner
  - 检测 CI 队列积压（pending 数量 + 持续时间）
  - 生成 runner 状态快照

说明:
  Gitea Actions API 直接返回的 runner 信息不包含心跳时间，
  因此"离线超过N分钟"的判断通过以下方式近似:
  1. status != "online" 的 runner 直接判定为离线
  2. busy=true 且长时间无 job 完成的 runner 标记为疑似挂起（待增强）
  3. 通过 pending job 数量和时长判断队列积压

用法:
  python3 scripts/ci/runner_monitor/runner_status.py --check
  python3 scripts/ci/runner_monitor/runner_status.py --snapshot
  python3 scripts/ci/runner_monitor/runner_status.py --list
"""

import argparse
import json
import os
import sys
import time
from datetime import UTC, datetime

# 复用 chatops 的 GiteaClient
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
_CI_DIR = os.path.dirname(_SCRIPT_DIR)
if _CI_DIR not in sys.path:
    sys.path.insert(0, _CI_DIR)

from chatops.gitea_client import GiteaClient  # noqa: E402


class RunnerStatusChecker:
    """Runner 状态巡检器"""

    def __init__(self, gitea_client=None):
        self.gitea = gitea_client or GiteaClient()

    # ── Runner 列表 ──────────────────────────────────

    def get_runners(self):
        """获取仓库所有 runner 列表

        Returns:
            list[dict]: runner 列表
        """
        # 直接调用 Gitea Actions runners API
        data = self.gitea._request("actions/runners")
        if not data:
            return []
        return data.get("runners", [])

    def get_runner_summary(self):
        """获取 runner 汇总信息

        Returns:
            dict: {total, online, offline, busy, disabled, runners}
        """
        runners = self.get_runners()
        if not runners:
            return {
                "total": 0,
                "online": 0,
                "offline": 0,
                "busy": 0,
                "disabled": 0,
                "runners": [],
            }

        online = sum(1 for r in runners if r.get("status") == "online" and not r.get("disabled"))
        offline = sum(1 for r in runners if r.get("status") != "online" and not r.get("disabled"))
        busy = sum(1 for r in runners if r.get("busy"))
        disabled = sum(1 for r in runners if r.get("disabled"))

        return {
            "total": len(runners),
            "online": online,
            "offline": offline,
            "busy": busy,
            "disabled": disabled,
            "runners": runners,
        }

    def get_offline_runners(self, offline_minutes=5):
        """获取离线的 runner 列表

        由于 Gitea API 不返回心跳时间，status != online 即视为离线。
        offline_minutes 参数保留用于后续 SSH 心跳检测增强。

        Returns:
            list[dict]: 离线 runner 列表
        """
        runners = self.get_runners()
        if not runners:
            return []

        offline = [r for r in runners if not r.get("disabled") and r.get("status") != "online"]
        # 补充离线时长字段（暂时用 None，后续增强）
        for r in offline:
            r["offline_minutes"] = None
            r["offline_reason"] = f"status={r.get('status', 'unknown')}"

        return offline

    # ── 队列积压检测 ──────────────────────────────────

    def get_pending_runs(self):
        """获取 pending / queued 状态的 workflow runs

        Returns:
            list[dict]: pending run 列表
        """
        # 尝试多种状态名（Gitea 可能用 queued / pending / waiting）
        pending = []
        for status in ["queued", "pending", "waiting"]:
            runs, _ = self.gitea.list_runs(status=status, limit=50)
            pending.extend(runs)

        # 去重
        seen = set()
        unique = []
        for r in pending:
            rid = r.get("id")
            if rid and rid not in seen:
                seen.add(rid)
                unique.append(r)

        return unique

    def get_queue_backlog(self, pending_threshold=10, duration_minutes=10):
        """检测队列积压

        Args:
            pending_threshold: pending 数量阈值
            duration_minutes: 持续时间阈值（分钟）

        Returns:
            dict: {is_backlogged, pending_count, oldest_pending_minutes, pending_runs}
        """
        pending = self.get_pending_runs()
        if not pending:
            return {
                "is_backlogged": False,
                "pending_count": 0,
                "oldest_pending_minutes": 0,
                "pending_runs": [],
            }

        now = datetime.now(UTC)
        oldest_minutes = 0
        for r in pending:
            created = r.get("created_at", "")
            if not created:
                continue
            try:
                t = datetime.fromisoformat(created.replace("Z", "+00:00"))
                age = (now - t).total_seconds() / 60
                oldest_minutes = max(oldest_minutes, age)
            except Exception:
                pass

        is_backlogged = len(pending) >= pending_threshold and oldest_minutes >= duration_minutes

        return {
            "is_backlogged": is_backlogged,
            "pending_count": len(pending),
            "oldest_pending_minutes": round(oldest_minutes, 1),
            "pending_runs": pending,
        }

    # ── 综合巡检 ──────────────────────────────────────

    def run_full_check(self, offline_minutes=5, pending_threshold=10, pending_duration=10):
        """执行完整的 runner 巡检

        Returns:
            dict: 巡检结果
        """
        summary = self.get_runner_summary()
        offline_runners = self.get_offline_runners(offline_minutes=offline_minutes)
        backlog = self.get_queue_backlog(
            pending_threshold=pending_threshold,
            duration_minutes=pending_duration,
        )

        return {
            "timestamp": datetime.now(UTC).isoformat(),
            "runner_summary": {
                "total": summary["total"],
                "online": summary["online"],
                "offline": summary["offline"],
                "busy": summary["busy"],
                "disabled": summary["disabled"],
            },
            "offline_runners": [
                {
                    "id": r.get("id"),
                    "name": r.get("name"),
                    "status": r.get("status"),
                    "busy": r.get("busy"),
                    "labels": [label.get("name") for label in r.get("labels", [])],
                    "offline_minutes": r.get("offline_minutes"),
                    "offline_reason": r.get("offline_reason"),
                }
                for r in offline_runners
            ],
            "queue_backlog": {
                "is_backlogged": backlog["is_backlogged"],
                "pending_count": backlog["pending_count"],
                "oldest_pending_minutes": backlog["oldest_pending_minutes"],
            },
            "issues_found": len(offline_runners) > 0 or backlog["is_backlogged"],
        }

    # ── 快照输出 ──────────────────────────────────────

    def save_snapshot(self, output_dir=None, data=None):
        """保存状态快照为 JSON 文件

        Returns:
            str: 快照文件路径
        """
        if data is None:
            data = self.run_full_check()
        if output_dir is None:
            from runner_monitor import config

            output_dir = config.OUTPUT_DIR

        os.makedirs(output_dir, exist_ok=True)
        ts = time.strftime("%Y%m%d_%H%M%S")
        filename = f"runner_snapshot_{ts}.json"
        filepath = os.path.join(output_dir, filename)

        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

        # 清理旧快照（保留最近 24 个）
        self._cleanup_old_snapshots(output_dir, keep=24)

        return filepath

    @staticmethod
    def _cleanup_old_snapshots(directory, keep=24):
        """清理旧快照文件"""
        try:
            files = sorted(
                [f for f in os.listdir(directory) if f.startswith("runner_snapshot_")],
                reverse=True,
            )
            for old in files[keep:]:
                os.remove(os.path.join(directory, old))
        except OSError:
            pass


# ── CLI 入口 ──────────────────────────────────────────


def main():
    parser = argparse.ArgumentParser(description="Runner 状态巡检")
    parser.add_argument("--list", action="store_true", help="列出所有 runner")
    parser.add_argument("--check", action="store_true", help="执行完整巡检")
    parser.add_argument("--snapshot", action="store_true", help="生成快照 JSON")
    parser.add_argument("--pending", action="store_true", help="查看 pending 队列")
    parser.add_argument("--output-dir", help="快照输出目录")

    args = parser.parse_args()

    checker = RunnerStatusChecker()

    if args.list:
        summary = checker.get_runner_summary()
        print(
            f"Runner 总览: {summary['online']}/{summary['total']} 在线, "
            f"{summary['busy']} 忙碌, {summary['offline']} 离线, "
            f"{summary['disabled']} 禁用"
        )
        print()
        for r in summary["runners"]:
            status_icon = "🟢" if r.get("status") == "online" else "🔴"
            if r.get("disabled"):
                status_icon = "⚪"
            busy_icon = "⚡" if r.get("busy") else "  "
            labels = ", ".join(label.get("name") for label in r.get("labels", [])[:4])
            print(f"  {status_icon}{busy_icon} {r['name']:<30} {r.get('status', '?'):<10} labels: {labels}")

    elif args.pending:
        pending = checker.get_pending_runs()
        print(f"Pending runs: {len(pending)}")
        for r in pending[:10]:
            print(
                f"  #{r.get('id')} {r.get('name', '?')} - {r.get('status', '?')} "
                f"({r.get('head_branch', '?')}) created: {r.get('created_at', '?')[:16]}"
            )

    elif args.check:
        result = checker.run_full_check()
        print(json.dumps(result, indent=2, ensure_ascii=False))

    elif args.snapshot:
        path = checker.save_snapshot(output_dir=args.output_dir)
        print(f"快照已保存: {path}")

    else:
        parser.print_help()
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())