diff --git a/main.py b/main.py new file mode 100644 index 0000000..760e2f6 --- /dev/null +++ b/main.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 + +import argparse +import subprocess +import re +import sys +import os +from pathlib import Path +import shutil + +VOLUME_BASE_PATH = Path("/var/lib/docker/volumes") + +def get_anonymous_volumes(): + result = subprocess.run( + ["docker", "volume", "ls", "--format", "{{.Name}}"], + stdout=subprocess.PIPE, + text=True + ) + return [ + vol for vol in result.stdout.splitlines() + if re.fullmatch(r"[a-f0-9]{64}", vol) + ] + +def get_mount_path(volume): + containers = subprocess.run(["docker", "ps", "-q"], stdout=subprocess.PIPE, text=True).stdout.split() + for container_id in containers: + mount_path = subprocess.run( + [ + "docker", "inspect", container_id, + "--format", f"{{{{range .Mounts}}}}{{{{if eq .Name \"{volume}\"}}}}{{{{.Destination}}}}{{{{end}}}}{{{{end}}}}" + ], + stdout=subprocess.PIPE, + text=True + ).stdout.strip() + if mount_path: + return mount_path + return None + +def is_volume_used(volume): + result = subprocess.run( + ["docker", "ps", "-aq", "--filter", f"volume={volume}"], + stdout=subprocess.PIPE, + text=True + ) + return bool(result.stdout.strip()) + +def cleanup_symlink(volume): + volume_path = VOLUME_BASE_PATH / volume + if volume_path.is_symlink(): + target_path = volume_path.resolve() + print(f"Volume directory {volume_path} is a symlink to {target_path}.") + try: + print(f"Removing symlink: {volume_path}") + volume_path.unlink() + if target_path.exists(): + print(f"Removing symlink target directory: {target_path}") + shutil.rmtree(target_path) + except Exception as e: + print(f"Failed to clean up symlink or target for {volume}: {e}") + +def delete_volume(volume): + cleanup_symlink(volume) + subprocess.run(["docker", "volume", "rm", volume]) + +def main(): + parser = argparse.ArgumentParser(description="Remove unused anonymous Docker volumes.") + parser.add_argument("whitelist", nargs="?", default="", help="Space-separated list of whitelisted volume IDs") + parser.add_argument("--no-confirmation", action="store_true", help="Skip confirmation before deleting volumes") + args = parser.parse_args() + + whitelist = set(args.whitelist.split()) + + anonymous_volumes = get_anonymous_volumes() + if not anonymous_volumes: + print("No anonymous volumes found.") + sys.exit(0) + + to_delete = [] + + print("Checking anonymous volumes...\n") + + for volume in anonymous_volumes: + if volume in whitelist: + print(f"Volume {volume} is whitelisted and will be skipped.") + continue + + mount_path = get_mount_path(volume) + if mount_path == "/var/www/bootstrap": + print(f"Volume {volume} is mounted at /var/www/bootstrap and will be skipped.") + continue + + if not is_volume_used(volume): + print(f"Volume {volume} is not used by any running containers.") + to_delete.append(volume) + else: + print(f"Volume {volume} is still used and will not be deleted.") + + if not to_delete: + print("\nNo unused anonymous volumes to delete.") + sys.exit(0) + + print("\nThe following volumes will be deleted:") + for vol in to_delete: + print(f" - {vol}") + + if not args.no_confirmation: + confirm = input("\nDo you want to proceed? [y/N]: ").lower() + if confirm != "y": + print("Aborted.") + sys.exit(1) + + for volume in to_delete: + delete_volume(volume) + + print("\nUnused anonymous volumes deleted.") + +if __name__ == "__main__": + main()