awoooi/apps/api/src/jobs/capacity_scanner_job.py

"""
Capacity Scanner Job — ADR-090 § Phase 4 NemoTron 容量巡檢 MVP
===============================================================
每日 02:00 Taipei 從 Prometheus 撈 node metrics → 寫 host_capacity_snapshot.

職責邊界:
  ✅ 撈 Prometheus node_exporter metrics (load / cpu / mem / swap)
  ✅ 為每個 host 寫一筆 host_capacity_snapshot + heuristic ai_verdict
  ✅ 超過硬閾值寫 capacity_violation_event
  ✅ 寫 automation_operation_log(capacity_recommendation)
  ❌ 不做 Holt-Winters 預測 (那是 Hermes 後續階段)
  ❌ 不自動執行修復 (只 recommend,統帥決策)

設計鐵律:
  - 每日一次 snapshot (歷史 30d 供 AI 趨勢分析)
  - ai_verdict heuristic: cpu>80 or mem>85 → critical; >60/70 → warning; else safe
  - Prometheus 失敗 → log + skip 該 host,不 crash 整 loop

資料來源:
  - PROMETHEUS_URL/api/v1/query (instant query)
  - 預期 instance 格式: '192.168.0.XXX:9100' 或 hostname

排程:
  - 首次延遲 120s
  - 後續每日 02:00 (Taipei) 對齊跑

2026-04-19 ogt + Claude Opus 4.7 (1M context) Asia/Taipei
ADR-090 § Phase 4
"""
from __future__ import annotations

import asyncio
import json as _json
import time as _time
from datetime import datetime, timedelta, timezone
from typing import Any

import httpx
import structlog

from src.core.config import settings

logger = structlog.get_logger(__name__)

# ============================================================================
# 排程 / 閾值
# ============================================================================
_FIRST_DELAY_SEC = 120
_HTTP_TIMEOUT_SEC = 10
_LOOP_BACKOFF_SEC = 1800

# Taipei = UTC+8,每日 02:00 Taipei = 18:00 UTC 前一天
_DAILY_TRIGGER_HOUR_TAIPEI = 2

# Heuristic 閾值 (ai_verdict 計算)
_CPU_CRITICAL = 80.0
_CPU_WARNING = 60.0
_MEM_CRITICAL = 85.0
_MEM_WARNING = 70.0
_SWAP_CRITICAL = 50.0
_LOAD1_CRITICAL_RATIO = 2.0  # load1 > 2x CPU cores = critical

# Prometheus 查詢 (instant query,每筆 host 一個 label)
_PROM_QUERIES = {
    "load1": "avg by(instance) (node_load1)",
    "load5": "avg by(instance) (node_load5)",
    "load15": "avg by(instance) (node_load15)",
    "cpu_used_pct": "100 - (avg by(instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[5m])) * 100)",
    "cpu_iowait_pct": "avg by(instance) (rate(node_cpu_seconds_total{mode=\"iowait\"}[5m])) * 100",
    "mem_used_pct": "(1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes) * 100",
    "swap_used_pct": "(node_memory_SwapTotal_bytes - node_memory_SwapFree_bytes) / (node_memory_SwapTotal_bytes > 0 or vector(1)) * 100",
}


# ============================================================================
# Public entry — main.py lifespan 呼叫
# ============================================================================

async def run_capacity_scanner_loop() -> None:
    """每日 02:00 Taipei 跑一次容量巡檢."""
    logger.info("capacity_scanner_loop_started")
    await asyncio.sleep(_FIRST_DELAY_SEC)

    while True:
        try:
            await scan_once()
        except Exception as e:
            logger.exception("capacity_scanner_loop_error", error=str(e))
            await asyncio.sleep(_LOOP_BACKOFF_SEC)
            continue

        # 算下次 02:00 Taipei 的 sleep 秒數
        sleep_sec = _seconds_until_next_trigger()
        logger.info("capacity_scanner_next_tick", sleep_sec=sleep_sec)
        await asyncio.sleep(sleep_sec)


async def scan_once(triggered_by: str = "cron") -> dict[str, int]:
    """執行一次容量巡檢,每 host 寫一筆 snapshot."""
    started_ms = _time.time()
    stats = {"hosts_scanned": 0, "violations": 0}
    error_msg: str | None = None

    try:
        metrics_by_host = await _fetch_all_metrics()
        for host, m in metrics_by_host.items():
            snapshot_id = await _write_snapshot(host, m)
            if snapshot_id:
                stats["hosts_scanned"] += 1
                viol = await _check_and_write_violations(host, m)
                stats["violations"] += viol
    except Exception as e:
        error_msg = f"{type(e).__name__}: {e}"[:1000]
        logger.exception("capacity_scan_once_failed", error=error_msg)

    duration_ms = int((_time.time() - started_ms) * 1000)
    await _log_aol(stats=stats, duration_ms=duration_ms, triggered_by=triggered_by, error=error_msg)

    logger.info(
        "capacity_scan_once_done",
        hosts=stats["hosts_scanned"],
        violations=stats["violations"],
        duration_ms=duration_ms,
    )
    return stats


# ============================================================================
# Prometheus 撈資料
# ============================================================================

async def _fetch_all_metrics() -> dict[str, dict[str, float]]:
    """
    對每個 _PROM_QUERIES 跑 instant query,回傳 {host: {metric: value}}.

    host 來自 query 結果 label 'instance' 的 IP 前綴 (去掉 :9100).
    """
    url = f"{settings.PROMETHEUS_URL.rstrip('/')}/api/v1/query"
    results: dict[str, dict[str, float]] = {}

    async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT_SEC, trust_env=False) as client:
        for metric_name, promql in _PROM_QUERIES.items():
            try:
                resp = await client.get(url, params={"query": promql})
                resp.raise_for_status()
                data = resp.json()
                if data.get("status") != "success":
                    logger.warning("prom_query_non_success", metric=metric_name)
                    continue
                for r in (data.get("data", {}) or {}).get("result", []) or []:
                    instance = (r.get("metric", {}) or {}).get("instance", "")
                    host = instance.split(":")[0] if instance else "unknown"
                    val = r.get("value", [None, None])
                    if val and len(val) >= 2:
                        try:
                            results.setdefault(host, {})[metric_name] = float(val[1])
                        except (ValueError, TypeError):
                            pass
            except Exception as e:
                logger.warning("prom_query_failed", metric=metric_name, error=str(e))
                continue
    return results


# ============================================================================
# DB 寫入
# ============================================================================

async def _write_snapshot(host: str, m: dict[str, float]) -> int | None:
    """寫 host_capacity_snapshot,回傳 snapshot_id."""
    if not host or host == "unknown":
        return None

    verdict, reasoning = _assess_verdict(m)

    try:
        from sqlalchemy import text as _sql
        from src.db.base import get_db_context

        async with get_db_context() as db:
            row = await db.execute(
                _sql("""
                    INSERT INTO host_capacity_snapshot (
                        host, captured_at,
                        load1, load5, load15,
                        cpu_used_pct, cpu_iowait_pct,
                        mem_used_pct, swap_used_pct,
                        ai_verdict, ai_reasoning,
                        written_by_agent
                    ) VALUES (
                        :host, NOW(),
                        :l1, :l5, :l15,
                        :cpu, :iowait,
                        :mem, :swap,
                        :verdict, :reason,
                        'capacity_scanner'
                    )
                    RETURNING snapshot_id
                """),
                {
                    "host": host,
                    "l1": m.get("load1"),
                    "l5": m.get("load5"),
                    "l15": m.get("load15"),
                    "cpu": m.get("cpu_used_pct"),
                    "iowait": m.get("cpu_iowait_pct"),
                    "mem": m.get("mem_used_pct"),
                    "swap": m.get("swap_used_pct"),
                    "verdict": verdict,
                    "reason": reasoning[:500],
                },
            )
            sid = row.scalar()
            return int(sid) if sid else None
    except Exception as e:
        logger.warning("capacity_snapshot_write_failed", host=host, error=str(e))
        return None


async def _check_and_write_violations(host: str, m: dict[str, float]) -> int:
    """超過硬閾值時寫 capacity_violation_event,回傳新增筆數."""
    from sqlalchemy import text as _sql
    from src.db.base import get_db_context

    violations: list[tuple[str, float, float]] = []
    cpu = m.get("cpu_used_pct")
    mem = m.get("mem_used_pct")
    swap = m.get("swap_used_pct")
    if cpu is not None and cpu > _CPU_CRITICAL:
        violations.append(("cpu_over_threshold", _CPU_CRITICAL, cpu))
    if mem is not None and mem > _MEM_CRITICAL:
        violations.append(("mem_over_threshold", _MEM_CRITICAL, mem))
    if swap is not None and swap > _SWAP_CRITICAL:
        violations.append(("swap_over_threshold", _SWAP_CRITICAL, swap))

    if not violations:
        return 0

    written = 0
    try:
        async with get_db_context() as db:
            for vtype, threshold, actual in violations:
                await db.execute(
                    _sql("""
                        INSERT INTO capacity_violation_event (
                            host, violation_type, threshold, actual_value,
                            detected_at
                        ) VALUES (
                            :host, :vt, :th, :av,
                            NOW()
                        )
                    """),
                    {"host": host, "vt": vtype, "th": threshold, "av": actual},
                )
                written += 1
    except Exception as e:
        logger.warning("capacity_violation_write_failed", host=host, error=str(e))
    return written


async def _log_aol(stats: dict[str, int], duration_ms: int, triggered_by: str, error: str | None) -> None:
    """寫 aol(capacity_recommendation)."""
    try:
        from sqlalchemy import text as _sql
        from src.db.base import get_db_context

        aol_status = "failed" if error else "success"
        input_payload = {"triggered_by": triggered_by, "source": "prometheus_node_exporter"}
        output_payload = dict(stats)

        async with get_db_context() as db:
            await db.execute(
                _sql("""
                    INSERT INTO automation_operation_log (
                        operation_type, actor, status,
                        input, output, duration_ms, error, tags
                    ) VALUES (
                        'capacity_recommendation',
                        'capacity_scanner',
                        :st,
                        CAST(:input AS jsonb),
                        CAST(:output AS jsonb),
                        :dur, :err, :tags
                    )
                """),
                {
                    "st": aol_status,
                    "input": _json.dumps(input_payload, ensure_ascii=False),
                    "output": _json.dumps(output_payload, ensure_ascii=False),
                    "dur": duration_ms,
                    "err": (error or "")[:2000] if error else None,
                    "tags": ["capacity", "scanner", "prometheus"],
                },
            )
    except Exception as e:
        logger.warning("capacity_aol_write_failed", error=str(e))


# ============================================================================
# Heuristic + 時間計算
# ============================================================================

def _assess_verdict(m: dict[str, float]) -> tuple[str, str]:
    """根據閾值給 ai_verdict (safe/warning/critical) + reasoning."""
    reasons = []
    max_level = 0  # 0=safe 1=warning 2=critical

    cpu = m.get("cpu_used_pct")
    mem = m.get("mem_used_pct")
    swap = m.get("swap_used_pct")

    if cpu is not None:
        if cpu > _CPU_CRITICAL:
            max_level = max(max_level, 2); reasons.append(f"cpu={cpu:.1f}% (>{_CPU_CRITICAL})")
        elif cpu > _CPU_WARNING:
            max_level = max(max_level, 1); reasons.append(f"cpu={cpu:.1f}% (>{_CPU_WARNING})")
    if mem is not None:
        if mem > _MEM_CRITICAL:
            max_level = max(max_level, 2); reasons.append(f"mem={mem:.1f}% (>{_MEM_CRITICAL})")
        elif mem > _MEM_WARNING:
            max_level = max(max_level, 1); reasons.append(f"mem={mem:.1f}% (>{_MEM_WARNING})")
    if swap is not None and swap > _SWAP_CRITICAL:
        max_level = max(max_level, 2); reasons.append(f"swap={swap:.1f}% (>{_SWAP_CRITICAL})")

    verdict = ("safe", "warning", "critical")[max_level]
    reasoning = "; ".join(reasons) if reasons else "all metrics within safe range"
    return verdict, reasoning


def _seconds_until_next_trigger() -> float:
    """算到下個 02:00 Taipei 的秒數."""
    tz_taipei = timezone(timedelta(hours=8))
    now = datetime.now(tz_taipei)
    today_trigger = now.replace(hour=_DAILY_TRIGGER_HOUR_TAIPEI, minute=0, second=0, microsecond=0)
    if now >= today_trigger:
        today_trigger = today_trigger + timedelta(days=1)
    delta = (today_trigger - now).total_seconds()
    # 上限保護: 至少 300s,至多 25h
    return max(300.0, min(delta, 25 * 3600))