mirror of
https://github.com/acedanger/shell.git
synced 2026-03-24 20:11:49 -07:00
599 lines
24 KiB
Python
Executable File
599 lines
24 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Docker Manager Application
|
|
|
|
This script manages Docker containers defined in subdirectories of `~/docker/`.
|
|
Each subdirectory is treated as a project and must contain a `docker-compose.yml`, `docker-compose.yaml`, `compose.yml`, or `compose.yaml`.
|
|
|
|
Usage Examples:
|
|
1. List all running containers:
|
|
$ dm list
|
|
|
|
2. Update a specific project (pulls latest images and recreates containers):
|
|
$ dm update media-server
|
|
|
|
3. Stop all projects (with confirmation prompt):
|
|
$ dm stop
|
|
|
|
4. Restart all projects without confirmation:
|
|
$ dm restart --all
|
|
|
|
5. Stop a specific project:
|
|
$ dm stop web-app
|
|
|
|
Purpose:
|
|
- Simplifies management of multiple Docker Compose projects.
|
|
- Provides a unified interface for common operations (list, stop, update, restart).
|
|
- Eliminates the need to navigate to specific directories to run docker commands.
|
|
|
|
To make the command available as dm from any directory, run:
|
|
chmod +x ~/shell/docker-manager/docker-manager.py
|
|
mkdir -p ~/.local/bin
|
|
ln -s ~/shell/docker-manager/docker-manager.py ~/.local/bin/dm
|
|
"""
|
|
|
|
import os
|
|
import argparse
|
|
import subprocess
|
|
import json
|
|
from pathlib import Path
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
from rich.panel import Panel
|
|
from rich.text import Text
|
|
from rich import box
|
|
|
|
# Configuration
|
|
DOCKER_ROOT = Path(os.path.expanduser("~/docker"))
|
|
console = Console()
|
|
|
|
|
|
def get_diun_container_name():
|
|
"""Find the running diun container name."""
|
|
# Try to find by image name first (most reliable if standard image is used)
|
|
cmd = ["docker", "ps", "--filter", "ancestor=crazymax/diun", "--format", "{{.Names}}"]
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
|
if res.returncode == 0 and res.stdout.strip():
|
|
return res.stdout.strip().split('\n')[0]
|
|
|
|
# Fallback to finding by name containing 'diun'
|
|
cmd = ["docker", "ps", "--filter", "name=diun", "--format", "{{.Names}}"]
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
|
if res.returncode == 0 and res.stdout.strip():
|
|
return res.stdout.strip().split('\n')[0]
|
|
|
|
return None
|
|
|
|
|
|
def get_diun_info():
|
|
"""Fetch image information from Diun."""
|
|
diun_info = {}
|
|
try:
|
|
diun_container = get_diun_container_name()
|
|
|
|
if not diun_container:
|
|
return diun_info
|
|
|
|
# Get diun image list
|
|
cmd = ["docker", "exec", diun_container, "diun", "image", "list", "--raw"]
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
|
|
|
if res.returncode != 0:
|
|
return diun_info
|
|
|
|
data = json.loads(res.stdout)
|
|
for img in data.get("images", []):
|
|
name = img.get("name", "")
|
|
latest = img.get("latest", {})
|
|
digest = latest.get("digest", "")
|
|
labels = latest.get("labels", {})
|
|
|
|
# Normalize name for matching
|
|
# Remove docker.io/ prefix
|
|
norm_name = name
|
|
if norm_name.startswith("docker.io/"):
|
|
norm_name = norm_name[10:]
|
|
|
|
# If no registry (no /), prepend library/
|
|
if "/" not in norm_name:
|
|
norm_name = f"library/{norm_name}"
|
|
|
|
diun_info[norm_name] = {
|
|
"digest": digest,
|
|
"labels": labels,
|
|
"tag": latest.get("tag", ""),
|
|
"original_name": name
|
|
}
|
|
|
|
except (subprocess.CalledProcessError, json.JSONDecodeError):
|
|
pass
|
|
|
|
return diun_info
|
|
|
|
|
|
def get_image_version(labels):
|
|
"""Extract version from image labels."""
|
|
version_keys = [
|
|
"org.opencontainers.image.version",
|
|
"version",
|
|
"org.label-schema.version",
|
|
"build_version"
|
|
]
|
|
for key in version_keys:
|
|
if key in labels:
|
|
return labels[key]
|
|
return "latest"
|
|
|
|
|
|
def get_projects():
|
|
"""Scan DOCKER_ROOT for subdirectories with compose files."""
|
|
projects = {}
|
|
if not DOCKER_ROOT.exists():
|
|
print(f"Error: Directory {DOCKER_ROOT} does not exist.")
|
|
return projects
|
|
|
|
for item in DOCKER_ROOT.iterdir():
|
|
if item.is_dir():
|
|
if (item / "docker-compose.yml").exists() or \
|
|
(item / "docker-compose.yaml").exists() or \
|
|
(item / "compose.yml").exists() or \
|
|
(item / "compose.yaml").exists():
|
|
projects[item.name] = item
|
|
return projects
|
|
|
|
|
|
def run_command(cmd, cwd, capture_output=False):
|
|
"""Run a shell command in a specific directory."""
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
cwd=cwd,
|
|
check=True,
|
|
text=True,
|
|
capture_output=capture_output
|
|
)
|
|
return result
|
|
except subprocess.CalledProcessError as e:
|
|
if capture_output:
|
|
return e
|
|
print(f"Error running command: {' '.join(cmd)}")
|
|
print(f"cwd: {cwd}")
|
|
return None
|
|
|
|
|
|
def list_containers(projects):
|
|
"""List running containers for all projects."""
|
|
table = Table(title="Docker Containers", box=box.ROUNDED)
|
|
table.add_column("Project", style="cyan", no_wrap=True)
|
|
table.add_column("Container Name", style="green")
|
|
table.add_column("State", style="yellow")
|
|
table.add_column("Image", style="blue")
|
|
table.add_column("Version", style="magenta")
|
|
table.add_column("Update", style="red")
|
|
|
|
# Get Diun info once
|
|
diun_info = get_diun_info()
|
|
|
|
found_any = False
|
|
for name, path in sorted(projects.items()):
|
|
# Get running container names
|
|
cmd = ["docker", "compose", "ps", "--format", "json"]
|
|
res = run_command(cmd, path, capture_output=True)
|
|
|
|
if res and res.returncode == 0:
|
|
lines = res.stdout.strip().split('\n')
|
|
for line in lines:
|
|
if not line:
|
|
continue
|
|
try:
|
|
container = json.loads(line)
|
|
c_name = container.get('Name', '')
|
|
state = container.get('State', '')
|
|
image = container.get('Image', '')
|
|
|
|
if not c_name:
|
|
continue
|
|
|
|
found_any = True
|
|
|
|
# Get version and update status
|
|
version = "unknown"
|
|
update_status = ""
|
|
|
|
# Normalize image name for lookup
|
|
norm_image = image
|
|
if norm_image.startswith("docker.io/"):
|
|
norm_image = norm_image[10:]
|
|
if "/" not in norm_image:
|
|
norm_image = f"library/{norm_image}"
|
|
|
|
# Check Diun info
|
|
if norm_image in diun_info:
|
|
info = diun_info[norm_image]
|
|
# Try to get version from Diun labels first
|
|
version = get_image_version(info.get("labels", {}))
|
|
|
|
# Check for updates
|
|
inspect_digest_cmd = ["docker", "inspect", c_name, "--format", "{{index .RepoDigests 0}}"]
|
|
inspect_digest_res = run_command(inspect_digest_cmd, path, capture_output=True)
|
|
if inspect_digest_res and inspect_digest_res.returncode == 0:
|
|
running_digest_full = inspect_digest_res.stdout.strip()
|
|
# running_digest is like name@sha256:hash
|
|
if "@" in running_digest_full:
|
|
running_digest = running_digest_full.split("@")[1]
|
|
latest_digest = info.get("digest", "")
|
|
if latest_digest and running_digest != latest_digest:
|
|
update_status = "Update Available"
|
|
|
|
# If version is still unknown, try to get from running container labels
|
|
if version == "latest" or version == "unknown":
|
|
inspect_cmd = ["docker", "inspect", c_name, "--format", "{{json .Config.Labels}}"]
|
|
inspect_res = run_command(inspect_cmd, path, capture_output=True)
|
|
if inspect_res and inspect_res.returncode == 0:
|
|
try:
|
|
labels = json.loads(inspect_res.stdout)
|
|
version = get_image_version(labels)
|
|
except:
|
|
pass
|
|
|
|
table.add_row(name, c_name, state, image, version, update_status)
|
|
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
if found_any:
|
|
console.print(table)
|
|
else:
|
|
console.print("[yellow]No running containers found in managed projects.[/yellow]")
|
|
|
|
|
|
def is_project_running(path):
|
|
"""Check if any containers in the project are running."""
|
|
cmd = ["docker", "compose", "ps", "--format",
|
|
"json", "--filter", "status=running"]
|
|
res = run_command(cmd, path, capture_output=True)
|
|
if res and res.returncode == 0:
|
|
# If output is not empty/just brackets, something is running
|
|
output = res.stdout.strip()
|
|
return output and output != "[]"
|
|
return False
|
|
|
|
|
|
def describe_project(projects, target):
|
|
"""Show detailed information about containers in a project."""
|
|
if target not in projects:
|
|
console.print(f"[red]Error: Project '{target}' not found.[/red]")
|
|
return
|
|
|
|
path = projects[target]
|
|
console.print(Panel(f"[bold blue]Describing project: {target}[/bold blue]", expand=False))
|
|
|
|
cmd = ["docker", "compose", "ps", "--format", "json"]
|
|
res = run_command(cmd, path, capture_output=True)
|
|
|
|
if res and res.returncode == 0:
|
|
lines = res.stdout.strip().split('\n')
|
|
found_any = False
|
|
for line in lines:
|
|
if not line:
|
|
continue
|
|
try:
|
|
container = json.loads(line)
|
|
found_any = True
|
|
name = container.get('Name', 'N/A')
|
|
|
|
# Create a table for this container
|
|
table = Table(show_header=False, box=box.SIMPLE)
|
|
table.add_column("Field", style="bold cyan")
|
|
table.add_column("Value")
|
|
|
|
table.add_row("Service", container.get('Service', 'N/A'))
|
|
table.add_row("Container", name)
|
|
table.add_row("Image", container.get('Image', 'N/A'))
|
|
table.add_row("State", container.get('State', 'N/A'))
|
|
table.add_row("Status", container.get('Status', 'N/A'))
|
|
|
|
ports = container.get('Ports', '')
|
|
if ports:
|
|
table.add_row("Ports", ports)
|
|
|
|
# Get details via inspect
|
|
inspect_cmd = ["docker", "inspect", name, "--format", "{{json .}}"]
|
|
inspect_res = run_command(inspect_cmd, path, capture_output=True)
|
|
if inspect_res and inspect_res.returncode == 0:
|
|
try:
|
|
details = json.loads(inspect_res.stdout.strip())
|
|
|
|
# Description
|
|
labels = details.get("Config", {}).get("Labels", {})
|
|
description = labels.get("org.opencontainers.image.description")
|
|
if description:
|
|
table.add_row("Description", description)
|
|
|
|
console.print(Panel(table, title=f"[bold]{name}[/bold]", border_style="green"))
|
|
|
|
# Volumes/Mounts
|
|
mounts = details.get("Mounts", [])
|
|
if mounts:
|
|
v_table = Table(title="Volumes", box=box.MINIMAL)
|
|
v_table.add_column("Type", style="cyan")
|
|
v_table.add_column("Source", style="yellow")
|
|
v_table.add_column("Destination", style="green")
|
|
|
|
for mount in mounts:
|
|
m_type = mount.get("Type")
|
|
source = mount.get("Source", "")
|
|
dest = mount.get("Destination", "")
|
|
v_table.add_row(m_type, source, dest)
|
|
|
|
console.print(v_table)
|
|
console.print("") # Spacing
|
|
|
|
except json.JSONDecodeError:
|
|
pass
|
|
else:
|
|
console.print(Panel(table, title=f"[bold]{name}[/bold]", border_style="green"))
|
|
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
if not found_any:
|
|
console.print("[yellow]No containers found for this project.[/yellow]")
|
|
else:
|
|
console.print("[red]Failed to get container info.[/red]")
|
|
|
|
|
|
def list_project_volumes(projects, target):
|
|
"""List volumes used by containers in a project."""
|
|
if target not in projects:
|
|
console.print(f"[red]Error: Project '{target}' not found.[/red]")
|
|
return
|
|
|
|
path = projects[target]
|
|
|
|
table = Table(title=f"Volumes for project: {target}", box=box.ROUNDED)
|
|
table.add_column("Service", style="cyan")
|
|
table.add_column("Container", style="blue")
|
|
table.add_column("Type", style="magenta")
|
|
table.add_column("Source", style="yellow")
|
|
table.add_column("Destination", style="green")
|
|
|
|
cmd = ["docker", "compose", "ps", "--format", "json"]
|
|
res = run_command(cmd, path, capture_output=True)
|
|
|
|
if res and res.returncode == 0:
|
|
lines = res.stdout.strip().split('\n')
|
|
for line in lines:
|
|
if not line:
|
|
continue
|
|
try:
|
|
container = json.loads(line)
|
|
name = container.get('Name', 'N/A')
|
|
service = container.get('Service', 'N/A')
|
|
|
|
inspect_cmd = ["docker", "inspect", name, "--format", "{{json .Mounts}}"]
|
|
inspect_res = run_command(inspect_cmd, path, capture_output=True)
|
|
|
|
if inspect_res and inspect_res.returncode == 0:
|
|
try:
|
|
mounts = json.loads(inspect_res.stdout.strip())
|
|
if mounts:
|
|
for mount in mounts:
|
|
m_type = mount.get("Type", "N/A")
|
|
source = mount.get("Source", "N/A")
|
|
dest = mount.get("Destination", "N/A")
|
|
table.add_row(service, name, m_type, source, dest)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
console.print(table)
|
|
|
|
|
|
|
|
def manage_project(projects, action, target, extra_args=None, container_filter=None):
|
|
"""
|
|
Execute the specified action (stop, update, restart, logs) on target project(s).
|
|
|
|
Args:
|
|
projects (dict): Dictionary of project names to their paths.
|
|
action (str): The action to perform ('stop', 'update', 'restart', 'logs').
|
|
target (str): The target project name or 'all'.
|
|
extra_args (list): Additional arguments for the command (e.g. for logs).
|
|
container_filter (str): Optional container/service name to filter logs.
|
|
"""
|
|
|
|
targets = []
|
|
# Determine which projects to target
|
|
if target == "all":
|
|
targets = sorted(projects.items())
|
|
elif target in projects:
|
|
targets = [(target, projects[target])]
|
|
else:
|
|
console.print(f"[red]Error: Project '{target}' not found in {DOCKER_ROOT}[/red]")
|
|
console.print(f"Available projects: {', '.join(sorted(projects.keys()))}")
|
|
return
|
|
|
|
# Logs are special, usually run on one project interactively
|
|
if action == "logs":
|
|
if len(targets) > 1:
|
|
console.print("[red]Error: Logs can only be viewed for one project at a time.[/red]")
|
|
return
|
|
name, path = targets[0]
|
|
|
|
# Filter by container/service if specified
|
|
services_to_log = []
|
|
if container_filter:
|
|
cmd_ps = ["docker", "compose", "ps", "--format", "json"]
|
|
res = run_command(cmd_ps, path, capture_output=True)
|
|
if res and res.returncode == 0:
|
|
try:
|
|
lines = res.stdout.strip().split('\n')
|
|
for line in lines:
|
|
if not line: continue
|
|
c = json.loads(line)
|
|
s_name = c.get('Service', '')
|
|
c_name = c.get('Name', '')
|
|
# Check if filter matches service or container name
|
|
if container_filter in s_name or container_filter in c_name:
|
|
services_to_log.append(s_name)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
if not services_to_log:
|
|
console.print(f"[yellow]No matching service/container found for '{container_filter}'. Showing all logs.[/yellow]")
|
|
else:
|
|
services_to_log = list(set(services_to_log))
|
|
console.print(f"[blue]Showing logs for service(s): {', '.join(services_to_log)}[/blue]")
|
|
|
|
console.print(f"[blue]Viewing logs for {name}...[/blue]")
|
|
cmd = ["docker", "compose", "logs"] + (extra_args or [])
|
|
|
|
if services_to_log:
|
|
cmd.extend(services_to_log)
|
|
|
|
# For logs, we want to stream output directly to stdout, so no capture_output
|
|
# and we might want to allow Ctrl+C to exit gracefully
|
|
try:
|
|
subprocess.run(cmd, cwd=path, check=False)
|
|
except KeyboardInterrupt:
|
|
console.print("\n[yellow]Log view stopped.[/yellow]")
|
|
return
|
|
|
|
console.print(f"[bold]Performing '{action}' on {len(targets)} project(s)...[/bold]")
|
|
|
|
for name, path in targets:
|
|
console.print(f"\n[bold cyan][{name}] -> {action}...[/bold cyan]")
|
|
|
|
# Execute the requested action
|
|
if action == "stop":
|
|
# Stop containers without removing them
|
|
run_command(["docker", "compose", "stop"], path)
|
|
elif action == "restart":
|
|
# Restart containers
|
|
run_command(["docker", "compose", "restart"], path)
|
|
elif action == "update":
|
|
# Update process: Check running -> Stop -> Pull -> Start if was running
|
|
was_running = is_project_running(path)
|
|
|
|
if was_running:
|
|
console.print(f" [yellow]Stopping running containers for {name}...[/yellow]")
|
|
run_command(["docker", "compose", "stop"], path)
|
|
|
|
console.print(f" [blue]Pulling latest images for {name}...[/blue]")
|
|
pull_res = run_command(["docker", "compose", "pull"], path)
|
|
|
|
if pull_res and pull_res.returncode == 0:
|
|
if was_running:
|
|
console.print(f" [green]Restarting containers for {name}...[/green]")
|
|
run_command(["docker", "compose", "up", "-d"], path)
|
|
else:
|
|
console.print(
|
|
f" [green]Project {name} was not running. Images updated, but not started.[/green]")
|
|
else:
|
|
console.print(f" [red]Failed to pull images for {name}. Skipping update.[/red]")
|
|
|
|
|
|
def main():
|
|
"""Main entry point for the application."""
|
|
parser = argparse.ArgumentParser(
|
|
description="Manage Docker containers in ~/docker/")
|
|
subparsers = parser.add_subparsers(
|
|
dest="command", help="Command to execute")
|
|
|
|
# List command configuration
|
|
subparsers.add_parser("list", help="List running containers")
|
|
|
|
# Describe command configuration
|
|
describe_parser = subparsers.add_parser("describe", help="Show details of a project's containers")
|
|
describe_parser.add_argument("project", help="Project name to describe")
|
|
|
|
# Volumes command configuration
|
|
volumes_parser = subparsers.add_parser("volumes", help="List volumes used by a project")
|
|
volumes_parser.add_argument("project", help="Project name to list volumes for")
|
|
|
|
# Stop command configuration
|
|
stop_parser = subparsers.add_parser("stop", help="Stop containers")
|
|
stop_parser.add_argument("project", nargs="?", default=None,
|
|
help="Project name (optional). If omitted, asks for confirmation to stop ALL.")
|
|
stop_parser.add_argument("--all", action="store_true",
|
|
help="Stop all projects without confirmation prompt if specified")
|
|
|
|
# Update command configuration
|
|
update_parser = subparsers.add_parser(
|
|
"update", help="Pull and update containers")
|
|
update_parser.add_argument("project", nargs="?", default=None,
|
|
help="Project name (optional). If omitted, asks for confirmation to update ALL.")
|
|
update_parser.add_argument("--all", action="store_true",
|
|
help="Update all projects without confirmation prompt if specified")
|
|
|
|
# Restart command configuration
|
|
restart_parser = subparsers.add_parser(
|
|
"restart", help="Restart containers")
|
|
restart_parser.add_argument("project", nargs="?", default=None,
|
|
help="Project name (optional). If omitted, asks for confirmation to restart ALL.")
|
|
restart_parser.add_argument("--all", action="store_true",
|
|
help="Restart all projects without confirmation prompt if specified")
|
|
|
|
# Logs command configuration
|
|
logs_parser = subparsers.add_parser("logs", help="View container logs")
|
|
logs_parser.add_argument("project", help="Project name to view logs for")
|
|
logs_parser.add_argument("container", nargs="?", help="Container/Service name to filter logs (optional)")
|
|
logs_parser.add_argument(
|
|
"-f", "--follow", action="store_true", help="Follow log output")
|
|
logs_parser.add_argument(
|
|
"--tail", default="all", help="Number of lines to show from the end of the logs (default: all)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Scan for projects
|
|
projects = get_projects()
|
|
if not projects:
|
|
print(f"No docker projects found in {DOCKER_ROOT}")
|
|
return
|
|
|
|
# Dispatch commands
|
|
if args.command == "list":
|
|
list_containers(projects)
|
|
elif args.command == "describe":
|
|
describe_project(projects, args.project)
|
|
elif args.command == "volumes":
|
|
list_project_volumes(projects, args.project)
|
|
elif args.command == "logs":
|
|
# Prepare extra args for logs
|
|
extra_args = []
|
|
if args.follow:
|
|
extra_args.append("-f")
|
|
if args.tail:
|
|
extra_args.extend(["--tail", args.tail])
|
|
|
|
manage_project(projects, "logs", args.project, extra_args=extra_args, container_filter=args.container)
|
|
|
|
elif args.command in ["stop", "update", "restart"]:
|
|
target = args.project
|
|
|
|
# Handle "all" logic safely with user confirmation
|
|
if target is None:
|
|
if args.all:
|
|
target = "all"
|
|
else:
|
|
# Interactive check if user wants to apply to all
|
|
print(
|
|
f"No project specified. Do you want to {args.command} ALL projects? (y/N)")
|
|
choice = input().lower()
|
|
if choice == 'y':
|
|
target = "all"
|
|
else:
|
|
print("Operation cancelled. Specify a project name or use --all.")
|
|
return
|
|
|
|
manage_project(projects, args.command, target)
|
|
else:
|
|
# Show help if no command provided
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|