From 411880842f2a62340a848bdce810a52cbe43a526 Mon Sep 17 00:00:00 2001 From: OG T Date: Wed, 1 Apr 2026 09:27:23 +0800 Subject: [PATCH] =?UTF-8?q?refactor(router):=20R4=20#129=20AlertAnalyzer?= =?UTF-8?q?=20=E9=81=B7=E7=A7=BB=E8=87=B3=20services=20=E5=B1=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR-024 Router 層瘦身 R4: 將業務邏輯從 Router 移出至正確層次。 變更: - 新增 src/models/webhook.py: AlertPayload + AlertResponse 移至 models 層 - 新增 src/services/alert_analyzer_service.py: AlertAnalyzer (141行) 移至 services 層 - RISK_MAPPING / ACTION_MAPPING / BLAST_RADIUS_MAPPING 對應表 - analyze() 方法含 K8s 資源名稱正規化 (ADR-016) - webhooks.py: 移除重複定義，改為 import，-243行 Router 層 webhooks.py 已符合 ADR-024 禁止清單規範： AlertAnalyzer 不再存在於 Router 層。 R4 狀態: #127✅ #128✅ #129✅ #130✅ (全部完成) Co-Authored-By: Claude Sonnet 4.6 --- apps/api/src/api/v1/webhooks.py | 251 +----------------- apps/api/src/models/webhook.py | 116 ++++++++ .../src/services/alert_analyzer_service.py | 175 ++++++++++++ 3 files changed, 299 insertions(+), 243 deletions(-) create mode 100644 apps/api/src/models/webhook.py create mode 100644 apps/api/src/services/alert_analyzer_service.py diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index 58c8f7681..82796888a 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -43,6 +43,9 @@ from src.models.approval import ( RiskLevel, ) from src.models.incident import Incident, IncidentStatus, Severity, Signal +# R4 #129 (2026-04-01 ogt): AlertPayload/AlertResponse 移至 models 層，AlertAnalyzer 移至 services 層 +from src.models.webhook import AlertPayload, AlertResponse +from src.services.alert_analyzer_service import AlertAnalyzer from src.services.approval_db import get_approval_service # Phase 17 P0: Service 層 (消除 Router 直接存取 Redis) @@ -55,8 +58,7 @@ from src.services.signal_producer import SignalData, get_signal_producer # Phase 5: Telegram Gateway (行動戰情室) from src.services.telegram_gateway import TelegramGatewayError, get_telegram_gateway -# Phase 18.1.7: K8s 資源名稱正規化 (ADR-016) -from src.utils.k8s_naming import normalize_resource_name +# Phase 18.1.7: K8s 資源名稱正規化 已移至 alert_analyzer_service (R4 #129) from src.utils.timezone import now_taipei router = APIRouter(prefix="/webhooks", tags=["Webhooks"]) @@ -370,101 +372,8 @@ DEBOUNCE_WINDOW_MINUTES = 5 # Request Models # ============================================================================= -class AlertPayload(BaseModel): - """ - 外部告警 Payload - - 接收來自 Prometheus AlertManager、K8s Event Watcher、Grafana 等 - 外部監控系統的告警通知。 - - OpenClaw AI 會自動分析告警並建立待簽核卡片。 - - Example: - ```json - { - "alert_type": "k8s_pod_crash", - "severity": "critical", - "source": "prometheus", - "target_resource": "harbor-core-7d4b8c9f5-xk2m3", - "namespace": "harbor", - "message": "Pod CrashLoopBackOff detected", - "metrics": {"restart_count": 5, "cpu_percent": 95} - } - ``` - """ - - alert_type: Literal[ - "k8s_node_failure", # K8s 節點故障 - "k8s_pod_crash", # Pod 崩潰 - "db_connection_timeout", # 資料庫連線超時 - "service_404", # 服務 404 錯誤 - "high_cpu", # CPU 飆高 - "high_memory", # 記憶體飆高 - "disk_full", # 磁碟滿 - "ssl_expiry", # SSL 憑證即將過期 - "custom", # 自訂告警 - ] = Field(..., description="告警類型") - - severity: Literal["info", "warning", "critical"] = Field( - "warning", - description="告警嚴重度", - ) - - source: str = Field( - ..., - description="告警來源 (例如: prometheus, k8s-event-watcher)", - ) - - target_resource: str = Field( - ..., - description="受影響的資源 (例如: harbor, nginx-ingress-7d4b8c9f5-xk2m3)", - ) - - namespace: str = Field( - "default", - description="K8s Namespace", - ) - - message: str = Field( - ..., - description="告警訊息", - ) - - metrics: dict | None = Field( - None, - description="相關指標數據 (例如: {cpu_percent: 95, memory_percent: 80})", - ) - - labels: dict | None = Field( - None, - description="告警標籤 (例如: {app: harbor, team: devops})", - ) - - -class AlertResponse(BaseModel): - """ - 告警處理回應 - - 包含 OpenClaw AI 分析結果： - - 風險等級 (risk_level) - - 爆炸半徑 (透過 approval_id 查詢) - - 建議修復腳本 (suggested_action) - - 戰略 B 新增： - - hit_count: 告警聚合次數 - - converged: 是否為收斂的重複告警 - """ - - success: bool = Field(..., description="處理是否成功") - message: str = Field(..., description="處理結果訊息") - alert_id: str | None = Field(None, description="告警唯一識別碼") - approval_created: bool = Field(False, description="是否已建立待簽核卡片") - approval_id: str | None = Field(None, description="待簽核卡片 ID (UUID)") - risk_level: str | None = Field(None, description="AI 判定風險等級 (low/medium/high/critical)") - suggested_action: str | None = Field(None, description="AI 建議修復腳本") - # 戰略 B: 告警風暴收斂 - hit_count: int = Field(1, description="告警聚合次數 (相同指紋觸發次數)") - converged: bool = Field(False, description="是否為收斂的重複告警 (跳過 LLM)") +# AlertPayload 和 AlertResponse 已移至 src/models/webhook.py (R4 #129, 2026-04-01 ogt) +# 由 import 區塊頂部的 from src.models.webhook import ... 引入 # ============================================================================= @@ -618,152 +527,8 @@ async def receive_signal( ) from e -# ============================================================================= -# Agent Logic - 告警分析大腦 -# ============================================================================= - -class AlertAnalyzer: - """ - 告警分析器 - AWOOOI 核心大腦 - - 根據告警類型、嚴重度、相關指標， - 自動判定風險等級、爆炸半徑、處置建議。 - """ - - # 告警類型 → 風險等級映射 - RISK_MAPPING: dict[str, RiskLevel] = { - "k8s_node_failure": RiskLevel.CRITICAL, - "k8s_pod_crash": RiskLevel.MEDIUM, - "db_connection_timeout": RiskLevel.CRITICAL, - "service_404": RiskLevel.MEDIUM, - "high_cpu": RiskLevel.MEDIUM, - "high_memory": RiskLevel.MEDIUM, - "disk_full": RiskLevel.CRITICAL, - "ssl_expiry": RiskLevel.LOW, - "custom": RiskLevel.MEDIUM, - } - - # 告警類型 → 處置建議映射 - ACTION_MAPPING: dict[str, str] = { - "k8s_node_failure": "kubectl drain {resource} --ignore-daemonsets", - "k8s_pod_crash": "kubectl delete pod {resource} -n {namespace}", - "db_connection_timeout": "重啟資料庫連線池並檢查網路", - "service_404": "kubectl rollout restart deployment/{resource} -n {namespace}", - "high_cpu": "kubectl scale deployment/{resource} --replicas=+2 -n {namespace}", - "high_memory": "kubectl delete pod {resource} -n {namespace} (記憶體洩漏清理)", - "disk_full": "清理 /var/log 與 /tmp 目錄", - "ssl_expiry": "更新 SSL 憑證", - "custom": "人工分析處置", - } - - # 告警類型 → 爆炸半徑映射 - BLAST_RADIUS_MAPPING: dict[str, dict] = { - "k8s_node_failure": {"pods": 10, "downtime": "~5 min", "services": ["all-on-node"]}, - "k8s_pod_crash": {"pods": 1, "downtime": "~30s", "services": []}, - "db_connection_timeout": {"pods": 0, "downtime": "~2 min", "services": ["api", "auth"]}, - "service_404": {"pods": 3, "downtime": "~1 min", "services": []}, - "high_cpu": {"pods": 0, "downtime": "0", "services": []}, - "high_memory": {"pods": 1, "downtime": "~30s", "services": []}, - "disk_full": {"pods": 0, "downtime": "~5 min", "services": ["logging"]}, - "ssl_expiry": {"pods": 0, "downtime": "0", "services": ["https"]}, - "custom": {"pods": 0, "downtime": "unknown", "services": []}, - } - - @classmethod - def analyze(cls, alert: AlertPayload) -> ApprovalRequestCreate: - """ - 分析告警並生成 ApprovalRequestCreate - - Phase 18.1.7: 整合 K8s 資源名稱正規化 (ADR-016) - - Returns: - ApprovalRequestCreate 用於建立待簽核卡片 - """ - # Phase 18.1.7: 先正規化資源名稱 - normalized = normalize_resource_name(alert.target_resource, alert.namespace) - resolved_resource = normalized.normalized or alert.target_resource - resolved_namespace = normalized.namespace or alert.namespace - - # 1. 判定風險等級 - base_risk = cls.RISK_MAPPING.get(alert.alert_type, RiskLevel.MEDIUM) - - # 嚴重度提升 - if alert.severity == "critical" and base_risk != RiskLevel.CRITICAL: - risk_level = RiskLevel.CRITICAL - else: - risk_level = base_risk - - # 2. 取得處置建議 (使用正規化後的資源名稱) - action_template = cls.ACTION_MAPPING.get(alert.alert_type, "人工分析處置") - action = action_template.format( - resource=resolved_resource, - namespace=resolved_namespace, - ) - - # 3. 取得爆炸半徑 - blast_info = cls.BLAST_RADIUS_MAPPING.get( - alert.alert_type, - {"pods": 0, "downtime": "unknown", "services": []}, - ) - - # 判定 data_impact - data_impact = DataImpact.NONE - if alert.alert_type in ["db_connection_timeout", "disk_full"]: - data_impact = DataImpact.WRITE - - # 4. 建立 Dry-run 檢查項目 - dry_run_checks = [ - DryRunCheck( - name="權限驗證", - passed=True, - message="cluster-admin", - ), - DryRunCheck( - name="語法驗證", - passed=True, - message=None, - ), - DryRunCheck( - name="告警來源驗證", - passed=True, - message=alert.source, - ), - ] - - # 如果有 metrics，加入 sigma 分析 - if alert.metrics: - cpu = alert.metrics.get("cpu_percent", 0) - sigma = alert.metrics.get("sigma_deviation", 0) - if sigma and abs(sigma) > 2: - dry_run_checks.append( - DryRunCheck( - name="基準線偏差分析", - passed=True, - message=f"CPU: {cpu:.0f}% (σ: {sigma:+.1f})", - ) - ) - - # 5. 組裝 description - description = f"[{alert.alert_type}] {alert.message}" - if alert.metrics: - metrics_str = ", ".join(f"{k}={v}" for k, v in alert.metrics.items()) - description += f" | 指標: {metrics_str}" - - # 6. 建立 ApprovalRequestCreate - return ApprovalRequestCreate( - action=action, - description=description, - risk_level=risk_level, - blast_radius=BlastRadius( - affected_pods=blast_info["pods"], - estimated_downtime=blast_info["downtime"], - related_services=blast_info["services"] + [alert.target_resource], - data_impact=data_impact, - ), - dry_run_checks=dry_run_checks, - requested_by="OpenClaw", - ) - +# AlertAnalyzer 已移至 src/services/alert_analyzer_service.py (R4 #129, 2026-04-01 ogt) +# 由 import 區塊頂部的 from src.services.alert_analyzer_service import ... 引入 # ============================================================================= # Endpoints diff --git a/apps/api/src/models/webhook.py b/apps/api/src/models/webhook.py new file mode 100644 index 000000000..fd2caf0bd --- /dev/null +++ b/apps/api/src/models/webhook.py @@ -0,0 +1,116 @@ +""" +Webhook API Schema - 告警接收 Pydantic 模型 +============================================= + +從 api/v1/webhooks.py 抽取至 models 層 (ADR-024 四層架構) + +設計原則: +- AlertPayload: 外部告警接收格式 (Prometheus, K8s, Alertmanager 等) +- AlertResponse: 告警處理回應格式 +- 不含業務邏輯，純資料結構 + +版本: v1.0 +建立: 2026-04-01 (台北時區) +建立者: Claude Code (R4 Router 瘦身 #129) +""" + +from typing import Literal + +from pydantic import BaseModel, Field + + +class AlertPayload(BaseModel): + """ + 外部告警 Payload + + 接收來自 Prometheus AlertManager、K8s Event Watcher、Grafana 等 + 外部監控系統的告警通知。 + + OpenClaw AI 會自動分析告警並建立待簽核卡片。 + + Example: + ```json + { + "alert_type": "k8s_pod_crash", + "severity": "critical", + "source": "prometheus", + "target_resource": "harbor-core-7d4b8c9f5-xk2m3", + "namespace": "harbor", + "message": "Pod CrashLoopBackOff detected", + "metrics": {"restart_count": 5, "cpu_percent": 95} + } + ``` + """ + + alert_type: Literal[ + "k8s_node_failure", # K8s 節點故障 + "k8s_pod_crash", # Pod 崩潰 + "db_connection_timeout", # 資料庫連線超時 + "service_404", # 服務 404 錯誤 + "high_cpu", # CPU 飆高 + "high_memory", # 記憶體飆高 + "disk_full", # 磁碟滿 + "ssl_expiry", # SSL 憑證即將過期 + "custom", # 自訂告警 + ] = Field(..., description="告警類型") + + severity: Literal["info", "warning", "critical"] = Field( + "warning", + description="告警嚴重度", + ) + + source: str = Field( + ..., + description="告警來源 (例如: prometheus, k8s-event-watcher)", + ) + + target_resource: str = Field( + ..., + description="受影響的資源 (例如: harbor, nginx-ingress-7d4b8c9f5-xk2m3)", + ) + + namespace: str = Field( + "default", + description="K8s Namespace", + ) + + message: str = Field( + ..., + description="告警訊息", + ) + + metrics: dict | None = Field( + None, + description="相關指標數據 (例如: {cpu_percent: 95, memory_percent: 80})", + ) + + labels: dict | None = Field( + None, + description="告警標籤 (例如: {app: harbor, team: devops})", + ) + + +class AlertResponse(BaseModel): + """ + 告警處理回應 + + 包含 OpenClaw AI 分析結果： + - 風險等級 (risk_level) + - 爆炸半徑 (透過 approval_id 查詢) + - 建議修復腳本 (suggested_action) + + 戰略 B 新增： + - hit_count: 告警聚合次數 + - converged: 是否為收斂的重複告警 + """ + + success: bool = Field(..., description="處理是否成功") + message: str = Field(..., description="處理結果訊息") + alert_id: str | None = Field(None, description="告警唯一識別碼") + approval_created: bool = Field(False, description="是否已建立待簽核卡片") + approval_id: str | None = Field(None, description="待簽核卡片 ID (UUID)") + risk_level: str | None = Field(None, description="AI 判定風險等級 (low/medium/high/critical)") + suggested_action: str | None = Field(None, description="AI 建議修復腳本") + # 戰略 B: 告警風暴收斂 + hit_count: int = Field(1, description="告警聚合次數 (相同指紋觸發次數)") + converged: bool = Field(False, description="是否為收斂的重複告警 (跳過 LLM)") diff --git a/apps/api/src/services/alert_analyzer_service.py b/apps/api/src/services/alert_analyzer_service.py new file mode 100644 index 000000000..16379d4e7 --- /dev/null +++ b/apps/api/src/services/alert_analyzer_service.py @@ -0,0 +1,175 @@ +""" +Alert Analyzer Service - 告警分析大腦 +====================================== + +從 api/v1/webhooks.py 抽取至 services 層 (ADR-024 四層架構，R4 #129) + +職責: +- 根據告警類型、嚴重度、相關指標，判定風險等級 +- 計算爆炸半徑 (Blast Radius) +- 組裝 ApprovalRequestCreate + +設計原則: +- 純業務邏輯層，不存取 Redis/DB +- 依賴 K8s 資源名稱正規化工具 (ADR-016) +- 可獨立測試 (無外部依賴) + +版本: v1.0 +建立: 2026-04-01 (台北時區) +建立者: Claude Code (R4 Router 瘦身 #129) +""" + +from src.models.approval import ( + ApprovalRequestCreate, + BlastRadius, + DataImpact, + DryRunCheck, + RiskLevel, +) +from src.models.webhook import AlertPayload +from src.utils.k8s_naming import normalize_resource_name + + +class AlertAnalyzer: + """ + 告警分析器 - AWOOOI 核心大腦 + + 根據告警類型、嚴重度、相關指標， + 自動判定風險等級、爆炸半徑、處置建議。 + + 搬移自: api/v1/webhooks.py (ADR-024 R4 #129, 2026-04-01 ogt) + """ + + # 告警類型 → 風險等級映射 + RISK_MAPPING: dict[str, RiskLevel] = { + "k8s_node_failure": RiskLevel.CRITICAL, + "k8s_pod_crash": RiskLevel.MEDIUM, + "db_connection_timeout": RiskLevel.CRITICAL, + "service_404": RiskLevel.MEDIUM, + "high_cpu": RiskLevel.MEDIUM, + "high_memory": RiskLevel.MEDIUM, + "disk_full": RiskLevel.CRITICAL, + "ssl_expiry": RiskLevel.LOW, + "custom": RiskLevel.MEDIUM, + } + + # 告警類型 → 處置建議映射 + ACTION_MAPPING: dict[str, str] = { + "k8s_node_failure": "kubectl drain {resource} --ignore-daemonsets", + "k8s_pod_crash": "kubectl delete pod {resource} -n {namespace}", + "db_connection_timeout": "重啟資料庫連線池並檢查網路", + "service_404": "kubectl rollout restart deployment/{resource} -n {namespace}", + "high_cpu": "kubectl scale deployment/{resource} --replicas=+2 -n {namespace}", + "high_memory": "kubectl delete pod {resource} -n {namespace} (記憶體洩漏清理)", + "disk_full": "清理 /var/log 與 /tmp 目錄", + "ssl_expiry": "更新 SSL 憑證", + "custom": "人工分析處置", + } + + # 告警類型 → 爆炸半徑映射 + BLAST_RADIUS_MAPPING: dict[str, dict] = { + "k8s_node_failure": {"pods": 10, "downtime": "~5 min", "services": ["all-on-node"]}, + "k8s_pod_crash": {"pods": 1, "downtime": "~30s", "services": []}, + "db_connection_timeout": {"pods": 0, "downtime": "~2 min", "services": ["api", "auth"]}, + "service_404": {"pods": 3, "downtime": "~1 min", "services": []}, + "high_cpu": {"pods": 0, "downtime": "0", "services": []}, + "high_memory": {"pods": 1, "downtime": "~30s", "services": []}, + "disk_full": {"pods": 0, "downtime": "~5 min", "services": ["logging"]}, + "ssl_expiry": {"pods": 0, "downtime": "0", "services": ["https"]}, + "custom": {"pods": 0, "downtime": "unknown", "services": []}, + } + + @classmethod + def analyze(cls, alert: AlertPayload) -> ApprovalRequestCreate: + """ + 分析告警並生成 ApprovalRequestCreate + + Phase 18.1.7: 整合 K8s 資源名稱正規化 (ADR-016) + + Returns: + ApprovalRequestCreate 用於建立待簽核卡片 + """ + # Phase 18.1.7: 先正規化資源名稱 + normalized = normalize_resource_name(alert.target_resource, alert.namespace) + resolved_resource = normalized.normalized or alert.target_resource + resolved_namespace = normalized.namespace or alert.namespace + + # 1. 判定風險等級 + base_risk = cls.RISK_MAPPING.get(alert.alert_type, RiskLevel.MEDIUM) + + # 嚴重度提升 + if alert.severity == "critical" and base_risk != RiskLevel.CRITICAL: + risk_level = RiskLevel.CRITICAL + else: + risk_level = base_risk + + # 2. 取得處置建議 (使用正規化後的資源名稱) + action_template = cls.ACTION_MAPPING.get(alert.alert_type, "人工分析處置") + action = action_template.format( + resource=resolved_resource, + namespace=resolved_namespace, + ) + + # 3. 取得爆炸半徑 + blast_info = cls.BLAST_RADIUS_MAPPING.get( + alert.alert_type, + {"pods": 0, "downtime": "unknown", "services": []}, + ) + + # 判定 data_impact + data_impact = DataImpact.NONE + if alert.alert_type in ["db_connection_timeout", "disk_full"]: + data_impact = DataImpact.WRITE + + # 4. 建立 Dry-run 檢查項目 + dry_run_checks = [ + DryRunCheck( + name="權限驗證", + passed=True, + message="cluster-admin", + ), + DryRunCheck( + name="語法驗證", + passed=True, + message=None, + ), + DryRunCheck( + name="告警來源驗證", + passed=True, + message=alert.source, + ), + ] + + # 如果有 metrics，加入 sigma 分析 + if alert.metrics: + cpu = alert.metrics.get("cpu_percent", 0) + sigma = alert.metrics.get("sigma_deviation", 0) + if sigma and abs(sigma) > 2: + dry_run_checks.append( + DryRunCheck( + name="基準線偏差分析", + passed=True, + message=f"CPU: {cpu:.0f}% (σ: {sigma:+.1f})", + ) + ) + + # 5. 組裝 description + description = f"[{alert.alert_type}] {alert.message}" + if alert.metrics: + metrics_str = ", ".join(f"{k}={v}" for k, v in alert.metrics.items()) + description += f" | 指標: {metrics_str}" + + # 6. 建立 ApprovalRequestCreate + return ApprovalRequestCreate( + action=action, + description=description, + risk_level=risk_level, + blast_radius=BlastRadius( + affected_pods=blast_info["pods"], + estimated_downtime=blast_info["downtime"], + related_services=blast_info["services"] + [alert.target_resource], + data_impact=data_impact, + ), + dry_run_checks=dry_run_checks, + requested_by="OpenClaw", + )