fix(aiops): escalate failed host verification
Some checks failed
CD Pipeline / tests (push) Successful in 1m27s
Code Review / ai-code-review (push) Successful in 29s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled

This commit is contained in:
Your Name
2026-05-01 10:47:29 +08:00
parent 046d598e88
commit 97be5dedd7
6 changed files with 137 additions and 1 deletions

View File

@@ -521,6 +521,13 @@ class AutoRepairService:
# 2026-04-27 P3.1-T1 by Claude — 三 Tier-1 服務整合
# PostExecutionVerifier 判斷失敗/降級 → 觸發自動 Rollback
if verification_result in ("failed", "degraded"):
if self._should_escalate_failed_verification(incident, playbook):
await self._escalate_failed_verification(
incident=incident,
playbook=playbook,
verification_result=verification_result,
)
return
try:
from src.services.rollback_manager import get_rollback_manager
from src.services.declarative_remediation import DeclarativeRemediation
@@ -716,6 +723,69 @@ class AutoRepairService:
return True
return False
def _should_escalate_failed_verification(self, incident: Incident, playbook: Playbook) -> bool:
"""非 K8s 修復或主機/備份事件驗證失敗時,禁止合成 K8s rollback。"""
return self._is_host_or_backup_incident(incident) or not self._playbook_has_k8s_steps(playbook)
async def _escalate_failed_verification(
self,
*,
incident: Incident,
playbook: Playbook,
verification_result: str,
) -> None:
"""Post-verification failed but rollback is unsafe; notify emergency channel."""
target = (incident.affected_services or ["unknown"])[0]
namespace = "awoooi-prod"
alert_type = self._incident_alert_type(incident)
reason = (
f"auto repair playbook {playbook.playbook_id} verification={verification_result}; "
"rollback is unsafe for host/backup or non-K8s remediation"
)
logger.warning(
"auto_repair_verification_failed_emergency",
incident_id=incident.incident_id,
playbook_id=playbook.playbook_id,
verification_result=verification_result,
target=target,
)
try:
from src.services.emergency_escalation_service import (
escalate_auto_repair_unavailable,
)
await escalate_auto_repair_unavailable(
incident_id=incident.incident_id,
approval_id=None,
alert_type=alert_type,
target_resource=target,
namespace=namespace,
failure_reason=reason,
attempted_actions=(
f"auto_repair:{playbook.playbook_id} -> verifier:{verification_result} "
"-> emergency_intervention"
),
)
except Exception as exc:
logger.warning(
"auto_repair_verification_emergency_failed",
incident_id=incident.incident_id,
playbook_id=playbook.playbook_id,
error=str(exc),
)
def _incident_alert_type(self, incident: Incident) -> str:
"""Best-effort alertname for emergency cards."""
for signal in incident.signals or []:
labels = signal.labels or {}
alertname = labels.get("alertname") or signal.alert_name
if alertname:
return str(alertname)
return "AutoRepairVerificationFailed"
def _risk_exceeds_threshold(self, risk: RiskLevel) -> bool:
"""檢查風險是否超過自動修復門檻"""
high_risks = {RiskLevel.HIGH, RiskLevel.CRITICAL}

View File

@@ -19,7 +19,7 @@ logger = structlog.get_logger(__name__)
async def escalate_auto_repair_unavailable(
*,
incident_id: str,
approval_id: str,
approval_id: str | None,
alert_type: str,
target_resource: str,
namespace: str,