204 lines
7.9 KiB
Python
204 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
backup_compose_volumes.py
|
|
-------------------------
|
|
Export (compress) every Docker *volume* that belongs to a Docker Compose project.
|
|
|
|
DISCOVERY MODES
|
|
- By label (default): looks for label com.docker.compose.project=<project>
|
|
- By name prefix (fallback/optional): looks for volume names starting with <project>_ or <project>-
|
|
|
|
This helps when volumes were created with a Compose project like "suitecoffee_dev" or "suitecoffee_prod"
|
|
and you're passing "SuiteCoffee" (capitalized) or when some volumes lack labels.
|
|
|
|
Usage examples
|
|
--------------
|
|
python3 backup_compose_volumes.py -p suitecoffee_dev
|
|
python3 backup_compose_volumes.py -p suitecoffee_prod -o /backups/suitecoffee
|
|
python3 backup_compose_volumes.py -p SuiteCoffee --discovery auto
|
|
python3 backup_compose_volumes.py -p suitecoffee --discovery name # treat -p as a name prefix
|
|
python3 backup_compose_volumes.py --list-only # just list what would be backed up
|
|
|
|
Notes
|
|
-----
|
|
- You generally want to pass the EXACT Compose project name (e.g., "suitecoffee_dev").
|
|
- Docker Compose sets project names in lowercase; labels are case-sensitive.
|
|
- If zero volumes are found by label, this script tries lowercase and name-prefix fallback automatically.
|
|
"""
|
|
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from typing import List, Dict
|
|
|
|
def run(cmd: List[str], check=True, capture_output=True, text=True) -> subprocess.CompletedProcess:
|
|
return subprocess.run(cmd, check=check, capture_output=capture_output, text=text)
|
|
|
|
def which(program: str) -> bool:
|
|
from shutil import which as _which
|
|
return _which(program) is not None
|
|
|
|
def detect_project_name(args_project: str) -> str:
|
|
if args_project:
|
|
return args_project
|
|
env_name = os.environ.get("COMPOSE_PROJECT_NAME")
|
|
if env_name:
|
|
return env_name
|
|
return pathlib.Path.cwd().name.replace(" ", "_")
|
|
|
|
def docker_volume_ls_json(filters: List[str]) -> List[Dict[str, str]]:
|
|
cmd = ["docker", "volume", "ls", "--format", "{{json .}}"]
|
|
for f in filters:
|
|
cmd += ["--filter", f]
|
|
cp = run(cmd)
|
|
out = []
|
|
for line in cp.stdout.splitlines():
|
|
try:
|
|
out.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return out
|
|
|
|
def list_by_label(project: str) -> List[Dict[str, str]]:
|
|
return docker_volume_ls_json([f"label=com.docker.compose.project={project}"])
|
|
|
|
def list_by_name_prefix(prefix: str) -> List[Dict[str, str]]:
|
|
# docker volume ls has no "name prefix" filter; we filter client-side.
|
|
vols = docker_volume_ls_json([])
|
|
keep = []
|
|
for v in vols:
|
|
name = v.get("Name") or v.get("Driver") # Name should be present
|
|
if not name:
|
|
continue
|
|
if name.startswith(prefix + "_") or name.startswith(prefix + "-") or name == prefix:
|
|
keep.append(v)
|
|
return keep
|
|
|
|
def ensure_alpine_image():
|
|
try:
|
|
run(["docker", "image", "inspect", "alpine:latest"])
|
|
except subprocess.CalledProcessError:
|
|
print("Pulling alpine:latest ...")
|
|
run(["docker", "pull", "alpine:latest"], check=True, capture_output=False)
|
|
|
|
def backup_volume(volume_name: str, out_dir: pathlib.Path, archive_name: str, dry_run: bool = False) -> int:
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
out_path = out_dir / archive_name
|
|
docker_cmd = [
|
|
"docker", "run", "--rm",
|
|
"-v", f"{volume_name}:/volume:ro",
|
|
"-v", f"{str(out_dir)}:/backup",
|
|
"alpine:latest",
|
|
"sh", "-lc",
|
|
f"tar czf /backup/{shlex.quote(out_path.name)} -C /volume ."
|
|
]
|
|
if dry_run:
|
|
print("[DRY RUN] Would run:", " ".join(shlex.quote(c) for c in docker_cmd))
|
|
return 0
|
|
cp = subprocess.run(docker_cmd)
|
|
return cp.returncode
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Export (compress) every Docker volume of a Docker Compose project.")
|
|
parser.add_argument("-p", "--project", help="Compose project or prefix (see --discovery).")
|
|
parser.add_argument("-o", "--output", help="Output directory (default: ./docker-volumes-backup-<timestamp>).")
|
|
parser.add_argument("--exclude", nargs="*", default=[], help="Volume names to exclude (space-separated).")
|
|
parser.add_argument("--dry-run", action="store_true", help="Show what would be done without doing it.")
|
|
parser.add_argument("--timestamp", default=datetime.datetime.now().strftime("%Y%m%d-%H%M%S"),
|
|
help="Timestamp to embed into filenames (default: current time).")
|
|
parser.add_argument("--discovery", choices=["auto","label","name"], default="auto",
|
|
help="How to discover volumes: 'label' (strict), 'name' (prefix), or 'auto' (default).")
|
|
parser.add_argument("--list-only", action="store_true", help="Only list volumes that would be backed up and exit.")
|
|
args = parser.parse_args()
|
|
|
|
if not which("docker"):
|
|
print("ERROR: 'docker' is not on PATH. Please install Docker and/or add it to PATH.", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
project_raw = detect_project_name(args.project)
|
|
project_norm = project_raw.replace(" ", "_")
|
|
project_lower = project_norm.lower()
|
|
ts = args.timestamp
|
|
out_dir = pathlib.Path(args.output) if args.output else pathlib.Path(f"./docker-volumes-backup-{ts}")
|
|
|
|
# Ensure daemon available
|
|
try:
|
|
run(["docker", "version"], check=True, capture_output=True)
|
|
except subprocess.CalledProcessError:
|
|
print("ERROR: Unable to talk to the Docker daemon. Are you in the 'docker' group? Is the daemon running?", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
selected = []
|
|
method_used = None
|
|
|
|
# Discovery
|
|
if args.discovery in ("auto","label"):
|
|
vols = list_by_label(project_norm)
|
|
if vols:
|
|
selected = vols; method_used = f"label:{project_norm}"
|
|
elif args.discovery == "auto":
|
|
vols2 = list_by_label(project_lower)
|
|
if vols2:
|
|
selected = vols2; method_used = f"label:{project_lower}"
|
|
|
|
if not selected and args.discovery in ("auto","name"):
|
|
# Treat project as a prefix
|
|
# Try exact, then lowercase
|
|
by_name = list_by_name_prefix(project_norm)
|
|
if by_name:
|
|
selected = by_name; method_used = f"name-prefix:{project_norm}"
|
|
else:
|
|
by_name2 = list_by_name_prefix(project_lower)
|
|
if by_name2:
|
|
selected = by_name2; method_used = f"name-prefix:{project_lower}"
|
|
|
|
if not selected:
|
|
print(f"No volumes found for project/prefix '{project_raw}'. Tried methods:")
|
|
print(f" - label:{project_norm}")
|
|
print(f" - label:{project_lower}")
|
|
print(f" - name-prefix:{project_norm} (prefix_*, prefix-*)")
|
|
print(f" - name-prefix:{project_lower} (prefix_*, prefix-*)")
|
|
sys.exit(0)
|
|
|
|
exclude_set = set(args.exclude or [])
|
|
selected = [v for v in selected if v.get("Name") not in exclude_set]
|
|
|
|
print(f"Discovery method: {method_used}")
|
|
print(f"Volumes discovered: {len(selected)}")
|
|
for v in selected:
|
|
print(" -", v.get("Name"))
|
|
|
|
if args.list_only:
|
|
return
|
|
|
|
if not args.dry_run:
|
|
ensure_alpine_image()
|
|
|
|
failures = []
|
|
for v in selected:
|
|
vname = v.get("Name")
|
|
if not vname:
|
|
continue
|
|
# Determine a 'project token' for filename: take the leading prefix before first '_' or '-'
|
|
prefix = project_lower
|
|
archive = f"{prefix}-{vname}-{ts}.tar.gz"
|
|
print(f"Backing up volume: {vname} -> {archive}")
|
|
rc = backup_volume(vname, out_dir, archive, dry_run=args.dry_run)
|
|
if rc != 0:
|
|
print(f" ERROR: backup failed for volume '{vname}' (exit code {rc})", file=sys.stderr)
|
|
failures.append(vname)
|
|
|
|
if failures:
|
|
print("\nCompleted with errors. Failed volumes:", ", ".join(failures))
|
|
sys.exit(1)
|
|
else:
|
|
print("\nAll done. Archives written to:", str(out_dir.resolve()))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|