2dac154d78
Executor nie ma klienta ssh ani klucza do floty (uid 1000 homelab, brak ~/.ssh, brak resolucji nazw wezlow) — container_restart przez subprocess ssh failowal w 6ms na kazdej probie. Zamiast dodawac SSH do executora, kierunek jest odwrocony: executor zapisuje zlecenie do /opt/homelab/actions/dispatch/<node>/<action_id>.json, a node-agent na docelowym wezle (ktory ma dzialajacy docker.sock i juz ma klucz SSH do VPS uzywany do shippingu eventow) sam je odbiera i wykonuje lokalnie. - executor: _dispatch_container_restart pisze zlecenie zamiast ssh; _reconcile_running_actions konsumuje zwrotne action_result eventy i timeoutuje akcje bez odpowiedzi (ACTION_TIMEOUT_SECS, domyslnie 300s). redeploy/disk_cleanup/alert_only bez zmian. - node-agent: nowy krok w petli — rsync-pull wlasnej podkatalogu dispatch z VPS (ten sam klucz co _ship_events_to_vps, w przeciwnym kierunku; no-op na VPS, gdzie katalog jest lokalny), walidacja (node_name, whitelist tylko container_restart, odmowa restartu wlasnego kontenera), wykonanie przez docker SDK, raport jako event action_result (istniejacy kanal shippingu). Idempotencja przez znacznik w /opt/homelab/state/processed-actions/. - 26 nowych testow (10 executor, 16 node-agent), pelny suite obu serwisow 183/183 zielony. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1116 lines
51 KiB
Python
1116 lines
51 KiB
Python
"""
|
|
node_agent.py — Homelab node health monitor daemon.
|
|
|
|
Runs as a Docker container on every managed node. Each cycle it:
|
|
1. Collects system metrics (disk, memory, CPU).
|
|
2. Checks Docker container health.
|
|
3. Emits structured event JSON files to /opt/homelab/events/<node-name>/.
|
|
4. Applies safe Docker / filesystem cleanup per the conservative policy.
|
|
5. Pulls any action the control-plane executor has dispatched to this node
|
|
(container_restart only — whitelisted, self-restart guarded, idempotent)
|
|
and executes it locally through the docker socket, reporting the result
|
|
as an action_result event.
|
|
6. Optionally rsyncs events (including action_result) to VPS so the
|
|
control-plane observer/executor can process them.
|
|
|
|
Cleanup policy (matches health-monitor.sh):
|
|
lte_node (chelsty-infra, chelsty-ha) : NO cleanup, NO image operations
|
|
sd_card (piha, saturn) : dangling images + stopped containers,
|
|
max once per 24 h
|
|
ai_node (solaria) : dangling + containers + build cache,
|
|
NEVER docker image prune -a
|
|
standard (vps) : dangling + containers + build cache +
|
|
control-plane filesystem rotation
|
|
|
|
NEVER TOUCHED on any node:
|
|
/opt/homelab/data/ Frigate recordings, Ollama models, HA db, MQTT state
|
|
/opt/homelab/config/ All hand-crafted and repo-seeded configuration
|
|
/opt/homelab/state/ Heartbeat files, observer checkpoint
|
|
actions/pending|approved|running Live work queue
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import docker as docker_sdk
|
|
DOCKER_AVAILABLE = True
|
|
except ImportError:
|
|
DOCKER_AVAILABLE = False
|
|
docker_sdk = None
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Logging
|
|
# ---------------------------------------------------------------------------
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger("node-agent")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Runtime paths
|
|
# ---------------------------------------------------------------------------
|
|
RUNTIME_PATH = Path(os.getenv("RUNTIME_PATH", "/opt/homelab"))
|
|
REPO_ROOT = Path(os.getenv("REPO_ROOT", "/repo"))
|
|
EVENTS_DIR = RUNTIME_PATH / "events"
|
|
STATE_DIR = RUNTIME_PATH / "state"
|
|
LOGS_DIR = RUNTIME_PATH / "logs"
|
|
ACTIONS_DIR = RUNTIME_PATH / "actions"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Node identity
|
|
# ---------------------------------------------------------------------------
|
|
NODE_NAME = os.getenv("NODE_NAME") or socket.gethostname()
|
|
NODE_TYPE = os.getenv("NODE_TYPE", "") # override auto-detection if set
|
|
|
|
VPS_NODE_NAME = "vps"
|
|
LTE_NODES = {"chelsty-infra", "chelsty-ha"}
|
|
SD_CARD_NODES = {"piha", "saturn"}
|
|
AI_NODES = {"solaria"}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Event shipping (optional — requires VPS_EVENTS_HOST + SSH key in container)
|
|
# ---------------------------------------------------------------------------
|
|
VPS_EVENTS_HOST = os.getenv("VPS_EVENTS_HOST", "")
|
|
VPS_EVENTS_USER = os.getenv("VPS_EVENTS_USER", "oskar")
|
|
VPS_EVENTS_PATH = os.getenv("VPS_EVENTS_PATH", "/opt/homelab/events")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Remediation dispatch (pull-based — reuses the same VPS_EVENTS_HOST / SSH key
|
|
# as event shipping above, just in the opposite direction). See
|
|
# docs/backlog.md "PROJEKT: remediacja bez SSH": the control-plane executor on
|
|
# VPS has no SSH client and no key to the fleet, so it never reaches out to a
|
|
# node directly. Instead it drops a small JSON file under
|
|
# /opt/homelab/actions/dispatch/<node>/ on VPS; the node-agent for that node
|
|
# pulls its own subdirectory (remote nodes, via rsync over the existing
|
|
# shipping key) or reads it directly (VPS's own node-agent — same filesystem,
|
|
# no network hop) and executes it locally through the docker socket it
|
|
# already holds.
|
|
# ---------------------------------------------------------------------------
|
|
VPS_DISPATCH_PATH = os.getenv("VPS_DISPATCH_PATH", "/opt/homelab/actions/dispatch")
|
|
|
|
# Action types the agent is willing to execute on its own. Deliberately just
|
|
# one to start: redeploy stays VPS/manual, disk_cleanup is out of scope for
|
|
# this change (see docs/backlog.md). Anything not in this set is refused with
|
|
# a clear action_result error rather than silently ignored.
|
|
ALLOWED_DISPATCH_ACTION_TYPES = {"container_restart"}
|
|
|
|
# Container names the agent will never restart, regardless of what a dispatch
|
|
# file asks for. Restarting node-agent's own container mid-execution would
|
|
# kill the very process performing the restart. Matches the CLAUDE.md
|
|
# convention that a service's container name equals its service name.
|
|
SELF_RESTART_GUARD_NAMES = {"node-agent"}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thresholds
|
|
# ---------------------------------------------------------------------------
|
|
DISK_WARN_PCT = 75
|
|
DISK_CRIT_PCT = 85
|
|
MEM_WARN_PCT = 85
|
|
MEM_CRIT_PCT = 95
|
|
|
|
# A container Docker reports as "restarting" that has accumulated at least this
|
|
# many restarts is treated as a genuine crash-loop (a real fault) rather than a
|
|
# benign post-deploy / one-off restart.
|
|
#
|
|
# Rationale for the threshold: Docker's restart backoff keeps a *genuinely*
|
|
# crash-looping container cycling through the "restarting" state, so a 60 s
|
|
# health cycle is very likely to catch it there. A container that merely
|
|
# restarted once or twice (fresh deploy, a single transient failure that then
|
|
# recovers) is almost always back in "running" by the next cycle. Requiring
|
|
# RestartCount >= 3 clears normal deploy churn while still catching a container
|
|
# that is actually stuck flapping. Configurable via env for tuning per fleet.
|
|
CRASH_LOOP_RESTART_THRESHOLD = int(os.getenv("CRASH_LOOP_RESTART_THRESHOLD", "3"))
|
|
|
|
# SD-card nodes: enforce 24-hour gap between Docker cleanup runs
|
|
CLEANUP_INTERVAL_SECS = 86_400
|
|
LAST_CLEANUP_FILE = STATE_DIR / "last-docker-cleanup"
|
|
|
|
# How long to wait between full health-check cycles
|
|
HEALTH_CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _resolve_node_type() -> str:
|
|
if NODE_TYPE:
|
|
return NODE_TYPE
|
|
if NODE_NAME in LTE_NODES:
|
|
return "lte_node"
|
|
if NODE_NAME in SD_CARD_NODES:
|
|
return "sd_card"
|
|
if NODE_NAME in AI_NODES:
|
|
return "ai_node"
|
|
return "standard"
|
|
|
|
|
|
def _utc_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
# Matches evt-<node>-<unixts>-<type>-<svc>.json (this module's own emit_event
|
|
# naming, mirrored by observer.py's _EVENT_TS_RE / _ts_from_event_name so both
|
|
# sides of the checkpoint agree on how to read a filename's embedded epoch).
|
|
_EVENT_TS_RE = re.compile(r"-(\d{9,11})-")
|
|
_EVENT_TYPE_RE = re.compile(r"^evt-.+?-\d{9,11}-(.+)$")
|
|
|
|
|
|
def _event_ts_from_filename(name: str):
|
|
"""Return the embedded <unixts> from an event filename, or None if absent."""
|
|
m = _EVENT_TS_RE.search(Path(name).stem)
|
|
return int(m.group(1)) if m else None
|
|
|
|
|
|
def _event_type_from_filename(name: str) -> str:
|
|
"""Best-effort event type from evt-<node>-<ts>-<type>-<svc>.json.
|
|
|
|
Used only to decide retention eligibility (never for dispatch), so an
|
|
imprecise match on an unrecognized type is safe: it simply keeps the file.
|
|
"""
|
|
stem = Path(name).stem
|
|
m = _EVENT_TYPE_RE.match(stem)
|
|
if not m:
|
|
return "unknown"
|
|
rest = m.group(1)
|
|
for t in _RETENTION_NOISE_TYPES:
|
|
if rest == t or rest.startswith(t + "-"):
|
|
return t
|
|
return rest
|
|
|
|
|
|
def _checkpoint_ts_from_value(value) -> int:
|
|
"""Coerce a stored observer_checkpoint.json value into an int epoch.
|
|
|
|
Mirrors observer.py's own _checkpoint_ts_from_value: current format is an
|
|
int/float epoch; older observer builds stored a lexical path string, from
|
|
which the embedded <unixts> is extracted. Unparseable/absent -> 0, which
|
|
compares as "nothing processed yet" and safely keeps every file for that
|
|
node rather than guessing a checkpoint too high and deleting unprocessed
|
|
events.
|
|
"""
|
|
if isinstance(value, bool):
|
|
return 0
|
|
if isinstance(value, (int, float)):
|
|
return int(value)
|
|
if isinstance(value, str) and value:
|
|
ts = _event_ts_from_filename(value)
|
|
if ts is not None:
|
|
return ts
|
|
return 0
|
|
|
|
|
|
# Event types eligible for the ongoing retention sweep in
|
|
# _cleanup_control_plane_fs: pure positive-confirmation noise, never the
|
|
# actionable signals (healthcheck_failed, containers_not_running, ha_*,
|
|
# incidents, disk/memory/cpu pressure, deployment/remediation records, node
|
|
# liveness transitions), which are kept indefinitely by this method — an
|
|
# operator or the panel event feed may need to look back at those.
|
|
_RETENTION_NOISE_TYPES = frozenset({"service_healthy", "node_health"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NodeAgent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class NodeAgent:
|
|
def __init__(self):
|
|
self.node_name = NODE_NAME
|
|
self.node_type = _resolve_node_type()
|
|
self._ensure_dirs()
|
|
|
|
# Tracks the last known health classification per service (by canonical
|
|
# name — container service names AND the synthetic "control-plane" key
|
|
# used by _check_control_plane_health share this dict) so service_healthy
|
|
# is emitted only on the unhealthy->healthy transition, not once per cycle
|
|
# for a service that is already healthy. In-memory only: reset on process
|
|
# restart is acceptable — it just re-emits one confirmation event, not a
|
|
# flood, and the observer treats that re-confirmation as idempotent (see
|
|
# process_event).
|
|
self._service_health_state: dict = {}
|
|
|
|
self.docker_client = None
|
|
if DOCKER_AVAILABLE:
|
|
try:
|
|
self.docker_client = docker_sdk.from_env()
|
|
logger.info("Docker SDK connected")
|
|
except Exception as exc:
|
|
logger.warning(f"Docker unavailable: {exc}")
|
|
|
|
logger.info(f"node-agent starting: node={self.node_name} type={self.node_type}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Directories
|
|
# ------------------------------------------------------------------
|
|
|
|
def _ensure_dirs(self):
|
|
for d in [EVENTS_DIR, STATE_DIR, LOGS_DIR,
|
|
EVENTS_DIR / self.node_name]:
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _node_events_dir(self) -> Path:
|
|
return EVENTS_DIR / self.node_name
|
|
|
|
# ------------------------------------------------------------------
|
|
# Event emission
|
|
# ------------------------------------------------------------------
|
|
|
|
def emit_event(self, event_type: str, severity: str, service,
|
|
message: str, payload: dict = None):
|
|
ts = int(time.time())
|
|
# Include service slug in the ID so that multiple events of the same type
|
|
# emitted within the same second (e.g. service_healthy for N containers)
|
|
# don't overwrite each other — each gets a unique filename.
|
|
svc_slug = re.sub(r"[^a-z0-9]", "-", (service or "node").lower())[:32].strip("-")
|
|
event_id = f"evt-{self.node_name}-{ts}-{event_type}-{svc_slug}"
|
|
event = {
|
|
"id": event_id,
|
|
"timestamp": ts,
|
|
"date": _utc_iso(),
|
|
"type": event_type,
|
|
"severity": severity,
|
|
"node": self.node_name,
|
|
"service": service or "",
|
|
"message": message,
|
|
"payload": payload or {},
|
|
}
|
|
path = self._node_events_dir() / f"{event_id}.json"
|
|
try:
|
|
path.write_text(json.dumps(event, indent=2))
|
|
except Exception as exc:
|
|
logger.error(f"Failed to write event {event_id}: {exc}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# System metrics
|
|
# ------------------------------------------------------------------
|
|
|
|
def check_disk(self) -> int:
|
|
"""
|
|
Check disk usage of the filesystem that hosts RUNTIME_PATH.
|
|
Using shutil.disk_usage on the mounted runtime path works both
|
|
natively on the host and inside a container that mounts /opt/homelab
|
|
from the host — the reported usage reflects the host partition.
|
|
"""
|
|
try:
|
|
usage = shutil.disk_usage(str(RUNTIME_PATH))
|
|
pct = int(usage.used * 100 / usage.total)
|
|
avail_mb = int(usage.free / (1024 * 1024))
|
|
total_mb = int(usage.total / (1024 * 1024))
|
|
payload = {
|
|
"usage_pct": pct, "avail_mb": avail_mb,
|
|
"total_mb": total_mb, "mount": str(RUNTIME_PATH),
|
|
}
|
|
if pct >= DISK_CRIT_PCT:
|
|
logger.warning(f"Disk critical: {pct}% used")
|
|
self.emit_event(
|
|
"disk_pressure", "high", None,
|
|
f"Disk usage critical: {pct}% on {RUNTIME_PATH} ({avail_mb} MB free)",
|
|
payload,
|
|
)
|
|
elif pct >= DISK_WARN_PCT:
|
|
logger.info(f"Disk elevated: {pct}% used")
|
|
self.emit_event(
|
|
"disk_pressure", "medium", None,
|
|
f"Disk usage elevated: {pct}% on {RUNTIME_PATH} ({avail_mb} MB free)",
|
|
payload,
|
|
)
|
|
return pct
|
|
except Exception as exc:
|
|
logger.error(f"Disk check failed: {exc}")
|
|
return 0
|
|
|
|
def check_memory(self) -> int:
|
|
"""Read host memory from /proc/meminfo (visible in container without special mounts)."""
|
|
try:
|
|
info: dict = {}
|
|
with open("/proc/meminfo") as fh:
|
|
for line in fh:
|
|
k, v = line.split(":")
|
|
info[k.strip()] = int(v.strip().split()[0])
|
|
|
|
total = info.get("MemTotal", 1)
|
|
avail = info.get("MemAvailable", total)
|
|
pct = int((total - avail) * 100 / total)
|
|
avail_mb = avail // 1024
|
|
total_mb = total // 1024
|
|
payload = {"usage_pct": pct, "avail_mb": avail_mb, "total_mb": total_mb}
|
|
|
|
if pct >= MEM_CRIT_PCT:
|
|
logger.warning(f"Memory critical: {pct}%")
|
|
self.emit_event("high_memory", "high", None,
|
|
f"Memory usage critical: {pct}% ({avail_mb} MB available)", payload)
|
|
elif pct >= MEM_WARN_PCT:
|
|
self.emit_event("high_memory", "medium", None,
|
|
f"Memory usage elevated: {pct}% ({avail_mb} MB available)", payload)
|
|
return pct
|
|
except Exception as exc:
|
|
logger.error(f"Memory check failed: {exc}")
|
|
return 0
|
|
|
|
def check_cpu(self) -> int:
|
|
"""Two-sample /proc/stat delta for accurate instantaneous CPU usage."""
|
|
try:
|
|
def _read():
|
|
with open("/proc/stat") as fh:
|
|
for line in fh:
|
|
if line.startswith("cpu "):
|
|
vals = list(map(int, line.split()[1:]))
|
|
return vals[3], sum(vals) # idle, total
|
|
return 0, 1
|
|
|
|
idle1, total1 = _read()
|
|
time.sleep(0.5)
|
|
idle2, total2 = _read()
|
|
|
|
d_total = total2 - total1
|
|
d_idle = idle2 - idle1
|
|
pct = 100 - int(d_idle * 100 / d_total) if d_total else 0
|
|
|
|
if pct >= 90:
|
|
self.emit_event("high_cpu", "medium", None,
|
|
f"CPU usage elevated: {pct}%", {"usage_pct": pct})
|
|
return pct
|
|
except Exception as exc:
|
|
logger.error(f"CPU check failed: {exc}")
|
|
return 0
|
|
|
|
# ------------------------------------------------------------------
|
|
# Docker container health
|
|
# ------------------------------------------------------------------
|
|
|
|
def _canonical_container_name(self, c) -> str:
|
|
"""Return a stable, human-readable service name for a container.
|
|
|
|
Priority:
|
|
1. com.docker.compose.service label — always the clean compose-file key
|
|
(e.g. "mosquitto", "zigbee2mqtt"), immune to the hash-prefix corruption
|
|
Docker uses for stale project-state tracking entries.
|
|
2. c.name with hash prefix stripped — fallback for non-Compose containers.
|
|
When a container is removed outside of compose and then recreated, Docker
|
|
stores the old container record as "<12-char-hex-id>_<original-name>".
|
|
c.name returns that corrupted form; we strip the prefix here.
|
|
|
|
Using c.name directly is the source of ghost service keys like
|
|
"vps/9e36297651e7_control-plane-observer" that accumulate in services.json
|
|
every time containers are rebuilt.
|
|
"""
|
|
labels = c.attrs.get("Config", {}).get("Labels", {}) or {}
|
|
compose_svc = labels.get("com.docker.compose.service", "").strip()
|
|
if compose_svc:
|
|
return compose_svc
|
|
# Strip Docker-internal stale-state prefix: "<12-char hex>_<real-name>"
|
|
name = c.name
|
|
if (len(name) > 13
|
|
and name[12] == "_"
|
|
and all(ch in "0123456789abcdef" for ch in name[:12])):
|
|
return name[13:]
|
|
return name
|
|
|
|
def check_containers(self):
|
|
if not self.docker_client:
|
|
return
|
|
try:
|
|
containers = self.docker_client.containers.list(all=True)
|
|
except Exception as exc:
|
|
logger.error(f"Docker container list failed: {exc}")
|
|
return
|
|
|
|
for c in containers:
|
|
try:
|
|
name = self._canonical_container_name(c)
|
|
status = c.status
|
|
host_config = c.attrs.get("HostConfig", {})
|
|
restart_policy = host_config.get("RestartPolicy", {}).get("Name", "")
|
|
health_status = (c.attrs.get("State", {})
|
|
.get("Health", {})
|
|
.get("Status", ""))
|
|
# RestartCount is a top-level inspect field (sibling of State),
|
|
# not nested under State. It monotonically counts how many times
|
|
# Docker has auto-restarted this container — the signal that
|
|
# distinguishes a crash-loop from a one-off restart.
|
|
restart_count = c.attrs.get("RestartCount", 0)
|
|
|
|
# Skip containers in "created" state — these are Docker Compose
|
|
# internal tracking artifacts (never started, often hash-prefixed)
|
|
# that appear when a container is rebuilt outside of compose.
|
|
# This is a conscious skip, not a blind spot: a "created" container
|
|
# is not a running service and has no fault to report.
|
|
if status == "created":
|
|
continue
|
|
|
|
# Only track containers with a restart policy (long-running services)
|
|
is_managed = restart_policy in ("unless-stopped", "always", "on-failure")
|
|
if not is_managed:
|
|
continue
|
|
|
|
base_payload = {
|
|
"container": name,
|
|
"status": status,
|
|
"restart_policy": restart_policy,
|
|
"restart_count": restart_count,
|
|
}
|
|
|
|
# Exited container that carries an auto-restart policy
|
|
if status in ("exited", "dead"):
|
|
self._service_health_state[name] = False
|
|
logger.warning(f"Container exited: {name} (restart={restart_policy})")
|
|
self.emit_event(
|
|
"containers_not_running", "high", name,
|
|
f"Container '{name}' has exited (restart={restart_policy})",
|
|
base_payload,
|
|
)
|
|
|
|
# Container stuck in Docker's "restarting" state. This is the gap
|
|
# that previously made crash-loops invisible: "restarting" matched
|
|
# neither the exited/dead branch nor the running branch, so a
|
|
# container flapping under its restart policy emitted ZERO events
|
|
# and the observer kept showing its last (stale) "healthy" state.
|
|
#
|
|
# We split on RestartCount so a benign post-deploy restart does not
|
|
# alarm, while a real crash-loop escalates to the same actionable
|
|
# signal as an exited container.
|
|
elif status == "restarting":
|
|
self._service_health_state[name] = False
|
|
if restart_count >= CRASH_LOOP_RESTART_THRESHOLD:
|
|
# Genuine crash-loop: reuse containers_not_running so it
|
|
# rides the existing, supervisor-wired remediation path
|
|
# (parity with exited/dead — the container is, by
|
|
# definition, not staying running).
|
|
logger.warning(
|
|
f"Container crash-looping: {name} "
|
|
f"(restarting, {restart_count} restarts, restart={restart_policy})"
|
|
)
|
|
self.emit_event(
|
|
"containers_not_running", "high", name,
|
|
f"Container '{name}' is crash-looping "
|
|
f"(restarting, {restart_count} restarts, restart={restart_policy})",
|
|
{**base_payload, "crash_loop": True},
|
|
)
|
|
else:
|
|
# Fresh / transient restart. Make it VISIBLE (so the state
|
|
# is no longer a silent hole) but do NOT trigger remediation
|
|
# — a restart action here would be premature and noisy. If
|
|
# it keeps flapping, RestartCount crosses the threshold on a
|
|
# later cycle and escalates to containers_not_running above.
|
|
# container_restarting is intentionally observational: it is
|
|
# not wired to any supervisor trigger.
|
|
logger.info(
|
|
f"Container restarting: {name} "
|
|
f"({restart_count} restarts) — watching for crash-loop"
|
|
)
|
|
self.emit_event(
|
|
"container_restarting", "low", name,
|
|
f"Container '{name}' is restarting "
|
|
f"({restart_count} restarts) — watching for crash-loop",
|
|
{**base_payload, "crash_loop": False},
|
|
)
|
|
|
|
# Running container with a failing health check
|
|
elif status == "running" and health_status == "unhealthy":
|
|
self._service_health_state[name] = False
|
|
logger.warning(f"Container unhealthy: {name}")
|
|
self.emit_event(
|
|
"healthcheck_failed", "high", name,
|
|
f"Container '{name}' is running but its health check is failing",
|
|
{**base_payload, "health_status": health_status},
|
|
)
|
|
|
|
# Running container that is healthy — confirm to observer so that
|
|
# services.json stays populated for the supervisor's drift detection.
|
|
# Without this, the supervisor sees services.json as empty and treats
|
|
# all desired services as "missing", flooding the action queue.
|
|
#
|
|
# Emitted ONLY on the unhealthy/unknown -> healthy transition, not
|
|
# every cycle: a healthy service otherwise re-confirms itself once
|
|
# per health-check interval forever, which is exactly the flood
|
|
# that paralyzed the supervisor's reconcile() glob (358k backlog
|
|
# files, 91% service_healthy). The positive "still healthy" state is
|
|
# already carried by the presence of a `healthy` services.json entry
|
|
# (written once here) plus the node_health heartbeat every cycle —
|
|
# so a stuck/frozen node_agent is still visible via node liveness,
|
|
# without needing a fresh service_healthy file per service per cycle.
|
|
elif status == "running":
|
|
if self._service_health_state.get(name) is not True:
|
|
self.emit_event(
|
|
"service_healthy", "info", name,
|
|
f"Container '{name}' is running",
|
|
{**base_payload, "health_status": health_status or "none"},
|
|
)
|
|
self._service_health_state[name] = True
|
|
|
|
# Deliberately paused. docker pause is always an operator/tool
|
|
# action (Docker never auto-pauses), so it is not a fault — but a
|
|
# managed service should not normally sit paused, so surface it
|
|
# observationally rather than swallowing it. Non-actionable.
|
|
elif status == "paused":
|
|
self._service_health_state[name] = False
|
|
logger.info(f"Container paused: {name}")
|
|
self.emit_event(
|
|
"container_state_unexpected", "medium", name,
|
|
f"Container '{name}' is paused (restart={restart_policy})",
|
|
base_payload,
|
|
)
|
|
|
|
# Ephemeral teardown state: the container is being removed and will
|
|
# be gone (or replaced) by the next cycle. Emitting a fault here
|
|
# would be a guaranteed false positive on every recreate/deploy, so
|
|
# this is a conscious, documented skip — not a silent hole.
|
|
elif status == "removing":
|
|
logger.debug(f"Container removing (transient teardown): {name}")
|
|
|
|
# FALLBACK — any Docker state we do not explicitly handle (including
|
|
# states a future Docker engine might introduce). Never fall through
|
|
# silently: emit a diagnostic so a new/unknown state becomes visible
|
|
# instead of quietly recreating the very blind spot this fix closes.
|
|
else:
|
|
self._service_health_state[name] = False
|
|
logger.warning(
|
|
f"Container in unhandled Docker state: {name} (state='{status}')"
|
|
)
|
|
self.emit_event(
|
|
"container_state_unexpected", "medium", name,
|
|
f"Container '{name}' is in unhandled Docker state '{status}' "
|
|
f"(restart={restart_policy})",
|
|
base_payload,
|
|
)
|
|
|
|
except Exception as exc:
|
|
logger.error(f"Error checking container {c.name}: {exc}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Safe Docker cleanup
|
|
# ------------------------------------------------------------------
|
|
|
|
def _sd_card_rate_ok(self) -> bool:
|
|
"""Return True only if 24 hours have elapsed since last cleanup."""
|
|
if LAST_CLEANUP_FILE.exists():
|
|
try:
|
|
last_ts = int(LAST_CLEANUP_FILE.read_text().strip())
|
|
elapsed = time.time() - last_ts
|
|
if elapsed < CLEANUP_INTERVAL_SECS:
|
|
logger.debug(
|
|
f"Docker cleanup rate-limited ({elapsed:.0f}s < {CLEANUP_INTERVAL_SECS}s)"
|
|
)
|
|
return False
|
|
except Exception:
|
|
pass
|
|
return True
|
|
|
|
def _mark_cleanup_done(self):
|
|
try:
|
|
LAST_CLEANUP_FILE.write_text(str(int(time.time())))
|
|
except Exception as exc:
|
|
logger.error(f"Failed to update cleanup timestamp: {exc}")
|
|
|
|
def _prune_dangling_images(self):
|
|
if not self.docker_client:
|
|
return
|
|
try:
|
|
result = self.docker_client.images.prune(filters={"dangling": True})
|
|
reclaimed = result.get("SpaceReclaimed", 0) // (1024 * 1024)
|
|
logger.info(f"Pruned dangling images ({reclaimed} MB reclaimed)")
|
|
except Exception as exc:
|
|
logger.error(f"Image prune failed: {exc}")
|
|
|
|
def _prune_stopped_containers(self):
|
|
if not self.docker_client:
|
|
return
|
|
try:
|
|
result = self.docker_client.containers.prune()
|
|
reclaimed = result.get("SpaceReclaimed", 0) // (1024 * 1024)
|
|
logger.info(f"Pruned stopped containers ({reclaimed} MB reclaimed)")
|
|
except Exception as exc:
|
|
logger.error(f"Container prune failed: {exc}")
|
|
|
|
def _prune_build_cache(self):
|
|
if not self.docker_client:
|
|
return
|
|
try:
|
|
result = self.docker_client.api.prune_builds()
|
|
reclaimed = result.get("SpaceReclaimed", 0) // (1024 * 1024)
|
|
logger.info(f"Pruned build cache ({reclaimed} MB reclaimed)")
|
|
except Exception as exc:
|
|
# prune_builds() was added in docker-py 5.0; log and continue
|
|
logger.warning(f"Build cache prune unavailable or failed: {exc}")
|
|
|
|
def run_safe_cleanup(self):
|
|
"""Apply the per-node-type Docker cleanup policy."""
|
|
if self.node_type == "lte_node":
|
|
# No cleanup on LTE nodes — any Docker op risks a pull over a
|
|
# metered/intermittent connection.
|
|
logger.debug("Skipping Docker cleanup: LTE node")
|
|
return
|
|
|
|
if self.node_type == "sd_card":
|
|
if not self._sd_card_rate_ok():
|
|
return
|
|
self._prune_dangling_images()
|
|
self._prune_stopped_containers()
|
|
# No builder prune: minimise write cycles on SD card
|
|
self._mark_cleanup_done()
|
|
return
|
|
|
|
# ai_node and standard: dangling + containers + build cache
|
|
# ai_node: NEVER -a (would remove Ollama runtime images)
|
|
self._prune_dangling_images()
|
|
self._prune_stopped_containers()
|
|
self._prune_build_cache()
|
|
|
|
# ------------------------------------------------------------------
|
|
# VPS-specific: control-plane filesystem rotation
|
|
# ------------------------------------------------------------------
|
|
|
|
def _cleanup_control_plane_fs(self):
|
|
"""
|
|
Rotate control-plane filesystem artefacts on VPS.
|
|
Safe targets only — never touches data/, config/, state/, or
|
|
live action directories (pending/approved/running).
|
|
"""
|
|
now = time.time()
|
|
seven_days = 7 * 86_400
|
|
thirty_days = 30 * 86_400
|
|
three_days = 3 * 86_400
|
|
|
|
# 1. Completed / failed actions older than 7 days
|
|
for status in ("completed", "failed"):
|
|
sdir = ACTIONS_DIR / status
|
|
if not sdir.exists():
|
|
continue
|
|
for f in sdir.glob("*.json"):
|
|
try:
|
|
if now - f.stat().st_mtime > seven_days:
|
|
f.unlink(missing_ok=True)
|
|
logger.info(f"Cleaned old {status} action: {f.name}")
|
|
except Exception as exc:
|
|
logger.error(f"Failed to remove {f}: {exc}")
|
|
|
|
# 2. Deploy logs older than 30 days
|
|
deploy_dir = LOGS_DIR / "deploy"
|
|
if deploy_dir.exists():
|
|
for f in deploy_dir.glob("*.log"):
|
|
try:
|
|
if now - f.stat().st_mtime > thirty_days:
|
|
f.unlink(missing_ok=True)
|
|
logger.info(f"Cleaned old deploy log: {f.name}")
|
|
except Exception as exc:
|
|
logger.error(f"Failed to remove {f}: {exc}")
|
|
|
|
# 3. Noise event files (service_healthy / node_health) that are BOTH
|
|
# already past the observer's per-node checkpoint (guaranteed
|
|
# processed into world_state — see _checkpoint_ts_from_value) AND
|
|
# older than 3 days (safety buffer past the checkpoint). The dual
|
|
# condition guarantees we never delete an unprocessed event, and
|
|
# the type restriction guarantees healthcheck_failed / incidents /
|
|
# ha_* / node-liveness events are kept indefinitely by this method.
|
|
#
|
|
# BUG FIX (event-flood cleanup): this previously compared str(f)
|
|
# (a full file path) against the checkpoint value with <=, which
|
|
# only ever worked when the checkpoint stored the pre-fix lexical
|
|
# path format. Since observer.py migrated node_checkpoints to int
|
|
# epoch timestamps, every comparison here raised
|
|
# "'<=' not supported between instances of 'str' and 'int'",
|
|
# silently caught by the broad except below — so this cleanup has
|
|
# been a no-op ever since, which is the direct cause of the 358k-file
|
|
# backlog that paralyzed the supervisor's reconcile(). Fixed by
|
|
# comparing epoch-to-epoch, same as observer.py's own checkpoint
|
|
# logic.
|
|
checkpoint_file = STATE_DIR / "observer_checkpoint.json"
|
|
node_checkpoints: dict = {}
|
|
if checkpoint_file.exists():
|
|
try:
|
|
cp = json.loads(checkpoint_file.read_text())
|
|
if "node_checkpoints" in cp:
|
|
node_checkpoints = {
|
|
node: _checkpoint_ts_from_value(val)
|
|
for node, val in (cp["node_checkpoints"] or {}).items()
|
|
}
|
|
elif "last_processed_file" in cp:
|
|
# Migrate old single-file format
|
|
old = cp.get("last_processed_file", "")
|
|
if old:
|
|
try:
|
|
node_dir = Path(old).relative_to(EVENTS_DIR).parts[0]
|
|
node_checkpoints = {node_dir: _checkpoint_ts_from_value(old)}
|
|
except Exception:
|
|
pass
|
|
except Exception as exc:
|
|
logger.error(f"Failed to read observer checkpoint: {exc}")
|
|
|
|
if node_checkpoints:
|
|
for f in EVENTS_DIR.glob("**/*.json"):
|
|
try:
|
|
etype = _event_type_from_filename(f.name)
|
|
if etype not in _RETENTION_NOISE_TYPES:
|
|
continue
|
|
# Determine which node directory this event belongs to
|
|
rel = Path(f).relative_to(EVENTS_DIR)
|
|
node_dir = str(rel.parts[0]) if rel.parts else "__unknown__"
|
|
checkpoint_ts = node_checkpoints.get(node_dir)
|
|
if not checkpoint_ts:
|
|
continue
|
|
event_ts = _event_ts_from_filename(f.name)
|
|
if event_ts is None:
|
|
event_ts = int(f.stat().st_mtime)
|
|
if (event_ts < checkpoint_ts
|
|
and now - f.stat().st_mtime > three_days):
|
|
f.unlink(missing_ok=True)
|
|
logger.info(f"Cleaned old event: {f.name}")
|
|
except Exception as exc:
|
|
logger.error(f"Failed to remove {f}: {exc}")
|
|
else:
|
|
logger.debug("No observer checkpoint present; skipping event cleanup")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Optional: ship events to VPS via rsync
|
|
# ------------------------------------------------------------------
|
|
|
|
def _ship_events_to_vps(self):
|
|
"""
|
|
Rsync local events to VPS so the observer can process them.
|
|
Requires:
|
|
- VPS_EVENTS_HOST env var set to the VPS hostname/IP
|
|
- SSH key accessible inside the container (mount via docker-compose)
|
|
- The node is NOT VPS itself
|
|
"""
|
|
if not VPS_EVENTS_HOST or self.node_name == VPS_NODE_NAME:
|
|
return
|
|
|
|
local_dir = str(self._node_events_dir()) + "/"
|
|
remote_dir = (f"{VPS_EVENTS_USER}@{VPS_EVENTS_HOST}:"
|
|
f"{VPS_EVENTS_PATH}/{self.node_name}/")
|
|
cmd = [
|
|
"rsync", "-az", "--remove-source-files",
|
|
# --omit-dir-times: the remote per-node event dir on VPS is owned
|
|
# by aerbot:aerbot; the ssh user (oskar) is only a group member,
|
|
# so it can write files into the dir but cannot chown/chmod/touch
|
|
# the dir itself. -a implies -t (preserve times) which also
|
|
# applies to directories, so rsync would otherwise try to set the
|
|
# dir's mtime, get EPERM, and return non-zero even though every
|
|
# file transferred fine. Files still keep their mtime via -t;
|
|
# this only skips the (harmless, doomed) directory mtime set.
|
|
# TODO tech-debt: fix ownership of /opt/homelab/events/<node> on
|
|
# VPS so this workaround isn't needed (see docs/backlog.md,
|
|
# "Tech-debt: globalny porządek uid/gid/uprawnień").
|
|
"--omit-dir-times",
|
|
# --no-perms/--no-owner/--no-group: same root cause as --omit-dir-times above.
|
|
# After omitting dir times, rsync next tried chmod on the same aerbot-owned
|
|
# dir and got EPERM ("failed to set permissions", exit 23) — then would try
|
|
# chown/chgrp for the same reason. -a implies -pgo; we drop all three for the
|
|
# destination dir. File CONTENT still transfers correctly (that is all the
|
|
# observer needs); only attribute-setting on the remote dir is skipped.
|
|
# TODO tech-debt: fix ownership of /opt/homelab/events/<node> on VPS so none
|
|
# of these workarounds are needed (see docs/backlog.md, uid/gid section).
|
|
"--no-perms",
|
|
"--no-owner",
|
|
"--no-group",
|
|
# -F /dev/null: skip ~/.ssh/config entirely. The .ssh dir is
|
|
# mounted from the host oskar user into the container which runs
|
|
# as root; OpenSSH rejects config files owned by a different UID.
|
|
# UserKnownHostsFile=/dev/null pairs with StrictHostKeyChecking=no
|
|
# so we never try to write a known_hosts inside a read-only mount.
|
|
"-e", ("ssh -F /dev/null"
|
|
" -o StrictHostKeyChecking=no"
|
|
" -o UserKnownHostsFile=/dev/null"
|
|
" -o ConnectTimeout=10"
|
|
" -o BatchMode=yes"),
|
|
local_dir,
|
|
remote_dir,
|
|
]
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
|
if result.returncode == 0:
|
|
logger.debug(f"Events shipped to {remote_dir}")
|
|
else:
|
|
logger.warning(f"Event shipping failed: {result.stderr.strip()}")
|
|
except Exception as exc:
|
|
logger.warning(f"Event shipping error: {exc}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Remediation dispatch: pull queued actions for this node and execute them
|
|
# ------------------------------------------------------------------
|
|
|
|
def _dispatch_inbox_dir(self) -> Path:
|
|
return ACTIONS_DIR / "dispatch" / self.node_name
|
|
|
|
def _processed_marker_path(self, action_id: str) -> Path:
|
|
return STATE_DIR / "processed-actions" / f"{action_id}.done"
|
|
|
|
def _already_processed(self, action_id: str) -> bool:
|
|
return self._processed_marker_path(action_id).exists()
|
|
|
|
def _mark_processed(self, action_id: str):
|
|
marker = self._processed_marker_path(action_id)
|
|
try:
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
marker.touch()
|
|
except Exception as exc:
|
|
logger.error(f"Failed to record action {action_id} as processed: {exc}")
|
|
|
|
def pull_dispatched_actions(self):
|
|
"""
|
|
Rsync-pull this node's dispatch inbox from VPS.
|
|
|
|
Reuses the exact same SSH key / connection settings as
|
|
_ship_events_to_vps, just as sender/receiver reversed: VPS is now the
|
|
source, this node the destination. --remove-source-files deletes the
|
|
action file on VPS once it has been fetched, so a dispatch file is
|
|
collected by exactly one node and cannot be re-pulled after that.
|
|
|
|
Requires VPS_EVENTS_HOST (same var event shipping uses) and is a
|
|
no-op on VPS itself, whose node-agent reads the dispatch dir directly
|
|
off the shared /opt/homelab mount — no network hop needed.
|
|
"""
|
|
if not VPS_EVENTS_HOST or self.node_name == VPS_NODE_NAME:
|
|
return
|
|
|
|
inbox = self._dispatch_inbox_dir()
|
|
inbox.mkdir(parents=True, exist_ok=True)
|
|
local_dir = str(inbox) + "/"
|
|
remote_dir = (f"{VPS_EVENTS_USER}@{VPS_EVENTS_HOST}:"
|
|
f"{VPS_DISPATCH_PATH}/{self.node_name}/")
|
|
cmd = [
|
|
"rsync", "-az", "--remove-source-files",
|
|
"--omit-dir-times", "--no-perms", "--no-owner", "--no-group",
|
|
"-e", ("ssh -F /dev/null"
|
|
" -o StrictHostKeyChecking=no"
|
|
" -o UserKnownHostsFile=/dev/null"
|
|
" -o ConnectTimeout=10"
|
|
" -o BatchMode=yes"),
|
|
remote_dir,
|
|
local_dir,
|
|
]
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
|
# rsync returns 23/24 ("partial transfer"/"vanished source files")
|
|
# when the remote dispatch dir is simply empty — the common case,
|
|
# not an error worth logging every cycle.
|
|
if result.returncode not in (0, 23, 24):
|
|
logger.warning(f"Dispatch pull failed: {result.stderr.strip()}")
|
|
except Exception as exc:
|
|
logger.warning(f"Dispatch pull error: {exc}")
|
|
|
|
def process_dispatched_actions(self):
|
|
"""Execute every action currently sitting in this node's dispatch inbox."""
|
|
inbox = self._dispatch_inbox_dir()
|
|
if not inbox.exists():
|
|
return
|
|
for action_file in sorted(inbox.glob("*.json")):
|
|
try:
|
|
action = json.loads(action_file.read_text())
|
|
except Exception as exc:
|
|
logger.error(f"Failed to read dispatched action {action_file.name}: {exc}")
|
|
action_file.unlink(missing_ok=True)
|
|
continue
|
|
self._execute_dispatched_action(action)
|
|
# Delete regardless of outcome: a rejected/failed action is not
|
|
# retried automatically (the whole point of the action_result
|
|
# report is that the executor decides what happens next — retry
|
|
# is a fresh dispatch with a fresh action_id, not this file living on).
|
|
action_file.unlink(missing_ok=True)
|
|
|
|
def _execute_dispatched_action(self, action: dict):
|
|
"""
|
|
Validate and execute one dispatched action, then report the result.
|
|
|
|
Security gates, in order (each rejection short-circuits execution and
|
|
reports a clear action_result error rather than running anything):
|
|
1. Idempotency — already processed this action_id? no-op.
|
|
2. Node scoping — action addressed to a DIFFERENT node? refuse.
|
|
(Defense in depth: the dispatch dir is already scoped by node
|
|
name, but a dispatch file could in principle be mis-delivered.)
|
|
3. Type whitelist — only container_restart, nothing else.
|
|
4. Self-restart guard — never restart node-agent's own container.
|
|
"""
|
|
action_id = action.get("action_id") or "unknown"
|
|
node = action.get("node")
|
|
action_type = action.get("type")
|
|
container_name = action.get("container_name") or action.get("service")
|
|
|
|
if self._already_processed(action_id):
|
|
logger.info(f"Action {action_id} already processed — skipping (idempotency)")
|
|
return
|
|
|
|
if node != self.node_name:
|
|
self._report_action_result(
|
|
action_id, container_name, False,
|
|
f"Action addressed to node '{node}', not '{self.node_name}' — refused",
|
|
)
|
|
self._mark_processed(action_id)
|
|
return
|
|
|
|
if action_type not in ALLOWED_DISPATCH_ACTION_TYPES:
|
|
self._report_action_result(
|
|
action_id, container_name, False,
|
|
f"Action type '{action_type}' is not whitelisted for agent-side "
|
|
f"execution (allowed: {sorted(ALLOWED_DISPATCH_ACTION_TYPES)})",
|
|
)
|
|
self._mark_processed(action_id)
|
|
return
|
|
|
|
if not container_name:
|
|
self._report_action_result(
|
|
action_id, container_name, False, "No container_name in dispatched action",
|
|
)
|
|
self._mark_processed(action_id)
|
|
return
|
|
|
|
if container_name in SELF_RESTART_GUARD_NAMES:
|
|
self._report_action_result(
|
|
action_id, container_name, False,
|
|
f"Refusing to restart '{container_name}': node-agent will not restart itself",
|
|
)
|
|
self._mark_processed(action_id)
|
|
return
|
|
|
|
if not self.docker_client:
|
|
self._report_action_result(
|
|
action_id, container_name, False, "Docker SDK unavailable on this node",
|
|
)
|
|
self._mark_processed(action_id)
|
|
return
|
|
|
|
try:
|
|
container = self.docker_client.containers.get(container_name)
|
|
container.restart()
|
|
logger.info(f"Restarted container '{container_name}' for action {action_id}")
|
|
self._report_action_result(action_id, container_name, True, "")
|
|
except Exception as exc:
|
|
logger.error(f"Failed to restart '{container_name}' for action {action_id}: {exc}")
|
|
self._report_action_result(action_id, container_name, False, str(exc))
|
|
|
|
self._mark_processed(action_id)
|
|
|
|
def _report_action_result(self, action_id: str, container_name, success: bool, error: str):
|
|
"""Emit an action_result event — picked up by the executor's
|
|
_reconcile_running_actions() to move the action to completed/failed.
|
|
Rides the existing event pipeline (and, for remote nodes, the existing
|
|
_ship_events_to_vps rsync) with no changes to either.
|
|
"""
|
|
self.emit_event(
|
|
"action_result",
|
|
"info" if success else "high",
|
|
container_name or action_id,
|
|
f"Action {action_id} {'succeeded' if success else 'failed'}",
|
|
{
|
|
"action_id": action_id,
|
|
"success": success,
|
|
"error": error,
|
|
"node": self.node_name,
|
|
},
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# VPS-specific: control-plane service health check
|
|
# ------------------------------------------------------------------
|
|
|
|
def _check_control_plane_health(self):
|
|
"""
|
|
VPS-only: probe the control-plane HTTP endpoint and emit a service
|
|
health event so the observer can populate services.json for the
|
|
'control-plane' entry in services.yaml.
|
|
|
|
The control-plane is a multi-container stack (observer, supervisor,
|
|
executor, ui), so individual container names don't match the service
|
|
name in services.yaml. Checking the HTTP endpoint gives a clean
|
|
boundary that maps 1-to-1 with the logical service.
|
|
"""
|
|
import urllib.request
|
|
endpoint = "http://localhost:18180/summary"
|
|
try:
|
|
resp = urllib.request.urlopen(endpoint, timeout=5)
|
|
if resp.status == 200:
|
|
# Transition-only, same rationale as check_containers(): without
|
|
# this gate, control-plane re-confirmed itself every cycle
|
|
# forever, contributing to the same service_healthy flood.
|
|
if self._service_health_state.get("control-plane") is not True:
|
|
self.emit_event(
|
|
"service_healthy", "info", "control-plane",
|
|
"Control-plane HTTP endpoint is reachable",
|
|
{"endpoint": endpoint},
|
|
)
|
|
self._service_health_state["control-plane"] = True
|
|
else:
|
|
self._service_health_state["control-plane"] = False
|
|
self.emit_event(
|
|
"service_unhealthy", "high", "control-plane",
|
|
f"Control-plane HTTP endpoint returned HTTP {resp.status}",
|
|
{"endpoint": endpoint, "http_status": resp.status},
|
|
)
|
|
except Exception as exc:
|
|
self._service_health_state["control-plane"] = False
|
|
self.emit_event(
|
|
"service_unhealthy", "high", "control-plane",
|
|
f"Control-plane HTTP endpoint unreachable: {exc}",
|
|
{"endpoint": endpoint, "error": str(exc)},
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Heartbeat
|
|
# ------------------------------------------------------------------
|
|
|
|
def _update_heartbeat(self):
|
|
try:
|
|
(STATE_DIR / "node-agent.heartbeat").touch()
|
|
except Exception as exc:
|
|
logger.error(f"Failed to update heartbeat: {exc}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Main loop
|
|
# ------------------------------------------------------------------
|
|
|
|
def run_once(self):
|
|
self._update_heartbeat()
|
|
|
|
disk_pct = self.check_disk()
|
|
mem_pct = self.check_memory()
|
|
cpu_pct = self.check_cpu()
|
|
self.check_containers()
|
|
|
|
self.run_safe_cleanup()
|
|
|
|
if self.node_name == VPS_NODE_NAME:
|
|
self._cleanup_control_plane_fs()
|
|
self._check_control_plane_health()
|
|
|
|
# Remediation dispatch: fetch and execute any action the executor has
|
|
# queued for this node, then report the outcome (via emit_event below,
|
|
# shipped in the same cycle by _ship_events_to_vps).
|
|
self.pull_dispatched_actions()
|
|
self.process_dispatched_actions()
|
|
|
|
# Emit a node_health heartbeat so the observer can update node status
|
|
# and the supervisor can correlate disk/memory metrics with service issues.
|
|
self.emit_event(
|
|
"node_health", "info", None,
|
|
f"Health check completed on {self.node_name}",
|
|
{"disk_pct": disk_pct, "mem_pct": mem_pct, "cpu_pct": cpu_pct},
|
|
)
|
|
|
|
self._ship_events_to_vps()
|
|
|
|
def loop(self, interval: int = HEALTH_CHECK_INTERVAL):
|
|
logger.info(
|
|
f"node-agent ready — node={self.node_name} type={self.node_type} "
|
|
f"interval={interval}s"
|
|
)
|
|
while True:
|
|
try:
|
|
self.run_once()
|
|
except Exception as exc:
|
|
logger.error(f"Health cycle error: {exc}")
|
|
time.sleep(interval)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
NodeAgent().loop()
|