290 lines
9.6 KiB
Python
290 lines
9.6 KiB
Python
"""
|
||
Incident Lifecycle Reconciler
|
||
=============================
|
||
|
||
把已有強證據的舊 stuck incident 收斂回 RESOLVED。
|
||
|
||
範圍刻意保守：
|
||
- auto_repair_executions.success = true
|
||
- approval_records.status = EXECUTION_SUCCESS
|
||
- approval_records.status = EXPIRED
|
||
|
||
不處理單純 APPROVED / NO_ACTION / manual_required，避免把仍需人工的事件
|
||
誤當作自動修復完成。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from dataclasses import dataclass
|
||
|
||
import httpx
|
||
import structlog
|
||
from sqlalchemy import text
|
||
|
||
from src.core.config import settings
|
||
from src.db.base import get_db_context
|
||
from src.utils.timezone import now_taipei
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
BATCH_LIMIT = 100
|
||
INTERVAL_SECONDS = 1800
|
||
_PROMETHEUS_TIMEOUT_SECONDS = 5.0
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class LifecycleCandidate:
|
||
incident_id: str
|
||
resolution_type: str
|
||
reason: str
|
||
direct_db_only: bool = False
|
||
|
||
|
||
async def run_incident_lifecycle_reconciler_loop() -> None:
|
||
"""每 30 分鐘收斂一小批已有完成證據的 stuck incident。"""
|
||
while True:
|
||
try:
|
||
resolved, errors = await reconcile_stuck_incidents()
|
||
if resolved > 0 or errors > 0:
|
||
logger.info(
|
||
"incident_lifecycle_reconciler_done",
|
||
resolved=resolved,
|
||
errors=errors,
|
||
batch_limit=BATCH_LIMIT,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("incident_lifecycle_reconciler_loop_failed", error=str(exc))
|
||
|
||
await asyncio.sleep(INTERVAL_SECONDS)
|
||
|
||
|
||
async def reconcile_stuck_incidents(limit: int = BATCH_LIMIT) -> tuple[int, int]:
|
||
"""
|
||
找出已完成但仍卡在 INVESTIGATING 的 incident，透過 IncidentService 統一路徑結案。
|
||
|
||
Returns:
|
||
(resolved_count, error_count)
|
||
"""
|
||
candidates = await _fetch_candidates(limit)
|
||
remaining = max(0, limit - len(candidates))
|
||
if remaining > 0:
|
||
active_alertnames = await _fetch_active_alertnames()
|
||
if active_alertnames is not None:
|
||
candidates.extend(
|
||
await _fetch_inactive_or_duplicate_alert_candidates(
|
||
limit=remaining,
|
||
active_alertnames=active_alertnames,
|
||
exclude_incident_ids={c.incident_id for c in candidates},
|
||
)
|
||
)
|
||
|
||
if not candidates:
|
||
return 0, 0
|
||
|
||
from src.services.incident_service import get_incident_service
|
||
|
||
incident_service = get_incident_service()
|
||
resolved = 0
|
||
errors = 0
|
||
|
||
for candidate in candidates:
|
||
try:
|
||
if candidate.direct_db_only:
|
||
result = await _resolve_db_only(candidate.incident_id)
|
||
else:
|
||
result = await incident_service.resolve_incident(
|
||
candidate.incident_id,
|
||
resolution_type=candidate.resolution_type,
|
||
emit_postmortem=False,
|
||
)
|
||
if not result:
|
||
continue
|
||
resolved += 1
|
||
logger.info(
|
||
"incident_lifecycle_reconciled",
|
||
incident_id=candidate.incident_id,
|
||
reason=candidate.reason,
|
||
resolution_type=candidate.resolution_type,
|
||
direct_db_only=candidate.direct_db_only,
|
||
)
|
||
except Exception as exc:
|
||
errors += 1
|
||
logger.warning(
|
||
"incident_lifecycle_reconcile_failed",
|
||
incident_id=candidate.incident_id,
|
||
reason=candidate.reason,
|
||
error=str(exc),
|
||
)
|
||
|
||
return resolved, errors
|
||
|
||
|
||
async def _fetch_active_alertnames() -> set[str] | None:
|
||
"""Read current firing alertnames from Prometheus. None means fail-closed."""
|
||
try:
|
||
async with httpx.AsyncClient(timeout=_PROMETHEUS_TIMEOUT_SECONDS) as client:
|
||
response = await client.get(
|
||
f"{settings.PROMETHEUS_URL.rstrip('/')}/api/v1/query",
|
||
params={"query": 'ALERTS{alertstate="firing"}'},
|
||
)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
except Exception as exc:
|
||
logger.warning("incident_lifecycle_active_alerts_fetch_failed", error=str(exc))
|
||
return None
|
||
|
||
result = payload.get("data", {}).get("result", [])
|
||
active_alertnames = {
|
||
item.get("metric", {}).get("alertname")
|
||
for item in result
|
||
if item.get("metric", {}).get("alertname")
|
||
}
|
||
logger.info(
|
||
"incident_lifecycle_active_alerts_loaded",
|
||
active_alert_count=len(active_alertnames),
|
||
)
|
||
return active_alertnames
|
||
|
||
|
||
async def _resolve_db_only(incident_id: str) -> bool:
|
||
from src.repositories.incident_repository import get_incident_repository
|
||
|
||
now = now_taipei()
|
||
return await get_incident_repository().update_status(
|
||
incident_id=incident_id,
|
||
status="resolved",
|
||
updated_at=now,
|
||
resolved_at=now,
|
||
)
|
||
|
||
|
||
async def _fetch_candidates(limit: int) -> list[LifecycleCandidate]:
|
||
async with get_db_context() as db:
|
||
result = await db.execute(
|
||
text(
|
||
"""
|
||
WITH stale AS (
|
||
SELECT
|
||
i.incident_id,
|
||
i.created_at,
|
||
EXISTS (
|
||
SELECT 1
|
||
FROM auto_repair_executions are
|
||
WHERE are.incident_id = i.incident_id
|
||
AND are.success IS TRUE
|
||
) AS has_success_auto_repair,
|
||
EXISTS (
|
||
SELECT 1
|
||
FROM approval_records ar
|
||
WHERE ar.incident_id = i.incident_id
|
||
AND ar.status::text = 'EXECUTION_SUCCESS'
|
||
) AS has_execution_success,
|
||
EXISTS (
|
||
SELECT 1
|
||
FROM approval_records ar
|
||
WHERE ar.incident_id = i.incident_id
|
||
AND ar.status::text = 'EXPIRED'
|
||
) AS has_expired_approval
|
||
FROM incidents i
|
||
WHERE i.status = 'INVESTIGATING'
|
||
AND i.created_at <= now() - interval '24 hours'
|
||
)
|
||
SELECT
|
||
incident_id,
|
||
CASE
|
||
WHEN has_success_auto_repair THEN 'auto_repair'
|
||
WHEN has_execution_success THEN 'auto_repair'
|
||
ELSE 'timeout'
|
||
END AS resolution_type,
|
||
CASE
|
||
WHEN has_success_auto_repair THEN 'auto_repair_execution_success'
|
||
WHEN has_execution_success THEN 'approval_execution_success'
|
||
ELSE 'approval_expired'
|
||
END AS reason
|
||
FROM stale
|
||
WHERE has_success_auto_repair
|
||
OR has_execution_success
|
||
OR has_expired_approval
|
||
ORDER BY created_at DESC
|
||
LIMIT :limit
|
||
"""
|
||
),
|
||
{
|
||
"limit": limit,
|
||
},
|
||
)
|
||
rows = result.mappings().all()
|
||
|
||
return [
|
||
LifecycleCandidate(
|
||
incident_id=str(row["incident_id"]),
|
||
resolution_type=str(row["resolution_type"]),
|
||
reason=str(row["reason"]),
|
||
)
|
||
for row in rows
|
||
]
|
||
|
||
|
||
async def _fetch_inactive_or_duplicate_alert_candidates(
|
||
*,
|
||
limit: int,
|
||
active_alertnames: set[str],
|
||
exclude_incident_ids: set[str],
|
||
) -> list[LifecycleCandidate]:
|
||
"""
|
||
收斂 Alertmanager 已不再 firing 的舊 incident，以及同一 active alertname 的舊重複案。
|
||
|
||
若 Prometheus/Alertmanager 讀不到 active alertnames，上層會 fail-closed 不呼叫本函式。
|
||
"""
|
||
active_list = list(active_alertnames) or ["__no_active_alertnames__"]
|
||
exclude_list = list(exclude_incident_ids) or ["__no_excluded_incidents__"]
|
||
async with get_db_context() as db:
|
||
result = await db.execute(
|
||
text(
|
||
"""
|
||
WITH ranked AS (
|
||
SELECT
|
||
i.incident_id,
|
||
i.alertname,
|
||
i.created_at,
|
||
row_number() OVER (
|
||
PARTITION BY i.alertname
|
||
ORDER BY i.created_at DESC, i.incident_id DESC
|
||
) AS rn
|
||
FROM incidents i
|
||
WHERE i.status = 'INVESTIGATING'
|
||
AND i.created_at <= now() - interval '24 hours'
|
||
AND NOT (i.incident_id = ANY(:exclude_incident_ids))
|
||
)
|
||
SELECT
|
||
incident_id,
|
||
CASE
|
||
WHEN alertname = ANY(:active_alertnames)
|
||
THEN 'active_duplicate_stale'
|
||
ELSE 'inactive_alert_stale'
|
||
END AS reason
|
||
FROM ranked
|
||
WHERE NOT (alertname = ANY(:active_alertnames) AND rn = 1)
|
||
ORDER BY created_at ASC
|
||
LIMIT :limit
|
||
"""
|
||
),
|
||
{
|
||
"active_alertnames": active_list,
|
||
"exclude_incident_ids": exclude_list,
|
||
"limit": limit,
|
||
},
|
||
)
|
||
rows = result.mappings().all()
|
||
|
||
return [
|
||
LifecycleCandidate(
|
||
incident_id=str(row["incident_id"]),
|
||
resolution_type="timeout",
|
||
reason=str(row["reason"]),
|
||
direct_db_only=True,
|
||
)
|
||
for row in rows
|
||
]
|