"""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() ), }