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

#!/usr/bin/env python3
"""
告警调度器 - 阈值判断 + 去重 + 飞书通知

功能:
  1. 从 runner_status 和 runner_metrics 获取数据
  2. 根据阈值判断是否触发告警
  3. 告警去重（同一问题 30 分钟内只报一次）
  4. 飞书卡片通知（复用 chatops FeishuNotifier）
  5. 生成状态快照 JSON（供看板用）

告警规则:
  P1（严重）:
    - Runner 离线超过 5 分钟
    - 磁盘使用率 > 90%

  P2（警告）:
    - 磁盘使用率 > 85%
    - 内存使用率 > 90% 持续 5 分钟
    - CI 队列积压 > 10 个 pending 超过 10 分钟

用法:
  python3 scripts/ci/runner_monitor/alert_manager.py --check
  python3 scripts/ci/runner_monitor/alert_manager.py --daemon  # 持续运行
  python3 scripts/ci/runner_monitor/alert_manager.py --snapshot
"""

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

# 复用 chatops 的飞书通知
_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 runner_monitor import config  # noqa: E402
from runner_monitor.runner_metrics import RunnerMetricsCollector  # noqa: E402
from runner_monitor.runner_status import RunnerStatusChecker  # noqa: E402


class Alert:
    """单条告警"""

    def __init__(self, alert_id, level, title, description, details=None, source="runner_monitor"):
        self.alert_id = alert_id  # 唯一标识，用于去重
        self.level = level  # P1 / P2 / INFO
        self.title = title
        self.description = description
        self.details = details or {}
        self.source = source
        self.timestamp = datetime.now(timezone.utc).isoformat()

    def to_dict(self):
        return {
            "alert_id": self.alert_id,
            "level": self.level,
            "title": self.title,
            "description": self.description,
            "details": self.details,
            "source": self.source,
            "timestamp": self.timestamp,
        }


class AlertManager:
    """告警调度器"""

    def __init__(
        self,
        status_checker=None,
        metrics_collector=None,
        dedupe_window=None,
    ):
        self.status_checker = status_checker or RunnerStatusChecker()
        self.metrics = metrics_collector or RunnerMetricsCollector()
        self.dedupe_window = dedupe_window or config.DEDUPE_WINDOW

        # 告警历史: {alert_id: last_triggered_timestamp}
        self._alert_history = {}
        # 内存持续超阈值记录: {host: first_detected_timestamp}
        self._mem_high_since = {}

    # ── 告警检测 ──────────────────────────────────────

    def detect_alerts(self):
        """执行所有检测规则，返回触发的告警列表

        Returns:
            list[Alert]: 新触发的告警（已去重）
        """
        all_alerts = []

        # 1. Runner 离线检测
        all_alerts.extend(self._check_runner_offline())

        # 2. 队列积压检测
        all_alerts.extend(self._check_queue_backlog())

        # 3. 系统指标检测（SSH，可能为空）
        all_alerts.extend(self._check_system_metrics())

        # 去重过滤
        new_alerts = [a for a in all_alerts if self._should_alert(a)]

        # 更新告警历史
        for alert in new_alerts:
            self._alert_history[alert.alert_id] = time.time()

        return new_alerts

    def _check_runner_offline(self):
        """检测离线 runner"""
        offline = self.status_checker.get_offline_runners(offline_minutes=config.RUNNER_OFFLINE_MINUTES)
        alerts = []

        for runner in offline:
            name = runner.get("name", "unknown")
            runner_id = runner.get("id", "?")
            alert_id = f"runner_offline_{runner_id}"

            # Gitea API 没有心跳时间，status != online 就告警（P1）
            alerts.append(
                Alert(
                    alert_id=alert_id,
                    level=config.P1,
                    title=f"Runner 离线: {name}",
                    description=(
                        f"Runner **{name}** (ID: {runner_id}) 状态为 "
                        f"{runner.get('status', 'unknown')}，已离线\n"
                        f"标签: {', '.join(label.get('name') for label in runner.get('labels', [])[:5])}"
                    ),
                    details={
                        "runner_id": runner_id,
                        "runner_name": name,
                        "status": runner.get("status"),
                        "labels": [label.get("name") for label in runner.get("labels", [])],
                    },
                )
            )

        return alerts

    def _check_queue_backlog(self):
        """检测队列积压"""
        backlog = self.status_checker.get_queue_backlog(
            pending_threshold=config.QUEUE_PENDING_COUNT,
            duration_minutes=config.QUEUE_PENDING_MINUTES,
        )

        if not backlog["is_backlogged"]:
            return []

        count = backlog["pending_count"]
        age = backlog["oldest_pending_minutes"]
        alert_id = f"queue_backlog_{int(age // 30)}"  # 每30分钟一个新告警id

        return [
            Alert(
                alert_id=alert_id,
                level=config.P2,
                title="CI 队列积压",
                description=(
                    f"当前有 **{count}** 个 pending run，最老的已等待 **{age:.0f} 分钟**\n"
                    f"阈值: >{config.QUEUE_PENDING_COUNT}个 且 超过{config.QUEUE_PENDING_MINUTES}分钟"
                ),
                details={
                    "pending_count": count,
                    "oldest_pending_minutes": age,
                },
            )
        ]

    def _check_system_metrics(self):
        """检测系统指标（磁盘/内存/CPU）"""
        metrics_list = self.metrics.collect_all()
        if not metrics_list:
            return []

        alerts = []
        now = time.time()

        for m in metrics_list:
            host = m.get("host", "unknown")
            if m.get("status") != "ok":
                continue

            # 磁盘告警
            disk_pct = m.get("disk_percent", 0)
            if disk_pct and disk_pct >= config.DISK_CRIT_PERCENT:
                alerts.append(
                    Alert(
                        alert_id=f"disk_crit_{host}",
                        level=config.P1,
                        title=f"磁盘使用率严重过高: {host}",
                        description=(
                            f"服务器 **{host}** 磁盘使用率 **{disk_pct:.1f}%** (P1阈值: {config.DISK_CRIT_PERCENT}%)\n"
                            f"已用: {m.get('disk_used_gb', '?')}G / {m.get('disk_total_gb', '?')}G"
                        ),
                        details={"host": host, "disk_percent": disk_pct},
                    )
                )
            elif disk_pct and disk_pct >= config.DISK_WARN_PERCENT:
                alerts.append(
                    Alert(
                        alert_id=f"disk_warn_{host}",
                        level=config.P2,
                        title=f"磁盘使用率过高: {host}",
                        description=(
                            f"服务器 **{host}** 磁盘使用率 **{disk_pct:.1f}%** (P2阈值: {config.DISK_WARN_PERCENT}%)\n"
                            f"已用: {m.get('disk_used_gb', '?')}G / {m.get('disk_total_gb', '?')}G"
                        ),
                        details={"host": host, "disk_percent": disk_pct},
                    )
                )

            # 内存告警（持续 N 分钟）
            mem_pct = m.get("mem_percent", 0)
            mem_key = f"mem_high_{host}"
            if mem_pct and mem_pct >= config.MEM_WARN_PERCENT:
                if mem_key not in self._mem_high_since:
                    self._mem_high_since[mem_key] = now
                else:
                    duration_min = (now - self._mem_high_since[mem_key]) / 60
                    if duration_min >= config.MEM_DURATION_MINUTES:
                        alerts.append(
                            Alert(
                                alert_id=f"mem_warn_{host}",
                                level=config.P2,
                                title=f"内存使用率持续过高: {host}",
                                description=(
                                    f"服务器 **{host}** 内存使用率 **{mem_pct:.1f}%**，"
                                    f"已持续 **{duration_min:.0f} 分钟**\n"
                                    f"阈值: {config.MEM_WARN_PERCENT}% 持续 {config.MEM_DURATION_MINUTES} 分钟"
                                ),
                                details={"host": host, "mem_percent": mem_pct, "duration_min": duration_min},
                            )
                        )
            else:
                # 恢复了，清除记录
                self._mem_high_since.pop(mem_key, None)

        return alerts

    # ── 去重 ──────────────────────────────────────────

    def _should_alert(self, alert):
        """判断是否应该发送告警（去重 + 等级开关）"""
        # 等级开关
        if alert.level == config.P1 and not config.P1_ENABLED:
            return False
        if alert.level == config.P2 and not config.P2_ENABLED:
            return False

        # 去重窗口
        last = self._alert_history.get(alert.alert_id, 0)
        if time.time() - last < self.dedupe_window:
            return False

        return True

    # ── 通知 ──────────────────────────────────────────

    def send_alerts(self, alerts):
        """发送告警到飞书

        复用 chatops 的 FeishuNotifier，这里直接构造卡片。
        不依赖 FeishuNotifier 实例方法，因为告警卡片格式不同。
        """
        if not alerts:
            return 0

        # 延迟导入
        # 直接用 urllib 发，走同一个 webhook
        import urllib.request

        from chatops.feishu_notify import FeishuNotifier  # noqa: F401

        webhook_url = config.__dict__.get("FEISHU_WEBHOOK_URL", "")
        if not webhook_url:
            # 从 chatops config 拿
            from chatops import config as chatops_config

            webhook_url = chatops_config.FEISHU_WEBHOOK_URL

        if not webhook_url:
            print("[WARN] 未配置飞书 webhook，跳过告警通知")
            return 0

        sent = 0
        for alert in alerts:
            card = self._build_alert_card(alert)
            payload = json.dumps({"msg_type": "interactive", "card": card}).encode("utf-8")
            req = urllib.request.Request(
                webhook_url,
                data=payload,
                headers={"Content-Type": "application/json"},
                method="POST",
            )
            try:
                with urllib.request.urlopen(req, timeout=10) as resp:
                    body = resp.read().decode()
                result = json.loads(body)
                if result.get("code", 0) == 0:
                    sent += 1
                    print(f"[INFO] 告警已发送: [{alert.level}] {alert.title}")
                else:
                    print(f"[WARN] 告警发送失败: {result.get('msg', body)}", file=sys.stderr)
            except Exception as e:
                print(f"[WARN] 告警发送异常: {e}", file=sys.stderr)

        return sent

    @staticmethod
    def _build_alert_card(alert):
        """构建飞书告警卡片"""
        color = config.LEVEL_COLOR.get(alert.level, "blue")
        emoji = config.LEVEL_EMOJI.get(alert.level, "ℹ️")

        fields = [
            {
                "is_short": True,
                "text": {"tag": "lark_md", "content": f"**等级**\n{alert.level}"},
            },
            {
                "is_short": True,
                "text": {"tag": "lark_md", "content": f"**来源**\n{alert.source}"},
            },
            {
                "is_short": False,
                "text": {"tag": "lark_md", "content": f"**详情**\n{alert.description}"},
            },
        ]

        return {
            "header": {
                "title": {"tag": "plain_text", "content": f"{emoji} Runner监控告警: {alert.title}"},
                "status": color,
            },
            "elements": [
                {"tag": "div", "fields": fields},
                {
                    "tag": "note",
                    "elements": [
                        {
                            "tag": "plain_text",
                            "content": f"告警ID: {alert.alert_id} | {alert.timestamp[:19].replace('T', ' ')}",
                        }
                    ],
                },
            ],
        }

    # ── 快照 ──────────────────────────────────────────

    def generate_snapshot(self, alerts=None):
        """生成完整的监控快照

        Returns:
            dict: 快照数据
        """
        status_result = self.status_checker.run_full_check()
        metrics = self.metrics.collect_all()

        if alerts is None:
            alerts = self.detect_alerts()

        snapshot = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "runner_summary": status_result["runner_summary"],
            "offline_runners": status_result["offline_runners"],
            "queue_backlog": status_result["queue_backlog"],
            "system_metrics": metrics,
            "active_alerts": [a.to_dict() for a in alerts],
            "alert_history_count": len(self._alert_history),
        }

        return snapshot

    def save_snapshot(self, output_dir=None):
        """保存快照到文件"""
        from runner_monitor.runner_status import RunnerStatusChecker as RSC

        snapshot = self.generate_snapshot()

        if output_dir is None:
            output_dir = config.OUTPUT_DIR

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

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

        # 清理旧快照
        RSC._cleanup_old_snapshots(output_dir, keep=24)

        return filepath

    # ── 单次检查 ──────────────────────────────────────

    def run_once(self):
        """执行一次完整检查 + 告警 + 快照

        Returns:
            dict: {alerts_count, sent_count, snapshot_path}
        """
        alerts = self.detect_alerts()
        sent = self.send_alerts(alerts)
        snapshot_path = self.save_snapshot()

        return {
            "alerts_detected": len(alerts),
            "alerts_sent": sent,
            "snapshot_path": snapshot_path,
            "alerts": [a.to_dict() for a in alerts],
        }


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


def main():
    parser = argparse.ArgumentParser(description="Runner 监控告警调度器")
    parser.add_argument("--check", action="store_true", help="执行一次检查")
    parser.add_argument("--snapshot", action="store_true", help="生成快照")
    parser.add_argument("--daemon", action="store_true", help="持续运行模式")
    parser.add_argument("--dry-run", action="store_true", help="只检测不发通知")
    parser.add_argument("--interval", type=int, help="检测间隔（秒），覆盖环境变量")

    args = parser.parse_args()

    if args.interval:
        config.CHECK_INTERVAL = args.interval

    manager = AlertManager()

    if args.daemon:
        print(f"[INFO] Runner 监控告警服务启动，检测间隔 {config.CHECK_INTERVAL} 秒")
        print(f"[INFO] P1告警: {'开启' if config.P1_ENABLED else '关闭'}")
        print(f"[INFO] P2告警: {'开启' if config.P2_ENABLED else '关闭'}")
        print(f"[INFO] 去重窗口: {config.DEDUPE_WINDOW} 秒")

        while True:
            try:
                result = (
                    manager.run_once()
                    if not args.dry_run
                    else {
                        "alerts_detected": len(manager.detect_alerts()),
                        "alerts_sent": 0,
                    }
                )
                now = time.strftime("%Y-%m-%d %H:%M:%S")
                print(
                    f"[{now}] 检测完成 - "
                    f"发现 {result['alerts_detected']} 个告警, "
                    f"发送 {result['alerts_sent']} 条通知"
                )
            except Exception as e:
                print(f"[ERROR] 检测异常: {e}", file=sys.stderr)

            time.sleep(config.CHECK_INTERVAL)

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

    elif args.check or args.dry_run:
        if args.dry_run:
            alerts = manager.detect_alerts()
            print(f"检测到 {len(alerts)} 个告警（dry-run，不发送）:")
            for a in alerts:
                print(f"  [{a.level}] {a.title}")
                print(f"       {a.description[:100]}")
        else:
            result = manager.run_once()
            print(json.dumps(result, indent=2, ensure_ascii=False))
    else:
        parser.print_help()
        return 1

    return 0


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