5f1528e4ab
Fixes the "dead node shown NOMINAL" silent outage: node status was set only by events and never expired, so a node that crashed/lost connectivity stayed "online" forever (chelsty-infra was online for 16d, piha ~6d). The only thing that flipped status to offline was a node_offline event, which an unreachable node can never emit. Now node status is derived from freshness (now - last_seen), recomputed every observer cycle (incl. cycles with no new events): - always-on: fresh <=180s, stale 180-600s, dead >600s (3x the 60s heartbeat) - remote/LTE (chelsty-*): fresh <=900s, stale 900-3600s, dead >3600s Thresholds + tier logic live in ONE shared helper, services/control-plane/src/ liveness.py, imported by the observer and both operator UIs (bind-mounted into the agent-system webui image). No 3x copy. Transitions are not silent: the observer emits node_stale / node_offline / node_online (recovery) events tagged source=observer (skipped on re-ingest so they never reset last_seen), routed by the supervisor to alert_only actions. Read-time safety net: both UIs recompute liveness from last_seen at request time, so a stalled observer still surfaces dead nodes. Services inherit their node's liveness (cascade, variant B) without mutating services.json. Replaces the earlier binary NODE_OFFLINE_TTL_SECS flip. Tests: liveness unit tests, observer 3-state + transitions/recovery/baseline + self-event skip, operator_ui read-time net + cascade, supervisor node-event routing. 89 passed. docker compose config valid for both stacks. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
123 lines
4 KiB
Python
123 lines
4 KiB
Python
"""Supervisor routing of observer-emitted node liveness events → alert_only."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
import supervisor as supervisor_module
|
|
from supervisor import Supervisor
|
|
|
|
|
|
def _setup_supervisor(tmp_path, monkeypatch):
|
|
actions = tmp_path / "actions"
|
|
events = tmp_path / "events"
|
|
world = tmp_path / "world"
|
|
repo = tmp_path / "repo"
|
|
state = tmp_path / "state"
|
|
for d in (actions, events, world, repo / "hosts", state):
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
monkeypatch.setattr(supervisor_module, "ACTIONS_DIR", actions)
|
|
monkeypatch.setattr(supervisor_module, "EVENTS_DIR", events)
|
|
monkeypatch.setattr(supervisor_module, "WORLD_DIR", world)
|
|
monkeypatch.setattr(supervisor_module, "REPO_ROOT", repo)
|
|
sup = Supervisor()
|
|
sup.desired_state = {"services": {}}
|
|
sup.actual_state = {"services": {}, "nodes": {}, "incidents": {}}
|
|
return sup
|
|
|
|
|
|
def _node_event(node, etype):
|
|
return {
|
|
"id": f"evt-{node}-{int(time.time())}-{etype}-node",
|
|
"type": etype,
|
|
"node": node,
|
|
"service": None,
|
|
"severity": "high",
|
|
"timestamp": int(time.time()),
|
|
"source": "observer",
|
|
"message": f"Node {node} {etype}",
|
|
"payload": {"affected_node": node, "from": "fresh", "to": "dead"},
|
|
}
|
|
|
|
|
|
def _write(events_dir, event):
|
|
p = events_dir / f"{event['id']}.json"
|
|
p.write_text(json.dumps(event))
|
|
return p
|
|
|
|
|
|
def _pending(tmp_path, action_id):
|
|
return tmp_path / "actions" / "pending" / f"{action_id}.json"
|
|
|
|
|
|
def test_node_offline_generates_alert_only(tmp_path, monkeypatch):
|
|
sup = _setup_supervisor(tmp_path, monkeypatch)
|
|
_write(tmp_path / "events", _node_event("chelsty-infra", "node_offline"))
|
|
|
|
sup._process_ha_events()
|
|
|
|
action_id = "alert-node-offline-chelsty-infra"
|
|
assert _pending(tmp_path, action_id).exists()
|
|
action = json.loads(_pending(tmp_path, action_id).read_text())
|
|
assert action["type"] == "alert_only"
|
|
assert action["node"] == "chelsty-infra"
|
|
assert action["payload"]["reason"] == "node_offline"
|
|
|
|
|
|
def test_node_stale_generates_alert_only(tmp_path, monkeypatch):
|
|
sup = _setup_supervisor(tmp_path, monkeypatch)
|
|
_write(tmp_path / "events", _node_event("piha", "node_stale"))
|
|
|
|
sup._process_ha_events()
|
|
|
|
assert _pending(tmp_path, "alert-node-stale-piha").exists()
|
|
|
|
|
|
def test_node_online_recovery_generates_alert_only(tmp_path, monkeypatch):
|
|
sup = _setup_supervisor(tmp_path, monkeypatch)
|
|
_write(tmp_path / "events", _node_event("piha", "node_online"))
|
|
|
|
sup._process_ha_events()
|
|
|
|
assert _pending(tmp_path, "alert-node-online-piha").exists()
|
|
|
|
|
|
def test_node_alert_deduped_within_cooldown(tmp_path, monkeypatch):
|
|
"""A second identical event must not create a second pending action."""
|
|
sup = _setup_supervisor(tmp_path, monkeypatch)
|
|
_write(tmp_path / "events", _node_event("chelsty-infra", "node_offline"))
|
|
sup._process_ha_events()
|
|
|
|
# Reset the in-memory processed-id set so the file is re-read, but the
|
|
# action already exists in pending → must be skipped.
|
|
sup._ha_processed_event_ids.clear()
|
|
sup._process_ha_events()
|
|
|
|
pending = list((tmp_path / "actions" / "pending").glob("*.json"))
|
|
assert len(pending) == 1
|
|
|
|
|
|
def test_node_alert_uses_affected_node_when_top_level_missing(tmp_path, monkeypatch):
|
|
"""If node is absent at top level, fall back to payload.affected_node."""
|
|
sup = _setup_supervisor(tmp_path, monkeypatch)
|
|
ev = _node_event("solaria", "node_offline")
|
|
ev["node"] = None
|
|
_write(tmp_path / "events", ev)
|
|
|
|
sup._process_ha_events()
|
|
|
|
assert _pending(tmp_path, "alert-node-offline-solaria").exists()
|
|
|
|
|
|
def test_non_node_non_ha_event_not_routed(tmp_path, monkeypatch):
|
|
sup = _setup_supervisor(tmp_path, monkeypatch)
|
|
ev = _node_event("vps", "node_health") # not in NODE_ALERT_EVENTS
|
|
_write(tmp_path / "events", ev)
|
|
|
|
sup._process_ha_events()
|
|
|
|
assert list((tmp_path / "actions" / "pending").glob("*.json")) == []
|