feat(ci): P2-3 Runner监控告警 - 离线/磁盘/内存/队列积压主动告警 (#537)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

scripts/ci/runner_monitor/runner_status.py

@@ -0,0 +1,320 @@
+#!/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 datetime, timezone
+
+# 复用 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(timezone.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(timezone.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())