oskar/homelab-codex-ws
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

feat(remediation): node-agent wykonuje zlecone akcje lokalnie — koniec SSH z executora

Browse source 
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>
This commit is contained in:
oskar 2026-07-22 16:58:07 +02:00
parent 1b48fad451
commit 2dac154d78
4 changed files with 850 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

191
services/control-plane/src/executor.py
View file

@ -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-<node>-<unixts>-<type>-<svc>.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 <container_name>`.
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:
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"Container '{container_name}' on {node} restarted successfully "
f"(attempt {attempt}/{max_attempts})"
f"Dispatched container_restart {action_id} "
f"(container={container_name}) to node-agent on {node}"
)
return True, ""
except Exception as e:
logger.error(f"Failed to dispatch {action_id} to {node}: {e}")
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}"
def _reconcile_running_actions(self):
"""Resolve container_restart actions previously dispatched to a node-agent.
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)
if attempt < max_attempts:
logger.info(f"Retrying in {retry_delay}s...")
time.sleep(retry_delay)
def _find_action_result(self, node, action_id, started_at):
"""Look for an action_result event from `node` reporting on `action_id`.
logger.error(
f"container_restart exhausted all {max_attempts} attempts "
f"for '{container_name}' on {node}"
)
return False, last_error
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):
"""

236
services/control-plane/tests/test_executor_dispatch.py Normal file
View file

@ -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-<node>-<service>) 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"))

213
services/node-agent/src/node_agent.py
View file

@ -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/<node-name>/.
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/<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
# ---------------------------------------------------------------------------
@ -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(

250
services/node-agent/tests/test_action_dispatch.py Normal file
View file

@ -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