From ac8fa33354aa81f0ccf674a454c5de907def6be6 Mon Sep 17 00:00:00 2001 From: OG T Date: Tue, 24 Mar 2026 19:34:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(webhooks):=20Alertmanager=20=E7=AB=AF?= =?UTF-8?q?=E9=BB=9E=E5=AE=8C=E6=95=B4=E6=B5=81=E7=A8=8B=20(LLM=20+=20Tele?= =?UTF-8?q?gram)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原問題：/alertmanager 只寫 Redis Stream，沒有觸發 Telegram 修正：遵循 phase5_telemetry_architecture.md 原始架構 流程：Alertmanager → Alert Normalizer → Fingerprint → LLM → Telegram Co-Authored-By: Claude Opus 4.5 --- apps/api/src/api/v1/webhooks.py | 283 +++++++++++++++++++++++++++----- 1 file changed, 245 insertions(+), 38 deletions(-) diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index da6e193b4..4303efcc8 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -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",