wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

refactor(router): R4 #129 AlertAnalyzer 遷移至 services 層

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-01 09:27:23 +08:00
parent 5086bafa36
commit 411880842f
3 changed files with 299 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

251
apps/api/src/api/v1/webhooks.py
View File

@@ -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

116
apps/api/src/models/webhook.py Normal file
View File

@@ -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)")

175
apps/api/src/services/alert_analyzer_service.py Normal file
View File

@@ -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",
)