33 lines
948 B
Python
33 lines
948 B
Python
|
"""DITA observability namespace helpers.
|
||
|
|
|
||
|
|
These helpers keep DITA diagnostics isolated by runtime namespace while still
|
||
|
|
allowing optional legacy key mirroring when explicitly requested.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
|
||
|
|
LEGACY_ANOMALY_SENSOR_KEY = "dita_anomaly_sensors"
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class DitaObservabilityNamespace:
|
||
|
|
"""Namespace contract for DITA observability payloads."""
|
||
|
|
|
||
|
|
runtime_namespace: str = "pink"
|
||
|
|
feature_map: str = "DOLPHIN_FEATURES"
|
||
|
|
meta_health_map: str = "DOLPHIN_META_HEALTH"
|
||
|
|
state_map: str = "DOLPHIN_STATE_PINK"
|
||
|
|
anomaly_sensor_key: str | None = None
|
||
|
|
mirror_legacy_key: bool = False
|
||
|
|
|
||
|
|
def resolved_sensor_key(self) -> str:
|
||
|
|
value = str(self.anomaly_sensor_key or "").strip()
|
||
|
|
if value:
|
||
|
|
return value
|
||
|
|
ns = str(self.runtime_namespace or "pink").strip().lower()
|
||
|
|
return f"{LEGACY_ANOMALY_SENSOR_KEY}_{ns}"
|
||
|
|
|