Something went wrong. Try again.
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
Something went wrong. Try again.
9.2 kB · 201 lines
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202#!/usr/bin/env python3"""Pull a consistent PostgreSQL archive; publish it only after an isolated restore."""
import datetimeimport fcntlimport hashlibimport jsonimport osfrom pathlib import Pathimport reimport selectorsimport shutilimport subprocessimport sysimport timeimport urllib.request
IMAGE = "postgres:17.10-alpine@sha256:742f40ea20b9ff2ff31db5458d127452988a2164df9e17441e191f3b72252193"HOSTS = ("stolas.mesh.wisp.place", "baal.mesh.wisp.place", "sjo1.mesh.wisp.place", "sin1.mesh.wisp.place")ROOT = Path("/storage/wisp-db-backups")CONFIG = Path("/home/regent/.config/wisp-db-backup")DOCKER = "/run/current-system/sw/bin/docker"MAPPINGS = ("domains", "custom_domains", "sites")SIGNATURE_SQL = "SELECT json_build_object(" + ",".join( f"'{table}',(SELECT json_build_object('rows',count(*),'digest'," f"md5(COALESCE(string_agg(to_jsonb(t)::text,E'\\n' ORDER BY to_jsonb(t)::text COLLATE \"C\"),''))) " f"FROM public.{table} t)" for table in MAPPINGS) + ");"
def command(args, **kwargs): return subprocess.run(args, check=True, timeout=900, stderr=subprocess.DEVNULL, **kwargs)
def primary(): leaders = [] http = urllib.request.build_opener(urllib.request.ProxyHandler({})) for host in HOSTS: try: with http.open(f"http://{host}:8008/primary", timeout=5) as response: body = json.load(response) if response.status == 200 and body.get("role") in ("primary", "master"): leaders.append(host) except (OSError, ValueError): pass if len(leaders) != 1: raise RuntimeError("expected exactly one reachable primary") return leaders[0]
def client(host): return [DOCKER, "run", "--rm", "-i", "--network", "host", "--label", "com.centurylinklabs.watchtower.enable=false", "--user", f"{os.getuid()}:{os.getgid()}", "-v", f"{CONFIG}/pgpass:/run/backup.pgpass:ro", "-e", "PGPASSFILE=/run/backup.pgpass", "-e", f"PGHOST={host}", "-e", "PGDATABASE=wisp", "-e", "PGUSER=wisp_backup", "-e", "PGCONNECT_TIMEOUT=10", "-e", "PGSSLMODE=disable", "-e", "PGOPTIONS=-c idle_in_transaction_session_timeout=600000 -c statement_timeout=600000", IMAGE]
def query(session, sql): session.stdin.write(sql + "\n") session.stdin.flush() with selectors.DefaultSelector() as selector: selector.register(session.stdout, selectors.EVENT_READ) if not selector.select(30): raise TimeoutError("snapshot query timed out") result = session.stdout.readline().strip() if not result: raise RuntimeError("snapshot session closed") return result
def dump(host, directory): args = client(host) session = subprocess.Popen(args + ["psql", "-XqAt", "-v", "ON_ERROR_STOP=1"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True) try: snapshot = query(session, "BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY; SELECT pg_export_snapshot();") if not re.fullmatch(r"[0-9A-F]+-[0-9A-F]+-[0-9]+", snapshot): raise ValueError("invalid snapshot") expected = json.loads(query(session, SIGNATURE_SQL)) (directory / "source-mappings.json").write_text(json.dumps(expected, indent=2) + "\n") with (directory / "wisp.dump").open("xb") as output: command(args + ["pg_dump", "--format=custom", "--compress=6", "--lock-wait-timeout=30s", "--snapshot=" + snapshot], stdout=output) output.flush() os.fsync(output.fileno()) session.communicate("ROLLBACK;\n", timeout=15) if session.returncode: raise RuntimeError("snapshot session failed") with (directory / "roles.sql").open("xb") as output: command(args + ["pg_dumpall", "--globals-only", "--no-role-passwords", "--database=wisp"], stdout=output) return expected finally: if session.poll() is None: session.kill() session.wait()
def restore_check(directory, expected): name = "wisp-db-restore-" + directory.name.removesuffix(".partial").lower() command([DOCKER, "run", "--rm", "-d", "--name", name, "--network", "none", "--label", "com.centurylinklabs.watchtower.enable=false", "--tmpfs", "/var/lib/postgresql/data:rw,size=512m", "-e", "POSTGRES_HOST_AUTH_METHOD=trust", IMAGE], stdout=subprocess.DEVNULL) try: for attempt in range(60): ready = subprocess.run([DOCKER, "exec", name, "pg_isready", "-U", "postgres"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10) if ready.returncode == 0: break time.sleep(1) else: raise TimeoutError("isolated restore database did not start") command([DOCKER, "exec", name, "createdb", "-U", "postgres", "restorecheck"]) with (directory / "wisp.dump").open("rb") as source: command([DOCKER, "exec", "-i", name, "pg_restore", "-U", "postgres", "--exit-on-error", "--no-owner", "--no-privileges", "-d", "restorecheck"], stdin=source) result = command([DOCKER, "exec", name, "psql", "-XAt", "-U", "postgres", "-d", "restorecheck", "-c", SIGNATURE_SQL], stdout=subprocess.PIPE, text=True) actual = json.loads(result.stdout) if actual != expected: raise ValueError("restored mapping counts or contents differ from source snapshot") return actual finally: # Only this disposable, network-isolated test container is stopped; its DB is tmpfs. command([DOCKER, "stop", "--time", "10", name], stdout=subprocess.DEVNULL)
def heartbeat(): destination = json.loads((CONFIG / "monitor.json").read_text()) url = "https://status.wisp.place/api/v1/endpoints/operations_database-backup/external?success=true" class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None request = urllib.request.Request(url, data=b"", method="POST", headers={"Authorization": "Bearer " + destination["token"]}) with urllib.request.build_opener(NoRedirect()).open(request, timeout=15) as response: if response.status != 200: raise RuntimeError("backup heartbeat rejected")
def verify_archive(directory): directory = directory.resolve() manifest = json.loads((directory / "manifest.json").read_text()) for name in ("wisp.dump", "roles.sql"): with (directory / name).open("rb") as source: digest = hashlib.file_digest(source, "sha256").hexdigest() if digest != manifest["sha256"][name]: raise ValueError("archive checksum mismatch") restore_check(directory, manifest["mapping_signatures"]) print("stored archive checksums and isolated restore verified", flush=True)
def main(): os.umask(0o077) if not os.path.ismount("/storage"): raise RuntimeError("backup storage is not mounted") ROOT.mkdir(mode=0o700, exist_ok=True) with (ROOT / ".lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) if shutil.disk_usage(ROOT).free < 2 * 1024 ** 3: raise RuntimeError("less than 2 GiB free on backup storage") started = datetime.datetime.now(datetime.timezone.utc) stamp = started.strftime("%Y%m%dT%H%M%S.%fZ") directory = ROOT / (stamp + ".partial") directory.mkdir(mode=0o700) host = primary() expected = dump(host, directory) restored = restore_check(directory, expected) checksums = {} for path in (directory / "wisp.dump", directory / "roles.sql"): with path.open("rb") as source: checksums[path.name] = hashlib.file_digest(source, "sha256").hexdigest() manifest = {"started_at": started.isoformat(), "primary": host, "client_image": IMAGE, "verified_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "mapping_signatures": restored, "sha256": checksums} (directory / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n") directory.rename(ROOT / stamp) (ROOT / "latest.next").symlink_to(stamp) os.replace(ROOT / "latest.next", ROOT / "latest") print(f"verified backup {stamp}; mapping rows " + json.dumps({k: v["rows"] for k, v in restored.items()}), flush=True) heartbeat() print("backup freshness reported", flush=True)
if __name__ == "__main__": try: if len(sys.argv) == 3 and sys.argv[1] == "--verify": verify_archive(Path(sys.argv[2])) elif len(sys.argv) == 1: main() else: raise ValueError("usage: backup.py [--verify backup-directory]") except Exception as error: print(f"backup failed ({type(error).__name__}); no success heartbeat sent", file=sys.stderr) sys.exit(1)