DataOps-platform/app/core/events/data_observability.py

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