beatz174-bit
207 lines
7.9 KiB
Python
207 lines
7.9 KiB
Python
#!/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}.<internal-domain>"
|
|
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 generate_physical_topology(compose: dict, out_dot: Path, out_svg: Path) -> None:
|
|
services = compose.get("services") or {}
|
|
hosts: dict[str, list[str]] = {}
|
|
for name, svc in services.items():
|
|
hosts.setdefault(infer_host(name, svc), []).append(name)
|
|
|
|
lines = [
|
|
"digraph PhysicalTopology {",
|
|
" graph [rankdir=LR, compound=true, splines=ortho, nodesep=0.45, ranksep=0.75, fontname=\"Helvetica\"];",
|
|
" node [fontname=\"Helvetica\", fontsize=10, style=\"rounded,filled\", fillcolor=\"#ffffff\"];",
|
|
" edge [fontname=\"Helvetica\", fontsize=9];",
|
|
]
|
|
|
|
for host, host_services in sorted(hosts.items()):
|
|
lines.extend(
|
|
[
|
|
f' subgraph "cluster_{host}" {{',
|
|
f' label="{host}";',
|
|
' style="rounded,filled";',
|
|
' color="#c7d6f5";',
|
|
' fillcolor="#eef3ff";',
|
|
f' "host:{host}" [label="{host}", shape=box3d, fillcolor="#d4e3ff"];',
|
|
]
|
|
)
|
|
for service in sorted(host_services):
|
|
lines.append(
|
|
f' "svc:{service}" [label="{service}", shape=box, fillcolor="#dff2e1"];'
|
|
)
|
|
lines.append(f' "host:{host}" -> "svc:{service}" [style=dashed, color="#6b7280"];')
|
|
lines.append(" }")
|
|
|
|
lines.append("}")
|
|
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=true, nodesep=0.5, ranksep=1.0, fontname=\"Helvetica\"];",
|
|
" node [fontname=\"Helvetica\", fontsize=10, style=\"rounded,filled\"];",
|
|
" edge [fontname=\"Helvetica\", fontsize=9];",
|
|
' "ext:dynu" [label="Dynu / Public DNS", shape=oval, fillcolor="#fde68a"];',
|
|
' "svc:traefik" [label="traefik", shape=box, fillcolor="#bfdbfe"];',
|
|
]
|
|
|
|
for net in sorted(networks.keys()):
|
|
lines.append(f' "net:{net}" [label="{net}", shape=ellipse, fillcolor="#f3f4f6"];')
|
|
|
|
for svc_name, svc in sorted(services.items()):
|
|
lines.append(f' "svc:{svc_name}" [label="{svc_name}", shape=box, fillcolor="#dcfce7"];')
|
|
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}" [color="#6b7280"];')
|
|
|
|
labels = parse_labels(svc)
|
|
router_prefix = "traefik.http.routers."
|
|
service_prefix = "traefik.http.services."
|
|
|
|
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", "")
|
|
router_service = labels.get(f"{router_prefix}{router}.service", svc_name)
|
|
middlewares = labels.get(f"{router_prefix}{router}.middlewares", "")
|
|
entrypoints = labels.get(f"{router_prefix}{router}.entrypoints", "")
|
|
tls = labels.get(f"{router_prefix}{router}.tls", "false")
|
|
lines.append(f' "router:{router}" [label="router:{router}\\nentry:{entrypoints} tls:{tls}", shape=diamond, fillcolor="#fbcfe8"];')
|
|
lines.append(f' "svc:traefik" -> "router:{router}";')
|
|
target_node = f"traefik-service:{router_service}"
|
|
lines.append(f' "{target_node}" [label="service:{router_service}", shape=component, fillcolor="#fecaca"];')
|
|
lines.append(f' "router:{router}" -> "{target_node}";')
|
|
|
|
for host in extract_hosts(rule):
|
|
clean = sanitize_domain(host, known_domains)
|
|
lines.append(f' "dns:{clean}" [label="{clean}", shape=note, fillcolor="#fde68a"];')
|
|
lines.append(f' "ext:dynu" -> "dns:{clean}";')
|
|
lines.append(f' "dns:{clean}" -> "router:{router}";')
|
|
|
|
for middleware in [m.strip() for m in middlewares.split(",") if m.strip()]:
|
|
lines.append(f' "mw:{middleware}" [label="{middleware}", shape=hexagon, fillcolor="#ddd6fe"];')
|
|
lines.append(f' "router:{router}" -> "mw:{middleware}" [style=dashed];')
|
|
|
|
lb_services = sorted({k[len(service_prefix):].split(".", 1)[0] for k in labels if k.startswith(service_prefix)})
|
|
for lb in lb_services:
|
|
port = labels.get(f"{service_prefix}{lb}.loadbalancer.server.port", "")
|
|
lines.append(f' "traefik-service:{lb}" [label="service:{lb}\\nport:{port}", shape=component, fillcolor="#fecaca"];')
|
|
lines.append(f' "traefik-service:{lb}" -> "svc:{svc_name}";')
|
|
|
|
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 <compose.yml> --out-dir <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_docker_traefik_dynu(compose, out_dir / "docker-compose.dot", out_dir / "docker-compose.svg")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|