feat(awooop): add provider upstream canary

2026-05-20 20:48:36 +08:00
parent e6cc008b87
commit f3fbd39898
8 changed files with 719 additions and 12 deletions
--- a/apps/api/src/api/v1/sentry_webhook.py
+++ b/apps/api/src/api/v1/sentry_webhook.py
@@ -14,12 +14,15 @@ AWOOOI API - Sentry Webhook Handler
 🔴 HARD RULE: 時間顯示使用 Asia/Taipei (UTC+8)
 """

+import json
 import uuid
+from typing import Any

 import structlog
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
 from pydantic import BaseModel

+from src.core.awooop_operator_auth import authenticate_awooop_operator_headers
 from src.core.circuit_breaker import get_openclaw_guard
 from src.core.metrics import (
    record_alert_chain_failure,
@@ -38,6 +41,7 @@ from src.services.approval_db import get_approval_service
 from src.services.channel_hub import record_external_alert_event
 from src.services.openclaw_http_service import get_openclaw_http_service
 from src.services.sentry_service import get_sentry_service
+
 # 2026-04-27 P3.1-T2 by Claude — Tier-2 三服務感知強化：補 SentryWebhookService 簽章驗證
 from src.services.sentry_webhook_service import (
    SentrySignatureError,
@@ -88,6 +92,114 @@ async def sentry_webhook_health() -> dict:
    return {"status": "ok", "webhook": "sentry"}


+def _sentry_event_tag(event_data: dict[str, Any], key: str) -> str | None:
+    tags = event_data.get("tags") or []
+    for tag in tags:
+        if isinstance(tag, list | tuple) and len(tag) >= 2 and str(tag[0]) == key:
+            return str(tag[1])
+        if isinstance(tag, dict) and str(tag.get("key")) == key:
+            value = tag.get("value")
+            return str(value) if value is not None else None
+    return None
+
+
+def _is_sentry_upstream_canary(payload: dict[str, Any]) -> bool:
+    data = payload.get("data") if isinstance(payload, dict) else None
+    if not isinstance(data, dict) or payload.get("action") != "triggered":
+        return False
+    issue_data = data.get("issue") if isinstance(data.get("issue"), dict) else {}
+    event_data = data.get("event") if isinstance(data.get("event"), dict) else {}
+    issue_id = str(issue_data.get("id") or "")
+    short_id = str(issue_data.get("shortId") or "")
+    title = str(issue_data.get("title") or "")
+    return (
+        issue_id.startswith("awoooi-canary-")
+        or short_id.upper().startswith("AWOOOI-CANARY")
+        or title == "AwoooPSourceProviderCanary"
+        or (_sentry_event_tag(event_data, "awoooi_canary") or "").lower() == "true"
+    )
+
+
+async def _record_sentry_upstream_canary(
+    payload: dict[str, Any],
+    request: Request,
+) -> dict[str, Any]:
+    operator = authenticate_awooop_operator_headers(
+        request.headers.get("x-awooop-operator-id"),
+        request.headers.get("x-awooop-operator-key"),
+    )
+    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
+    issue_data = data.get("issue") if isinstance(data.get("issue"), dict) else {}
+    event_data = data.get("event") if isinstance(data.get("event"), dict) else {}
+    issue_id = str(
+        issue_data.get("id")
+        or issue_data.get("shortId")
+        or _sentry_event_tag(event_data, "run_ref")
+        or "awoooi-canary-unknown"
+    )
+    source_url = (
+        issue_data.get("permalink")
+        or issue_data.get("web_url")
+        or issue_data.get("url")
+    )
+    event_uuid = await record_external_alert_event(
+        project_id="awoooi",
+        provider="sentry",
+        event_id=issue_id,
+        stage="upstream_canary",
+        title=str(issue_data.get("title") or "AwoooPSourceProviderCanary"),
+        severity=str(issue_data.get("level") or "info"),
+        namespace="awoooi-prod",
+        target_resource=str(issue_data.get("culprit") or "source-provider-ingestion"),
+        fingerprint=f"source-provider-canary:sentry:{issue_id}",
+        source_url=source_url,
+        labels={
+            "project": issue_data.get("project", {}),
+            "level": issue_data.get("level", "info"),
+            "awoooi_canary": "true",
+            "operator_id": operator.operator_id,
+            "telegram": "not_sent",
+            "incident": "not_created",
+            "approval": "not_created",
+        },
+        annotations={
+            "message": event_data.get("message"),
+            "summary": (
+                "Operator-signed Sentry webhook canary; records upstream "
+                "source evidence without creating incident, approval, or Telegram."
+            ),
+        },
+        payload={
+            "raw_canary": payload,
+            "operator_id": operator.operator_id,
+            "auth_method": operator.auth_method,
+            "side_effects": {
+                "incident_created": False,
+                "approval_created": False,
+                "telegram_sent": False,
+                "openclaw_called": False,
+            },
+        },
+    )
+    if event_uuid is None:
+        raise HTTPException(
+            status_code=500,
+            detail="sentry upstream canary was not recorded",
+        )
+    return {
+        "status": "canary_recorded",
+        "provider": "sentry",
+        "event_id": issue_id,
+        "conversation_event_id": str(event_uuid),
+        "side_effects": {
+            "incident_created": False,
+            "approval_created": False,
+            "telegram_sent": False,
+            "openclaw_called": False,
+        },
+    }
+
+
@router.post("/error")
 async def handle_sentry_error(
    request: Request,
@@ -109,6 +221,14 @@ async def handle_sentry_error(
    try:
        # 2026-04-27 P3.1-T2 by Claude — Tier-2 三服務感知強化：接入 SentryWebhookService 簽章驗證
        body = await request.body()
+        try:
+            payload_from_body = json.loads(body.decode("utf-8") or "{}")
+        except json.JSONDecodeError:
+            payload_from_body = {}
+
+        if isinstance(payload_from_body, dict) and _is_sentry_upstream_canary(payload_from_body):
+            return await _record_sentry_upstream_canary(payload_from_body, request)
+
        sig_header = request.headers.get("sentry-hook-signature", "")
        try:
            verify_sentry_signature(body, sig_header)
@@ -214,6 +334,8 @@ async def handle_sentry_error(
            "message": "Analysis scheduled"
        }

+    except HTTPException:
+        raise
    except Exception as e:
        logger.exception("Sentry webhook processing failed")
        raise HTTPException(status_code=500, detail=str(e)) from e
--- a/apps/api/src/api/v1/signoz_webhook.py
+++ b/apps/api/src/api/v1/signoz_webhook.py
@@ -1,7 +1,3 @@
-from __future__ import annotations
-
-import asyncio
-
 """
 AWOOOI API - SignOz Webhook Handler
 ====================================
@@ -17,6 +13,9 @@ AWOOOI API - SignOz Webhook Handler
 🔴 HARD RULE: 時間顯示使用 Asia/Taipei (UTC+8)
 """

+from __future__ import annotations
+
+import asyncio
 import uuid
 from typing import TYPE_CHECKING

@@ -24,6 +23,7 @@ import structlog
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
 from pydantic import BaseModel

+from src.core.awooop_operator_auth import authenticate_awooop_operator_headers
 from src.core.metrics import (
    record_alert_chain_failure,
    record_alert_chain_success,
@@ -72,6 +72,101 @@ class SignOzAlertPayload(BaseModel):
    generatorURL: str | None = None


+def _is_signoz_upstream_canary(alert: dict) -> bool:
+    labels = alert.get("labels", {}) if isinstance(alert.get("labels"), dict) else {}
+    annotations = (
+        alert.get("annotations", {})
+        if isinstance(alert.get("annotations"), dict)
+        else {}
+    )
+    alert_name = str(alert.get("alertname") or labels.get("alertname") or "")
+    return (
+        str(labels.get("awoooi_canary", "")).lower() == "true"
+        or alert_name == "AwoooPSourceProviderCanary"
+        or str(annotations.get("awooop_canary", "")).lower() == "true"
+    )
+
+
+async def _record_signoz_upstream_canary(
+    alert: dict,
+    request: Request,
+) -> dict:
+    operator = authenticate_awooop_operator_headers(
+        request.headers.get("x-awooop-operator-id"),
+        request.headers.get("x-awooop-operator-key"),
+    )
+    labels = alert.get("labels", {}) if isinstance(alert.get("labels"), dict) else {}
+    annotations = (
+        alert.get("annotations", {})
+        if isinstance(alert.get("annotations"), dict)
+        else {}
+    )
+    alert_name = str(alert.get("alertname") or labels.get("alertname") or "AwoooPSourceProviderCanary")
+    run_ref = str(labels.get("run_ref") or labels.get("fingerprint") or "unknown")
+    event_id = f"awooop-canary-{run_ref}"
+    severity = str(labels.get("severity") or "info")
+    service_name = str(labels.get("service_name") or labels.get("service") or "source-provider-ingestion")
+    namespace = str(labels.get("namespace") or "awoooi-prod")
+    fingerprint = str(labels.get("fingerprint") or f"source-provider-canary:signoz:{run_ref}")
+    event_uuid = await record_external_alert_event(
+        project_id="awoooi",
+        provider="signoz",
+        event_id=event_id,
+        stage="upstream_canary",
+        title=alert_name,
+        severity=severity,
+        namespace=namespace,
+        target_resource=service_name,
+        fingerprint=fingerprint,
+        source_url=alert.get("generatorURL"),
+        labels={
+            **labels,
+            "awoooi_canary": "true",
+            "operator_id": operator.operator_id,
+            "telegram": "not_sent",
+            "incident": "not_created",
+            "approval": "not_created",
+        },
+        annotations={
+            **annotations,
+            "summary": annotations.get("summary")
+            or (
+                "Operator-signed SignOz webhook canary; records upstream "
+                "source evidence without creating incident, approval, or Telegram."
+            ),
+        },
+        payload={
+            "raw_canary": alert,
+            "operator_id": operator.operator_id,
+            "auth_method": operator.auth_method,
+            "side_effects": {
+                "incident_created": False,
+                "approval_created": False,
+                "telegram_sent": False,
+                "openclaw_called": False,
+            },
+        },
+    )
+    if event_uuid is None:
+        raise HTTPException(
+            status_code=500,
+            detail="signoz upstream canary was not recorded",
+        )
+    return {
+        "status": "canary_recorded",
+        "provider": "signoz",
+        "event_id": event_id,
+        "alert_name": alert_name,
+        "conversation_event_id": str(event_uuid),
+        "side_effects": {
+            "incident_created": False,
+            "approval_created": False,
+            "telegram_sent": False,
+            "openclaw_called": False,
+        },
+    }
+
+
@router.post("/alert")
 async def handle_signoz_alert(
    request: Request,
@@ -104,6 +199,10 @@ async def handle_signoz_alert(
                results.append({"status": "ignored", "reason": "not firing"})
                continue

+            if _is_signoz_upstream_canary(alert):
+                results.append(await _record_signoz_upstream_canary(alert, request))
+                continue
+
            # 提取告警資訊
            alert_name = alert.get("alertname", alert.get("labels", {}).get("alertname", "unknown"))
            labels = alert.get("labels", {})
@@ -149,6 +248,8 @@ async def handle_signoz_alert(

        return {"status": "ok", "processed": len(results), "results": results}

+    except HTTPException:
+        raise
    except Exception as e:
        logger.exception("signoz_webhook_error", error=str(e))
        raise HTTPException(status_code=500, detail=str(e)) from e
@@ -336,7 +437,7 @@ async def create_signoz_approval(
    severity: str,
    incident_id: str,
    anomaly_frequency: dict | None = None,
-    analysis_result: "LLMAnalysisResult" | None = None,
+    analysis_result: LLMAnalysisResult | None = None,
 ) -> str:
    """
    為 SignOz 告警建立 Approval 記錄
@@ -433,7 +534,7 @@ async def send_signoz_telegram(
    annotations: dict,
    severity: str,
    anomaly_frequency: dict | None = None,
-    analysis_result: "LLMAnalysisResult" | None = None,
+    analysis_result: LLMAnalysisResult | None = None,
    ai_provider: str = "none",
 ):
    """
@@ -496,6 +597,7 @@ async def _send_log_summary_notification(
    帶 5s 軟超時：超時後摘要繼續生成並存 Redis，不阻塞告警主流程
    """
    import html as _html
+
    from src.services.log_summary_service import get_log_summary_service
    from src.services.telegram_gateway import get_telegram_gateway