#!/usr/bin/env python3 """ Per-machine monitoring agent (sidecar container). Tails the rclone sync container's /logs/sync.log (mounted read-only), parses it into structured events, and POSTs them to the central collector every few seconds. The POST itself doubles as the machine's heartbeat — an empty event list still proves the machine (and this agent) is alive. Works against the byte-identical production sync.sh: no changes to the sync container are required. Python stdlib only — no pip installs. Env: MACHINE_ID unique id for this machine (e.g. GCBOT-01) SITE site/org label (e.g. GCBOT) COLLECTOR_URL e.g. http://collector:8000 SYNC_LOG default /logs/sync.log POLL_SECONDS default 5 """ import json import os import re import time import urllib.request MACHINE_ID = os.environ.get("MACHINE_ID", "unknown") SITE = os.environ.get("SITE", "unknown") COLLECTOR = os.environ.get("COLLECTOR_URL", "http://collector:8000").rstrip("/") SYNC_LOG = os.environ.get("SYNC_LOG", "/logs/sync.log") POLL = float(os.environ.get("POLL_SECONDS", "5")) AGENT_TOKEN = os.environ.get("AGENT_TOKEN", "") # must match the collector's # Source dirs of the UP pairs, mounted ro at the SAME paths the sync container # uses (e.g. /sources/csv_files) so inventory keys match pair labels. The # collector diffs this against the server listing => pending/missing files. WATCH_DIRS = [d for d in os.environ.get("WATCH_DIRS", "").split(",") if d] MAX_INV_FILES = 2000 # safety cap per dir def take_inventory(): inv = {} for d in WATCH_DIRS: files = [] try: for root, _, names in os.walk(d): for n in names: p = os.path.join(root, n) try: st = os.stat(p) except OSError: continue files.append({"name": os.path.relpath(p, d).replace("\\", "/"), "size": st.st_size, "mtime": st.st_mtime}) if len(files) >= MAX_INV_FILES: break except OSError: pass inv[d] = files return inv # --- line patterns for sync.sh's log format --------------------------------- RE_ROUND_START = re.compile(r"^=== Round #(\d+) started at (.+) ===") RE_ROUND_DONE = re.compile(r"^=== Round #(\d+) done at (.+) ===") RE_PAIR_START = re.compile(r"^--- \[(\S+ (?:UP|DOWN))\] (.+?) ---$") RE_PAIR_OK = re.compile(r"^OK: \[(\S+ (?:UP|DOWN))\] (.+)$") RE_PAIR_FAIL = re.compile(r"^ERROR: \[(\S+ (?:UP|DOWN))\] (.+) failed \(exit (\d+)\)$") RE_COPIED = re.compile(r"^([\d/]+ [\d:]+) INFO\s+: (.+?): ((?:Copied|Moved).*)$") RE_RCLONE_ERR = re.compile(r"^([\d/]+ [\d:]+) ERROR\s*: (.*)$") # one-line stats: "2026/08/05 10:48:22 INFO : 8.5 MiB / 157 MiB, 5%, 4.2 MiB/s, ETA 32s" RE_STATS = re.compile( r"^([\d/]+ [\d:]+) INFO\s+:\s+(\S+ \S*B) / (\S+ \S*B), (\d+)%, (\S+ ?\S*B/s), ETA (\S+)" ) class LogParser: """Stateful line parser: remembers current round/pair for event context.""" def __init__(self): self.round = None self.pair = None # current pair label self.direction = None # "up" | "down" def _dir(self, arrow): return "up" if "UP" in arrow else "down" def feed(self, line): """Return an event dict for this line, or None if it's noise.""" line = line.rstrip("\r\n") ctx = {"round": self.round, "pair": self.pair, "direction": self.direction} m = RE_ROUND_START.match(line) if m: self.round = int(m.group(1)) self.pair = self.direction = None return {"type": "round_start", "round": self.round, "log_ts": m.group(2)} m = RE_ROUND_DONE.match(line) if m: self.pair = self.direction = None return {"type": "round_end", "round": int(m.group(1)), "log_ts": m.group(2)} m = RE_PAIR_START.match(line) if m: self.direction = self._dir(m.group(1)) self.pair = m.group(2) return {"type": "pair_start", "round": self.round, "pair": self.pair, "direction": self.direction} m = RE_PAIR_OK.match(line) if m: return {"type": "pair_ok", "round": self.round, "pair": m.group(2), "direction": self._dir(m.group(1))} m = RE_PAIR_FAIL.match(line) if m: return {"type": "pair_fail", "round": self.round, "pair": m.group(2), "direction": self._dir(m.group(1)), "exit_code": int(m.group(3))} m = RE_COPIED.match(line) if m: return {"type": "file_synced", "log_ts": m.group(1), "file": m.group(2), "action": m.group(3), **ctx} m = RE_STATS.match(line) if m: return {"type": "progress", "log_ts": m.group(1), "done": m.group(2), "total": m.group(3), "percent": int(m.group(4)), "speed": m.group(5), "eta": m.group(6), **ctx} m = RE_RCLONE_ERR.match(line) if m: return {"type": "error", "log_ts": m.group(1), "message": m.group(2), **ctx} return None def post_events(events): payload = { "machine_id": MACHINE_ID, "site": SITE, "sent_at": time.time(), "events": events, } if WATCH_DIRS: payload["inventory"] = take_inventory() headers = {"Content-Type": "application/json"} if AGENT_TOKEN: headers["Authorization"] = f"Bearer {AGENT_TOKEN}" req = urllib.request.Request( COLLECTOR + "/api/ingest", data=json.dumps(payload).encode(), headers=headers, ) urllib.request.urlopen(req, timeout=10).read() def main(): parser = LogParser() f = None offset = 0 print(f"[agent] {MACHINE_ID} ({SITE}) -> {COLLECTOR}, log={SYNC_LOG}", flush=True) while True: events = [] try: if f is None: if os.path.exists(SYNC_LOG): f = open(SYNC_LOG, "r", errors="replace") f.seek(0, os.SEEK_END) # start from "now"; no historical replay offset = f.tell() else: # handle rotation/truncation: file shrank -> reopen from start size = os.path.getsize(SYNC_LOG) if size < offset: f.close() f = open(SYNC_LOG, "r", errors="replace") offset = 0 for line in f: ev = parser.feed(line) if ev: events.append(ev) offset = f.tell() except OSError as e: print(f"[agent] log read error: {e}", flush=True) f = None try: post_events(events) # empty list == pure heartbeat except Exception as e: print(f"[agent] collector unreachable: {e}", flush=True) time.sleep(POLL) if __name__ == "__main__": main()