diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index 9494026b..a3c6ab5e 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -30,8 +30,8 @@ from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request, from pydantic import BaseModel, Field from src.core.config import settings -from src.constants.alert_types import ALERTNAME_TO_TYPE from src.core.constants import is_cicd_alertname, is_heartbeat_alertname +from src.services.alert_rule_engine import get_incident_type from src.core.logging import get_logger from src.core.metrics import record_alert_chain_success @@ -1100,9 +1100,8 @@ async def alertmanager_webhook( approval_created=False, ) - # M3 重構 2026-04-11: alertname_to_type 抽至 src/constants/alert_types.py - # TODO I1: 下 Sprint 整合 ADR-064 Rule Engine 動態推斷 - alert_type = ALERTNAME_TO_TYPE.get(alertname, "custom") + # I1 整合 ADR-064 Rule Engine 2026-04-11: YAML 規則動態推斷,ALERTNAME_TO_TYPE 為 fallback + alert_type = get_incident_type(alertname) severity_map = {"critical": "critical", "warning": "warning", "info": "info"} severity = severity_map.get( diff --git a/apps/api/src/services/alert_rule_engine.py b/apps/api/src/services/alert_rule_engine.py index 8df2f64f..83b5d299 100644 --- a/apps/api/src/services/alert_rule_engine.py +++ b/apps/api/src/services/alert_rule_engine.py @@ -261,6 +261,42 @@ _AUTO_RULE_PROMPT = """\ """ +def get_incident_type(alertname: str) -> str: + """ + I1 整合 ADR-064 Rule Engine (2026-04-11): + 從 YAML 規則動態推斷 incident_type,取代 webhooks.py 靜態 dict。 + + 優先順序: + 1. alert_rules.yaml 中 match.alertname 完全匹配 → 使用 rule.incident_type + 2. ALERTNAME_TO_TYPE 靜態 dict fallback(constants/alert_types.py) + 3. "custom"(兜底) + + rule.incident_type 可選欄位;若 YAML 規則未設則跳過進 fallback。 + """ + from src.constants.alert_types import ALERTNAME_TO_TYPE + + try: + rules = _load_rules() + for rule in rules: + if _is_generic(rule): + continue + alertnames = rule.get("match", {}).get("alertname", []) + if alertname in alertnames: + # YAML 有 incident_type 欄位優先用 + incident_type = rule.get("incident_type") + if incident_type: + return incident_type + # 無 incident_type 欄位 → 用 rule id 作為 incident_type(語意一致) + rule_id = rule.get("id", "") + if rule_id: + return rule_id + break + except Exception: + pass + + return ALERTNAME_TO_TYPE.get(alertname, "custom") + + def _rule_id_exists(alertname: str) -> bool: """檢查 alertname 是否已有規則(排除通用兜底)""" try: