feat(I1): ADR-064 Rule Engine 整合 — 動態推斷 incident_type
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 11m28s
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 11m28s
- alert_rule_engine.py: 新增 get_incident_type(alertname) 優先從 YAML 規則 match.alertname 查找 incident_type/rule_id Fallback: ALERTNAME_TO_TYPE 靜態 dict → "custom" - webhooks.py: alert_type 改用 get_incident_type(alertname) 取代 ALERTNAME_TO_TYPE.get() 靜態查找 - YAML 規則 19 條 alertname 覆蓋自動生效(無需手改 dict) - 新 alertname 觸發 generic_fallback → auto_generate_rule() 後自動加入 YAML Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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(
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user