diff --git a/Dockerfile b/Dockerfile index 7bd5e69..28f8da0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,11 +5,12 @@ FROM debian:bookworm-slim # postgresql-client-17 from PGDG (Debian's default is 15, and pg_dump refuses a newer server -# major — nc-db and most modern Postgres are 17). default-mysql-client covers MySQL/MariaDB. +# major — nc-db and most modern Postgres are 17). default-mysql-client covers MySQL/MariaDB; +# sqlite3 lets the agent back up SQLite files (e.g. Vaultwarden) via `.dump`. RUN set -eux; \ apt-get update; \ apt-get install -y --no-install-recommends \ - ca-certificates curl gnupg python3 cron default-mysql-client; \ + ca-certificates curl gnupg python3 cron default-mysql-client sqlite3; \ install -d /usr/share/postgresql-common/pgdg; \ curl -fsSL https://www.postgresql.org/media/keys/ACCC4CF8.asc \ -o /usr/share/postgresql-common/pgdg/apt.postgresql.org.asc; \ diff --git a/backhaul-agent.py b/backhaul-agent.py index f4e3f8c..07f69e8 100644 --- a/backhaul-agent.py +++ b/backhaul-agent.py @@ -22,18 +22,27 @@ import subprocess import sys import urllib.request -VERSION = "1.0" +VERSION = "1.3" RESULT_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".dump-result.json") def targets(): - """Parse BACKHAUL_TARGETS into dicts. Each: engine:host:port:db:user:pass.""" + """Parse BACKHAUL_TARGETS into dicts. A networked DB is engine:host:port:db:user:pass; + a SQLite file is sqlite: (a file bind-mounted into the agent, not a server).""" out = [] for spec in os.environ.get("BACKHAUL_TARGETS", "").split(","): spec = spec.strip() if not spec: continue parts = spec.split(":") + if parts[0] == "sqlite": + path = ":".join(parts[1:]).strip() # rejoin in case the path holds a ':' + if not path: + print(f"skipping sqlite target without a path: {spec}", file=sys.stderr) + continue + out.append({"engine": "sqlite", "path": path, + "name": f"sqlite/{os.path.basename(path)}"}) + continue if len(parts) < 6: print(f"skipping malformed target: {spec}", file=sys.stderr) continue @@ -47,6 +56,10 @@ def targets(): def db_version(t): """Fetch the engine version string so the app can pick a matching restore image.""" try: + if t["engine"] == "sqlite": + out = subprocess.run(("sqlite3", "--version"), + capture_output=True, text=True, timeout=15).stdout + return "SQLite " + ((out.strip().split(" ")[0]) or "3") if t["engine"] == "postgres": env = dict(os.environ, PGPASSWORD=t["pw"]) out = subprocess.run(("psql", "-h", t["host"], "-p", t["port"], "-U", t["user"], @@ -65,8 +78,79 @@ def db_version(t): return t["engine"] +def db_stats(t): + """Read-only live metrics for the dashboard's 'nerd stats' — size, tables, rows, index + footprint, connections, cache-hit ratio, the biggest tables, server version + uptime. + Best-effort: any failure returns None and never disturbs the heartbeat. Postgres only.""" + if t["engine"] == "sqlite": + return _sqlite_stats(t) + if t["engine"] != "postgres": + return None + env = dict(os.environ, PGPASSWORD=t["pw"]) + + def q(sql): + r = subprocess.run(("psql", "-h", t["host"], "-p", t["port"], "-U", t["user"], + "-d", t["db"], "-tAc", sql), + capture_output=True, text=True, timeout=30, env=env) + return r.stdout.strip() + + try: + top = q("SELECT COALESCE(json_agg(x),'[]') FROM (SELECT relname AS name, " + "pg_total_relation_size(relid) AS bytes, n_live_tup AS rows " + "FROM pg_stat_user_tables ORDER BY pg_total_relation_size(relid) DESC " + "LIMIT 10) x") + hit = q("SELECT round(100.0*sum(blks_hit)/NULLIF(sum(blks_hit)+sum(blks_read),0),2) " + "FROM pg_stat_database WHERE datname=current_database()") + up = q("SELECT EXTRACT(EPOCH FROM (now()-pg_postmaster_start_time()))::bigint") + return { + "size_bytes": int(q("SELECT pg_database_size(current_database())") or 0), + "tables": int(q("SELECT count(*) FROM pg_stat_user_tables") or 0), + "rows_est": int(float(q("SELECT COALESCE(SUM(n_live_tup),0) " + "FROM pg_stat_user_tables") or 0)), + "index_bytes": int(q("SELECT COALESCE(SUM(pg_indexes_size(relid)),0) " + "FROM pg_stat_user_tables") or 0), + "connections": int(q("SELECT count(*) FROM pg_stat_activity " + "WHERE datname=current_database()") or 0), + "cache_hit_pct": float(hit) if hit else None, + "server_version": q("SHOW server_version"), + "uptime_s": int(up) if up else None, + "top_tables": json.loads(top or "[]"), + } + except Exception: + return None + + +def _sqlite_stats(t): + """Live metrics for a SQLite file: file size, table count, row totals, largest tables.""" + def q(sql): + r = subprocess.run(("sqlite3", t["path"], sql), + capture_output=True, text=True, timeout=30) + return r.stdout.strip() + try: + names = [n for n in q("SELECT name FROM sqlite_master WHERE type='table' " + "AND name NOT LIKE 'sqlite_%' ORDER BY name").splitlines() if n] + top = [] + rows = 0 + for n in names: + c = int(q(f'SELECT count(*) FROM "{n}"') or 0) + rows += c + top.append({"name": n, "bytes": 0, "rows": c}) + top.sort(key=lambda x: x["rows"], reverse=True) + size = os.path.getsize(t["path"]) if os.path.exists(t["path"]) else 0 + return {"size_bytes": size, "tables": len(names), "rows_est": rows, + "index_bytes": 0, "connections": 0, "cache_hit_pct": None, + "server_version": "SQLite " + q("SELECT sqlite_version()"), + "uptime_s": None, "top_tables": top[:10]} + except Exception: + return None + + def dump_cmd(t): """The dump command + environment for an engine. Streams compressed to stdout.""" + if t["engine"] == "sqlite": + # `.dump` runs in a read transaction → a consistent SQL snapshot even under writes. + # The file (and its -wal/-shm) must be readable in the agent container (bind-mount). + return (("sqlite3", t["path"], ".dump"), dict(os.environ)) if t["engine"] == "postgres": env = dict(os.environ, PGPASSWORD=t["pw"]) # custom format is already compressed and is what pg_restore wants @@ -83,22 +167,60 @@ def dump_cmd(t): raise ValueError(f"unknown engine {t['engine']}") +UPLOAD_CHUNK = 64 * 1024 * 1024 # keep each request well under Cloudflare's 100 MB body cap + + +def _post_chunk(url, token, name, meta, upload_id, index, final, data): + """POST one slice of a dump. Each request has a normal Content-Length under the CDN limit, + so this works over the public HTTPS URL from any server, not just co-located ones.""" + req = urllib.request.Request( + url.rstrip("/") + "/api/dump", data=bytes(data), + headers={"X-Agent-Token": token, "X-DB-Name": name, "X-DB-Meta": meta, + "X-Upload-Id": upload_id, "X-Chunk-Index": str(index), + "X-Chunk-Final": "1" if final else "0", + "Content-Type": "application/octet-stream", + "User-Agent": f"backhaul-agent/{VERSION}"}) + with urllib.request.urlopen(req, timeout=3600) as resp: + return resp.status == 200 + + def run_dump(t, url, token): - """Stream a dump of one target straight to the app's ingest endpoint — never buffer the - whole dump in memory or a temp file.""" + """Stream a dump home in <100 MB slices so it passes the CDN's upload limit from any + server; the app reassembles by upload id and finalizes on the last slice. The dump is never + fully buffered — only one slice is held at a time.""" cmd, env = dump_cmd(t) meta = json.dumps({"engine": t["engine"], "version": db_version(t)}) - req = urllib.request.Request( - url.rstrip("/") + "/api/dump", - headers={"X-Agent-Token": token, "X-DB-Name": t["name"], - "X-DB-Meta": meta, "Content-Type": "application/octet-stream", - "Transfer-Encoding": "chunked", "User-Agent": f"backhaul-agent/{VERSION}"}) + upload_id = os.urandom(8).hex() proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, env=env) - # feed the dump process's stdout straight into the HTTP request body - req.data = iter(lambda: proc.stdout.read(1 << 20), b"") - with urllib.request.urlopen(req, timeout=3600) as resp: + buf = bytearray() + index = 0 + ok = True + try: + while True: + b = proc.stdout.read(1 << 20) + if not b: + break + buf += b + while len(buf) >= UPLOAD_CHUNK: + if not _post_chunk(url, token, t["name"], meta, upload_id, index, False, + buf[:UPLOAD_CHUNK]): + ok = False + break + del buf[:UPLOAD_CHUNK] + index += 1 + if not ok: + break proc.wait() - return {"name": t["name"], "ok": resp.status == 200 and proc.returncode == 0} + if ok: # final slice finalizes the backup (even if empty) + ok = _post_chunk(url, token, t["name"], meta, upload_id, index, True, bytes(buf)) + except Exception as e: + print("dump upload failed:", e, file=sys.stderr) + ok = False + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + return {"name": t["name"], "ok": ok and proc.returncode == 0} def maybe_self_update(answer, url, token): @@ -110,8 +232,10 @@ def maybe_self_update(answer, url, token): return me = os.path.abspath(__file__) try: - req = urllib.request.Request(url.rstrip("/") + "/api/agent-script", - headers={"X-Agent-Token": token}) + req = urllib.request.Request( + url.rstrip("/") + "/api/agent-script", + headers={"X-Agent-Token": token, + "User-Agent": f"backhaul-agent/{VERSION}"}) # else Cloudflare 403s urllib with urllib.request.urlopen(req, timeout=30) as r: source = r.read() text = source.decode() @@ -131,9 +255,14 @@ def maybe_self_update(answer, url, token): def checkin(url, token, tgts, pending): """POST the heartbeat: which DBs we see + any finished dump result. The reply tells us what to do next and the latest agent version.""" + stats = {} + for t in tgts: + s = db_stats(t) + if s: + stats[t["name"]] = s body = json.dumps({"agent_version": VERSION, "databases": [{"name": t["name"], "engine": t["engine"]} for t in tgts], - "results": pending}).encode() + "results": pending, "stats": stats}).encode() req = urllib.request.Request(url.rstrip("/") + "/api/checkin", data=body, headers={"X-Agent-Token": token, "Content-Type": "application/json",