fix(webhooks): Alertmanager 端點完整流程 (LLM + Telegram)
原問題：/alertmanager 只寫 Redis Stream，沒有觸發 Telegram 修正：遵循 phase5_telemetry_architecture.md 原始架構 流程：Alertmanager → Alert Normalizer → Fingerprint → LLM → Telegram Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
@@ -1018,20 +1018,24 @@ def is_internal_ip(client_ip: str) -> bool:
|
||||
|
||||
@router.post(
|
||||
"/alertmanager",
|
||||
response_model=SignalResponse,
|
||||
response_model=AlertResponse,
|
||||
summary="Phase 10: Alertmanager 原生格式 (內網免 HMAC)",
|
||||
description="接收 Alertmanager Webhook，內網來源免 HMAC 驗證。",
|
||||
description="接收 Alertmanager Webhook，內網來源免 HMAC 驗證，觸發 LLM 分析 + Telegram 通知。",
|
||||
)
|
||||
async def alertmanager_webhook(
|
||||
request: Request,
|
||||
payload: AlertmanagerPayload,
|
||||
) -> SignalResponse:
|
||||
background_tasks: BackgroundTasks,
|
||||
) -> AlertResponse:
|
||||
"""
|
||||
接收 Alertmanager 格式告警並轉換為 Signal
|
||||
接收 Alertmanager 格式告警並觸發完整 AWOOOI 流程
|
||||
|
||||
原始架構流程 (phase5_telemetry_architecture.md):
|
||||
Alertmanager → /alertmanager → Alert Normalizer → Fingerprint → LLM → Telegram
|
||||
|
||||
安全策略:
|
||||
- 內網 IP (192.168.x.x, 10.x.x.x): 免 HMAC
|
||||
- 外網 IP: 拒絕 (需使用 /signals 端點)
|
||||
- 內網 IP (192.168.x.x, 10.x.x.x, 172.x.x.x): 免 HMAC
|
||||
- 外網 IP: 拒絕
|
||||
"""
|
||||
# 取得客戶端 IP
|
||||
client_ip = request.client.host if request.client else "unknown"
|
||||
@@ -1043,11 +1047,11 @@ async def alertmanager_webhook(
|
||||
logger.warning(
|
||||
"alertmanager_external_rejected",
|
||||
client_ip=actual_ip,
|
||||
reason="External IP must use /signals with HMAC",
|
||||
reason="External IP must use /alerts with HMAC",
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="External sources must use /signals endpoint with HMAC signature",
|
||||
detail="External sources must use /alerts endpoint with HMAC signature",
|
||||
)
|
||||
|
||||
logger.info(
|
||||
@@ -1057,42 +1061,245 @@ async def alertmanager_webhook(
|
||||
alert_count=len(payload.alerts),
|
||||
)
|
||||
|
||||
# 處理每個告警
|
||||
message_ids = []
|
||||
for alert in payload.alerts:
|
||||
if alert.status != "firing":
|
||||
continue # 只處理 firing 狀態
|
||||
|
||||
# 轉換為 SignalPayload
|
||||
severity_map = {"critical": "critical", "warning": "warning", "info": "info"}
|
||||
severity = severity_map.get(
|
||||
alert.labels.get("severity", "warning").lower(),
|
||||
"warning"
|
||||
# 只處理第一個 firing 告警 (避免告警風暴)
|
||||
firing_alerts = [a for a in payload.alerts if a.status == "firing"]
|
||||
if not firing_alerts:
|
||||
return AlertResponse(
|
||||
success=True,
|
||||
message="No firing alerts to process",
|
||||
alert_id="",
|
||||
approval_created=False,
|
||||
)
|
||||
|
||||
signal = SignalPayload(
|
||||
source="alertmanager",
|
||||
alert_name=alert.labels.get("alertname", "UnknownAlert"),
|
||||
severity=severity,
|
||||
namespace=alert.labels.get("namespace", "default"),
|
||||
target=alert.labels.get("pod", alert.labels.get("instance", "unknown")),
|
||||
message=alert.annotations.get("summary", alert.annotations.get("description", "")),
|
||||
labels=alert.labels,
|
||||
annotations=alert.annotations,
|
||||
)
|
||||
alert = firing_alerts[0]
|
||||
alert_id = f"alert-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
|
||||
|
||||
try:
|
||||
message_id = await produce_signal_to_stream(signal)
|
||||
message_ids.append(message_id)
|
||||
except Exception as e:
|
||||
logger.error("alertmanager_signal_produce_error", error=str(e))
|
||||
# ==========================================================================
|
||||
# Alert Normalizer: 轉換 Alertmanager 格式 → AWOOOI AlertPayload
|
||||
# ==========================================================================
|
||||
alertname = alert.labels.get("alertname", "UnknownAlert")
|
||||
|
||||
return SignalResponse(
|
||||
success=True,
|
||||
message_id=message_ids[0] if message_ids else None,
|
||||
stream=SIGNAL_STREAM_KEY,
|
||||
# 映射 alertname → alert_type
|
||||
alertname_to_type = {
|
||||
"KubePodCrashLooping": "k8s_pod_crash",
|
||||
"KubePodNotReady": "k8s_pod_crash",
|
||||
"KubeNodeNotReady": "k8s_node_failure",
|
||||
"KubeNodeUnreachable": "k8s_node_failure",
|
||||
"HighCPUUsage": "high_cpu",
|
||||
"HighMemoryUsage": "high_memory",
|
||||
"DiskSpaceLow": "disk_full",
|
||||
"SSLCertExpiringSoon": "ssl_expiry",
|
||||
"TargetDown": "service_404",
|
||||
}
|
||||
alert_type = alertname_to_type.get(alertname, "custom")
|
||||
|
||||
severity_map = {"critical": "critical", "warning": "warning", "info": "info"}
|
||||
severity = severity_map.get(
|
||||
alert.labels.get("severity", "warning").lower(),
|
||||
"warning"
|
||||
)
|
||||
|
||||
target_resource = alert.labels.get("pod") or alert.labels.get("instance") or alertname
|
||||
namespace = alert.labels.get("namespace", "default")
|
||||
message = alert.annotations.get("summary") or alert.annotations.get("description") or alertname
|
||||
|
||||
# 建立正規化的 AlertPayload
|
||||
normalized_alert = AlertPayload(
|
||||
alert_type=alert_type,
|
||||
severity=severity,
|
||||
source="alertmanager",
|
||||
target_resource=target_resource,
|
||||
namespace=namespace,
|
||||
message=message,
|
||||
metrics={},
|
||||
labels=alert.labels,
|
||||
)
|
||||
|
||||
# ==========================================================================
|
||||
# 告警指紋 + 收斂
|
||||
# ==========================================================================
|
||||
fingerprint = generate_alert_fingerprint(normalized_alert)
|
||||
|
||||
logger.info(
|
||||
"alertmanager_normalized",
|
||||
alert_id=alert_id,
|
||||
alert_type=alert_type,
|
||||
severity=severity,
|
||||
target=target_resource,
|
||||
fingerprint=fingerprint,
|
||||
)
|
||||
|
||||
try:
|
||||
service = get_approval_service()
|
||||
|
||||
# 查詢是否有同指紋的現有記錄
|
||||
existing_approval = await service.find_by_fingerprint(
|
||||
fingerprint=fingerprint,
|
||||
debounce_minutes=DEBOUNCE_WINDOW_MINUTES,
|
||||
)
|
||||
|
||||
if existing_approval:
|
||||
# 收斂告警 - 跳過 LLM
|
||||
logger.info(
|
||||
"alertmanager_converged",
|
||||
alert_id=alert_id,
|
||||
fingerprint=fingerprint,
|
||||
existing_id=str(existing_approval.id),
|
||||
)
|
||||
|
||||
updated_approval = await service.increment_hit_count(existing_approval.id)
|
||||
if updated_approval:
|
||||
background_tasks.add_task(
|
||||
_push_to_telegram_background,
|
||||
approval_id=str(updated_approval.id),
|
||||
risk_level=updated_approval.risk_level.value,
|
||||
resource_name=target_resource,
|
||||
root_cause=message,
|
||||
suggested_action=updated_approval.action,
|
||||
estimated_downtime="~30s",
|
||||
hit_count=updated_approval.hit_count,
|
||||
primary_responsibility="COLLAB",
|
||||
confidence=0.70,
|
||||
namespace=namespace,
|
||||
)
|
||||
|
||||
return AlertResponse(
|
||||
success=True,
|
||||
message=f"🛡️ 告警收斂 (x{updated_approval.hit_count})",
|
||||
alert_id=alert_id,
|
||||
approval_created=False,
|
||||
approval_id=str(updated_approval.id),
|
||||
risk_level=updated_approval.risk_level.value,
|
||||
suggested_action=updated_approval.action,
|
||||
hit_count=updated_approval.hit_count,
|
||||
converged=True,
|
||||
)
|
||||
|
||||
# ==========================================================================
|
||||
# 新告警 - LLM 分析
|
||||
# ==========================================================================
|
||||
alert_context = {
|
||||
"alert_type": alert_type,
|
||||
"severity": severity,
|
||||
"source": "alertmanager",
|
||||
"target_resource": target_resource,
|
||||
"namespace": namespace,
|
||||
"message": message,
|
||||
"metrics": {},
|
||||
"labels": alert.labels,
|
||||
}
|
||||
|
||||
openclaw = get_openclaw()
|
||||
analysis_result, ai_provider, raw_response, signoz_metrics, signoz_trace_url = await openclaw.analyze_alert(alert_context)
|
||||
|
||||
if analysis_result:
|
||||
risk_level = analysis_result.get("risk_level", "medium")
|
||||
action = analysis_result.get("action", "OBSERVE")
|
||||
root_cause = analysis_result.get("root_cause", message)
|
||||
estimated_downtime = analysis_result.get("estimated_downtime", "~30s")
|
||||
primary_responsibility = analysis_result.get("primary_responsibility", "COLLAB")
|
||||
confidence = analysis_result.get("confidence", 0.5)
|
||||
|
||||
# 建立 ApprovalRecord
|
||||
approval = await service.create_approval(
|
||||
alert_type=alert_type,
|
||||
target=target_resource,
|
||||
action=action,
|
||||
risk_level=risk_level,
|
||||
details={
|
||||
"source": "alertmanager",
|
||||
"alertname": alertname,
|
||||
"namespace": namespace,
|
||||
"message": message,
|
||||
"ai_analysis": analysis_result,
|
||||
"ai_provider": ai_provider,
|
||||
},
|
||||
fingerprint=fingerprint,
|
||||
)
|
||||
|
||||
# 推送 Telegram
|
||||
background_tasks.add_task(
|
||||
_push_to_telegram_background,
|
||||
approval_id=str(approval.id),
|
||||
risk_level=risk_level,
|
||||
resource_name=target_resource,
|
||||
root_cause=root_cause,
|
||||
suggested_action=action,
|
||||
estimated_downtime=estimated_downtime,
|
||||
hit_count=1,
|
||||
primary_responsibility=primary_responsibility,
|
||||
confidence=confidence,
|
||||
namespace=namespace,
|
||||
signoz_rps=signoz_metrics.get("rps", 0) if signoz_metrics else 0,
|
||||
signoz_rps_trend=signoz_metrics.get("rps_trend", "stable") if signoz_metrics else "stable",
|
||||
signoz_error_rate=signoz_metrics.get("error_rate", 0) if signoz_metrics else 0,
|
||||
signoz_p99_latency=signoz_metrics.get("p99_latency", 0) if signoz_metrics else 0,
|
||||
signoz_latency_trend=signoz_metrics.get("latency_trend", "stable") if signoz_metrics else "stable",
|
||||
signoz_trace_url=signoz_trace_url or "",
|
||||
)
|
||||
|
||||
return AlertResponse(
|
||||
success=True,
|
||||
message=f"✅ LLM 分析完成 (via {ai_provider})",
|
||||
alert_id=alert_id,
|
||||
approval_created=True,
|
||||
approval_id=str(approval.id),
|
||||
risk_level=risk_level,
|
||||
suggested_action=action,
|
||||
hit_count=1,
|
||||
converged=False,
|
||||
)
|
||||
else:
|
||||
# LLM 失敗 - 使用預設值
|
||||
approval = await service.create_approval(
|
||||
alert_type=alert_type,
|
||||
target=target_resource,
|
||||
action="OBSERVE",
|
||||
risk_level="medium",
|
||||
details={
|
||||
"source": "alertmanager",
|
||||
"alertname": alertname,
|
||||
"namespace": namespace,
|
||||
"message": message,
|
||||
"ai_analysis": None,
|
||||
"ai_error": "LLM analysis failed",
|
||||
},
|
||||
fingerprint=fingerprint,
|
||||
)
|
||||
|
||||
background_tasks.add_task(
|
||||
_push_to_telegram_background,
|
||||
approval_id=str(approval.id),
|
||||
risk_level="medium",
|
||||
resource_name=target_resource,
|
||||
root_cause=message,
|
||||
suggested_action="OBSERVE",
|
||||
estimated_downtime="unknown",
|
||||
hit_count=1,
|
||||
primary_responsibility="HUMAN",
|
||||
confidence=0.0,
|
||||
namespace=namespace,
|
||||
)
|
||||
|
||||
return AlertResponse(
|
||||
success=True,
|
||||
message="⚠️ LLM 分析失敗，使用預設值",
|
||||
alert_id=alert_id,
|
||||
approval_created=True,
|
||||
approval_id=str(approval.id),
|
||||
risk_level="medium",
|
||||
suggested_action="OBSERVE",
|
||||
hit_count=1,
|
||||
converged=False,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error("alertmanager_error", error=str(e))
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to process alert: {str(e)}",
|
||||
) from e
|
||||
|
||||
|
||||
@router.get(
|
||||
"/health",
|
||||
|
||||
Reference in New Issue
Block a user