awoooi/ops/monitoring/validate_coverage.py

#!/usr/bin/env python3
"""
AWOOOI 監控覆蓋率驗證
====================
CI/CD 階段執行，確保所有服務都有對應的監控配置

用法:
    python ops/monitoring/validate_coverage.py

退出碼:
    0 - 所有服務都已註冊
    1 - 發現未監控的服務
"""

import subprocess
import sys
import yaml
from pathlib import Path
from typing import NamedTuple


class ValidationResult(NamedTuple):
    """驗證結果"""
    passed: bool
    errors: list[str]
    warnings: list[str]
    coverage: dict[str, float]


def load_registry() -> dict:
    """載入服務註冊表"""
    registry_path = Path(__file__).parent / 'service-registry.yaml'
    with open(registry_path) as f:
        return yaml.safe_load(f)


def get_k8s_deployments() -> list[dict]:
    """取得所有 K8s Deployments"""
    try:
        result = subprocess.run(
            [
                'kubectl', 'get', 'deployments', '-A',
                '-o', 'jsonpath={range .items[*]}{.metadata.namespace}/{.metadata.name}{\"\\n\"}{end}'
            ],
            capture_output=True, text=True, timeout=30
        )
        deployments = []
        for line in result.stdout.strip().split('\n'):
            if line and '/' in line:
                ns, name = line.split('/', 1)
                deployments.append({'namespace': ns, 'name': name})
        return deployments
    except Exception as e:
        print(f"Warning: Cannot get K8s deployments: {e}")
        return []


def get_k8s_services() -> list[dict]:
    """取得所有 K8s Services"""
    try:
        result = subprocess.run(
            [
                'kubectl', 'get', 'services', '-A',
                '-o', 'jsonpath={range .items[*]}{.metadata.namespace}/{.metadata.name}{\"\\n\"}{end}'
            ],
            capture_output=True, text=True, timeout=30
        )
        services = []
        for line in result.stdout.strip().split('\n'):
            if line and '/' in line:
                ns, name = line.split('/', 1)
                services.append({'namespace': ns, 'name': name})
        return services
    except Exception as e:
        print(f"Warning: Cannot get K8s services: {e}")
        return []


def check_docker_containers(host: str) -> list[str]:
    """檢查主機上的 Docker 容器"""
    try:
        result = subprocess.run(
            ['ssh', '-o', 'ConnectTimeout=5', host, 'docker', 'ps', '--format', '{{.Names}}'],
            capture_output=True, text=True, timeout=10
        )
        return [c for c in result.stdout.strip().split('\n') if c]
    except Exception as e:
        print(f"Warning: Cannot check Docker on {host}: {e}")
        return []


def validate_registry(registry: dict) -> ValidationResult:
    """驗證服務註冊表完整性"""
    errors = []
    warnings = []

    registered_services = {s['name'] for s in registry.get('services', [])}
    registered_k8s = {
        (s['namespace'], s['name'])
        for s in registry.get('services', [])
        if s.get('type') == 'k8s-deployment'
    }

    # ==========================================================================
    # 1. 檢查 K8s Deployments
    # ==========================================================================
    k8s_deployments = get_k8s_deployments()
    ignored_namespaces = {'kube-system', 'kube-public', 'kube-node-lease', 'local-path-storage'}
    ignored_prefixes = {'coredns', 'metrics-server', 'local-path-provisioner'}

    for deploy in k8s_deployments:
        ns, name = deploy['namespace'], deploy['name']

        # 跳過系統命名空間
        if ns in ignored_namespaces:
            continue

        # 跳過系統元件
        if any(name.startswith(p) for p in ignored_prefixes):
            continue

        if (ns, name) not in registered_k8s:
            errors.append(f"K8s Deployment '{ns}/{name}' 未在 service-registry.yaml 中註冊")

    # ==========================================================================
    # 2. 檢查 Docker 容器
    # ==========================================================================
    docker_hosts = ['192.168.0.188', '192.168.0.110']
    docker_services = {
        s['name']
        for s in registry.get('services', [])
        if s.get('type') == 'docker'
    }

    ignored_containers = {
        'k3s', 'pause', 'registry', 'nginx', 'traefik',
        # SignOz 相關容器群組
        'signoz-alertmanager', 'signoz-query-service', 'signoz-otel-collector-metrics',
        'zookeeper', 'clickhouse',
        # Sentry 相關容器群組
        'sentry-web', 'sentry-worker', 'sentry-cron', 'sentry-kafka', 'sentry-redis',
        'sentry-postgres', 'sentry-zookeeper', 'sentry-snuba',
    }

    for host in docker_hosts:
        containers = check_docker_containers(host)
        for container in containers:
            if not container:
                continue

            # 跳過已知系統容器
            if any(ignored in container for ignored in ignored_containers):
                continue

            # 提取主要名稱 (去除 _1, -1 等後綴)
            base_name = container.split('_')[0].split('-')[0]

            if container not in docker_services and base_name not in docker_services:
                warnings.append(f"Docker 容器 '{container}' on {host} 未在 registry 中 (可能需要加入)")

    # ==========================================================================
    # 3. 檢查 API 端點覆蓋
    # ==========================================================================
    api_endpoints = registry.get('api_endpoints', [])
    critical_endpoints = [e for e in api_endpoints if e.get('critical')]

    if len(critical_endpoints) < 5:
        warnings.append(f"僅定義了 {len(critical_endpoints)} 個關鍵 API 端點，建議至少 5 個")

    # ==========================================================================
    # 4. 檢查前端頁面覆蓋
    # ==========================================================================
    pages = registry.get('pages', [])
    if len(pages) < 3:
        warnings.append(f"僅定義了 {len(pages)} 個前端頁面監控，建議至少 3 個")

    # ==========================================================================
    # 5. 計算覆蓋率
    # ==========================================================================
    services = registry.get('services', [])
    total = len(services) if services else 1

    coverage = {
        'prometheus_source_declared': sum(
            1
            for s in services
            if (
                s.get('monitoring', {}).get('prometheus')
                or s.get('monitoring', {}).get('blackbox_target_enabled')
                or s.get('monitoring', {}).get('agent_verifier')
            )
        ) / total,
        'prometheus_runtime_verified': sum(
            1
            for s in services
            if s.get('monitoring', {}).get('runtime_verified') is True
        ) / total,
        'sentry': sum(1 for s in services if s.get('monitoring', {}).get('sentry')) / total,
        'otel': sum(1 for s in services if s.get('monitoring', {}).get('otel')) / total,
        'alerts': sum(1 for s in services if s.get('alerts')) / total,
        'auto_repair': sum(1 for s in services if s.get('auto_repair', {}).get('enabled')) / total,
    }

    # 覆蓋率低於 80% 產生警告
    for metric, rate in coverage.items():
        if rate < 0.8:
            warnings.append(f"{metric} 覆蓋率僅 {rate:.0%}，建議提升至 80% 以上")

    passed = len(errors) == 0
    return ValidationResult(passed=passed, errors=errors, warnings=warnings, coverage=coverage)


def print_report(result: ValidationResult):
    """輸出驗證報告"""
    print("\n" + "=" * 60)
    print("AWOOOI 監控覆蓋率驗證報告")
    print("=" * 60)

    # 覆蓋率
    print("\n📊 覆蓋率:")
    for metric, rate in result.coverage.items():
        status = "✅" if rate >= 0.8 else "⚠️" if rate >= 0.5 else "❌"
        print(f"   {status} {metric}: {rate:.0%}")

    # 錯誤
    if result.errors:
        print(f"\n❌ 錯誤 ({len(result.errors)}):")
        for err in result.errors:
            print(f"   • {err}")

    # 警告
    if result.warnings:
        print(f"\n⚠️ 警告 ({len(result.warnings)}):")
        for warn in result.warnings:
            print(f"   • {warn}")

    # 結論
    print("\n" + "-" * 60)
    if result.passed:
        print("✅ 驗證通過 - 所有關鍵服務都已註冊監控")
    else:
        print("❌ 驗證失敗 - 請更新 ops/monitoring/service-registry.yaml")
    print("=" * 60 + "\n")


def main():
    """主函數"""
    registry = load_registry()
    result = validate_registry(registry)
    print_report(result)

    # 錯誤時退出碼 1
    if not result.passed:
        sys.exit(1)

    sys.exit(0)


if __name__ == '__main__':
    main()