fix(node-agent): batch rsync, backlog trim, timeout 120s, backlog warn
Root cause of fleet staleness since ~OOM 2026-06-01: events/<node>/ grew to 291k files (1.2G on piha); rsync of the whole dir exceeded the 30s subprocess timeout every cycle; --remove-source-files never ran; backlog compounded silently. Four fixes: 1. Batch shipping (SHIP_BATCH_SIZE=1000): rsync sends only the oldest 1000 files per cycle via --files-from=- on stdin instead of the whole directory. Backlog drains across cycles; each push fits within timeout. 2. Timeout 30s → 120s: wider margin for large batches and slow links (piha → vps over Tailscale). 3. Backlog trim safety-net (_trim_events_backlog): if events/<node>/ exceeds BACKLOG_MAX (5000) files, oldest files are deleted to bring count back to 5000. Called each cycle before shipping. Breaks the death-spiral independently of rsync success. VPS excluded (uses _cleanup_control_plane_fs). 4. Backlog visibility: WARNING log with file count when unsent events exceed BACKLOG_WARN (2000). "events backlog: N unsent files" — no more silent accumulation. SHIP_BATCH_SIZE is env-configurable for tuning per-node. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
58ac6edd7d
commit
d483274037
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Reference in a new issue