Poor man's copy with re-chunking for restic

restic-to-restic.py

#! /usr/bin/env nix-shell
#! nix-shell -i python -p bash restic bubblewrap jq

import json
import logging
import os
import shlex
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
from tempfile import TemporaryDirectory

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger()
# logger.setLevel(level=logging.DEBUG)

USERNAME = "sk"
HOST = "alien"
S3_ENV = "/run/secrets/restic-domashka-env"

RESTIC_PACK_SIZE = 128
RESTIC_PASSWORD_FILE = os.environ["RESTIC_PASSWORD_FILE"]
RESTIC_REPOSITORY_FROM = os.environ["RESTIC_REPOSITORY_FROM"]
RESTIC_REPOSITORY_TO = os.environ["RESTIC_REPOSITORY_TO"]
RESTIC_READ_CONCURRENCY = "8"
GODEBUG = "asyncpreemptoff=1"


def populate_env() -> None:
    for line in Path(S3_ENV).read_text().splitlines():
        k, v = line.split("=", 2)
        os.environ[k] = v
    os.environ["RESTIC_PASSWORD_FILE"] = RESTIC_PASSWORD_FILE
    os.environ["RESTIC_PACK_SIZE"] = str(RESTIC_PACK_SIZE)
    os.environ["RESTIC_READ_CONCURRENCY"] = RESTIC_READ_CONCURRENCY
    os.environ["GODEBUG"] = GODEBUG
    os.environ["LANG"] = "C"
    os.environ["LC_ALL"] = "C"


def mount_source(mountpoint: str):
    command = shlex.split(
        f"restic -r {shlex.quote(RESTIC_REPOSITORY_FROM)} mount {shlex.quote(mountpoint)}"
    )
    return subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        # shell=True,
    )


def list_snapshots():
    command = shlex.split(
        f"restic -r {shlex.quote(RESTIC_REPOSITORY_FROM)} snapshots --json"
    )
    proc = subprocess.run(command, capture_output=True, check=True)
    return json.loads(proc.stdout.decode())


def mk_wrapper_command(root: str, paths: list[str]):
    command = ["bwrap", "--bind", "/", "/"]
    if len(paths) == 1:
        command += ["--bind", root, paths[0]]
    else:
        for p in paths:
            command += ["--bind", root + p, p]
    command += ["--"]

    return command


def backup(root: str, snapshot):
    # snap_time = datetime.strptime(snapshot["time"], "%Y-%m-%dT%H:%M:%S.f%z")
    snap_time = datetime.fromisoformat(snapshot["time"])
    snap_paths = snapshot["paths"]
    snap_root = Path(root) / "ids" / snapshot["short_id"]
    logger.info(
        "Processing snapshot ID=%s, DT=%s, PATHS=%s",
        snapshot["short_id"],
        snap_time,
        ",".join(snap_paths),
    )
    command = (
        mk_wrapper_command(str(snap_root), snap_paths)
        + ["restic", "-r", RESTIC_REPOSITORY_TO, "backup"]
        + ["--ignore-ctime", "--ignore-inode", "--no-scan"]
        + ["--host", HOST, "--tag", "imported"]
        + ["--time", snap_time.strftime("%Y-%m-%d %H:%M:%S")]
        # + ["-n"]
        + snap_paths
    )
    logger.debug(f"Run: {' '.join(command)}")

    subprocess.run(command, stdout=sys.stdout, stderr=sys.stderr, check=True)


def main():
    populate_env()
    logger.debug("List snapshots")
    snapshots = list_snapshots()

    logger.info("Found %s snapshots", len(snapshots))
    len(snapshots)

    with TemporaryDirectory() as tmp_dir:
        logger.debug("Mount source repository")
        mount_proc = mount_source(tmp_dir)
        while len(list(Path(tmp_dir).iterdir())) == 0:
            logger.debug("Wait when source repository is ready")
            time.sleep(1)

        try:
            for snapshot in snapshots:
                backup(tmp_dir, snapshot)
        except KeyboardInterrupt:
            pass
        except Exception as e:
            logger.error(e)
        finally:
            logger.debug("Terminate source repository mount process")
            mount_proc.terminate()
            mount_proc.wait(timeout=60)


if __name__ == "__main__":
    main()