49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
from backend.events.bus import EventBus
|
|
from backend.metrics.node_state import NodeStateRegistry
|
|
|
|
|
|
class ObservabilityHub:
|
|
"""
|
|
Single fusion layer for:
|
|
- events (what happened)
|
|
- metrics (what is happening)
|
|
- state (what is true now)
|
|
"""
|
|
|
|
def __init__(self, registry: NodeStateRegistry, bus: EventBus):
|
|
self.registry = registry
|
|
self.bus = bus
|
|
self.subscribers = []
|
|
|
|
# -----------------------------
|
|
# STREAM OUT TO UI / PROMETHEUS / GRAFANA
|
|
# -----------------------------
|
|
def subscribe(self, fn):
|
|
self.subscribers.append(fn)
|
|
|
|
def emit(self, event: dict):
|
|
|
|
# enrich event with live state snapshot
|
|
if event.get("node"):
|
|
node = self.registry.get(event["node"])
|
|
if node:
|
|
event["node_state"] = {
|
|
"cpu": node.cpu_load,
|
|
"temp": node.temp,
|
|
"instability": node.instability,
|
|
"active_jobs": node.active_jobs,
|
|
}
|
|
|
|
for sub in self.subscribers:
|
|
sub(event)
|
|
|
|
# -----------------------------
|
|
# CONNECT EVENT BUS
|
|
# -----------------------------
|
|
def attach_bus(self):
|
|
|
|
def forward(event):
|
|
self.emit(event)
|
|
|
|
self.bus.subscribe(forward)
|