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

#!/usr/bin/env python3
"""
Runner 系统指标采集 - CPU/内存/磁盘（通过 SSH 连接构建服务器）

⚠️ 第一版：骨架 + 接口定义，SSH 实装后续补充
原因：跨机器 SSH 需要密钥管理和网络权限，先把监控框架搭好。

接口约定（与 alert_manager 对接）:
  metrics = RunnerMetricsCollector().collect_all()
  # 返回: [{"host": "...", "cpu_percent": 75.2, "mem_percent": 80.1, "disk_percent": 65.0,
  #         "disk_total_gb": 500, "disk_used_gb": 325, "status": "ok"}, ...]

当 SSH 不可用时，返回空列表，不影响其他监控功能。
"""

from runner_monitor import config


class RunnerMetricsCollector:
    """Runner 系统指标采集器

    第一版：返回空数据（SSH 实装待后续迭代）
    接口已定义好，alert_manager 直接消费。
    """

    def __init__(self, ssh_hosts=None, ssh_key_path=None, ssh_user=None):
        self.ssh_hosts = ssh_hosts or config.SSH_HOSTS
        self.ssh_key_path = ssh_key_path or config.SSH_KEY_PATH
        self.ssh_user = ssh_user or config.SSH_USER

    def collect_all(self):
        """采集所有 runner 的系统指标

        Returns:
            list[dict]: 每台机器的指标数据
        """
        if not self.ssh_hosts:
            # 没有配置 SSH 主机，返回空列表
            return []

        results = []
        for host in self.ssh_hosts:
            try:
                metrics = self._collect_one(host)
                results.append(metrics)
            except Exception as e:
                results.append(
                    {
                        "host": host,
                        "status": "error",
                        "error": str(e),
                    }
                )

        return results

    def _collect_one(self, host):
        """采集单台机器的指标（SSH 实装待后续）

        当前直接返回 not_available 状态。
        后续实现方案：用 paramiko 或 subprocess + ssh 命令执行：
          - top / mpstat  取 CPU
          - free          取内存
          - df -h         取磁盘
        """
        return {
            "host": host,
            "status": "not_available",
            "cpu_percent": None,
            "mem_percent": None,
            "disk_percent": None,
            "disk_total_gb": None,
            "disk_used_gb": None,
            "note": "SSH metrics collection not implemented yet",
        }

    # ── 便捷方法 ──────────────────────────────────────

    @staticmethod
    def is_available():
        """是否有可用的指标采集（SSH 已配置）"""
        return bool(config.SSH_HOSTS and config.SSH_KEY_PATH)