From 2dac154d78b859d9e93cec1740b27b4ec26745da Mon Sep 17 00:00:00 2001 From: oskar Date: Wed, 22 Jul 2026 16:58:07 +0200 Subject: [PATCH] =?UTF-8?q?feat(remediation):=20node-agent=20wykonuje=20zl?= =?UTF-8?q?econe=20akcje=20lokalnie=20=E2=80=94=20koniec=20SSH=20z=20execu?= =?UTF-8?q?tora?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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//.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 --- services/control-plane/src/executor.py | 197 ++++++++++---- .../tests/test_executor_dispatch.py | 236 +++++++++++++++++ services/node-agent/src/node_agent.py | 213 ++++++++++++++- .../node-agent/tests/test_action_dispatch.py | 250 ++++++++++++++++++ 4 files changed, 850 insertions(+), 46 deletions(-) create mode 100644 services/control-plane/tests/test_executor_dispatch.py create mode 100644 services/node-agent/tests/test_action_dispatch.py diff --git a/services/control-plane/src/executor.py b/services/control-plane/src/executor.py index 642f721..710c34e 100644 --- a/services/control-plane/src/executor.py +++ b/services/control-plane/src/executor.py @@ -1,4 +1,5 @@ import os +import re import json import time import logging @@ -18,10 +19,15 @@ def _atomic_write_json(path: Path, data) -> None: # Constants and Paths RUNTIME_PATH = os.getenv("RUNTIME_PATH", "/opt/homelab") ACTIONS_DIR = Path(RUNTIME_PATH) / "actions" +EVENTS_DIR = Path(RUNTIME_PATH) / "events" +DISPATCH_DIR = ACTIONS_DIR / "dispatch" REPO_ROOT = Path(os.getenv("REPO_ROOT", "/repo")) # SSH configuration # SSH_USER can be overridden per-deployment environment. +# Still used by _execute_disk_cleanup (out of scope for this change — see +# docs/backlog.md "shadow_mode -> remediacja"). container_restart no longer +# uses SSH: see _dispatch_container_restart / _reconcile_running_actions. SSH_USER = os.getenv("SSH_USER", "oskar") SSH_OPTIONS = [ "-o", "StrictHostKeyChecking=no", @@ -29,6 +35,16 @@ SSH_OPTIONS = [ "-o", "BatchMode=yes", ] +# How long a container_restart action may sit in "running" waiting for the +# target node-agent to report a result before the executor gives up and marks +# it failed. Env-overridable so a slow/flaky LTE node can be tuned without a +# code change. +ACTION_TIMEOUT_SECS = int(os.getenv("ACTION_TIMEOUT_SECS", "300")) + +# Matches evt----.json, same convention as +# node_agent.py / observer.py / operator_ui.py. +_EVENT_TS_RE = re.compile(r"-(\d{9,11})-") + # Logging setup logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger("executor") @@ -41,6 +57,7 @@ class Executor: def _ensure_dirs(self): for s in ["approved", "running", "completed", "failed", "rejected"]: (ACTIONS_DIR / s).mkdir(parents=True, exist_ok=True) + DISPATCH_DIR.mkdir(parents=True, exist_ok=True) def process_actions(self): # Update heartbeat @@ -50,6 +67,11 @@ class Executor: except Exception as e: logger.error(f"Failed to touch heartbeat file: {e}") + # Resolve container_restart actions dispatched to a node-agent on a + # previous cycle before dispatching new work, so a result that landed + # this cycle is reflected immediately rather than waiting a cycle. + self._reconcile_running_actions() + approved_dir = ACTIONS_DIR / "approved" action_files = sorted(approved_dir.glob("*.json")) @@ -97,10 +119,17 @@ class Executor: error_msg = result.stderr or result.stdout elif action_type == "container_restart": - # Lightweight restart: SSH to node and docker restart the container. + # No SSH from the executor (see CLAUDE.md / docs/backlog.md): the + # action is handed to the node-agent running on the target node, + # which restarts the container locally via its own docker socket. # container_name is set by the supervisor; falls back to service name. container_name = data.get("container_name") or service - success, error_msg = self._execute_container_restart(node, container_name) + self._dispatch_container_restart(action_id, node, service, container_name) + # Stays in "running" — _reconcile_running_actions() resolves it + # on a later cycle once the node-agent reports a result (or the + # action times out). Do not fall through to the completed/failed + # move below. + return elif action_type == "disk_cleanup": # Operator-approved aggressive Docker cleanup (image prune -a + @@ -122,7 +151,10 @@ class Executor: success = False error_msg = str(e) - # Move to completed/failed + self._finalize_action(action_id, running_path, data, success, error_msg) + + def _finalize_action(self, action_id, running_path, data, success, error_msg): + """Move a running action to completed/ or failed/, recording the result.""" target_status = "completed" if success else "failed" target_path = ACTIONS_DIR / target_status / f"{action_id}.json" try: @@ -136,52 +168,127 @@ class Executor: except Exception as e: logger.error(f"Failed to move {action_id} to {target_status}: {e}") - def _execute_container_restart(self, node, container_name, retry_delay=10): + # ------------------------------------------------------------------ + # container_restart: dispatch to node-agent, no SSH from the executor + # ------------------------------------------------------------------ + # + # The executor has no SSH client and no key to the fleet (deliberate — + # see docs/backlog.md "PROJEKT: remediacja bez SSH"). Instead, it writes a + # small dispatch file that the node-agent running ON the target node picks + # up and executes locally through its own docker socket. This works + # identically whether the target is a remote node (piha/solaria/... reach + # the file via their existing rsync-pull, reusing the SSH key node-agent + # already has for event shipping) or the local VPS node (its node-agent + # reads the same on-disk path directly — no network hop at all). + + def _dispatch_container_restart(self, action_id, node, service, container_name): + """Write a dispatch file for the node-agent on `node` to pick up. + + Does not resolve the action itself — it stays in running/ until + _reconcile_running_actions() sees an action_result event or the + action times out. """ - SSH to the target node and run `docker restart `. - - Attempts the restart up to 2 times (initial + 1 retry). If the first - attempt fails, waits retry_delay seconds then tries once more before - declaring the action failed. - - Returns (success: bool, error_msg: str). - """ - cmd = [ - "ssh", - *SSH_OPTIONS, - f"{SSH_USER}@{node}", - f"docker restart {container_name}", - ] - logger.info(f"SSH container restart: {' '.join(cmd)}") - - max_attempts = 2 - last_error = "" - - for attempt in range(1, max_attempts + 1): - result = subprocess.run(cmd, capture_output=True, text=True) - - if result.returncode == 0: - logger.info( - f"Container '{container_name}' on {node} restarted successfully " - f"(attempt {attempt}/{max_attempts})" - ) - return True, "" - - last_error = (result.stderr or result.stdout).strip() - logger.warning( - f"container_restart attempt {attempt}/{max_attempts} failed " - f"for '{container_name}' on {node}: {last_error}" + if not node: + logger.error(f"Action {action_id}: container_restart with no node set") + return + inbox = DISPATCH_DIR / node + inbox.mkdir(parents=True, exist_ok=True) + payload = { + "action_id": action_id, + "type": "container_restart", + "node": node, + "service": service, + "container_name": container_name, + "dispatched_at": time.time(), + } + try: + _atomic_write_json(inbox / f"{action_id}.json", payload) + logger.info( + f"Dispatched container_restart {action_id} " + f"(container={container_name}) to node-agent on {node}" ) + except Exception as e: + logger.error(f"Failed to dispatch {action_id} to {node}: {e}") - if attempt < max_attempts: - logger.info(f"Retrying in {retry_delay}s...") - time.sleep(retry_delay) + def _reconcile_running_actions(self): + """Resolve container_restart actions previously dispatched to a node-agent. - logger.error( - f"container_restart exhausted all {max_attempts} attempts " - f"for '{container_name}' on {node}" - ) - return False, last_error + Other action types (redeploy/disk_cleanup/alert_only) resolve + synchronously inside _execute_action and never linger in running/, so + they are not touched here. + """ + running_dir = ACTIONS_DIR / "running" + if not running_dir.exists(): + return + + for action_file in sorted(running_dir.glob("*.json")): + try: + with open(action_file, "r") as f: + data = json.load(f) + except Exception as e: + logger.error(f"Failed to read running action {action_file.name}: {e}") + continue + + if data.get("type") != "container_restart": + continue + + action_id = data.get("action_id") or action_file.stem + node = data.get("node") + started_at = data.get("started_at") or 0 + + result = self._find_action_result(node, action_id, started_at) + if result is not None: + success = bool(result.get("success")) + error_msg = "" if success else (result.get("error") or "node-agent reported failure") + self._finalize_action(action_id, action_file, data, success, error_msg) + continue + + if started_at and (time.time() - started_at) > ACTION_TIMEOUT_SECS: + error_msg = ( + f"Timed out after {ACTION_TIMEOUT_SECS}s waiting for node-agent " + f"on '{node}' to report a result for container_restart " + f"(action_id={action_id})" + ) + logger.error(f"Action {action_id} timed out: {error_msg}") + self._finalize_action(action_id, action_file, data, False, error_msg) + + def _find_action_result(self, node, action_id, started_at): + """Look for an action_result event from `node` reporting on `action_id`. + + Only considers events at or after `started_at` so a stale action_result + left over from a PREVIOUS run of the same (deterministic) action_id can + never be mistaken for the current run's outcome. Returns the event's + payload dict, or None if no matching result has arrived yet. + """ + if not node: + return None + node_events_dir = EVENTS_DIR / node + if not node_events_dir.exists(): + return None + + # Event timestamps are always whole seconds (int(time.time()) in + # node_agent.emit_event); started_at is a float. Floor started_at to + # whole seconds before comparing so a result emitted in the SAME + # wall-clock second as the dispatch (started_at's fractional part + # ahead of the truncated event second) is not mistaken for "stale". + started_at_floor = int(started_at) + + for event_file in node_events_dir.glob("evt-*-action_result-*.json"): + m = _EVENT_TS_RE.search(event_file.stem) + if m and int(m.group(1)) < started_at_floor: + continue + try: + with open(event_file, "r") as f: + event = json.load(f) + except Exception: + continue + payload = event.get("payload", {}) + if payload.get("action_id") == action_id: + event_ts = event.get("timestamp") or 0 + if event_ts and event_ts < started_at_floor: + continue + return payload + return None def _execute_disk_cleanup(self, node: str, payload: dict): """ diff --git a/services/control-plane/tests/test_executor_dispatch.py b/services/control-plane/tests/test_executor_dispatch.py new file mode 100644 index 0000000..f1ef3b4 --- /dev/null +++ b/services/control-plane/tests/test_executor_dispatch.py @@ -0,0 +1,236 @@ +"""Tests for Executor container_restart dispatch — the no-SSH remediation path. + +Covers docs/backlog.md "PROJEKT: remediacja bez SSH": the executor never +shells out to SSH for container_restart. Instead it drops a dispatch file for +the target node's node-agent to pick up, and later resolves the action from +either a matching action_result event or a timeout — never leaving it running +forever. +""" +from __future__ import annotations + +import json +import sys +import time +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +import executor as executor_module +from executor import Executor + + +def _setup_executor(tmp_path: Path, monkeypatch) -> Executor: + actions = tmp_path / "actions" + events = tmp_path / "events" + repo = tmp_path / "repo" + state = tmp_path / "state" + for d in (actions, events, repo, state): + d.mkdir(parents=True, exist_ok=True) + + monkeypatch.setattr(executor_module, "ACTIONS_DIR", actions) + monkeypatch.setattr(executor_module, "EVENTS_DIR", events) + monkeypatch.setattr(executor_module, "DISPATCH_DIR", actions / "dispatch") + monkeypatch.setattr(executor_module, "REPO_ROOT", repo) + return Executor() + + +def _write_approved(tmp_path, action_id, node="piha", service="zigbee2mqtt", + container_name=None, action_type="container_restart"): + action = { + "action_id": action_id, + "type": action_type, + "node": node, + "service": service, + "container_name": container_name or service, + "status": "approved", + "timestamp": time.time(), + } + path = tmp_path / "actions" / "approved" / f"{action_id}.json" + path.write_text(json.dumps(action)) + return path + + +def _write_action_result_event(tmp_path, node, action_id, success, error="", ts=None): + ts = int(ts if ts is not None else time.time()) + node_dir = tmp_path / "events" / node + node_dir.mkdir(parents=True, exist_ok=True) + event = { + "id": f"evt-{node}-{ts}-action_result-{action_id}", + "timestamp": ts, + "type": "action_result", + "node": node, + "payload": {"action_id": action_id, "success": success, "error": error, "node": node}, + } + path = node_dir / f"evt-{node}-{ts}-action_result-{action_id}.json" + path.write_text(json.dumps(event)) + return path + + +def _read(tmp_path, state, action_id): + return json.loads((tmp_path / "actions" / state / f"{action_id}.json").read_text()) + + +def _exists(tmp_path, state, action_id): + return (tmp_path / "actions" / state / f"{action_id}.json").exists() + + +# --------------------------------------------------------------------------- +# Dispatch: no SSH, writes to DISPATCH_DIR, stays in running +# --------------------------------------------------------------------------- + +def test_container_restart_dispatched_not_ssh(tmp_path, monkeypatch): + ex = _setup_executor(tmp_path, monkeypatch) + called = {"ssh": False} + + def fake_run(cmd, **kwargs): + if cmd and cmd[0] == "ssh": + called["ssh"] = True + raise AssertionError("executor must never invoke subprocess for container_restart") + + monkeypatch.setattr(executor_module.subprocess, "run", fake_run) + + action_file = _write_approved(tmp_path, "cr-1", node="piha", container_name="zigbee2mqtt") + ex._execute_action(action_file) + + assert not called["ssh"] + dispatch_file = tmp_path / "actions" / "dispatch" / "piha" / "cr-1.json" + assert dispatch_file.exists() + payload = json.loads(dispatch_file.read_text()) + assert payload["action_id"] == "cr-1" + assert payload["container_name"] == "zigbee2mqtt" + assert payload["node"] == "piha" + + # Stays in running — not resolved synchronously. + assert _exists(tmp_path, "running", "cr-1") + assert not _exists(tmp_path, "completed", "cr-1") + assert not _exists(tmp_path, "failed", "cr-1") + + +def test_container_restart_with_no_node_does_not_crash(tmp_path, monkeypatch): + ex = _setup_executor(tmp_path, monkeypatch) + action_file = _write_approved(tmp_path, "cr-none", node="", container_name="x") + + ex._execute_action(action_file) # must not raise + + assert list((tmp_path / "actions" / "dispatch").glob("*/cr-none.json")) == [] + + +# --------------------------------------------------------------------------- +# Reconcile: action_result event resolves the action +# --------------------------------------------------------------------------- + +def test_reconcile_moves_to_completed_on_success_result(tmp_path, monkeypatch): + ex = _setup_executor(tmp_path, monkeypatch) + action_file = _write_approved(tmp_path, "cr-2", node="piha") + ex._execute_action(action_file) + + _write_action_result_event(tmp_path, "piha", "cr-2", success=True) + ex._reconcile_running_actions() + + assert _exists(tmp_path, "completed", "cr-2") + assert not _exists(tmp_path, "running", "cr-2") + + +def test_reconcile_moves_to_failed_on_failure_result(tmp_path, monkeypatch): + ex = _setup_executor(tmp_path, monkeypatch) + action_file = _write_approved(tmp_path, "cr-3", node="piha") + ex._execute_action(action_file) + + _write_action_result_event(tmp_path, "piha", "cr-3", success=False, error="no such container") + ex._reconcile_running_actions() + + assert _exists(tmp_path, "failed", "cr-3") + data = _read(tmp_path, "failed", "cr-3") + assert data["error"] == "no such container" + + +def test_reconcile_ignores_stale_result_from_previous_run(tmp_path, monkeypatch): + """A same-action_id result left over from an earlier run (before this run's + started_at) must never resolve the CURRENT run — action_ids are + deterministic (container-restart--) and can repeat days + apart.""" + ex = _setup_executor(tmp_path, monkeypatch) + + # Stale result from "yesterday" + stale_ts = int(time.time()) - 86_400 + _write_action_result_event(tmp_path, "piha", "cr-4", success=True, ts=stale_ts) + + action_file = _write_approved(tmp_path, "cr-4", node="piha") + ex._execute_action(action_file) # started_at is "now", after the stale event + + ex._reconcile_running_actions() + + assert _exists(tmp_path, "running", "cr-4") + assert not _exists(tmp_path, "completed", "cr-4") + + +def test_reconcile_no_op_when_no_result_and_not_timed_out(tmp_path, monkeypatch): + ex = _setup_executor(tmp_path, monkeypatch) + action_file = _write_approved(tmp_path, "cr-5", node="piha") + ex._execute_action(action_file) + + ex._reconcile_running_actions() + + assert _exists(tmp_path, "running", "cr-5") + + +# --------------------------------------------------------------------------- +# Timeout +# --------------------------------------------------------------------------- + +def test_reconcile_times_out_stuck_action(tmp_path, monkeypatch): + monkeypatch.setattr(executor_module, "ACTION_TIMEOUT_SECS", 5) + ex = _setup_executor(tmp_path, monkeypatch) + action_file = _write_approved(tmp_path, "cr-6", node="piha") + ex._execute_action(action_file) + + running_path = tmp_path / "actions" / "running" / "cr-6.json" + data = json.loads(running_path.read_text()) + data["started_at"] = time.time() - 10 # older than the 5s timeout + running_path.write_text(json.dumps(data)) + + ex._reconcile_running_actions() + + assert _exists(tmp_path, "failed", "cr-6") + data = _read(tmp_path, "failed", "cr-6") + assert "Timed out" in data["error"] + + +def test_reconcile_does_not_time_out_before_deadline(tmp_path, monkeypatch): + monkeypatch.setattr(executor_module, "ACTION_TIMEOUT_SECS", 300) + ex = _setup_executor(tmp_path, monkeypatch) + action_file = _write_approved(tmp_path, "cr-7", node="piha") + ex._execute_action(action_file) + + ex._reconcile_running_actions() + + assert _exists(tmp_path, "running", "cr-7") + + +# --------------------------------------------------------------------------- +# Other action types are untouched by the reconcile loop +# --------------------------------------------------------------------------- + +def test_reconcile_ignores_non_container_restart_running_actions(tmp_path, monkeypatch): + ex = _setup_executor(tmp_path, monkeypatch) + running_path = tmp_path / "actions" / "running" / "redeploy-x.json" + running_path.write_text(json.dumps({ + "action_id": "redeploy-x", "type": "redeploy", "node": "piha", + "started_at": time.time() - 999_999, + })) + + ex._reconcile_running_actions() + + assert running_path.exists() # untouched: redeploy resolves synchronously elsewhere + + +def test_alert_only_action_resolves_synchronously_not_via_dispatch(tmp_path, monkeypatch): + ex = _setup_executor(tmp_path, monkeypatch) + action_file = _write_approved(tmp_path, "alert-1", node="piha", action_type="alert_only") + + ex._execute_action(action_file) + + assert _exists(tmp_path, "completed", "alert-1") + assert not (tmp_path / "actions" / "dispatch").exists() or \ + not list((tmp_path / "actions" / "dispatch").glob("**/*.json")) diff --git a/services/node-agent/src/node_agent.py b/services/node-agent/src/node_agent.py index a457914..78db505 100644 --- a/services/node-agent/src/node_agent.py +++ b/services/node-agent/src/node_agent.py @@ -6,7 +6,12 @@ Runs as a Docker container on every managed node. Each cycle it: 2. Checks Docker container health. 3. Emits structured event JSON files to /opt/homelab/events//. 4. Applies safe Docker / filesystem cleanup per the conservative policy. - 5. Optionally rsyncs events to VPS so the control-plane observer can process them. + 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 @@ -79,6 +84,32 @@ 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// 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 # --------------------------------------------------------------------------- @@ -804,6 +835,180 @@ class NodeAgent: 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 # ------------------------------------------------------------------ @@ -877,6 +1082,12 @@ class NodeAgent: 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( diff --git a/services/node-agent/tests/test_action_dispatch.py b/services/node-agent/tests/test_action_dispatch.py new file mode 100644 index 0000000..f7b033c --- /dev/null +++ b/services/node-agent/tests/test_action_dispatch.py @@ -0,0 +1,250 @@ +"""Tests for NodeAgent remediation dispatch: pull_dispatched_actions / +process_dispatched_actions / _execute_dispatched_action. + +Covers the security gates required by docs/backlog.md "PROJEKT: remediacja +bez SSH": the agent executes only actions addressed to its own node, refuses +anything outside the container_restart whitelist, never restarts its own +container, and treats a repeated dispatch of the same action_id as a no-op. +""" +from __future__ import annotations + +import json +import shutil +import time +from unittest.mock import MagicMock + +import pytest + +import node_agent + + +@pytest.fixture(autouse=True) +def _isolated_runtime_state(): + """ACTIONS_DIR/STATE_DIR/EVENTS_DIR are a single temp dir shared for the + whole test session (conftest.py sets RUNTIME_PATH once at import time). + Clear the dispatch/processed-actions/events state before and after each + test so tests don't leak into each other or other test modules.""" + def _clear(): + for d in (node_agent.ACTIONS_DIR, node_agent.EVENTS_DIR, node_agent.STATE_DIR): + if d.exists(): + shutil.rmtree(d) + d.mkdir(parents=True, exist_ok=True) + + _clear() + yield + _clear() + + +def _write_dispatch(agent, action_id, node, action_type="container_restart", + container_name="zigbee2mqtt"): + inbox = agent._dispatch_inbox_dir() + inbox.mkdir(parents=True, exist_ok=True) + path = inbox / f"{action_id}.json" + path.write_text(json.dumps({ + "action_id": action_id, + "type": action_type, + "node": node, + "service": container_name, + "container_name": container_name, + "dispatched_at": time.time(), + })) + return path + + +def _fake_docker_client(): + client = MagicMock() + container = MagicMock() + client.containers.get.return_value = container + return client, container + + +def _last_action_result_payload(agent): + events_dir = agent._node_events_dir() + files = sorted(events_dir.glob("evt-*-action_result-*.json")) + assert files, "expected an action_result event to have been emitted" + return json.loads(files[-1].read_text())["payload"] + + +# --------------------------------------------------------------------------- +# Node scoping +# --------------------------------------------------------------------------- + +def test_executes_action_addressed_to_own_node(agent): + agent.docker_client, container = _fake_docker_client() + _write_dispatch(agent, "act-1", node=agent.node_name) + + agent.process_dispatched_actions() + + container.restart.assert_called_once() + payload = _last_action_result_payload(agent) + assert payload["action_id"] == "act-1" + assert payload["success"] is True + + +def test_refuses_action_addressed_to_another_node(agent): + agent.docker_client, container = _fake_docker_client() + _write_dispatch(agent, "act-2", node="some-other-node") + # Force it into THIS node's inbox directly (simulates a mis-delivery — + # the dispatch dir is normally already scoped by node name). + inbox = agent._dispatch_inbox_dir() + action_file = inbox / "act-2.json" + data = json.loads(action_file.read_text()) + assert data["node"] == "some-other-node" + + agent.process_dispatched_actions() + + container.restart.assert_not_called() + payload = _last_action_result_payload(agent) + assert payload["success"] is False + assert "not '" + agent.node_name + "'" in payload["error"] + + +# --------------------------------------------------------------------------- +# Type whitelist +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("bad_type", ["redeploy", "disk_cleanup", "shell_exec", "alert_only"]) +def test_rejects_non_whitelisted_action_types(agent, bad_type): + agent.docker_client, container = _fake_docker_client() + _write_dispatch(agent, f"act-{bad_type}", node=agent.node_name, action_type=bad_type) + + agent.process_dispatched_actions() + + container.restart.assert_not_called() + payload = _last_action_result_payload(agent) + assert payload["success"] is False + assert "not whitelisted" in payload["error"] + + +def test_accepts_container_restart(agent): + assert "container_restart" in node_agent.ALLOWED_DISPATCH_ACTION_TYPES + assert node_agent.ALLOWED_DISPATCH_ACTION_TYPES == {"container_restart"} + + +# --------------------------------------------------------------------------- +# Self-restart guard +# --------------------------------------------------------------------------- + +def test_refuses_to_restart_itself(agent): + agent.docker_client, container = _fake_docker_client() + _write_dispatch(agent, "act-self", node=agent.node_name, container_name="node-agent") + + agent.process_dispatched_actions() + + container.restart.assert_not_called() + agent.docker_client.containers.get.assert_not_called() + payload = _last_action_result_payload(agent) + assert payload["success"] is False + assert "restart itself" in payload["error"] + + +# --------------------------------------------------------------------------- +# Idempotency +# --------------------------------------------------------------------------- + +def test_same_action_id_executes_only_once(agent): + agent.docker_client, container = _fake_docker_client() + _write_dispatch(agent, "act-dup", node=agent.node_name) + + agent.process_dispatched_actions() + assert container.restart.call_count == 1 + + # Dispatch file is consumed after processing (real pipeline: rsync + # --remove-source-files means it can't be re-pulled either), but even if + # the SAME action_id is re-delivered, the processed marker must block it. + _write_dispatch(agent, "act-dup", node=agent.node_name) + agent.process_dispatched_actions() + + assert container.restart.call_count == 1, "action_id re-delivery must be a no-op" + + +def test_dispatch_file_removed_after_processing(agent): + agent.docker_client, _ = _fake_docker_client() + action_file = _write_dispatch(agent, "act-cleanup", node=agent.node_name) + + agent.process_dispatched_actions() + + assert not action_file.exists() + + +# --------------------------------------------------------------------------- +# Docker restart failure surfaces as a failed action_result +# --------------------------------------------------------------------------- + +def test_docker_restart_exception_reports_failure(agent): + client = MagicMock() + client.containers.get.side_effect = Exception("no such container") + agent.docker_client = client + _write_dispatch(agent, "act-fail", node=agent.node_name) + + agent.process_dispatched_actions() + + payload = _last_action_result_payload(agent) + assert payload["success"] is False + assert "no such container" in payload["error"] + + +def test_no_docker_client_reports_failure(agent): + agent.docker_client = None + _write_dispatch(agent, "act-nodocker", node=agent.node_name) + + agent.process_dispatched_actions() + + payload = _last_action_result_payload(agent) + assert payload["success"] is False + assert "Docker SDK unavailable" in payload["error"] + + +# --------------------------------------------------------------------------- +# pull_dispatched_actions: rsync gating (mirrors test_ship_events_to_vps.py) +# --------------------------------------------------------------------------- + +def test_pull_skips_when_vps_host_not_set(agent, monkeypatch): + monkeypatch.setattr(node_agent, "VPS_EVENTS_HOST", "") + fake_run = MagicMock() + monkeypatch.setattr(node_agent.subprocess, "run", fake_run) + + agent.pull_dispatched_actions() + + fake_run.assert_not_called() + + +def test_pull_skips_on_vps_node(agent, monkeypatch): + agent.node_name = node_agent.VPS_NODE_NAME + fake_run = MagicMock() + monkeypatch.setattr(node_agent.subprocess, "run", fake_run) + + agent.pull_dispatched_actions() + + fake_run.assert_not_called() + + +def test_pull_invokes_rsync_pull_direction(agent, monkeypatch): + captured = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + return MagicMock(returncode=0, stderr="") + + monkeypatch.setattr(node_agent.subprocess, "run", fake_run) + agent.pull_dispatched_actions() + + cmd = captured["cmd"] + assert cmd[0] == "rsync" + assert "--remove-source-files" in cmd + # Source (remote VPS) comes before destination (local inbox) — pull, not push. + remote_arg = f"{node_agent.VPS_EVENTS_USER}@{node_agent.VPS_EVENTS_HOST}:" + assert any(a.startswith(remote_arg) for a in cmd[:-1]) + assert cmd[-1] == str(agent._dispatch_inbox_dir()) + "/" + + +def test_pull_treats_empty_source_returncodes_as_non_error(agent, monkeypatch, caplog): + def fake_run(cmd, **kwargs): + return MagicMock(returncode=23, stderr="rsync: some vanished-source message") + + monkeypatch.setattr(node_agent.subprocess, "run", fake_run) + + with caplog.at_level("WARNING"): + agent.pull_dispatched_actions() + + assert "Dispatch pull failed" not in caplog.text