#!/usr/bin/env python3 import argparse import re import subprocess import shutil from pathlib import Path import yaml INTERNAL_DOMAIN_RE = re.compile(r"\b[a-zA-Z0-9.-]+\.lan\.ddnsgeek\.com\b") HOST_MATCH_RE = re.compile(r"Host\(([^)]*)\)") TICKED_HOST_RE = re.compile(r"`([^`]+)`") def require_dot() -> None: if not shutil.which("dot"): raise SystemExit( "Graphviz 'dot' not found in environment. Install graphviz before running docs generation." ) def load_compose(path: Path) -> dict: with path.open() as handle: return yaml.safe_load(handle) or {} def sanitize_domain(value: str, known: dict[str, str]) -> str: if value in known: return known[value] if INTERNAL_DOMAIN_RE.search(value): mapped = f"service-{len(known)+1}." known[value] = mapped return mapped return value def parse_labels(service: dict) -> dict[str, str]: labels = service.get("labels") or {} if isinstance(labels, list): mapped = {} for item in labels: if "=" in str(item): k, v = str(item).split("=", 1) mapped[k] = v return mapped if isinstance(labels, dict): return {str(k): str(v) for k, v in labels.items()} return {} def extract_hosts(rule: str) -> list[str]: hosts: list[str] = [] for m in HOST_MATCH_RE.findall(rule or ""): for host in TICKED_HOST_RE.findall(m): hosts.append(host.strip()) return hosts def render_svg(dot_path: Path, svg_path: Path) -> None: subprocess.run(["dot", "-Tsvg", str(dot_path), "-o", str(svg_path)], check=True) def write_dot(path: Path, lines: list[str]) -> None: path.write_text("\n".join(lines) + "\n") def infer_host(service_name: str, service: dict) -> str: labels = parse_labels(service) for key in ("com.docker.compose.project", "infra.host", "host"): value = labels.get(key) if value: return value.lower() s = service_name.lower() if "raspi" in s or "pi" in s: return "raspberrypi" if "proxmox" in s: return "proxmox" return "docker" def categorize_service(service_name: str) -> str: s = service_name.lower() if any(k in s for k in ["traefik", "authelia", "oauth", "auth", "proxy", "nginx", "caddy"]): return "edge/proxy/auth" if any(k in s for k in ["prometheus", "grafana", "loki", "promtail", "alert", "node-exporter", "cadvisor"]): return "monitoring" if any(k in s for k in ["watchtower", "diun", "ansible", "cron", "runner", "backup"]): return "automation" if any(k in s for k in ["postgres", "mariadb", "mysql", "redis", "minio", "nfs", "storage", "db", "queue"]): return "storage/database/support" return "apps" def generate_physical_topology(compose: dict, out_dot: Path, out_svg: Path) -> None: services = compose.get("services") or {} hosts: dict[str, dict[str, list[str]]] = {} for name, svc in services.items(): host = infer_host(name, svc) cat = categorize_service(name) hosts.setdefault(host, {}).setdefault(cat, []).append(name) lines = [ "digraph PhysicalTopology {", " graph [rankdir=LR, compound=true, splines=polyline, nodesep=0.7, ranksep=1.2, fontname=\"Helvetica\", concentrate=true];", " node [fontname=\"Helvetica\", fontsize=10, style=\"rounded,filled\", fillcolor=\"#ffffff\"];", " edge [fontname=\"Helvetica\", fontsize=9, color=\"#64748b\"];", ] for host, cat_map in sorted(hosts.items()): lines.extend([ f' subgraph "cluster_{host}" {{', f' label="{host} host";', ' style="rounded,filled";', ' color="#93c5fd";', ' fillcolor="#eff6ff";', ]) for category, svcs in sorted(cat_map.items()): cluster_id = f"cluster_{host}_{re.sub(r'[^a-zA-Z0-9]+', '_', category)}" lines.extend([ f' subgraph "{cluster_id}" {{', f' label="{category}";', ' style="rounded,dashed";', ' color="#bfdbfe";', ' fillcolor="#f8fbff";', ]) for service in sorted(svcs): lines.append(f' "svc:{service}" [label="{service}", shape=box, fillcolor="#dcfce7"];') lines.append(' }') lines.append(' }') lines.extend([ ' subgraph "cluster_legend" {', ' label="Legend"; style="rounded"; color="#d1d5db";', ' "leg_host" [label="Host cluster", shape=box3d, fillcolor="#eff6ff"];', ' "leg_cat" [label="Service category", shape=folder, fillcolor="#f8fbff"];', ' "leg_svc" [label="Container/service", shape=box, fillcolor="#dcfce7"];', ' "leg_host" -> "leg_cat" [style=dashed];', ' "leg_cat" -> "leg_svc" [style=dashed];', ' }', "}", ]) write_dot(out_dot, lines) render_svg(out_dot, out_svg) def generate_docker_traefik_dynu(compose: dict, out_dot: Path, out_svg: Path) -> None: services = compose.get("services") or {} networks = compose.get("networks") or {} known_domains: dict[str, str] = {} lines = [ "digraph DockerTraefikDynu {", " graph [rankdir=LR, compound=true, splines=ortho, nodesep=0.7, ranksep=1.2, fontname=\"Helvetica\", concentrate=true];", " node [fontname=\"Helvetica\", fontsize=10, style=\"rounded,filled\"];", " edge [fontname=\"Helvetica\", fontsize=9, color=\"#334155\"];", ' "svc:traefik" [label="Traefik\n(entrypoint)", shape=box, fillcolor="#bfdbfe"];', ] routes: dict[str, dict] = {} dns_nodes: set[str] = set() for svc_name, svc in sorted(services.items()): labels = parse_labels(svc) router_prefix = "traefik.http.routers." service_prefix = "traefik.http.services." lb_ports = {} for k, v in labels.items(): if k.startswith(service_prefix) and k.endswith(".loadbalancer.server.port"): lb_ports[k[len(service_prefix):].split(".", 1)[0]] = v routers = sorted({k[len(router_prefix):].split(".", 1)[0] for k in labels if k.startswith(router_prefix)}) for router in routers: rule = labels.get(f"{router_prefix}{router}.rule", "") target = labels.get(f"{router_prefix}{router}.service", svc_name) middlewares = labels.get(f"{router_prefix}{router}.middlewares", "") tls = labels.get(f"{router_prefix}{router}.tls", "false") port = lb_ports.get(target, "") badges = [] if str(tls).lower() in ("true", "1"): badges.append("TLS") mw_low = middlewares.lower() if "authelia" in mw_low: badges.append("authelia") if "mtls" in mw_low: badges.append("mTLS") hosts = [sanitize_domain(h, known_domains) for h in extract_hosts(rule)] if not hosts: continue info = routes.setdefault(svc_name, {"hosts": set(), "port": port, "badges": set()}) info["hosts"].update(hosts) if port: info["port"] = port info["badges"].update(badges) for svc_name, info in sorted(routes.items()): label = svc_name if info.get("port"): label += f"\n:{info['port']}" if info.get("badges"): label += "\n[" + ", ".join(sorted(info["badges"])) + "]" lines.append(f' "svc:{svc_name}" [label="{label}", shape=box, fillcolor="#dcfce7"];') lines.append(f' "svc:traefik" -> "svc:{svc_name}" [penwidth=1.4];') for host in sorted(info["hosts"]): dns_nodes.add(host) lines.append(f' "dns:{host}" [label="{host}", shape=note, fillcolor="#fef3c7"];') lines.append(f' "dns:{host}" -> "svc:traefik";') lines.append(' { rank=same; ' + '; '.join([f'"dns:{d}"' for d in sorted(dns_nodes)]) + '; }' if dns_nodes else '') lines.append(' subgraph "cluster_networks" {') lines.append(' label="Docker backend networks"; style="rounded,dashed"; color="#d1d5db";') for net in sorted(networks.keys()): lines.append(f' "net:{net}" [label="{net}", shape=ellipse, fillcolor="#f8fafc"];') lines.append(' }') for svc_name in sorted(routes.keys()): svc = services.get(svc_name, {}) svc_nets = svc.get("networks") or [] if isinstance(svc_nets, dict): svc_nets = svc_nets.keys() for net in svc_nets: lines.append(f' "svc:{svc_name}" -> "net:{net}" [style=dashed, color="#94a3b8", arrowsize=0.7];') lines.append("}") write_dot(out_dot, lines) render_svg(out_dot, out_svg) def generate_compose_topology(compose: dict, out_dot: Path, out_svg: Path) -> None: services = compose.get("services") or {} networks = compose.get("networks") or {} lines = [ "digraph Compose {", " rankdir=LR;", ' node [fontname="Helvetica"];', ] for service in sorted(services): lines.append(f' "svc:{service}" [label="{service}", shape=box, style=filled, fillcolor="#dfefff"];') for net in sorted(networks): lines.append(f' "net:{net}" [label="{net}", shape=ellipse, style=filled, fillcolor="#f4f4f4"];') for service, svc in sorted(services.items()): svc_nets = svc.get("networks") or [] if isinstance(svc_nets, dict): svc_nets = svc_nets.keys() for net in svc_nets: lines.append(f' "svc:{service}" -> "net:{net}";') lines.append("}") write_dot(out_dot, lines) render_svg(out_dot, out_svg) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("legacy", nargs="*") parser.add_argument("--compose") parser.add_argument("--out-dir") args = parser.parse_args() require_dot() if args.compose and args.out_dir: compose_path = Path(args.compose) out_dir = Path(args.out_dir) elif len(args.legacy) == 3: compose_path = Path(args.legacy[0]) out_dir = Path(args.legacy[1]).parent else: raise SystemExit("Usage: generate-diagrams.py --compose --out-dir ") out_dir.mkdir(parents=True, exist_ok=True) compose = load_compose(compose_path) generate_docker_traefik_dynu(compose, out_dir / "docker-traefik-dynu.dot", out_dir / "docker-traefik-dynu.svg") generate_physical_topology(compose, out_dir / "physical-topology.dot", out_dir / "physical-topology.svg") generate_compose_topology(compose, out_dir / "docker-compose.dot", out_dir / "docker-compose.svg") if __name__ == "__main__": main()