oskar/homelab-codex-ws
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity
homelab-codex-ws/services/control-plane/src/supervisor.py
oskar d03288e5b3 fix(supervisor): drop dead mqtt_unreachable restart trigger 
The observer never creates incidents with trigger_type=mqtt_unreachable,
so the CONTAINER_RESTART_TRIGGERS branch for it could never fire (recon
D15). stability-agent keeps emitting the event; it just never becomes an
incident. Tests: 145 passed.

SERVICE_NAMES in ai-cluster's service_ops_worker.py (the other dead
constant from the plan) is NOT touched: that code is legacy-frozen in
the unmerged task/ai-cluster-solaria worktree and nothing on this branch
references it (verified by grep — only the recon and plan docs mention it).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-29 19:15:55 +02:00

930 lines
40 KiB
Python
Raw Blame History

import os
import json
import time
import logging
import yaml
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
def _atomic_write_json(path: Path, data) -> None:
"""Write JSON atomically: write to a sibling .tmp, fsync, then os.replace."""
tmp = path.with_suffix(".tmp")
with open(tmp, "w") as f:
json.dump(data, f, indent=2)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
# Constants and Paths
RUNTIME_PATH = os.getenv("RUNTIME_PATH", "/opt/homelab")
WORLD_DIR = Path(RUNTIME_PATH) / "world"
ACTIONS_DIR = Path(RUNTIME_PATH) / "actions"
EVENTS_DIR = Path(RUNTIME_PATH) / "events"
REPO_ROOT = Path(os.getenv("REPO_ROOT", "/repo"))
# Node alias map: maps alternative node names (as they appear in events/world state)
# to canonical topology node names (as they appear in hosts/*/services.yaml and topology.yaml).
# Override at runtime via NODE_ALIAS_MAP env var as a JSON string, e.g.:
# NODE_ALIAS_MAP='{"node-2": "chelsty", "node-1": "piha"}'
_NODE_ALIAS_ENV = os.getenv("NODE_ALIAS_MAP", "{}")
try:
NODE_ALIAS_MAP = json.loads(_NODE_ALIAS_ENV)
except Exception:
NODE_ALIAS_MAP = {}
# Incident trigger types that should result in a lightweight container_restart
# rather than a full redeploy: the container is present but not running.
# mqtt_unreachable was removed 2026-07-28: the observer never creates incidents
# with that trigger_type, so the branch was dead code (recon
# docs/architecture/RECON-multiagent-2026-07-27.md, D15). stability-agent still
# emits the mqtt_unreachable *event*; it just never becomes an incident.
CONTAINER_RESTART_TRIGGERS = {"containers_not_running"}
# Nodes where automatic disk_cleanup actions must NOT be generated.
# On chelsty nodes disk fullness is overwhelmingly caused by Frigate recordings
# or the HA database — Docker cleanup will not help and the operator must
# decide explicitly (e.g. adjust Frigate retain policy or purge HA recorder).
NO_DISK_CLEANUP_NODES = {"chelsty-infra", "chelsty-ha"}
# ---------------------------------------------------------------------------
# HA diagnostic event routing (ha-diag-agent events)
# ---------------------------------------------------------------------------
# ha_websocket_dead: HA WebSocket unresponsive → restart the homeassistant container.
# Separate from CONTAINER_RESTART_TRIGGERS because these events are routed directly
# from the events dir (not via the world-state drift loop) to avoid conflicts with
# the stability-agent's independent container health tracking on the same service key.
HA_CONTAINER_RESTART_EVENTS = {"ha_websocket_dead"}
# Alert-only events — operator notification, no automated action.
HA_ALERT_ONLY_EVENTS = {
"ha_integration_failed",
"ha_entity_unavailable_long",
"ha_automation_failing",
"ha_update_available",
"ha_recorder_lag",
"ha_system_health_degraded",
}
# Stable action-ID suffix for each alert-only type
_HA_ALERT_ID_SUFFIX = {
"ha_integration_failed": "integration-failed",
"ha_entity_unavailable_long": "entity-unavailable",
"ha_automation_failing": "automation-failing",
"ha_update_available": "update-available",
"ha_recorder_lag": "recorder-lag",
"ha_system_health_degraded": "system-health-degraded",
}
# 30-min cooldown after a container_restart completes; prevents restart loops
# when HA repeatedly fails to connect (e.g. bad config, slow startup).
HA_WEBSOCKET_RESTART_COOLDOWN = 1800
# 1-hour cooldown for alert-only events; avoids repeated Telegram noise for
# persistent conditions (e.g. an entity that stays unavailable for hours).
HA_ALERT_COOLDOWN = 3600
# Suppress ha_* events if homeassistant had a containers_not_running incident
# within this window — HA is in a planned restart/update and alerts would be noise.
HA_TRANSITION_WINDOW = 300 # 5 minutes
# ---------------------------------------------------------------------------
# Node liveness event routing (observer-emitted node_offline/node_stale/node_online)
# ---------------------------------------------------------------------------
# A node we cannot reach cannot be auto-remediated (you can't docker-restart a
# host that is offline), so these are alert-only — they exist to make a silent
# outage loud. node_online is the recovery notice.
NODE_ALERT_EVENTS = {"node_offline", "node_stale", "node_online"}
NODE_ALERT_COOLDOWN = 3600 # 1-hour cooldown to avoid repeated Telegram noise
# When True, events that would generate container_restart are downgraded to alert_only
# with a "[SHADOW MODE]" note. Safe default for initial deployment; set
# HA_DIAG_SHADOW_MODE=false on the control-plane node when ready for live actions.
HA_DIAG_SHADOW_MODE = os.getenv("HA_DIAG_SHADOW_MODE", "true").lower() == "true"
# ---------------------------------------------------------------------------
# Loop resilience
# ---------------------------------------------------------------------------
# A single reconcile() cycle must never be able to freeze the process forever.
# reconcile() does only synchronous local filesystem I/O (open/fsync/os.replace,
# directory globs) — none of it has a language-level timeout — so the cycle is
# run in a single-worker executor and bounded by RECONCILE_TIMEOUT. If a cycle
# blocks past the timeout (e.g. a stalled fsync on the /opt/homelab bind mount,
# or an ever-growing EVENTS_DIR walk), the main loop logs it and keeps ticking
# instead of hanging silently; the stuck worker thread is abandoned in place
# (Python cannot forcibly cancel a blocked syscall) and the executor's single
# worker naturally serializes the next cycle behind it, so two cycles can never
# write the same action file concurrently.
RECONCILE_TIMEOUT = float(os.getenv("SUPERVISOR_RECONCILE_TIMEOUT", "90"))
# Every Nth cycle logs an INFO "tick" line even when nothing actionable
# happened, so silence in `docker logs` is itself a meaningful signal rather
# than being indistinguishable from a healthy, quiet loop.
TICK_LOG_EVERY = int(os.getenv("SUPERVISOR_TICK_LOG_EVERY", "10"))
# Logging setup
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("supervisor")
class Supervisor:
def __init__(self):
self.desired_state = {"services": {}}
self.actual_state = {"services": {}, "nodes": {}, "incidents": {}}
# Nodes declared `status: dormant` in inventory/topology.yaml (e.g. the
# chelsty site, hardware down since ~2026-06-01). The supervisor
# generates NO actions of any kind for them: their services are excluded
# from desired state (which also auto-cancels their stale pending
# actions), and node/HA events from them are not routed to alerts.
# See docs/architecture/ARCHITEKTURA.md.
self.dormant_nodes: set = set()
# In-memory set of already-routed HA event IDs; prevents re-processing
# on each reconcile cycle. Grows to at most ~hundreds of entries/day.
self._ha_processed_event_ids: set = set()
self._ensure_dirs()
logger.info(
"shadow_mode=%s — HA container_restart actions %s",
HA_DIAG_SHADOW_MODE,
"downgraded to alert_only" if HA_DIAG_SHADOW_MODE else "enabled",
)
def _ensure_dirs(self):
ACTIONS_DIR.mkdir(parents=True, exist_ok=True)
(ACTIONS_DIR / "pending").mkdir(parents=True, exist_ok=True)
# ------------------------------------------------------------------
# Node name resolution
# ------------------------------------------------------------------
def _resolve_node(self, name):
"""Resolve an event/world-state node name to its canonical topology name."""
return NODE_ALIAS_MAP.get(name, name)
# ------------------------------------------------------------------
# Container name lookup
# ------------------------------------------------------------------
def _get_container_name(self, service):
"""
Determine the Docker container name for a service.
Parses container_name from the service's docker-compose.yml.
Falls back to the service name if not found.
"""
compose_path = REPO_ROOT / "services" / service / "docker-compose.yml"
if compose_path.exists():
try:
with open(compose_path, "r") as f:
compose = yaml.safe_load(f)
for svc_block in compose.get("services", {}).values():
cname = svc_block.get("container_name")
if cname:
return cname
except Exception as e:
logger.warning(f"Could not parse docker-compose for {service}: {e}")
# Convention: container name matches service name
return service
# ------------------------------------------------------------------
# State loading
# ------------------------------------------------------------------
def _load_dormant_nodes(self):
"""Refresh the set of dormant nodes from inventory/topology.yaml."""
dormant = set()
topo_file = REPO_ROOT / "inventory" / "topology.yaml"
try:
if topo_file.exists():
with open(topo_file, "r") as f:
topo = yaml.safe_load(f) or {}
for name, info in (topo.get("nodes") or {}).items():
if (info or {}).get("status") == "dormant":
dormant.add(name)
except Exception as e:
logger.error(f"Failed to load dormant nodes from {topo_file}: {e}")
self.dormant_nodes = dormant
def _load_desired_state(self):
services = {}
hosts_dir = REPO_ROOT / "hosts"
if not hosts_dir.exists():
logger.warning(f"Hosts directory {hosts_dir} does not exist")
return
for host_dir in hosts_dir.iterdir():
if host_dir.is_dir():
svc_file = host_dir / "services.yaml"
if svc_file.exists():
try:
with open(svc_file, "r") as f:
data = yaml.safe_load(f)
host_name = data.get("host")
if host_name in self.dormant_nodes:
logger.debug(
f"Skipping desired services of {host_name}: node dormant"
)
continue
for svc_name, svc_info in data.get("services", {}).items():
svc_info = svc_info or {}
# monitor: false — service is documented as desired but
# intentionally excluded from supervisor action generation.
# Use this when a service is not yet bootstrapped on an
# offline/LTE node so the queue stays clean until it is.
if svc_info.get("monitor") is False:
logger.debug(
f"Skipping {host_name}/{svc_name}: monitor=false"
)
continue
svc_key = f"{host_name}/{svc_name}"
services[svc_key] = {
"node": host_name,
"service": svc_name,
"desired": "running"
}
except Exception as e:
logger.error(f"Failed to load {svc_file}: {e}")
self.desired_state["services"] = services
def _load_actual_state(self) -> bool:
"""Load world state from disk. Returns False if any file is unreadable
(empty / mid-write truncation), in which case actual_state is NOT updated
so the caller can skip this reconcile cycle rather than treating missing
data as a real drift signal."""
files = {
"services": WORLD_DIR / "services.json",
"nodes": WORLD_DIR / "nodes.json",
"incidents": WORLD_DIR / "incidents.json"
}
raw = {}
for key, path in files.items():
if path.exists():
try:
with open(path, "r") as f:
raw[key] = json.load(f)
except Exception as e:
logger.warning(
f"World state {path.name} unreadable (truncated write?): {e} "
f"— skipping reconcile cycle, keeping last known state"
)
return False
else:
raw[key] = {}
# Normalize node names in services using alias map so that
# event-sourced names (e.g. "node-2") resolve to canonical
# topology names (e.g. "chelsty") before comparison with desired state.
normalized_services = {}
for svc_key, svc_info in raw.get("services", {}).items():
svc_info = dict(svc_info)
raw_node = svc_info.get("node", "")
canonical_node = self._resolve_node(raw_node)
if canonical_node != raw_node:
logger.debug(f"Resolved node alias: {raw_node} → {canonical_node}")
svc_info["node"] = canonical_node
svc_name = svc_info.get("service") or svc_key.split("/", 1)[-1]
svc_key = f"{canonical_node}/{svc_name}"
normalized_services[svc_key] = svc_info
# Normalize node names in incidents as well
normalized_incidents = {}
for inc_id, inc in raw.get("incidents", {}).items():
inc = dict(inc)
raw_node = inc.get("node", "")
inc["node"] = self._resolve_node(raw_node)
normalized_incidents[inc_id] = inc
self.actual_state["services"] = normalized_services
self.actual_state["nodes"] = raw.get("nodes", {})
self.actual_state["incidents"] = normalized_incidents
return True
# ------------------------------------------------------------------
# Incident helpers
# ------------------------------------------------------------------
def _get_incident_trigger(self, svc_key):
"""
Return the trigger_type of the active incident for a service, or None.
trigger_type is set by the observer when it creates an incident from
a specific event type (e.g. 'containers_not_running').
"""
svc_info = self.actual_state["services"].get(svc_key, {})
incident_id = svc_info.get("incident_id")
if not incident_id:
return None
incident = self.actual_state["incidents"].get(incident_id, {})
if incident.get("status") == "active":
return incident.get("trigger_type")
return None
# ------------------------------------------------------------------
# Reconciliation loop
# ------------------------------------------------------------------
def reconcile(self):
# Update heartbeat
heartbeat_file = WORLD_DIR.parent / "state" / "supervisor.heartbeat"
try:
heartbeat_file.touch()
except Exception as e:
logger.error(f"Failed to touch heartbeat file: {e}")
self._load_dormant_nodes()
self._load_desired_state()
if not self._load_actual_state():
return # world state unreadable this cycle — skip to avoid false drift
drifts = []
# 1. Check for missing or unhealthy services
for svc_key, desired_info in self.desired_state["services"].items():
actual_info = self.actual_state["services"].get(svc_key)
if not actual_info:
drifts.append({
"type": "missing_service",
"svc_key": svc_key,
"node": desired_info["node"],
"service": desired_info["service"],
"trigger_type": None,
})
elif actual_info.get("status") != "healthy":
trigger_type = self._get_incident_trigger(svc_key)
drifts.append({
"type": "unhealthy_service",
"svc_key": svc_key,
"node": desired_info["node"],
"service": desired_info["service"],
"status": actual_info.get("status"),
"trigger_type": trigger_type,
})
# 2. Generate service-level recommendations
for drift in drifts:
self._generate_recommendation(drift)
# 3. Generate node-level recommendations (disk pressure)
for node_name, node_info in self.actual_state["nodes"].items():
if node_name in NO_DISK_CLEANUP_NODES or node_name in self.dormant_nodes:
continue
if node_info.get("disk_pressure") == "high":
self._generate_disk_cleanup_recommendation(node_name)
# 4. Cancel pending actions whose drift has been resolved.
# When a service becomes healthy again (because node-agent emits
# service_healthy and the observer updates services.json), any
# previously queued redeploy/container_restart action for that
# service is no longer needed. Move it to "cancelled/" so the
# operator can see it was auto-resolved rather than silently dropped.
self._cancel_resolved_pending_actions()
# 5. Route HA diagnostic events emitted by ha-diag-agent.
# Processed directly from the events directory — not via the world-state
# drift loop — to avoid conflicts with stability-agent's independent
# container health tracking for the homeassistant service.
self._process_ha_events()
# ------------------------------------------------------------------
# Recommendation generation
# ------------------------------------------------------------------
def _generate_recommendation(self, drift):
node = drift["node"]
service = drift["service"]
trigger_type = drift.get("trigger_type")
# Choose action type first so we can build the stable, deterministic ID.
# Stable IDs mean reconcile is truly idempotent: the same drift always
# produces the same filename, so we never create duplicates even across
# restarts of the supervisor.
if trigger_type in CONTAINER_RESTART_TRIGGERS:
action_id = f"container-restart-{node}-{service}"
else:
action_id = f"redeploy-{node}-{service}"
# Skip if an action for this ID is already live in any active state
# (pending → approved → running). This prevents re-creation after
# a human approves an action that hasn't executed yet.
for state in ("pending", "approved", "running"):
if (ACTIONS_DIR / state / f"{action_id}.json").exists():
logger.debug(f"Skipping {action_id}: already in state '{state}'")
return
if trigger_type in CONTAINER_RESTART_TRIGGERS:
# Lightweight remediation: the container exists but is not running
# (containers_not_running). A docker restart is sufficient and low-risk.
container_name = self._get_container_name(service)
action = {
"action_id": action_id,
"timestamp": time.time(),
"type": "container_restart",
"node": node,
"service": service,
"container_name": container_name,
"risk_level": "low",
"confidence": 0.95,
"description": (
f"Restart container '{container_name}' on {node} "
f"(service: {service}, reason: {trigger_type})"
),
"status": "pending",
"payload": {
"reason": trigger_type,
"svc_key": drift["svc_key"],
},
}
else:
# Full redeploy: container is running but service is broken,
# or the cause is unknown / not a simple restart candidate.
action = {
"action_id": action_id,
"timestamp": time.time(),
"type": "redeploy",
"node": node,
"service": service,
"risk_level": "guarded",
"confidence": 0.9,
"description": f"Redeploy {service} on {node} due to {drift['type']}",
"status": "pending",
"payload": {
"reason": drift["type"],
"svc_key": drift["svc_key"],
},
}
action_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
try:
_atomic_write_json(action_path, action)
logger.info(
f"Generated recommendation: {action_id} "
f"(type={action['type']}, risk={action['risk_level']})"
)
except Exception as e:
logger.error(f"Failed to save recommendation {action_id}: {e}")
def _generate_disk_cleanup_recommendation(self, node: str):
"""
Generate a disk_cleanup action when node-agent reports critical disk
pressure (>85 %) on a node that supports automated Docker cleanup.
This is an OPERATOR-APPROVED action (risk=guarded): it runs
`docker image prune -a -f` and `docker volume prune -f`, which are
more aggressive than the safe auto-cleanup the node-agent runs itself.
Nodes in NO_DISK_CLEANUP_NODES never reach this method (filtered in
reconcile) because their disk fullness is caused by application data
(Frigate, HA) that the operator must handle manually.
"""
action_id = f"disk-cleanup-{node}"
for state in ("pending", "approved", "running"):
if (ACTIONS_DIR / state / f"{action_id}.json").exists():
logger.debug(f"Skipping {action_id}: already in state '{state}'")
return
action = {
"action_id": action_id,
"timestamp": time.time(),
"type": "disk_cleanup",
"node": node,
"service": "",
"risk_level": "guarded",
"confidence": 0.85,
"description": (
f"Aggressive disk cleanup on {node}: docker image prune -a "
f"and docker volume prune (requires operator approval)"
),
"status": "pending",
"payload": {
"reason": "disk_pressure",
"commands": [
"docker image prune -a -f",
"docker volume prune -f",
],
},
}
action_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
try:
_atomic_write_json(action_path, action)
logger.info(
f"Generated disk cleanup recommendation: {action_id} "
f"(node={node}, risk=guarded)"
)
except Exception as e:
logger.error(f"Failed to save disk cleanup recommendation {action_id}: {e}")
def _cancel_resolved_pending_actions(self):
"""
Auto-cancel pending service actions (redeploy / container_restart) whose
target service is now healthy in the actual state.
This keeps the action queue clean: when node-agent starts reporting
service_healthy for a container that previously had no world-state entry,
the pending 'missing_service' redeploy action that was generated before
the first health confirmation should be removed automatically rather than
sitting in the queue until an operator manually rejects it.
Only pending actions are considered — approved/running actions have already
been committed to by the operator and must not be cancelled automatically.
"""
cancelled_dir = ACTIONS_DIR / "cancelled"
cancelled_dir.mkdir(parents=True, exist_ok=True)
pending_dir = ACTIONS_DIR / "pending"
if not pending_dir.exists():
return
for action_file in list(pending_dir.glob("*.json")):
try:
with open(action_file, "r") as f:
action = json.load(f)
except Exception as e:
logger.error(f"Failed to read action {action_file.name}: {e}")
continue
action_type = action.get("type")
node = action.get("node")
service = action.get("service")
# Only auto-cancel service-level actions (not disk_cleanup)
if action_type not in ("redeploy", "container_restart"):
continue
if not node or not service:
continue
svc_key = f"{node}/{service}"
cancel_reason = None
# Case 1: service is no longer in desired state (removed from services.yaml
# or marked monitor:false). The action was generated under old config.
if svc_key not in self.desired_state["services"]:
cancel_reason = "service_removed_from_desired_state"
# Case 2: drift resolved — service is now healthy in actual state.
elif self.actual_state["services"].get(svc_key, {}).get("status") == "healthy":
cancel_reason = "drift_resolved_auto"
if cancel_reason:
dest = cancelled_dir / action_file.name
try:
action["status"] = "cancelled"
action["cancelled_reason"] = cancel_reason
action["cancelled_at"] = time.time()
_atomic_write_json(dest, action)
action_file.unlink()
logger.info(
f"Auto-cancelled {action_file.name}: "
f"{svc_key} — {cancel_reason}"
)
except Exception as e:
logger.error(f"Failed to cancel action {action_file.name}: {e}")
# ------------------------------------------------------------------
# HA diagnostic event routing
# ------------------------------------------------------------------
def _process_ha_events(self):
"""Scan the events directory for unprocessed ha_* events and route them."""
if not EVENTS_DIR.exists():
return
for event_file in sorted(EVENTS_DIR.glob("**/*.json")):
event_id = event_file.stem
if event_id in self._ha_processed_event_ids:
continue
self._ha_processed_event_ids.add(event_id)
try:
with open(event_file) as f:
event = json.load(f)
except Exception as e:
logger.debug(f"Could not read event {event_file}: {e}")
continue
etype = event.get("type", "")
if etype in NODE_ALERT_EVENTS:
self._route_node_event(event)
continue
if not etype.startswith("ha_"):
continue
self._route_ha_event(event)
def _route_ha_event(self, event: dict):
event_type = event.get("type", "")
node = event.get("node", "")
if not node:
return
if node in self.dormant_nodes:
logger.debug(f"Suppressing {event_type} on {node}: node dormant")
return
if event_type in HA_CONTAINER_RESTART_EVENTS:
if self._is_ha_in_transition(node):
logger.debug(
f"Suppressing {event_type} on {node}: homeassistant in transition"
)
return
if HA_DIAG_SHADOW_MODE:
logger.info(
"shadow_mode: suppressed container_restart for %s", event_type
)
self._generate_ha_shadow_alert(node, event)
else:
self._generate_ha_container_restart(node, event)
elif event_type == "ha_websocket_recovered":
self._cancel_ha_container_restart(node)
elif event_type in HA_ALERT_ONLY_EVENTS:
if self._is_ha_in_transition(node):
logger.debug(
f"Suppressing {event_type} on {node}: homeassistant in transition"
)
return
self._generate_ha_alert_only(node, event)
def _is_ha_in_transition(self, node: str) -> bool:
"""Return True if homeassistant container had a recent containers_not_running incident.
Suppresses ha_* alerts during planned HA restarts/updates to avoid
flooding the operator with secondary diagnostic alerts.
"""
svc_key = f"{node}/homeassistant"
svc_info = self.actual_state["services"].get(svc_key, {})
incident_id = svc_info.get("incident_id")
if not incident_id:
return False
incident = self.actual_state["incidents"].get(incident_id, {})
return (
incident.get("status") == "active"
and incident.get("trigger_type") == "containers_not_running"
and time.time() - (incident.get("last_occurrence") or 0) < HA_TRANSITION_WINDOW
)
def _ha_action_recently_completed(self, action_id: str, cooldown: int) -> bool:
"""Return True if action completed/rejected/cancelled within the cooldown window."""
for state in ("completed", "rejected", "cancelled"):
path = ACTIONS_DIR / state / f"{action_id}.json"
if path.exists():
try:
with open(path) as f:
data = json.load(f)
finished = (
data.get("finished_at")
or data.get("cancelled_at")
or data.get("updated_at")
or 0
)
if time.time() - finished < cooldown:
return True
except Exception:
pass
return False
def _generate_ha_container_restart(self, node: str, event: dict):
service = "homeassistant"
action_id = f"container-restart-{node}-{service}"
for state in ("pending", "approved", "running"):
if (ACTIONS_DIR / state / f"{action_id}.json").exists():
logger.debug(f"Skipping {action_id}: already in state '{state}'")
return
if self._ha_action_recently_completed(action_id, HA_WEBSOCKET_RESTART_COOLDOWN):
logger.debug(
f"Skipping {action_id}: within {HA_WEBSOCKET_RESTART_COOLDOWN}s cooldown"
)
return
payload = dict(event.get("payload", {}))
payload["reason"] = "ha_websocket_dead"
payload["svc_key"] = f"{node}/{service}"
container_name = self._get_container_name(service)
action = {
"action_id": action_id,
"timestamp": time.time(),
"type": "container_restart",
"node": node,
"service": service,
"container_name": container_name,
"risk_level": "low",
"confidence": 0.9,
"description": (
f"Restart '{container_name}' on {node}: HA WebSocket unresponsive"
),
"status": "pending",
"payload": payload,
}
self._write_pending_action(action)
def _generate_ha_shadow_alert(self, node: str, event: dict):
"""Shadow-mode downgrade: emit alert_only instead of container_restart.
Uses the same action_id and cooldown as the real restart so that
cooldown semantics are identical regardless of shadow mode state.
"""
service = "homeassistant"
action_id = f"container-restart-{node}-{service}"
for state in ("pending", "approved", "running"):
if (ACTIONS_DIR / state / f"{action_id}.json").exists():
logger.debug(f"Skipping {action_id}: already in state '{state}'")
return
if self._ha_action_recently_completed(action_id, HA_WEBSOCKET_RESTART_COOLDOWN):
logger.debug(
f"Skipping {action_id}: within {HA_WEBSOCKET_RESTART_COOLDOWN}s cooldown"
)
return
payload = dict(event.get("payload", {}))
payload["reason"] = "ha_websocket_dead"
payload["svc_key"] = f"{node}/{service}"
payload["shadow_mode"] = True
action = {
"action_id": action_id,
"timestamp": time.time(),
"type": "alert_only",
"node": node,
"service": service,
"risk_level": "info",
"confidence": 0.9,
"description": (
f"[SHADOW MODE] would have triggered container_restart "
f"for {service} on {node}: HA WebSocket unresponsive"
),
"status": "pending",
"payload": payload,
}
self._write_pending_action(action)
def _generate_ha_alert_only(self, node: str, event: dict):
event_type = event.get("type", "")
suffix = _HA_ALERT_ID_SUFFIX.get(event_type, event_type.replace("_", "-"))
action_id = f"alert-ha-{suffix}-{node}"
for state in ("pending", "approved", "running"):
if (ACTIONS_DIR / state / f"{action_id}.json").exists():
logger.debug(f"Skipping {action_id}: already in state '{state}'")
return
if self._ha_action_recently_completed(action_id, HA_ALERT_COOLDOWN):
logger.debug(
f"Skipping {action_id}: within {HA_ALERT_COOLDOWN}s cooldown"
)
return
payload = dict(event.get("payload", {}))
payload["reason"] = event_type
action = {
"action_id": action_id,
"timestamp": time.time(),
"type": "alert_only",
"node": node,
"service": event.get("service", "homeassistant"),
"risk_level": "info",
"confidence": 1.0,
"description": event.get(
"message", f"HA diagnostic alert: {event_type} on {node}"
),
"status": "pending",
"payload": payload,
}
self._write_pending_action(action)
def _route_node_event(self, event: dict):
"""Route an observer-emitted node liveness event to an alert_only action.
node_offline / node_stale make a silent outage loud; node_online is the
recovery notice. No auto-remediation — an unreachable node can't be
restarted from here. Dedup via stable action_id + cooldown, mirroring
the HA alert path.
"""
event_type = event.get("type", "")
node = event.get("node") or event.get("payload", {}).get("affected_node")
if not node:
return
if node in self.dormant_nodes:
logger.debug(f"Suppressing {event_type} on {node}: node dormant")
return
action_id = f"alert-{event_type.replace('_', '-')}-{node}"
for state in ("pending", "approved", "running"):
if (ACTIONS_DIR / state / f"{action_id}.json").exists():
logger.debug(f"Skipping {action_id}: already in state '{state}'")
return
if self._ha_action_recently_completed(action_id, NODE_ALERT_COOLDOWN):
logger.debug(f"Skipping {action_id}: within {NODE_ALERT_COOLDOWN}s cooldown")
return
payload = dict(event.get("payload", {}))
payload["reason"] = event_type
action = {
"action_id": action_id,
"timestamp": time.time(),
"type": "alert_only",
"node": node,
"service": None,
"risk_level": "info",
"confidence": 1.0,
"description": event.get(
"message", f"Node liveness alert: {event_type} on {node}"
),
"status": "pending",
"payload": payload,
}
self._write_pending_action(action)
def _cancel_ha_container_restart(self, node: str):
"""Move a pending ha_websocket_dead container_restart to cancelled on recovery."""
action_id = f"container-restart-{node}-homeassistant"
pending_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
if not pending_path.exists():
return
cancelled_dir = ACTIONS_DIR / "cancelled"
cancelled_dir.mkdir(parents=True, exist_ok=True)
dest = cancelled_dir / f"{action_id}.json"
try:
with open(pending_path) as f:
action = json.load(f)
action["status"] = "cancelled"
action["cancelled_reason"] = "ha_websocket_recovered"
action["cancelled_at"] = time.time()
_atomic_write_json(dest, action)
pending_path.unlink()
logger.info(f"Cancelled {action_id}: ha_websocket_recovered on {node}")
except Exception as e:
logger.error(f"Failed to cancel {action_id}: {e}")
def _write_pending_action(self, action: dict):
action_id = action["action_id"]
action_path = ACTIONS_DIR / "pending" / f"{action_id}.json"
try:
_atomic_write_json(action_path, action)
logger.info(
f"Generated HA action: {action_id} "
f"(type={action['type']}, risk={action['risk_level']})"
)
except Exception as e:
logger.error(f"Failed to save action {action_id}: {e}")
def _run_cycle_safely(self):
"""Run one reconcile() cycle, never letting an exception escape.
An uncaught exception here would previously propagate out of loop()
and kill the process outright — a crash is at least visible (the
container exits and restart:unless-stopped brings it back). This
makes that failure mode explicit and non-fatal: log the full
traceback and let the loop continue on the next cycle.
"""
try:
self.reconcile()
except Exception:
logger.exception(
"reconcile cycle raised an unhandled exception — logging and "
"continuing to the next cycle"
)
def loop(self, interval=30, max_cycles=None, reconcile_timeout=None):
"""Run reconcile() every `interval` seconds, forever.
max_cycles: stop after N cycles instead of looping forever (tests only).
reconcile_timeout: override RECONCILE_TIMEOUT (tests only).
"""
logger.info("Starting supervisor loop")
timeout = RECONCILE_TIMEOUT if reconcile_timeout is None else reconcile_timeout
cycle = 0
# max_workers=1 serializes cycles: if one is abandoned after a timeout,
# the next submit() queues behind it rather than running concurrently
# and racing on the same action files.
executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="supervisor-reconcile")
try:
while max_cycles is None or cycle < max_cycles:
cycle += 1
future = executor.submit(self._run_cycle_safely)
try:
future.result(timeout=timeout)
except TimeoutError:
logger.error(
"reconcile cycle #%d did not complete within %ss — "
"likely blocked on I/O (fsync/glob/file open) with no "
"language-level timeout. Abandoning this cycle; the loop "
"continues. The stuck worker thread keeps running in the "
"background and the next cycle will queue behind it.",
cycle, timeout,
)
if cycle % TICK_LOG_EVERY == 0:
logger.info("tick: supervisor loop alive, cycle #%d", cycle)
time.sleep(interval)
finally:
executor.shutdown(wait=False, cancel_futures=True)
if __name__ == "__main__":
supervisor = Supervisor()
supervisor.loop()
Reference in a new issue View git blame Copy permalink