diff --git a/services/node-agent/src/node_agent.py b/services/node-agent/src/node_agent.py index 2cbb3e2..6a6d778 100644 --- a/services/node-agent/src/node_agent.py +++ b/services/node-agent/src/node_agent.py @@ -94,6 +94,12 @@ 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")) +# Event shipping: max files per rsync batch so each push fits within timeout +SHIP_BATCH_SIZE = int(os.getenv("SHIP_BATCH_SIZE", "1000")) +# Warn when unsent backlog exceeds this; hard-trim when exceeding BACKLOG_MAX +BACKLOG_WARN = 2000 +BACKLOG_MAX = 5000 + # --------------------------------------------------------------------------- # Helpers @@ -524,42 +530,117 @@ class NodeAgent: logger.debug("No observer checkpoint present; skipping event cleanup") # ------------------------------------------------------------------ - # Optional: ship events to VPS via rsync + # Event backlog safety-net # ------------------------------------------------------------------ - 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 + def _trim_events_backlog(self): + """Remove oldest local events when backlog exceeds BACKLOG_MAX. + + Guards against the rsync-death-spiral: if shipping fails for an + extended period, local events accumulate without bound. When the + directory exceeds BACKLOG_MAX files, the oldest (sorted ascending + by filename, which encodes a Unix timestamp) are deleted first — + week-old health-check JSON is worthless; keeping the queue bounded + is more important. Runs only on shipping nodes (VPS_EVENTS_HOST + set); VPS cleans its own events via _cleanup_control_plane_fs. """ if not VPS_EVENTS_HOST or self.node_name == VPS_NODE_NAME: return - local_dir = str(self._node_events_dir()) + "/" + events_dir = self._node_events_dir() + try: + files = sorted(events_dir.glob("*.json")) + count = len(files) + except Exception as exc: + logger.error(f"backlog trim: failed to scan {events_dir}: {exc}") + return + + if count <= BACKLOG_MAX: + return + + to_delete = files[:count - BACKLOG_MAX] # oldest first (path sorts by ts) + deleted = 0 + for f in to_delete: + try: + f.unlink() + deleted += 1 + except Exception as exc: + logger.error(f"backlog trim: failed to delete {f.name}: {exc}") + + logger.warning( + f"backlog trim: deleted {deleted} oldest events " + f"(had {count}, trimmed to {BACKLOG_MAX})" + ) + + # ------------------------------------------------------------------ + # Optional: ship events to VPS via rsync + # ------------------------------------------------------------------ + + def _ship_events_to_vps(self): + """Rsync a batch of local events to VPS. + + Sends at most SHIP_BATCH_SIZE (default 1000) files per cycle so + each push completes within the subprocess timeout even when a large + backlog has accumulated. Files are selected oldest-first (sorted + by filename, which encodes a Unix timestamp); the observer processes + them in order via its per-node checkpoint. + + Requires VPS_EVENTS_HOST set and an SSH key accessible inside the + container. Does nothing on VPS itself. + """ + if not VPS_EVENTS_HOST or self.node_name == VPS_NODE_NAME: + return + + events_dir = self._node_events_dir() + try: + all_files = sorted(events_dir.glob("*.json")) + except Exception as exc: + logger.error(f"Failed to list events dir {events_dir}: {exc}") + return + + total = len(all_files) + if total == 0: + return + if total > BACKLOG_WARN: + logger.warning(f"events backlog: {total} unsent files in {events_dir}") + + batch = all_files[:SHIP_BATCH_SIZE] + # Pass filenames via stdin with --files-from=- to avoid ARG_MAX limits + # and to ship only the selected batch (not the whole directory). + files_input = "\n".join(f.name for f in batch) + remote_dir = (f"{VPS_EVENTS_USER}@{VPS_EVENTS_HOST}:" f"{VPS_EVENTS_PATH}/{self.node_name}/") cmd = [ "rsync", "-az", "--remove-source-files", + "--files-from=-", # -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. + # mounted from the host user into the container; 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. + # so we never 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, + str(events_dir) + "/", remote_dir, ] try: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + result = subprocess.run( + cmd, + input=files_input, + capture_output=True, + text=True, + timeout=120, + ) if result.returncode == 0: - logger.debug(f"Events shipped to {remote_dir}") + remaining = total - len(batch) + logger.debug( + f"Shipped {len(batch)}/{total} events to {remote_dir}" + + (f" ({remaining} remaining)" if remaining else "") + ) else: logger.warning(f"Event shipping failed: {result.stderr.strip()}") except Exception as exc: @@ -639,6 +720,7 @@ class NodeAgent: {"disk_pct": disk_pct, "mem_pct": mem_pct, "cpu_pct": cpu_pct}, ) + self._trim_events_backlog() self._ship_events_to_vps() def loop(self, interval: int = HEALTH_CHECK_INTERVAL):