fix(sre): type backup alerts before generic routing

This commit is contained in:
Your Name
2026-07-19 01:02:09 +08:00
parent dfe75456ab
commit e127c8b56d
4 changed files with 85 additions and 44 deletions

View File

@@ -336,7 +336,7 @@ async def test_backup_source_only_v2_lookup_reuses_only_full_legacy_owner() -> N
@pytest.mark.asyncio
async def test_alertmanager_type1_backup_queues_durable_owner_before_info_shortcut(
async def test_alertmanager_backup_failure_queues_typed_durable_owner_before_generic_flow(
monkeypatch: pytest.MonkeyPatch,
) -> None:
grouped = SimpleNamespace(is_grouped=False)
@@ -360,7 +360,7 @@ async def test_alertmanager_type1_backup_queues_durable_owner_before_info_shortc
webhooks,
"get_approval_service",
lambda: (_ for _ in ()).throw(
AssertionError("generic approval/TYPE-1 branch must not run")
AssertionError("generic approval branch must not run")
),
)
@@ -397,8 +397,8 @@ async def test_alertmanager_type1_backup_queues_durable_owner_before_info_shortc
call = process.await_args.kwargs
assert call["alertname"] == "HostBackupFailed"
assert call["source_event_fingerprint"] == "native-backup-fingerprint"
assert call["notification_type"] == "TYPE-1"
assert call["alert_category"] == "backup"
assert call["notification_type"] == "TYPE-3"
assert call["alert_category"] == "backup_restore"
assert call["target_resource"] == "backup_restore"

View File

@@ -63,15 +63,43 @@ class TestInfoAlerts:
assert ac == "info"
def test_backup_keyword_info_only(self):
# severity=info → severity 規則先命中TYPE-1
# 成功通知仍是 TYPE-1但不可丟失備份領域分類。
ac, nt = classify_alert_early("BackupJobComplete", "info", {})
assert ac == "backup_restore"
assert nt == "TYPE-1"
def test_backup_keyword_warning_not_type1(self):
# BackupJobFailed severity=warning → 繼續走 prefix 規則,不應是 TYPE-1
# 備份失敗立即進唯讀 BackupCheck不等 24 小時。
ac, nt = classify_alert_early("BackupJobFailed", "warning", {})
assert ac == "backup_restore"
assert nt == "TYPE-3"
@pytest.mark.parametrize(
"alertname",
(
"Backup.awoooi-frequent",
"Backup.langfuse",
"Backup.signoz",
"HostBackupFailed",
"BackupCredentialEscrowEvidenceMissing",
),
)
def test_backup_products_share_typed_readback_domain(self, alertname):
ac, nt = classify_alert_early(alertname, "warning", {})
assert (ac, nt) == ("backup_restore", "TYPE-3")
def test_backup_domain_label_routes_even_without_backup_alertname(self):
ac, nt = classify_alert_early(
"DataProtectionEvidenceMissing",
"critical",
{"asset_domain": "backup_restore"},
)
assert (ac, nt) == ("backup_restore", "TYPE-3")
def test_generic_service_restored_is_not_a_backup_identity(self):
ac, nt = classify_alert_early("ServiceRestored", "info", {})
assert (ac, nt) == ("info", "TYPE-1")
def test_watchdog_heartbeat(self):
# Watchdog (Alertmanager 心跳) severity=none → severity 規則先命中TYPE-1
ac, nt = classify_alert_early("Watchdog", "none", {})
@@ -84,9 +112,9 @@ class TestInfoAlerts:
assert nt == "TYPE-1"
def test_backup_critical_not_type1(self):
# critical backup 告警應走各自 prefix不是純資訊
# critical backup 告警必須是明確的備份領域唯讀調查。
ac, nt = classify_alert_early("BACKUP_MISSING", "critical", {})
assert nt == "TYPE-3"
assert (ac, nt) == ("backup_restore", "TYPE-3")
# --------------------------------------------------------------------------- #
@@ -270,10 +298,9 @@ class TestKubernetes:
assert ac == "kubernetes"
def test_velero_backup_failed_is_kubernetes(self):
# VeleroBackupFailed severity=critical → backup 規則不命中,走 Velero prefix → kubernetes TYPE-3
# Velero 備份是資料保護事件,不可誤掉到 K8s 寫入 executor。
ac, nt = classify_alert_early("VeleroBackupFailed", "critical", {})
assert nt == "TYPE-3"
assert ac == "kubernetes"
assert (ac, nt) == ("backup_restore", "TYPE-3")
def test_velero_backup_success_info_is_type1(self):
# VeleroBackupSuccess severity=info → TYPE-1

View File

@@ -66,23 +66,23 @@ class TestHighCpuVariants:
class TestBackup:
def test_host_backup_failed_fresh(self):
"""< 24h 備份失敗 → TYPE-1pure info"""
"""< 24h 備份失敗也要立即進唯讀 BackupCheck。"""
cat, ntype = classify_alert_early("HostBackupFailed", "warning", age_hours=10.0)
assert cat == "backup" and ntype == "TYPE-1"
assert cat == "backup_restore" and ntype == "TYPE-3"
def test_host_backup_failed_stale_upgrade(self):
"""> 24h 備份失敗 → 升級為 TYPE-3P0 故障)"""
"""> 24h 仍保持同一 typed backup/restore lane。"""
cat, ntype = classify_alert_early("HostBackupFailed", "warning", age_hours=25.0)
assert cat == "backup_failure" and ntype == "TYPE-3"
assert cat == "backup_restore" and ntype == "TYPE-3"
def test_host_backup_stale_upgrade(self):
cat, ntype = classify_alert_early("HostBackupStale", "warning", age_hours=30.0)
assert cat == "backup_failure" and ntype == "TYPE-3"
assert cat == "backup_restore" and ntype == "TYPE-3"
def test_backup_restore_test_not_upgraded(self):
"""BackupRestoreTestFailed 不受 age 升級影響"""
"""Restore drill 失敗也是唯讀調查,不誤標資訊。"""
cat, ntype = classify_alert_early("BackupRestoreTestFailed", "warning", age_hours=48.0)
assert ntype == "TYPE-1"
assert cat == "backup_restore" and ntype == "TYPE-3"
class TestDatabase: