|
|
@@ -0,0 +1,818 @@
|
|
|
+"""Cross-domain data observability, alert aggregation and incident operations."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import hashlib
|
|
|
+import json
|
|
|
+import math
|
|
|
+import re
|
|
|
+import uuid
|
|
|
+from collections.abc import Callable
|
|
|
+from datetime import datetime
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+from app.core.common.timezone_utils import now_china
|
|
|
+
|
|
|
+SLI_CATALOG = (
|
|
|
+ {
|
|
|
+ "type": "freshness",
|
|
|
+ "name": "新鲜度",
|
|
|
+ "unit": "seconds",
|
|
|
+ "description": "源数据观测时间到当前时间的延迟",
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "type": "completeness",
|
|
|
+ "name": "完整性",
|
|
|
+ "unit": "ratio",
|
|
|
+ "description": "约定字段和记录满足完整性要求的比例",
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "type": "quality",
|
|
|
+ "name": "质量",
|
|
|
+ "unit": "score",
|
|
|
+ "description": "确定性质量规则计算得到的质量分数",
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "type": "delivery",
|
|
|
+ "name": "交付",
|
|
|
+ "unit": "ratio",
|
|
|
+ "description": "约定窗口内成功完成采集或交付的比例",
|
|
|
+ },
|
|
|
+)
|
|
|
+SLI_TYPES = frozenset(item["type"] for item in SLI_CATALOG)
|
|
|
+LAYERS = ("service", "task", "data", "capacity")
|
|
|
+SEVERITIES = ("info", "warning", "error", "critical")
|
|
|
+IMPACT_TYPES = frozenset(
|
|
|
+ {"asset", "data_product", "business_domain", "user_group", "service"}
|
|
|
+)
|
|
|
+OPERATORS = frozenset({">=", "<="})
|
|
|
+SECRET_TOKENS = frozenset(
|
|
|
+ {
|
|
|
+ "apikey",
|
|
|
+ "authorization",
|
|
|
+ "connectionstring",
|
|
|
+ "credential",
|
|
|
+ "dsn",
|
|
|
+ "password",
|
|
|
+ "secret",
|
|
|
+ "token",
|
|
|
+ }
|
|
|
+)
|
|
|
+CODE_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{2,119}$")
|
|
|
+
|
|
|
+
|
|
|
+def _normalized_key(value: Any) -> str:
|
|
|
+ return re.sub(r"[^a-z0-9]", "", str(value).casefold())
|
|
|
+
|
|
|
+
|
|
|
+def _reject_secret_material(value: Any, path: str = "$") -> None:
|
|
|
+ if isinstance(value, dict):
|
|
|
+ for key, item in value.items():
|
|
|
+ normalized = _normalized_key(key)
|
|
|
+ if any(token in normalized for token in SECRET_TOKENS):
|
|
|
+ raise ValueError(f"secret material is not allowed at {path}.{key}")
|
|
|
+ _reject_secret_material(item, f"{path}.{key}")
|
|
|
+ elif isinstance(value, list):
|
|
|
+ for index, item in enumerate(value):
|
|
|
+ _reject_secret_material(item, f"{path}[{index}]")
|
|
|
+
|
|
|
+
|
|
|
+def _closed_object(
|
|
|
+ value: Any,
|
|
|
+ allowed: set[str] | frozenset[str],
|
|
|
+ label: str,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ if not isinstance(value, dict):
|
|
|
+ raise ValueError(f"{label} must be an object")
|
|
|
+ _reject_secret_material(value)
|
|
|
+ unknown = sorted(set(value) - set(allowed))
|
|
|
+ if unknown:
|
|
|
+ raise ValueError(
|
|
|
+ f"{label} contains unsupported fields: {', '.join(unknown)}"
|
|
|
+ )
|
|
|
+ return dict(value)
|
|
|
+
|
|
|
+
|
|
|
+def _string(value: Any, label: str, maximum: int = 500) -> str:
|
|
|
+ if not isinstance(value, str) or not value.strip():
|
|
|
+ raise ValueError(f"{label} is required")
|
|
|
+ result = value.strip()
|
|
|
+ if len(result) > maximum:
|
|
|
+ raise ValueError(f"{label} exceeds {maximum} characters")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _uid(value: Any, label: str) -> str:
|
|
|
+ try:
|
|
|
+ return str(uuid.UUID(str(value)))
|
|
|
+ except (TypeError, ValueError, AttributeError) as error:
|
|
|
+ raise ValueError(f"{label} must be a UUID") from error
|
|
|
+
|
|
|
+
|
|
|
+def _number(value: Any, label: str) -> float:
|
|
|
+ if isinstance(value, bool):
|
|
|
+ raise ValueError(f"{label} must be numeric")
|
|
|
+ try:
|
|
|
+ result = float(value)
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError(f"{label} must be numeric") from error
|
|
|
+ if not math.isfinite(result):
|
|
|
+ raise ValueError(f"{label} must be finite")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _timestamp(value: Any, label: str) -> datetime:
|
|
|
+ if isinstance(value, datetime):
|
|
|
+ result = value
|
|
|
+ else:
|
|
|
+ try:
|
|
|
+ result = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
|
+ except (TypeError, ValueError) as error:
|
|
|
+ raise ValueError(f"{label} must be an ISO-8601 datetime") from error
|
|
|
+ if result.tzinfo is None:
|
|
|
+ raise ValueError(f"{label} must include a timezone")
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def _canonical_hash(value: Any) -> str:
|
|
|
+ serialized = json.dumps(
|
|
|
+ value,
|
|
|
+ ensure_ascii=False,
|
|
|
+ sort_keys=True,
|
|
|
+ separators=(",", ":"),
|
|
|
+ )
|
|
|
+ return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+def _severity_rank(value: str) -> int:
|
|
|
+ try:
|
|
|
+ return SEVERITIES.index(value)
|
|
|
+ except ValueError as error:
|
|
|
+ raise ValueError("severity is invalid") from error
|
|
|
+
|
|
|
+
|
|
|
+class DataObservabilityService:
|
|
|
+ """Operate deterministic SLO evidence without inventing root causes."""
|
|
|
+
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ repository,
|
|
|
+ *,
|
|
|
+ uid_factory: Callable[[], str] = new_governance_uid,
|
|
|
+ now_factory: Callable[[], datetime] = now_china,
|
|
|
+ commit: Callable[[], None] = lambda: None,
|
|
|
+ rollback: Callable[[], None] = lambda: None,
|
|
|
+ ):
|
|
|
+ self.repository = repository
|
|
|
+ self.uid_factory = uid_factory
|
|
|
+ self.now_factory = now_factory
|
|
|
+ self.commit = commit
|
|
|
+ self.rollback = rollback
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def sli_catalog() -> list[dict[str, str]]:
|
|
|
+ return [dict(item) for item in SLI_CATALOG]
|
|
|
+
|
|
|
+ def create_slo(self, payload: Any, *, actor_uid: str) -> dict[str, Any]:
|
|
|
+ body = _closed_object(
|
|
|
+ payload,
|
|
|
+ {
|
|
|
+ "code",
|
|
|
+ "name",
|
|
|
+ "sli_type",
|
|
|
+ "scope_type",
|
|
|
+ "scope_uid",
|
|
|
+ "operator",
|
|
|
+ "target",
|
|
|
+ "window_seconds",
|
|
|
+ "owner_uid",
|
|
|
+ },
|
|
|
+ "SLO policy",
|
|
|
+ )
|
|
|
+ code = _string(body.get("code"), "code", 120).upper()
|
|
|
+ if not CODE_PATTERN.fullmatch(code):
|
|
|
+ raise ValueError("SLO policy code is invalid")
|
|
|
+ sli_type = _string(body.get("sli_type"), "sli_type", 30)
|
|
|
+ if sli_type not in SLI_TYPES:
|
|
|
+ raise ValueError("SLO policy sli_type is invalid")
|
|
|
+ operator = _string(body.get("operator"), "operator", 2)
|
|
|
+ if operator not in OPERATORS:
|
|
|
+ raise ValueError("SLO policy operator is invalid")
|
|
|
+ target = _number(body.get("target"), "target")
|
|
|
+ window_seconds = int(_number(body.get("window_seconds"), "window_seconds"))
|
|
|
+ if window_seconds < 60 or window_seconds > 31_536_000:
|
|
|
+ raise ValueError("window_seconds must be between 60 and 31536000")
|
|
|
+ now = self.now_factory().isoformat()
|
|
|
+ record = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "code": code,
|
|
|
+ "name": _string(body.get("name"), "name", 300),
|
|
|
+ "sli_type": sli_type,
|
|
|
+ "scope_type": _string(body.get("scope_type"), "scope_type", 40),
|
|
|
+ "scope_uid": _string(body.get("scope_uid"), "scope_uid", 500),
|
|
|
+ "operator": operator,
|
|
|
+ "target": target,
|
|
|
+ "window_seconds": window_seconds,
|
|
|
+ "owner_uid": _uid(body.get("owner_uid"), "owner_uid"),
|
|
|
+ "status": "active",
|
|
|
+ "created_by": _uid(actor_uid, "actor_uid"),
|
|
|
+ "created_at": now,
|
|
|
+ "updated_at": now,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ stored = self.repository.create_slo(record)
|
|
|
+ self.commit()
|
|
|
+ return stored
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def list_slos(self) -> list[dict[str, Any]]:
|
|
|
+ return self.repository.list_slos()
|
|
|
+
|
|
|
+ def collect(self, *, actor_uid: str) -> dict[str, Any]:
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ counters = {
|
|
|
+ "processed": 0,
|
|
|
+ "ignored": 0,
|
|
|
+ "created_alerts": 0,
|
|
|
+ "aggregated_alerts": 0,
|
|
|
+ "recovered_alerts": 0,
|
|
|
+ }
|
|
|
+ incident_uids: list[str] = []
|
|
|
+ try:
|
|
|
+ for raw_signal in self.repository.pending_signals():
|
|
|
+ signal = self._signal(raw_signal)
|
|
|
+ if self.repository.source_event_exists(
|
|
|
+ signal["source_event_key"]
|
|
|
+ ):
|
|
|
+ counters["ignored"] += 1
|
|
|
+ continue
|
|
|
+ claimed = self.repository.record_source_event(
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "source_event_key": signal["source_event_key"],
|
|
|
+ "source_type": signal["source_type"],
|
|
|
+ "source_uid": signal["source_uid"],
|
|
|
+ "status": signal["status"],
|
|
|
+ "signal": signal,
|
|
|
+ "processed_at": self.now_factory().isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ if claimed is False:
|
|
|
+ counters["ignored"] += 1
|
|
|
+ continue
|
|
|
+ result = self._process_signal(signal, actor_uid=actor)
|
|
|
+ counters["processed"] += 1
|
|
|
+ counters[result["counter"]] += 1
|
|
|
+ if (
|
|
|
+ result.get("incident_uid")
|
|
|
+ and result["incident_uid"] not in incident_uids
|
|
|
+ ):
|
|
|
+ incident_uids.append(result["incident_uid"])
|
|
|
+ self.commit()
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+ return {**counters, "incident_uids": incident_uids}
|
|
|
+
|
|
|
+ def _signal(self, value: Any) -> dict[str, Any]:
|
|
|
+ body = _closed_object(
|
|
|
+ value,
|
|
|
+ {
|
|
|
+ "source_event_key",
|
|
|
+ "source_type",
|
|
|
+ "source_uid",
|
|
|
+ "correlation_key",
|
|
|
+ "layer",
|
|
|
+ "sli_type",
|
|
|
+ "status",
|
|
|
+ "severity",
|
|
|
+ "title",
|
|
|
+ "owner_uid",
|
|
|
+ "observed_at",
|
|
|
+ "actual",
|
|
|
+ "target",
|
|
|
+ "evidence",
|
|
|
+ "impacts",
|
|
|
+ },
|
|
|
+ "observability signal",
|
|
|
+ )
|
|
|
+ layer = _string(body.get("layer"), "layer", 20)
|
|
|
+ if layer not in LAYERS:
|
|
|
+ raise ValueError("observability signal layer is invalid")
|
|
|
+ sli_type = _string(body.get("sli_type"), "sli_type", 30)
|
|
|
+ if sli_type not in SLI_TYPES:
|
|
|
+ raise ValueError("observability signal sli_type is invalid")
|
|
|
+ status = _string(body.get("status"), "status", 20)
|
|
|
+ if status not in {"violated", "recovered"}:
|
|
|
+ raise ValueError("observability signal status is invalid")
|
|
|
+ severity = _string(body.get("severity"), "severity", 20)
|
|
|
+ _severity_rank(severity)
|
|
|
+ evidence = body.get("evidence")
|
|
|
+ if not isinstance(evidence, dict) or not evidence.get("deterministic"):
|
|
|
+ raise ValueError("signal evidence must be deterministic")
|
|
|
+ impacts = body.get("impacts")
|
|
|
+ if not isinstance(impacts, list) or not impacts:
|
|
|
+ raise ValueError("signal impacts are required")
|
|
|
+ normalized_impacts = []
|
|
|
+ for raw_impact in impacts:
|
|
|
+ impact = _closed_object(
|
|
|
+ raw_impact,
|
|
|
+ {"target_type", "target_uid", "label"},
|
|
|
+ "signal impact",
|
|
|
+ )
|
|
|
+ target_type = _string(
|
|
|
+ impact.get("target_type"), "impact target_type", 40
|
|
|
+ )
|
|
|
+ if target_type not in IMPACT_TYPES:
|
|
|
+ raise ValueError("impact target_type is invalid")
|
|
|
+ normalized_impacts.append(
|
|
|
+ {
|
|
|
+ "target_type": target_type,
|
|
|
+ "target_uid": _string(
|
|
|
+ impact.get("target_uid"), "impact target_uid", 500
|
|
|
+ ),
|
|
|
+ "label": _string(impact.get("label"), "impact label", 300),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ observed_at = _timestamp(body.get("observed_at"), "observed_at")
|
|
|
+ return {
|
|
|
+ "source_event_key": _string(
|
|
|
+ body.get("source_event_key"), "source_event_key", 500
|
|
|
+ ),
|
|
|
+ "source_type": _string(
|
|
|
+ body.get("source_type"), "source_type", 80
|
|
|
+ ),
|
|
|
+ "source_uid": _string(body.get("source_uid"), "source_uid", 500),
|
|
|
+ "correlation_key": _string(
|
|
|
+ body.get("correlation_key"), "correlation_key", 500
|
|
|
+ ),
|
|
|
+ "layer": layer,
|
|
|
+ "sli_type": sli_type,
|
|
|
+ "status": status,
|
|
|
+ "severity": severity,
|
|
|
+ "title": _string(body.get("title"), "title", 300),
|
|
|
+ "owner_uid": _uid(body.get("owner_uid"), "owner_uid"),
|
|
|
+ "observed_at": observed_at.isoformat(),
|
|
|
+ "actual": _number(body.get("actual"), "actual"),
|
|
|
+ "target": _number(body.get("target"), "target"),
|
|
|
+ "evidence": dict(evidence),
|
|
|
+ "impacts": normalized_impacts,
|
|
|
+ }
|
|
|
+
|
|
|
+ def _process_signal(
|
|
|
+ self,
|
|
|
+ signal: dict[str, Any],
|
|
|
+ *,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ now = self.now_factory()
|
|
|
+ dedup_key = _canonical_hash(
|
|
|
+ {
|
|
|
+ "correlation_key": signal["correlation_key"],
|
|
|
+ "sli_type": signal["sli_type"],
|
|
|
+ }
|
|
|
+ )
|
|
|
+ alert = self.repository.find_alert(dedup_key)
|
|
|
+ if signal["status"] == "recovered":
|
|
|
+ result = self._recover(
|
|
|
+ signal,
|
|
|
+ alert=alert,
|
|
|
+ dedup_key=dedup_key,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ now=now,
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ result = self._violate(
|
|
|
+ signal,
|
|
|
+ alert=alert,
|
|
|
+ dedup_key=dedup_key,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ now=now,
|
|
|
+ )
|
|
|
+ return result
|
|
|
+
|
|
|
+ def _violate(
|
|
|
+ self,
|
|
|
+ signal: dict[str, Any],
|
|
|
+ *,
|
|
|
+ alert: dict[str, Any] | None,
|
|
|
+ dedup_key: str,
|
|
|
+ actor_uid: str,
|
|
|
+ now: datetime,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ if alert is None or alert["status"] == "recovered":
|
|
|
+ alert = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "dedup_key": dedup_key,
|
|
|
+ "incident_uid": None,
|
|
|
+ "layer": signal["layer"],
|
|
|
+ "sli_type": signal["sli_type"],
|
|
|
+ "title": signal["title"],
|
|
|
+ "severity": signal["severity"],
|
|
|
+ "status": "open",
|
|
|
+ "occurrence_count": 1,
|
|
|
+ "escalation_level": 1,
|
|
|
+ "owner_uid": signal["owner_uid"],
|
|
|
+ "first_observed_at": signal["observed_at"],
|
|
|
+ "last_observed_at": signal["observed_at"],
|
|
|
+ "suppressed_until": None,
|
|
|
+ "suppression_reason": None,
|
|
|
+ "delivery_status": "pending",
|
|
|
+ "delivery_receipt": {},
|
|
|
+ "evidence": signal["evidence"],
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ alert_counter = "created_alerts"
|
|
|
+ else:
|
|
|
+ alert["occurrence_count"] = int(alert["occurrence_count"]) + 1
|
|
|
+ alert["escalation_level"] = min(
|
|
|
+ 3, int(alert["escalation_level"]) + 1
|
|
|
+ )
|
|
|
+ if _severity_rank(signal["severity"]) > _severity_rank(
|
|
|
+ alert["severity"]
|
|
|
+ ):
|
|
|
+ alert["severity"] = signal["severity"]
|
|
|
+ alert["last_observed_at"] = signal["observed_at"]
|
|
|
+ suppressed_until = alert.get("suppressed_until")
|
|
|
+ if not suppressed_until or _timestamp(
|
|
|
+ suppressed_until, "suppressed_until"
|
|
|
+ ) <= now:
|
|
|
+ alert["status"] = "open"
|
|
|
+ alert["delivery_status"] = "pending"
|
|
|
+ alert["evidence"] = signal["evidence"]
|
|
|
+ alert["updated_at"] = now.isoformat()
|
|
|
+ alert_counter = "aggregated_alerts"
|
|
|
+
|
|
|
+ incident = self.repository.find_open_incident(dedup_key)
|
|
|
+ if incident is None:
|
|
|
+ incident_uid = self.uid_factory()
|
|
|
+ incident = {
|
|
|
+ "uid": incident_uid,
|
|
|
+ "code": f"INC-{now:%Y%m%d}-{incident_uid[-6:].upper()}",
|
|
|
+ "dedup_key": dedup_key,
|
|
|
+ "title": signal["title"],
|
|
|
+ "severity": signal["severity"],
|
|
|
+ "status": "open",
|
|
|
+ "owner_uid": signal["owner_uid"],
|
|
|
+ "escalation_level": alert["escalation_level"],
|
|
|
+ "first_detected_at": signal["observed_at"],
|
|
|
+ "last_observed_at": signal["observed_at"],
|
|
|
+ "created_by": actor_uid,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ "closed_by": None,
|
|
|
+ "closed_at": None,
|
|
|
+ }
|
|
|
+ self.repository.create_incident(incident)
|
|
|
+ else:
|
|
|
+ incident["last_observed_at"] = signal["observed_at"]
|
|
|
+ incident["escalation_level"] = max(
|
|
|
+ int(incident["escalation_level"]),
|
|
|
+ int(alert["escalation_level"]),
|
|
|
+ )
|
|
|
+ if _severity_rank(signal["severity"]) > _severity_rank(
|
|
|
+ incident["severity"]
|
|
|
+ ):
|
|
|
+ incident["severity"] = signal["severity"]
|
|
|
+ incident["status"] = "open"
|
|
|
+ incident["updated_at"] = now.isoformat()
|
|
|
+ self.repository.update_incident(incident)
|
|
|
+ alert["incident_uid"] = incident["uid"]
|
|
|
+ if alert_counter == "created_alerts":
|
|
|
+ self.repository.create_alert(alert)
|
|
|
+ action = "alert_created"
|
|
|
+ else:
|
|
|
+ self.repository.update_alert(alert)
|
|
|
+ action = "alert_aggregated"
|
|
|
+ self.repository.save_impacts(incident["uid"], signal["impacts"])
|
|
|
+ self._timeline(
|
|
|
+ incident["uid"],
|
|
|
+ action,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ evidence={
|
|
|
+ **signal["evidence"],
|
|
|
+ "source_event_key": signal["source_event_key"],
|
|
|
+ "alert_uid": alert["uid"],
|
|
|
+ },
|
|
|
+ now=now,
|
|
|
+ )
|
|
|
+ return {"counter": alert_counter, "incident_uid": incident["uid"]}
|
|
|
+
|
|
|
+ def _recover(
|
|
|
+ self,
|
|
|
+ signal: dict[str, Any],
|
|
|
+ *,
|
|
|
+ alert: dict[str, Any] | None,
|
|
|
+ dedup_key: str,
|
|
|
+ actor_uid: str,
|
|
|
+ now: datetime,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ if alert is None or alert.get("incident_uid") is None:
|
|
|
+ return {"counter": "ignored", "incident_uid": None}
|
|
|
+ alert["status"] = "recovered"
|
|
|
+ alert["last_observed_at"] = signal["observed_at"]
|
|
|
+ alert["delivery_status"] = "recovery_pending"
|
|
|
+ alert["evidence"] = signal["evidence"]
|
|
|
+ alert["updated_at"] = now.isoformat()
|
|
|
+ self.repository.update_alert(alert)
|
|
|
+ incident = self.repository.get_incident(alert["incident_uid"])
|
|
|
+ if incident is None:
|
|
|
+ return {"counter": "ignored", "incident_uid": None}
|
|
|
+ incident["status"] = "monitoring"
|
|
|
+ incident["last_observed_at"] = signal["observed_at"]
|
|
|
+ incident["updated_at"] = now.isoformat()
|
|
|
+ self.repository.update_incident(incident)
|
|
|
+ self._timeline(
|
|
|
+ incident["uid"],
|
|
|
+ "recovered",
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ evidence={
|
|
|
+ **signal["evidence"],
|
|
|
+ "source_event_key": signal["source_event_key"],
|
|
|
+ "alert_uid": alert["uid"],
|
|
|
+ },
|
|
|
+ now=now,
|
|
|
+ )
|
|
|
+ return {
|
|
|
+ "counter": "recovered_alerts",
|
|
|
+ "incident_uid": incident["uid"],
|
|
|
+ }
|
|
|
+
|
|
|
+ def suppress_alert(
|
|
|
+ self,
|
|
|
+ alert_uid: str,
|
|
|
+ *,
|
|
|
+ until: datetime | str,
|
|
|
+ reason: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ alert = self.repository.get_alert(_uid(alert_uid, "alert_uid"))
|
|
|
+ if alert is None:
|
|
|
+ raise LookupError("alert was not found")
|
|
|
+ if alert["status"] == "recovered":
|
|
|
+ raise RuntimeError("recovered alert cannot be suppressed")
|
|
|
+ deadline = _timestamp(until, "until")
|
|
|
+ now = self.now_factory()
|
|
|
+ if deadline <= now:
|
|
|
+ raise ValueError("suppression deadline must be in the future")
|
|
|
+ alert.update(
|
|
|
+ {
|
|
|
+ "status": "suppressed",
|
|
|
+ "suppressed_until": deadline.isoformat(),
|
|
|
+ "suppression_reason": _string(reason, "reason", 500),
|
|
|
+ "delivery_status": "suppressed",
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return self._save_alert_action(
|
|
|
+ alert,
|
|
|
+ action="suppressed",
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ evidence={"reason": alert["suppression_reason"], "deterministic": True},
|
|
|
+ )
|
|
|
+
|
|
|
+ def acknowledge_delivery(
|
|
|
+ self,
|
|
|
+ alert_uid: str,
|
|
|
+ *,
|
|
|
+ channel: str,
|
|
|
+ receipt_id: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ alert = self.repository.get_alert(_uid(alert_uid, "alert_uid"))
|
|
|
+ if alert is None:
|
|
|
+ raise LookupError("alert was not found")
|
|
|
+ now = self.now_factory()
|
|
|
+ alert.update(
|
|
|
+ {
|
|
|
+ "delivery_status": "acknowledged",
|
|
|
+ "delivery_receipt": {
|
|
|
+ "channel": _string(channel, "channel", 80),
|
|
|
+ "receipt_id": _string(receipt_id, "receipt_id", 200),
|
|
|
+ "acknowledged_by": _uid(actor_uid, "actor_uid"),
|
|
|
+ "acknowledged_at": now.isoformat(),
|
|
|
+ },
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return self._save_alert_action(
|
|
|
+ alert,
|
|
|
+ action="delivery_acknowledged",
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ evidence={
|
|
|
+ "receipt_id": alert["delivery_receipt"]["receipt_id"],
|
|
|
+ "deterministic": True,
|
|
|
+ },
|
|
|
+ )
|
|
|
+
|
|
|
+ def _save_alert_action(
|
|
|
+ self,
|
|
|
+ alert: dict[str, Any],
|
|
|
+ *,
|
|
|
+ action: str,
|
|
|
+ actor_uid: str,
|
|
|
+ evidence: dict[str, Any],
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ try:
|
|
|
+ stored = self.repository.update_alert(alert)
|
|
|
+ self._timeline(
|
|
|
+ alert["incident_uid"],
|
|
|
+ action,
|
|
|
+ actor_uid=_uid(actor_uid, "actor_uid"),
|
|
|
+ evidence=evidence,
|
|
|
+ now=self.now_factory(),
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return stored
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def escalate_incident(
|
|
|
+ self,
|
|
|
+ incident_uid: str,
|
|
|
+ *,
|
|
|
+ reason: str,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ uid = _uid(incident_uid, "incident_uid")
|
|
|
+ incident = self.repository.get_incident(uid)
|
|
|
+ if incident is None:
|
|
|
+ raise LookupError("incident was not found")
|
|
|
+ if incident["status"] == "closed":
|
|
|
+ raise RuntimeError("closed incident cannot be escalated")
|
|
|
+ incident["escalation_level"] = min(
|
|
|
+ 3, int(incident["escalation_level"]) + 1
|
|
|
+ )
|
|
|
+ incident["updated_at"] = self.now_factory().isoformat()
|
|
|
+ try:
|
|
|
+ stored = self.repository.update_incident(incident)
|
|
|
+ self._timeline(
|
|
|
+ uid,
|
|
|
+ "escalated",
|
|
|
+ actor_uid=_uid(actor_uid, "actor_uid"),
|
|
|
+ evidence={
|
|
|
+ "reason": _string(reason, "reason", 500),
|
|
|
+ "level": incident["escalation_level"],
|
|
|
+ "deterministic": True,
|
|
|
+ },
|
|
|
+ now=self.now_factory(),
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return stored
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def close_incident(
|
|
|
+ self,
|
|
|
+ incident_uid: str,
|
|
|
+ payload: Any,
|
|
|
+ *,
|
|
|
+ actor_uid: str,
|
|
|
+ ) -> dict[str, Any]:
|
|
|
+ uid = _uid(incident_uid, "incident_uid")
|
|
|
+ body = _closed_object(
|
|
|
+ payload,
|
|
|
+ {
|
|
|
+ "root_cause",
|
|
|
+ "user_impact",
|
|
|
+ "corrective_actions",
|
|
|
+ "closure_evidence",
|
|
|
+ },
|
|
|
+ "incident postmortem",
|
|
|
+ )
|
|
|
+ required = {
|
|
|
+ "root_cause",
|
|
|
+ "user_impact",
|
|
|
+ "corrective_actions",
|
|
|
+ "closure_evidence",
|
|
|
+ }
|
|
|
+ if set(body) != required:
|
|
|
+ raise ValueError("incident postmortem evidence is incomplete")
|
|
|
+ detail = self.repository.incident_detail(uid)
|
|
|
+ if detail is None:
|
|
|
+ raise LookupError("incident was not found")
|
|
|
+ if detail["status"] == "closed":
|
|
|
+ return detail
|
|
|
+ if not detail.get("owner_uid") or not detail.get("impacts"):
|
|
|
+ raise RuntimeError("incident responsibility and impact are required")
|
|
|
+ if not detail.get("alerts") or any(
|
|
|
+ item["status"] != "recovered" for item in detail["alerts"]
|
|
|
+ ):
|
|
|
+ raise RuntimeError("all incident alerts must recover before closure")
|
|
|
+ actions = body["corrective_actions"]
|
|
|
+ evidence = body["closure_evidence"]
|
|
|
+ if (
|
|
|
+ not isinstance(actions, list)
|
|
|
+ or not actions
|
|
|
+ or not isinstance(evidence, list)
|
|
|
+ or not evidence
|
|
|
+ ):
|
|
|
+ raise ValueError("incident postmortem evidence is incomplete")
|
|
|
+ normalized_actions = [
|
|
|
+ _string(item, "corrective action", 500) for item in actions
|
|
|
+ ]
|
|
|
+ normalized_evidence = []
|
|
|
+ for item in evidence:
|
|
|
+ value = _closed_object(
|
|
|
+ item, {"kind", "ref"}, "closure evidence"
|
|
|
+ )
|
|
|
+ normalized_evidence.append(
|
|
|
+ {
|
|
|
+ "kind": _string(value.get("kind"), "evidence kind", 80),
|
|
|
+ "ref": _string(value.get("ref"), "evidence ref", 500),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ actor = _uid(actor_uid, "actor_uid")
|
|
|
+ now = self.now_factory()
|
|
|
+ postmortem = {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "incident_uid": uid,
|
|
|
+ "root_cause": _string(
|
|
|
+ body.get("root_cause"), "root_cause", 2_000
|
|
|
+ ),
|
|
|
+ "user_impact": _string(
|
|
|
+ body.get("user_impact"), "user_impact", 2_000
|
|
|
+ ),
|
|
|
+ "corrective_actions": normalized_actions,
|
|
|
+ "closure_evidence": normalized_evidence,
|
|
|
+ "created_by": actor,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ incident = self.repository.get_incident(uid)
|
|
|
+ incident.update(
|
|
|
+ {
|
|
|
+ "status": "closed",
|
|
|
+ "closed_by": actor,
|
|
|
+ "closed_at": now.isoformat(),
|
|
|
+ "updated_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ self.repository.save_postmortem(postmortem)
|
|
|
+ self.repository.update_incident(incident)
|
|
|
+ self._timeline(
|
|
|
+ uid,
|
|
|
+ "closed",
|
|
|
+ actor_uid=actor,
|
|
|
+ evidence={
|
|
|
+ "postmortem_uid": postmortem["uid"],
|
|
|
+ "closure_evidence_count": len(normalized_evidence),
|
|
|
+ "deterministic": True,
|
|
|
+ },
|
|
|
+ now=now,
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return self.repository.incident_detail(uid)
|
|
|
+ except Exception:
|
|
|
+ self.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def _timeline(
|
|
|
+ self,
|
|
|
+ incident_uid: str,
|
|
|
+ action: str,
|
|
|
+ *,
|
|
|
+ actor_uid: str,
|
|
|
+ evidence: dict[str, Any],
|
|
|
+ now: datetime,
|
|
|
+ ) -> None:
|
|
|
+ _reject_secret_material(evidence)
|
|
|
+ self.repository.append_timeline(
|
|
|
+ {
|
|
|
+ "uid": self.uid_factory(),
|
|
|
+ "incident_uid": incident_uid,
|
|
|
+ "action": action,
|
|
|
+ "actor_uid": actor_uid,
|
|
|
+ "evidence": evidence,
|
|
|
+ "created_at": now.isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+
|
|
|
+ def list_alerts(self) -> list[dict[str, Any]]:
|
|
|
+ return self.repository.list_alerts()
|
|
|
+
|
|
|
+ def list_incidents(self) -> list[dict[str, Any]]:
|
|
|
+ return self.repository.list_incidents()
|
|
|
+
|
|
|
+ def incident_detail(self, uid: str) -> dict[str, Any]:
|
|
|
+ value = self.repository.incident_detail(_uid(uid, "incident_uid"))
|
|
|
+ if value is None:
|
|
|
+ raise LookupError("incident was not found")
|
|
|
+ return value
|
|
|
+
|
|
|
+ def overview(self) -> dict[str, Any]:
|
|
|
+ counts = self.repository.overview_counts()
|
|
|
+ return {
|
|
|
+ "layers": {layer: dict(counts.get(layer) or {}) for layer in LAYERS},
|
|
|
+ "sli_catalog": self.sli_catalog(),
|
|
|
+ "slos": self.list_slos(),
|
|
|
+ "open_incident_count": sum(
|
|
|
+ item["status"] != "closed" for item in self.list_incidents()
|
|
|
+ ),
|
|
|
+ }
|