From 4e887ab0284882e514cb4230886e1a6c03b8fd97 Mon Sep 17 00:00:00 2001 From: Kaushik Date: Tue, 18 Aug 2026 11:34:29 +0530 Subject: [PATCH] Add native rclone sync monitoring (agent v3.6-sync + sync-monitor engine) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Absorbs the standalone rclone fleet monitor into the RMM, per plan: - deploy_agent.sh: embedded agent v3.6-sync gains a fail-soft sync-monitor module — tails the rclone-synology-sync container's log (docker inspect discovery), parses rounds/pairs/skips/transfers, inventories upload dirs, POSTs to /api/sync-ingest every 10s. Machines without the container report no_container and stay silent. Heartbeat loop untouched. - central_api_prototype.py: POST /api/sync-ingest (agent-token auth) into new sync_* collections; native /api/sync-monitor/* read API (overview, alerts, stats, timeseries, per-machine events, delete) with JWT auth; NAS verification loop (rclone lsjson, Mongo-shared listings across workers) + per-volume NAS health via rclone about (NAS_VOLUMES). - .env: NAS creds + sync-monitor tuning keys. Rolled out fleet-wide 2026-08-18 (25 clients on 3.6-sync). Standalone collector still runs in parallel pending retirement. Co-Authored-By: Claude Fable 5 --- .env | 16 ++ central_api_prototype.py | 554 +++++++++++++++++++++++++++++++++++++++ deploy_agent.sh | 246 ++++++++++++++++- 3 files changed, 815 insertions(+), 1 deletion(-) diff --git a/.env b/.env index 53e1c38..3386c6d 100644 --- a/.env +++ b/.env @@ -10,3 +10,19 @@ JWT_SECRET_KEY="Gme4fl7mCxTQv9b7H7iahacoEG6j5R4Av7kLAPQpzgVzHjINNScKIDWZ_tJsf93a # Dashboard login. CHANGE THESE, then restart the backend. DASHBOARD_USERNAME="root" DASHBOARD_PASSWORD="seekright159@" + +# --- rclone sync-monitor proxy (added 2026-08-17) --- +# Collector on this same VM; viewer creds are its READ-ONLY login. +SYNC_MONITOR_URL=http://127.0.0.1:4400 +SYNC_MONITOR_USER=admin +SYNC_MONITOR_PASS=HjiYnDgkbclVPfI102CU + +# --- NAS (Synology) for sync verification + health panel (Phase 2) --- +REAL_SYNOLOGY_HOST=takeleap.in +REAL_SYNOLOGY_PORT=26 +REAL_SYNOLOGY_USER=sr-upload +REAL_SYNOLOGY_PASS=T@keleap@123 +SYNC_VERIFY_INTERVAL=300 +SYNC_MISSING_GRACE_S=600 +NAS_MIN_FREE_PCT=10 +NAS_VOLUMES=ClientSync,Saudi_Video_Sync diff --git a/central_api_prototype.py b/central_api_prototype.py index 93945f5..edfb76a 100644 --- a/central_api_prototype.py +++ b/central_api_prototype.py @@ -1306,3 +1306,557 @@ async def receive_heartbeat(client_id: str, payload: TelemetryPayload, request: "search": search, "shift_path": shift_path } + + +# (Phase-1 sync-monitor proxy removed 2026-08-17 — replaced by the native +# /api/sync-monitor/* implementation below, same URL contract.) + + +# ============================================================================= +# Sync-monitor NATIVE ingest — Phase 1 of absorbing the rclone collector +# (2026-08-17). The v3.6-sync RMM agent POSTs parsed sync-log events here. +# ADDITIVE: own sync_* collections in rmm_db, own endpoint, existing agent +# token auth. Runs in PARALLEL with the standalone collector until Phase 4. +# Logic ported from R-clone monitor/collector/main.py incl. its fixes: +# skip-aware pairs, restart-safe round summaries, transfer-byte accounting. +# ============================================================================= +import re +from datetime import datetime as _sdt, timedelta as _std, timezone as _stz + +sync_machines = db["sync_machines"] +sync_pairs = db["sync_pairs"] +sync_events = db["sync_events"] +sync_events.create_index([("machine_id", 1), ("received_at", -1)]) +sync_events.create_index([("machine_id", 1), ("type", 1), ("direction", 1), + ("received_at", -1)]) +sync_pairs.create_index([("machine_id", 1), ("pair", 1)], unique=True) +try: + sync_events.create_index([("received_at", 1)], name="sync_events_ttl", + expireAfterSeconds=30 * 86400) +except Exception: + pass # window changed — harmless, old TTL keeps working + +_SYNC_KEEP = {"round_start", "round_end", "pair_ok", "pair_fail", + "file_synced", "error"} +_SYNC_LOG_TS = "%Y/%m/%d %H:%M:%S" +_SYNC_SIZES = {"": 1, "K": 1e3, "Ki": 1024, "M": 1e6, "Mi": 1024 ** 2, + "G": 1e9, "Gi": 1024 ** 3, "T": 1e12, "Ti": 1024 ** 4} + + +def _sync_size_bytes(s): + m = re.match(r"([\d.]+)\s*([KMGT]i?)?B$", (s or "").strip()) + return float(m.group(1)) * _SYNC_SIZES[m.group(2) or ""] if m else None + + +def _sync_summarise_round(mid, end_ev, inventory, ts): + rnd = end_ev.get("round") + if db is None or rnd is None: + return + # dedup on round AND log_ts: container restarts reset the round counter + if sync_events.find_one({"machine_id": mid, "type": "round_summary", + "round": rnd, "log_ts": end_ev.get("log_ts")}): + return + start_ev = sync_events.find_one( + {"machine_id": mid, "type": "round_start", "round": rnd}, + sort=[("received_at", -1)]) + since = start_ev["received_at"] if start_ev else ts - _std(minutes=30) + synced = list(sync_events.find( + {"machine_id": mid, "type": "file_synced", "round": rnd, + "received_at": {"$gte": since}}, {"file": 1, "direction": 1})) + up = [e for e in synced if e.get("direction") == "up"] + sizes = {f["name"]: f.get("size", 0) + for files in (inventory or {}).values() for f in files} + bytes_up = sum(sizes.get(e.get("file"), 0) for e in up) + duration = None + try: + t0 = _sdt.strptime(start_ev["log_ts"].strip(), _SYNC_LOG_TS) + t1 = _sdt.strptime(end_ev["log_ts"].strip(), _SYNC_LOG_TS) + duration = (t1 - t0).total_seconds() + except Exception: + pass + summary = {"type": "round_summary", "machine_id": mid, "round": rnd, + "log_ts": end_ev.get("log_ts"), "files": len(synced), + "files_up": len(up), "files_down": len(synced) - len(up), + "bytes_up": bytes_up, "duration_s": duration, + "avg_bps": (bytes_up / duration) if duration and bytes_up else None, + "received_at": ts} + sync_events.insert_one(dict(summary)) + if up: + summary.pop("_id", None) + sync_machines.update_one({"machine_id": mid}, + {"$set": {"last_transfer": summary}}) + + +@app.post("/api/sync-ingest") +async def sync_ingest(payload: Dict[str, Any], request: Request, + _auth: None = Depends(require_agent_token)): + """Heartbeat + parsed sync-log events from the RMM agent's sync module.""" + mid = payload.get("machine_id") + if not mid: + raise HTTPException(status_code=400, detail="machine_id required") + ts = _sdt.now(_stz.utc) + mset = {"machine_id": mid, "site": payload.get("site"), "last_seen": ts, + "no_container": bool(payload.get("no_container"))} + if "inventory" in payload: + mset["inventory"] = payload["inventory"] + if payload.get("events"): + mset["last_activity"] = ts + + to_insert = [] + for ev in payload.get("events", []): + etype = ev.get("type") + if etype == "round_start": + mset["current_round"] = ev.get("round") + mset["progress"] = None + elif etype == "round_end": + mset["last_round"] = ev.get("round") + mset["last_round_at"] = ts + mset["progress"] = None + elif etype == "progress": + mset["progress"] = {k: ev.get(k) for k in + ("pair", "direction", "percent", "speed", + "eta", "done", "total")} + elif etype == "pair_skip": + sync_pairs.update_one( + {"machine_id": mid, "pair": ev.get("pair")}, + {"$set": {"direction": ev.get("direction"), + "last_skip_at": ts, "last_idle_s": ev.get("idle_s")}, + "$setOnInsert": {"last_status": "idle"}}, upsert=True) + elif etype in ("pair_start", "pair_ok", "pair_fail"): + status = {"pair_start": "running", "pair_ok": "ok", + "pair_fail": "fail"}[etype] + ev["transferred_bytes"] = _sync_size_bytes(ev.get("transferred")) + upd = {"machine_id": mid, "pair": ev.get("pair"), + "direction": ev.get("direction"), "last_status": status, + "updated_at": ts} + if etype == "pair_fail": + upd["last_exit"] = ev.get("exit_code") + upd["last_fail_at"] = ts + elif etype == "pair_ok": + upd["last_exit"] = 0 + upd["last_ok_at"] = ts + sync_pairs.update_one({"machine_id": mid, "pair": ev.get("pair")}, + {"$set": upd}, upsert=True) + if etype in _SYNC_KEEP: + to_insert.append({**ev, "machine_id": mid, "received_at": ts}) + + sync_machines.update_one({"machine_id": mid}, + {"$set": mset, + "$setOnInsert": {"first_seen": ts}}, upsert=True) + if to_insert: + sync_events.insert_many(to_insert) + for ev in payload.get("events", []): + if ev.get("type") == "round_end": + inv = payload.get("inventory") + if inv is None: + inv = (sync_machines.find_one({"machine_id": mid}, + {"inventory": 1}) or {}).get("inventory") + _sync_summarise_round(mid, ev, inv, ts) + return {"ok": True, "stored": len(to_insert)} + + +# ============================================================================= +# Sync-monitor NATIVE engine (Phase 2, 2026-08-17): verification against the +# NAS, NAS health, fleet view/alerts, stats & timeseries — same URL contract +# the SyncMonitor UI already uses. Listings/health live in Mongo so the 4 +# uvicorn workers share one verification effort instead of quadrupling it. +# rclone is invoked with env-injected config (no config file on disk). +# ============================================================================= +import subprocess +import threading +import time + +_SYNC_NAS_HOST = os.getenv("REAL_SYNOLOGY_HOST", "") +_SYNC_NAS_PORT = os.getenv("REAL_SYNOLOGY_PORT", "22") +_SYNC_NAS_USER = os.getenv("REAL_SYNOLOGY_USER", "") +_SYNC_NAS_PASS = os.getenv("REAL_SYNOLOGY_PASS", "") +_SYNC_VERIFY_IVL = int(os.getenv("SYNC_VERIFY_INTERVAL", "300")) +_SYNC_GRACE_S = int(os.getenv("SYNC_MISSING_GRACE_S", "600")) +_SYNC_NAS_MIN_FREE = float(os.getenv("NAS_MIN_FREE_PCT", "10")) +_SYNC_ONLINE_S = 60 +_SYNC_STALLED_S = 900 +_sync_listings = db["sync_listings"] # {_id: remote_path, files: [...], at} +_sync_rclone_env = None + + +def _sync_now(): + # their Mongo client is NOT tz_aware: reads come back naive-UTC, so all + # read-path arithmetic uses naive-UTC now + return _sdt.utcnow() + + +def _sync_rclone(args, timeout): + """Run rclone against the NAS remote via env-injected config.""" + global _sync_rclone_env + if not _SYNC_NAS_HOST: + return None + if _sync_rclone_env is None: + obscured = subprocess.run(["rclone", "obscure", _SYNC_NAS_PASS], + capture_output=True, text=True, timeout=15) + env = dict(os.environ) + env.update({"RCLONE_CONFIG_SYNOREAL_TYPE": "sftp", + "RCLONE_CONFIG_SYNOREAL_HOST": _SYNC_NAS_HOST, + "RCLONE_CONFIG_SYNOREAL_PORT": _SYNC_NAS_PORT, + "RCLONE_CONFIG_SYNOREAL_USER": _SYNC_NAS_USER, + "RCLONE_CONFIG_SYNOREAL_PASS": obscured.stdout.strip()}) + _sync_rclone_env = env + return subprocess.run(["rclone"] + args, capture_output=True, + timeout=timeout, env=_sync_rclone_env) + + +def _sync_verify_loop(): + time.sleep(60 + (os.getpid() % 180)) # stagger the 4 workers + while True: + try: + # NAS health first — reachability + capacity, shared via Mongo + nas_doc = _sync_listings.find_one({"_id": "@nas"}) + if not nas_doc or (_sync_now() - nas_doc["at"]).total_seconds() > _SYNC_VERIFY_IVL: + # per-VOLUME capacity: `about` on the SFTP root reports the + # chrooted home (2.4 GB!), not the data shares the uploads + # land on — each share sits on its own volume + health = {"_id": "@nas", "at": _sync_now(), "ok": False, + "error": None, "volumes": []} + try: + for vol in [v for v in os.getenv( + "NAS_VOLUMES", "").split(",") if v.strip()] or [""]: + out = _sync_rclone(["about", f"synoreal:{vol.strip()}", + "--json"], 60) + if out is None: + health["error"] = "NAS creds not configured" + break + if out.returncode == 0: + about = json.loads(out.stdout) + health["ok"] = True # at least one volume answered + health["volumes"].append( + {"path": vol.strip() or "/", + "total": about.get("total"), + "free": about.get("free")}) + else: + health["error"] = out.stderr.decode()[:200] + except FileNotFoundError: + health["error"] = "rclone binary not installed on server" + except Exception as e: + health["error"] = str(e)[:200] + _sync_listings.update_one({"_id": "@nas"}, {"$set": health}, + upsert=True) + + # listings for every known up-pair remote path + for p in sync_pairs.find({"direction": "up"}): + label = p.get("pair", "") + if "server:" not in label: + continue + rpath = label.split("server:", 1)[1].strip() + doc = _sync_listings.find_one({"_id": rpath}) + if doc and (_sync_now() - doc["at"]).total_seconds() < _SYNC_VERIFY_IVL: + continue + try: + out = _sync_rclone(["lsjson", "--recursive", "--files-only", + f"synoreal:{rpath}"], 300) + if out is not None and out.returncode == 0: + files = [{"n": f["Path"], "s": f["Size"]} + for f in json.loads(out.stdout)] + _sync_listings.update_one( + {"_id": rpath}, + {"$set": {"files": files, "at": _sync_now()}}, + upsert=True) + except FileNotFoundError: + break # no rclone binary — health doc already says so + except Exception: + pass + except Exception: + pass + time.sleep(_SYNC_VERIFY_IVL) + + +threading.Thread(target=_sync_verify_loop, daemon=True).start() + + +def _sync_annotate(m, pairs): + inv = m.get("inventory") or {} + total_pending = total_bytes = 0 + verified_any = False + for p in pairs: + label = p.get("pair", "") + if p.get("direction") != "up" or "server:" not in label: + continue + local = label.split("→")[0].strip() + rpath = label.split("server:", 1)[1].strip() + listing = _sync_listings.find_one({"_id": rpath}) + files = inv.get(local) + if listing is None or files is None: + continue + verified_any = True + server_names = {f["n"] for f in listing.get("files", [])} + pending = [f for f in files if f["name"] not in server_names] + p["pending_count"] = len(pending) + p["pending_bytes"] = sum(f.get("size", 0) for f in pending) + p["verified_at"] = listing["at"] + p["missing"] = [f["name"] for f in pending + if time.time() - f["mtime"] > _SYNC_GRACE_S][:20] + total_pending += len(pending) + total_bytes += p["pending_bytes"] + if not verified_any: + return None, None + return total_pending, total_bytes + + +def _sync_machine_view(m, ts): + mid = m["machine_id"] + last_seen = m.get("last_seen") + online = bool(last_seen) and (ts - last_seen).total_seconds() < _SYNC_ONLINE_S + pairs = [dict(p, _id=None) for p in sync_pairs.find({"machine_id": mid})] + for p in pairs: + p.pop("_id", None) + pending_total, pending_bytes = _sync_annotate(m, pairs) + hour_ago = ts - _std(hours=1) + up_pairs = [p for p in pairs if p.get("direction") == "up"] + progress = m.get("progress") + return { + "machine_id": mid, "site": m.get("site"), "online": online, + "last_seen": last_seen, "last_activity": m.get("last_activity"), + "first_seen": m.get("first_seen"), + "no_container": m.get("no_container", False), + "current_round": m.get("current_round"), + "last_round": m.get("last_round"), + "last_round_at": m.get("last_round_at"), + "progress": progress if (progress or {}).get("direction") == "up" else None, + "last_transfer": m.get("last_transfer"), + "pairs": pairs, "up_pairs": up_pairs, + "pending_total": pending_total, "pending_bytes": pending_bytes, + "in_progress": len([p for p in up_pairs if p.get("last_status") == "running"]), + "failed_up": len([p for p in up_pairs if p.get("last_status") == "fail"]), + "uploads_1h": sync_events.count_documents( + {"machine_id": mid, "type": "file_synced", "direction": "up", + "received_at": {"$gt": hour_ago}}), + "uploads_24h": sync_events.count_documents( + {"machine_id": mid, "type": "file_synced", "direction": "up", + "received_at": {"$gt": ts - _std(hours=24)}}), + "errors_1h": sync_events.count_documents( + {"machine_id": mid, "type": "error", "direction": "up", + "received_at": {"$gt": hour_ago}}), + } + + +def _sync_fleet(): + ts = _sync_now() + machines = sorted((_sync_machine_view(m, ts) for m in sync_machines.find()), + key=lambda x: x["machine_id"]) + alerts = [] + nas = _sync_listings.find_one({"_id": "@nas"}) or {} + nas.pop("_id", None) + if nas and not nas.get("ok"): + alerts.append({"severity": "critical", "machine_id": "NAS", + "message": f"NAS check failed: {nas.get('error')}"}) + for vol in nas.get("volumes") or []: + if vol.get("total"): + vol["free_pct"] = round(100.0 * (vol.get("free") or 0) / vol["total"], 1) + if vol["free_pct"] < _SYNC_NAS_MIN_FREE: + alerts.append({"severity": "critical", "machine_id": "NAS", + "message": f"NAS volume {vol['path']} nearly full — " + f"{vol['free_pct']}% free"}) + for m in machines: + if m["no_container"]: + continue # roster-style visibility, but no sync rules apply + if not m["online"]: + alerts.append({"severity": "critical", "machine_id": m["machine_id"], + "message": "machine offline — no heartbeat"}) + if (m["online"] and not m.get("last_round_at") and not m.get("last_activity") + and m.get("first_seen") + and (ts - m["first_seen"]).total_seconds() > _SYNC_STALLED_S): + alerts.append({"severity": "serious", "machine_id": m["machine_id"], + "message": "agent alive but sync log SILENT since install " + "— is the sync container running?"}) + for p in m["up_pairs"]: + if p.get("last_status") == "fail": + alerts.append({"severity": "serious", "machine_id": m["machine_id"], + "message": f"upload pair failing (exit {p.get('last_exit')}): {p['pair']}"}) + if m["online"] and m["errors_1h"] > 0: + alerts.append({"severity": "warning", "machine_id": m["machine_id"], + "message": f"{m['errors_1h']} upload error(s) in the last hour"}) + busy_now = (m.get("last_activity") + and (ts - m["last_activity"]).total_seconds() < 120) + for p in m["up_pairs"]: + if p.get("missing") and not (p.get("last_status") == "running" and busy_now): + alerts.append({"severity": "serious", "machine_id": m["machine_id"], + "message": f"{len(p['missing'])} file(s) NOT on server after " + f"{_SYNC_GRACE_S // 60}min grace: " + f"{', '.join(p['missing'][:3])}…"}) + if m["online"] and m.get("last_round_at"): + marks = [m["last_round_at"]] + if m.get("last_activity"): + marks.append(m["last_activity"]) + stalled = (ts - max(marks)).total_seconds() + if stalled > _SYNC_STALLED_S: + age = f"{stalled / 3600:.1f}h" if stalled >= 3600 else f"{int(stalled // 60)}min" + alerts.append({"severity": "serious", "machine_id": m["machine_id"], + "message": f"agent alive but NO sync round completed in {age}"}) + return ts, machines, alerts, nas + + +@app.get("/api/sync-monitor/health") +def sync_native_health(): + return {"ok": True, "time": _sync_now(), "native": True} + + +@app.get("/api/sync-monitor/overview") +def sync_native_overview(current_user: str = Depends(get_current_user)): + ts, machines, alerts, nas = _sync_fleet() + return {"generated_at": ts, "machines": machines, "alerts": alerts, "nas": nas} + + +@app.get("/api/sync-monitor/alerts") +def sync_native_alerts(current_user: str = Depends(get_current_user)): + ts, _, alerts, _nas = _sync_fleet() + return {"generated_at": ts, "count": len(alerts), "alerts": alerts} + + +_SYNC_WINDOWS = [("1h", 3600), ("24h", 86400), ("3d", 259200), + ("7d", 604800), ("30d", 2592000)] +_SYNC_BUCKET = {"1h": 300, "24h": 3600, "3d": 10800, "7d": 21600, "30d": 86400} + + +@app.get("/api/sync-monitor/stats") +def sync_native_stats(current_user: str = Depends(get_current_user)): + ts = _sync_now() + cuts = {lbl: ts - _std(seconds=s) for lbl, s in _SYNC_WINDOWS} + counters = {"uploaded": {"type": "file_synced", "direction": "up"}, + "downloaded": {"type": "file_synced", "direction": "down"}, + "errors": {"type": "error", "direction": "up"}, + "failures": {"type": "pair_fail", "direction": "up"}, + "rounds": {"type": "round_summary"}} + sums = {"bytes": ({"type": "round_summary"}, "bytes_up"), + "seconds": ({"type": "round_summary"}, "duration_s"), + "up_bytes": ({"type": "pair_ok", "direction": "up"}, "transferred_bytes"), + "down_bytes": ({"type": "pair_ok", "direction": "down"}, "transferred_bytes")} + + def cond(cut, conds): + cl = [{"$gte": ["$received_at", cut]}] + cl += [{"$eq": [{"$ifNull": [f"${k}", None]}, v]} for k, v in conds.items()] + return {"$and": cl} + + group = {"_id": "$machine_id"} + for lbl, _s in _SYNC_WINDOWS: + for name, conds in counters.items(): + group[f"{name}|{lbl}"] = {"$sum": {"$cond": [cond(cuts[lbl], conds), 1, 0]}} + for name, (conds, field) in sums.items(): + group[f"{name}|{lbl}"] = {"$sum": {"$cond": [ + cond(cuts[lbl], conds), {"$ifNull": [f"${field}", 0]}, 0]}} + + per_machine = {} + for row in sync_events.aggregate([ + {"$match": {"received_at": {"$gte": cuts["30d"]}}}, + {"$group": group}]): + windows = {} + for lbl, _s in _SYNC_WINDOWS: + byts = row.get(f"bytes|{lbl}") or 0 + secs = row.get(f"seconds|{lbl}") or 0 + windows[lbl] = {k: row.get(f"{k}|{lbl}", 0) for k in + ("uploaded", "downloaded", "errors", "failures", + "rounds", "up_bytes", "down_bytes")} + windows[lbl]["bytes"] = byts + windows[lbl]["seconds"] = secs + windows[lbl]["avg_bps"] = (byts / secs) if secs > 0 and byts else None + per_machine[row["_id"]] = windows + + fleet = {} + for lbl, _s in _SYNC_WINDOWS: + agg = {k: 0 for k in ("uploaded", "downloaded", "errors", "failures", + "rounds", "bytes", "up_bytes", "down_bytes", "seconds")} + for wins in per_machine.values(): + w = wins.get(lbl) or {} + for k in agg: + agg[k] += w.get(k) or 0 + agg["avg_bps"] = (agg["bytes"] / agg["seconds"]) if agg["seconds"] else None + fleet[lbl] = agg + return {"generated_at": ts, "windows": [w for w, _ in _SYNC_WINDOWS], + "retention_days": 30, "fleet": fleet, "machines": per_machine} + + +@app.get("/api/sync-monitor/timeseries") +def sync_native_timeseries(range: str = "24h", machine_id: str = None, + current_user: str = Depends(get_current_user)): + if range not in _SYNC_BUCKET: + raise HTTPException(status_code=400, detail="bad range") + step = _SYNC_BUCKET[range] + ts = _sync_now() + start = ts - _std(seconds=dict(_SYNC_WINDOWS)[range]) + match = {"received_at": {"$gte": start}, + "type": {"$in": ["file_synced", "error"]}} + volm = {"received_at": {"$gte": start}, "type": "pair_ok"} + if machine_id: + match["machine_id"] = machine_id + volm["machine_id"] = machine_id + bucket = {"$toDate": {"$subtract": [{"$toLong": "$received_at"}, + {"$mod": [{"$toLong": "$received_at"}, step * 1000]}]}} + is_up = {"$eq": [{"$ifNull": ["$direction", ""]}, "up"]} + is_down = {"$eq": [{"$ifNull": ["$direction", ""]}, "down"]} + series = {"$switch": {"branches": [ + {"case": {"$and": [{"$eq": ["$type", "error"]}, is_up]}, "then": "errors"}, + {"case": {"$and": [{"$eq": ["$type", "file_synced"]}, is_up]}, "then": "uploaded"}, + {"case": {"$and": [{"$eq": ["$type", "file_synced"]}, is_down]}, "then": "downloaded"}, + ], "default": "skip"}} + counts, volumes = {}, {} + for row in sync_events.aggregate([{"$match": match}, + {"$group": {"_id": {"b": bucket, "s": series}, "n": {"$sum": 1}}}]): + if row["_id"]["s"] != "skip": + counts[(int(row["_id"]["b"].replace(tzinfo=_stz.utc).timestamp()), row["_id"]["s"])] = row["n"] + for row in sync_events.aggregate([{"$match": volm}, + {"$group": {"_id": {"b": bucket, "d": "$direction"}, + "bytes": {"$sum": {"$ifNull": ["$transferred_bytes", 0]}}}}]): + volumes[(int(row["_id"]["b"].replace(tzinfo=_stz.utc).timestamp()), row["_id"]["d"])] = row["bytes"] + first = int(start.replace(tzinfo=_stz.utc).timestamp()) // step * step + last = int(ts.replace(tzinfo=_stz.utc).timestamp()) // step * step + points = [] + b = first + while b <= last: + points.append({"t": _sdt.fromtimestamp(b, _stz.utc), + "uploaded": counts.get((b, "uploaded"), 0), + "downloaded": counts.get((b, "downloaded"), 0), + "errors": counts.get((b, "errors"), 0), + "up_bytes": volumes.get((b, "up"), 0), + "down_bytes": volumes.get((b, "down"), 0)}) + b += step + return {"generated_at": ts, "range": range, "bucket_s": step, + "machine_id": machine_id, "points": points} + + +@app.get("/api/sync-monitor/machines/{machine_id}/events") +def sync_native_events(machine_id: str, type: str = None, direction: str = None, + q: str = None, since: _sdt = None, until: _sdt = None, + busy: bool = False, limit: int = 50, skip: int = 0, + current_user: str = Depends(get_current_user)): + query = {"machine_id": machine_id} + if type: + query["type"] = {"$in": type.split(",")} + if direction: + query["direction"] = direction + if busy: + query.setdefault("$and", []).append( + {"$or": [{"files_up": {"$gt": 0}}, {"files": {"$gt": 0}}, + {"type": {"$ne": "round_summary"}}]}) + if since or until: + rng = {} + if since: + rng["$gte"] = since.replace(tzinfo=None) if since.tzinfo else since + if until: + rng["$lt"] = until.replace(tzinfo=None) if until.tzinfo else until + query["received_at"] = rng + if q: + rx = {"$regex": re.escape(q), "$options": "i"} + query.setdefault("$and", []).append( + {"$or": [{"file": rx}, {"message": rx}, {"pair": rx}]}) + limit = max(1, min(limit, 500)) + total = sync_events.count_documents(query) + evs = list(sync_events.find(query, {"_id": 0}) + .sort("received_at", -1).skip(max(0, skip)).limit(limit)) + return {"machine_id": machine_id, "total": total, "skip": skip, + "limit": limit, "events": evs} + + +@app.delete("/api/sync-monitor/machines/{machine_id}") +def sync_native_forget(machine_id: str, + current_user: str = Depends(get_current_user)): + sync_machines.delete_one({"machine_id": machine_id}) + pairs = sync_pairs.delete_many({"machine_id": machine_id}).deleted_count + events = sync_events.delete_many({"machine_id": machine_id}).deleted_count + return {"ok": True, "machine_id": machine_id, "pairs": pairs, "events": events} diff --git a/deploy_agent.sh b/deploy_agent.sh index 6b9d740..df4e10c 100644 --- a/deploy_agent.sh +++ b/deploy_agent.sh @@ -116,7 +116,7 @@ def get_cpu_temp(): pass return None -AGENT_VERSION = "3.5-net" +AGENT_VERSION = "3.6-sync" CLIENT_ID = "TEMPLATE_CLIENT_ID" CENTRAL_BASE = "TEMPLATE_CENTRAL_URL" @@ -782,6 +782,247 @@ def run_speed_test(): print(f" -> [!] Upload speed test failed: {e}") print(f" -> Link speed: down {speed_download_mbps} Mbps / up {speed_upload_mbps} Mbps") +# ========================================================================= +# rclone sync monitor module (agent v3.6-sync) — watches the site's +# rclone-synology-sync container by tailing its log + inventorying the +# upload source folders, reporting to /api/sync-ingest every SYNC_POLL s. +# FAIL-SOFT BY CONTRACT: every loop iteration is wrapped; no exception in +# here may ever reach the heartbeat loop. Machines without the sync +# container report {"no_container": true} and otherwise stay silent. +# Ported from the standalone rclone-agent incl. skip parsing, mid-round +# context priming, inventory dedup and per-pair transfer accounting. +# ========================================================================= +import hashlib +import re + +SYNC_CONTAINER = "rclone-synology-sync" +SYNC_POLL = 10 +SYNC_INV_INTERVAL = 60 +SYNC_MAX_INV = 20000 +SYNC_INGEST_URL = f"{CENTRAL_BASE}/api/sync-ingest" + +_SYNC_RE_ROUND_START = re.compile(r"^=== Round #(\d+) started at (.+) ===") +_SYNC_RE_ROUND_DONE = re.compile(r"^=== Round #(\d+) done at (.+) ===") +_SYNC_RE_PAIR_START = re.compile(r"^--- \[(\S+ (?:UP|DOWN))\] (.+?) ---$") +_SYNC_RE_PAIR_OK = re.compile(r"^OK: \[(\S+ (?:UP|DOWN))\] (.+)$") +_SYNC_RE_PAIR_FAIL = re.compile(r"^ERROR: \[(\S+ (?:UP|DOWN))\] (.+) failed \(exit (\d+)\)$") +_SYNC_RE_PAIR_SKIP = re.compile(r"^SKIP: \[(\S+ (?:UP|DOWN))\] (.+) — unchanged \(idle (\d+)s\)$") +_SYNC_RE_COPIED = re.compile(r"^([\d/]+ [\d:]+) INFO\s+: (.+?): ((?:Copied|Moved).*)$") +_SYNC_RE_ERR = re.compile(r"^([\d/]+ [\d:]+) ERROR\s*: (.*)$") +_SYNC_RE_STATS = re.compile( + r"^([\d/]+ [\d:]+) INFO\s+:\s+(\S+ \S*B) / (\S+ \S*B), (\d+)%, (\S+ ?\S*B/s), ETA (\S+)") + + +class _SyncParser: + def __init__(self): + self.round = None + self.pair = None + self.direction = None + self.last_done = None + + def _dir(self, arrow): + return "up" if "UP" in arrow else "down" + + def feed(self, line): + line = line.rstrip("\r\n") + ctx = {"round": self.round, "pair": self.pair, "direction": self.direction} + m = _SYNC_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 = _SYNC_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 = _SYNC_RE_PAIR_START.match(line) + if m: + self.direction = self._dir(m.group(1)) + self.pair = m.group(2) + self.last_done = None + return {"type": "pair_start", "round": self.round, + "pair": self.pair, "direction": self.direction} + m = _SYNC_RE_PAIR_OK.match(line) + if m: + done, self.last_done = self.last_done, None + return {"type": "pair_ok", "round": self.round, "transferred": done, + "pair": m.group(2), "direction": self._dir(m.group(1))} + m = _SYNC_RE_PAIR_FAIL.match(line) + if m: + done, self.last_done = self.last_done, None + return {"type": "pair_fail", "round": self.round, "pair": m.group(2), + "transferred": done, "direction": self._dir(m.group(1)), + "exit_code": int(m.group(3))} + m = _SYNC_RE_PAIR_SKIP.match(line) + if m: + return {"type": "pair_skip", "round": self.round, "pair": m.group(2), + "direction": self._dir(m.group(1)), "idle_s": int(m.group(3))} + m = _SYNC_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 = _SYNC_RE_STATS.match(line) + if m: + self.last_done = m.group(2) + 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 = _SYNC_RE_ERR.match(line) + if m: + return {"type": "error", "log_ts": m.group(1), "message": m.group(2), **ctx} + return None + + +def _sync_mounts(): + """Host paths of the sync container's mounts, or None if absent/stopped.""" + try: + out = subprocess.run( + ["docker", "inspect", SYNC_CONTAINER, + "--format", "{{json .Mounts}}"], + capture_output=True, text=True, timeout=10) + if out.returncode != 0: + return None + return {m["Destination"]: m["Source"] for m in json.loads(out.stdout)} + except Exception: + return None + + +def _sync_site(): + """Site label from the container's SYNC_5 remote path (…//).""" + try: + out = subprocess.run( + ["docker", "inspect", SYNC_CONTAINER, "--format", + "{{range .Config.Env}}{{println .}}{{end}}"], + capture_output=True, text=True, timeout=10) + for env_line in out.stdout.splitlines(): + if env_line.startswith("SYNC_5="): + return env_line.rsplit("/", 1)[-1] or CLIENT_ID + except Exception: + pass + return CLIENT_ID + + +def _sync_inventory(watch): + inv = {} + for label, d in watch.items(): + files = [] + capped = False + try: + for root, _dirs, names in os.walk(d): + for n in names: + if len(files) >= SYNC_MAX_INV: + capped = True + break + 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 capped: + break + except OSError: + pass + inv[label] = files + return inv + + +def _sync_prime(f, parser, chunks=(65536, 1048576, 8388608)): + """Recover round/pair context from the log tail without replaying events.""" + end = f.tell() + for back in chunks: + start = max(0, end - back) + f.seek(start) + if start: + f.readline() + scratch = _SyncParser() + for line in f: + scratch.feed(line) + if scratch.pair or scratch.round is not None or start == 0: + parser.round, parser.pair = scratch.round, scratch.pair + parser.direction = scratch.direction + return + + +def _sync_post(payload): + req = urllib.request.Request( + SYNC_INGEST_URL, data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req, timeout=10).read() + + +def sync_monitor_loop(): + time.sleep(30) # let the agent settle first + parser = _SyncParser() + f = None + site = None + last_inv_walk = 0.0 + last_inv_digest = None + last_none_post = 0.0 + while True: + try: + mounts = _sync_mounts() + if not mounts or "/logs" not in mounts: + if f: + f.close() + f = None + # tell the server this machine has no (running) sync container, + # but only once a minute — it is a state, not an event stream + if time.time() - last_none_post > 60: + last_none_post = time.time() + _sync_post({"machine_id": CLIENT_ID, "site": CLIENT_ID, + "no_container": True, "events": []}) + time.sleep(SYNC_POLL) + continue + + log_path = os.path.join(mounts["/logs"], "sync.log") + watch = {c: mounts[c] for c in + ("/sources/image_root_dir", "/sources/csv_files") if c in mounts} + if site is None: + site = _sync_site() + + events = [] + if f is None and os.path.exists(log_path): + f = open(log_path, "r", errors="replace") + f.seek(0, os.SEEK_END) + _sync_prime(f, parser) + if parser.pair: + events.append({"type": "pair_start", "round": parser.round, + "pair": parser.pair, "direction": parser.direction}) + elif f is not None: + if os.path.getsize(log_path) < f.tell(): + f.close() + f = open(log_path, "r", errors="replace") + for line in f: + ev = parser.feed(line) + if ev: + events.append(ev) + + inventory = None + if watch and time.time() - last_inv_walk >= SYNC_INV_INTERVAL: + last_inv_walk = time.time() + inv = _sync_inventory(watch) + digest = hashlib.md5( + json.dumps(inv, sort_keys=True).encode()).hexdigest() + if digest != last_inv_digest: + inventory = inv + last_inv_digest = digest + + payload = {"machine_id": CLIENT_ID, "site": site, "events": events} + if inventory is not None: + payload["inventory"] = inventory + try: + _sync_post(payload) + except Exception: + if inventory is not None: + last_inv_digest = None # resend inventory next walk + except Exception: + pass # fail-soft: sync monitoring must never break the agent + time.sleep(SYNC_POLL) +# ===================== end rclone sync monitor module ==================== + + def speed_test_loop(): time.sleep(120) # let the agent settle before the first probe while True: @@ -1110,6 +1351,9 @@ if __name__ == "__main__": t = threading.Thread(target=speed_test_loop) t.daemon = True t.start() + ts_sync = threading.Thread(target=sync_monitor_loop) + ts_sync.daemon = True + ts_sync.start() try: while True: send_heartbeat()