#!/usr/bin/env python3 """service-map.py — the only reader of the service inventory. Four lists used to agree by discipline (the planner's path rules, the deployer's allow-list, the changed-test mapper, and the scrape/probe/DNS facts written by hand in three config files). They now all read `shared/services.yml` through this tool: hosts the deploy targets services [--host H] deployables a target accepts, canonical order allows --host H --service S exit 0 when H accepts S (silent) plan --host H PATH... services selected by changed paths, order-safe seams [--host H] PATH... the Bats seams a changed path must re-run owner --host H PATH the service that owns a file (empty: none) units --host H --service S " " lines the handler installs stack --host H --service S the deployable + members one handler deploys get --service S [--host H] F one field as JSON render pihole-hosts the Pi-hole addn-hosts file check [NAME...] integrity + drift gates (exit 1 on drift) Two laws the engine enforces structurally: * A path selects only services that RUN on the planned target. The old planner matched paths host-blind and filtered at the end, so a frankbook-pro file could plan a dankserver service (the stale *arr entries made that reachable). * The netns family is DERIVED, never declared: a unit file that says `Network=gluetun.container` rides that tunnel, so a gluetun change selects every sharer. The unit files are the source of that coupling. """ from __future__ import annotations import argparse import json import subprocess import sys from functools import lru_cache from pathlib import Path, PurePosixPath from typing import Any import yaml REPO_ROOT = Path(__file__).resolve().parent.parent DEFAULT_INVENTORY = REPO_ROOT / "shared" / "services.yml" REQUIRED_KEYS = ("name", "hosts", "kind", "tests") KINDS = {"quadlet", "root-unit", "vm-quadlet", "host"} UNIT_SUFFIXES = (".container", ".network") # Repo material that is never a deployed runtime's state. NON_RUNTIME_ROOTS = {"docs", "report", "old", ".githooks", ".obsidian", ".github", "tests", "serverAppConfigs-docs"} NON_RUNTIME_NAMES = {".env.example"} NON_RUNTIME_PATHS = {"README.md", "AGENTS.md", ".gitignore"} def is_trash(path: str) -> bool: return ".AGENTTRASH" in PurePosixPath(path).parts # -------------------------------------------------------------------------- # inventory # -------------------------------------------------------------------------- class Inventory: def __init__(self, doc: dict[str, Any], origin: str) -> None: self.origin = origin self.hosts: dict[str, dict[str, Any]] = doc["hosts"] self.services: list[dict[str, Any]] = doc["services"] self.unowned: dict[str, str] = doc.get("unowned", {}) for index, svc in enumerate(self.services): svc["_index"] = index @classmethod def load(cls, path: str | Path = DEFAULT_INVENTORY) -> "Inventory": doc = yaml.safe_load(Path(path).read_text()) if not isinstance(doc, dict) or "services" not in doc: raise SystemExit(f"{path}: not a service inventory document") return cls(doc, str(path)) def on(self, host: str) -> list[dict[str, Any]]: return [s for s in self.services if host in s["hosts"]] def named(self, host: str, name: str) -> list[dict[str, Any]]: return [s for s in self.services if s["name"] == name and host in s["hosts"]] def any_named(self, name: str) -> list[dict[str, Any]]: return [s for s in self.services if s["name"] == name] @property def order(self) -> list[str]: seen: list[str] = [] for svc in self.services: if svc["name"] not in seen: seen.append(svc["name"]) return seen def edge_ip(self) -> str: """The single address every DNS name resolves at (Caddy's edge).""" for host in self.hosts.values(): if "dns_edge" in host: return str(host["dns_edge"]) raise SystemExit(f"{self.origin}: no host declares dns_edge") def unit_dirs(self, svc: dict[str, Any], host: str | None = None) -> list[str]: """The repo dirs holding this service's unit files, for one host.""" dirs = svc.get("unit_dir") if not dirs: return [] if isinstance(dirs, dict): if host is not None: return [dirs[host]] if host in dirs else [] return [dirs[h] for h in svc["hosts"] if h in dirs] return [dirs] def owns_unit(self, svc: dict[str, Any], host: str, unit: str, directory: str) -> bool: return unit in svc.get("units", []) and directory in self.unit_dirs(svc, host) # -------------------------------------------------------------------------- # repo facts # -------------------------------------------------------------------------- @lru_cache(maxsize=1) def tracked_files() -> tuple[str, ...]: # Tracked + untracked (a check that only saw HEAD would ignore the very # files a work-in-progress commit is adding). out = subprocess.run( ["git", "-C", str(REPO_ROOT), "ls-files", "--cached", "--others", "--exclude-standard"], text=True, stdout=subprocess.PIPE, check=True) return tuple(line for line in out.stdout.splitlines() if line) def _network_values(container: Path) -> list[str]: """Every `Network=` value a quadlet unit declares (comma lists included).""" values: list[str] = [] try: text = container.read_text() except OSError: return values for line in text.splitlines(): if line.startswith("Network="): values.extend(v.strip() for v in line[len("Network="):].split(",") if v.strip()) return values class Graph: """The unit-file coupling, read out of the tree: netns + netns-net users. Quadlet installs every user unit flat, so `Network=gluetun.container` names a unit by name, not by path; the same host is the scope, which is also the deployment scope. """ def __init__(self, hosts: list[str]) -> None: self.units: dict[tuple[str, str], str] = {} # (host, unit) -> path self.networks: dict[tuple[str, str], str] = {} # (host, net) -> path self.netns_children: dict[tuple[str, str], set[tuple[str, str]]] = {} self.network_users: dict[tuple[str, str], set[tuple[str, str]]] = {} for path in tracked_files(): parts = PurePosixPath(path).parts if not parts or parts[0] not in hosts or is_trash(path): continue host, directory, name = parts[0], str(PurePosixPath(*parts[1:-1])), parts[-1] if name.endswith(".container"): self.units[(host, name[: -len(".container")])] = path elif name.endswith(".network"): self.networks[(host, name[: -len(".network")])] = path for (host, unit), path in self.units.items(): for value in _network_values(REPO_ROOT / path): if value.endswith(".container"): parent = (host, value[: -len(".container")]) self.netns_children.setdefault(parent, set()).add((host, unit)) elif value.endswith(".network"): net = (host, value[: -len(".network")]) self.network_users.setdefault(net, set()).add((host, unit)) def rides(self, start: list[tuple[str, str]]) -> set[tuple[str, str]]: """The transitive netns closure: a unit, everything riding it, and everything riding those (a netns inside a netns).""" found: set[tuple[str, str]] = set() frontier = list(start) while frontier: node = frontier.pop() if node in found: continue found.add(node) frontier.extend(self.netns_children.get(node, ())) return found @lru_cache(maxsize=8) def graph_for(hosts: tuple[str, ...]) -> Graph: return Graph(list(hosts)) # -------------------------------------------------------------------------- # selection # -------------------------------------------------------------------------- def _owner_index(inv: Inventory) -> dict[tuple[str, str, int]]: """(host, unit) -> service index, for unit-file ownership. A quadlet unit name is unique per host (``check units`` enforces one claim), so the coupling closure can resolve a riding unit that lives in a DIFFERENT directory than the changed file — gluetun's netns children are the point. """ index: dict[tuple[str, str], int] = {} for svc in inv.services: for host in svc["hosts"]: for unit in svc.get("units", []): key = (host, unit) if key in index and index[key] != svc["_index"]: raise SystemExit( f"{inv.origin}: unit {unit!r} on {host} claimed by " f"{inv.services[index[key]]['name']} and {svc['name']}" ) index[key] = svc["_index"] return index def select(inv: Inventory, host: str, paths: list[str]) -> list[str]: """Canonical-order service names the changed paths select for one host.""" if host not in inv.hosts: raise SystemExit(f"{inv.origin}: no host {host!r}") graph = graph_for(tuple(inv.hosts)) owners = _owner_index(inv) hits: set[int] = set() def add(index: int) -> None: hits.add(index) for raw in paths: path = raw.replace("\\", "/") parts = PurePosixPath(path).parts if not parts or is_trash(path) or path in NON_RUNTIME_PATHS: continue if PurePosixPath(path).name in NON_RUNTIME_NAMES: continue head = parts[0] if head in NON_RUNTIME_ROOTS: continue # A host's directory is the scope: a frankbook-pro file can never # select a dankserver service. scoped_host = head if head in inv.hosts else None if scoped_host is not None and scoped_host != host: continue name = parts[-1] directory = str(PurePosixPath(*parts[1:-1])) if len(parts) > 1 else "" if scoped_host and name.endswith(UNIT_SUFFIXES): stem = name.rsplit(".", 1)[0] key = (scoped_host, stem) if name.endswith(".network") and key in graph.networks: users = graph.network_users.get(key, set()) for unit in graph.rides(sorted(users)): index = owners.get(unit) if index is not None: add(index) elif key in graph.units: for unit in graph.rides([key]): index = owners.get(unit) if index is not None: add(index) for index in list(hits): for dependent in inv.services[index].get("dependents", []): for svc in inv.named(host, dependent): add(svc["_index"]) for svc in inv.on(host): if path in svc.get("files", []): add(svc["_index"]) if path in svc.get("retires", []): add(svc["_index"]) for directory_rule in svc.get("dirs", []): if path == directory_rule or path.startswith(directory_rule + "/"): add(svc["_index"]) return [svc["name"] for svc in sorted((inv.services[i] for i in hits), key=lambda s: s["_index"])] def seams_for(inv: Inventory, host: str, paths: list[str]) -> list[str]: """The Bats seams the selected services require, deduplicated, in order.""" seams: list[str] = [] for name in select(inv, host, paths): for svc in inv.named(host, name): for seam in svc.get("tests", []): if seam not in seams: seams.append(seam) if seams and "tests/deployment_scoping.bats" not in seams: seams.append("tests/deployment_scoping.bats") return seams # -------------------------------------------------------------------------- # renderers + facts # -------------------------------------------------------------------------- def render_pihole_hosts(inv: Inventory) -> str: edge = inv.edge_ip() lines = [ "# A records for the flat franksdank.cloud scheme (dnsmasq addn-hosts", "# format: ). RENDERED — do not edit by hand:", "#", "# scripts/service-map.py render pihole-hosts > dankclaw/pihole/addn-hosts", "#", "# The source of truth is the `dns:` field of shared/services.yml; the", "# drift gate in tests/service_inventory.bats fails when this file and", "# the inventory disagree. Caddy is consolidated on Dankclaw, so every", "# name — including the services running on dankserver and", "# frankbook-pro — resolves at the Dankclaw edge.", "#", "# dnsmasq strips ONLY whole-line comments in addn-hosts: a trailing", "# comment on a data line becomes a HOSTNAME token. No inline notes", "# here; the service attribution lives in the inventory.", "", ] for svc in inv.services: for name in svc.get("dns", []): lines.append(f"{edge} {name}") return "\n".join(lines) + "\n" def scrape_facts(inv: Inventory) -> list[dict[str, str]]: facts = [] for svc in inv.services: for job in svc.get("scrape", []): facts.append({"job": job["job"], "target": job["target"], "metrics_path": job.get("metrics_path", "/")}) return facts def prometheus_facts(path: Path) -> list[dict[str, str]]: doc = yaml.safe_load(path.read_text()) facts = [] for job in doc.get("scrape_configs", []): for sc in job.get("static_configs", []): for target in sc.get("targets", []): facts.append({"job": job["job_name"], "target": target, "metrics_path": job.get("metrics_path", "/")}) return facts def probe_facts(inv: Inventory) -> list[str]: facts: list[str] = [] for svc in inv.services: probe = svc.get("probe") if isinstance(probe, dict): facts.extend(str(v) for v in probe.values()) elif probe: facts.append(str(probe)) return facts def homepage_probes(path: Path) -> list[str]: return sorted({s["probe"] for s in json.loads(path.read_text()) if s.get("probe")}) # -------------------------------------------------------------------------- # checks # -------------------------------------------------------------------------- def check_ownership(inv: Inventory, graph: Graph) -> list[str]: """Every repo file belongs to a service, a rule, or the unowned list.""" problems: list[str] = [] exact: set[str] = set() directories: set[str] = set() for svc in inv.services: for host in svc["hosts"]: for directory in inv.unit_dirs(svc, host): for unit in svc.get("units", []): exact.add(f"{directory}/{unit}.container") for field in ("files", "retires"): exact.update(svc.get(field, [])) directories.update(svc.get("dirs", [])) # A shared network file is claimed by the units that ride it; the unit # graph is the referrer, and check_units proves every reference resolves. exact.update(graph.networks.values()) def is_claimed(path: str) -> bool: if path in exact: return True first = PurePosixPath(path).parts[0] if first in NON_RUNTIME_ROOTS or path in NON_RUNTIME_PATHS or is_trash(path): return True if PurePosixPath(path).name in NON_RUNTIME_NAMES: return True for owned in directories | set(inv.unowned): if path == owned or path.startswith(owned + "/"): return True return False for path in tracked_files(): if not is_claimed(path): problems.append(f"file belongs to no service and is not in unowned: {path}") # The reverse direction: an unowned entry that names nothing is stale. for directory in inv.unowned: if not any(p == directory or p.startswith(directory + "/") for p in tracked_files()): problems.append(f"unowned entry names no tracked file: {directory}") # Every inventory file/dir claim names a tracked file. for svc in inv.services: # `retires` names a file whose DELETION selects the service, so by # definition it is not in the tree. for field in ("files", "dirs"): for entry in svc.get(field, []): if entry not in tracked_files() and not any( p.startswith(entry + "/") for p in tracked_files()): problems.append(f"{svc['name']}: {field} entry names no tracked file: {entry}") del graph return problems def check_units(inv: Inventory, graph: Graph) -> list[str]: """Every unit file has exactly one owner; every unit reference resolves.""" problems: list[str] = [] owners: dict[str, list[str]] = {} for svc in inv.services: for host in svc["hosts"]: for directory in inv.unit_dirs(svc, host): for unit in svc.get("units", []): path = f"{directory}/{unit}.container" owners.setdefault(path, []).append(f"{svc['name']}@{host}") if path not in tracked_files(): problems.append(f"{svc['name']}: unit file is not tracked: {path}") for path in tracked_files(): if not path.endswith(".container") or is_trash(path): continue found = owners.get(path, []) if not found: problems.append(f"unit file claimed by no service: {path}") elif len(found) > 1: problems.append(f"unit file claimed by {len(found)} services: {path}: {found}") # Root units (scripts/*.service) are claimed by exactly one `files:` list. for path in tracked_files(): if not (path.startswith("scripts/") and path.endswith(".service")) or is_trash(path): continue found = [s["name"] for s in inv.services if path in s.get("files", [])] if not found: problems.append(f"root unit claimed by no service: {path}") elif len(found) > 1: problems.append(f"root unit claimed by {found}: {path}") # Every Network= reference in a unit file names a file on the same host. for (host, unit), path in graph.units.items(): for value in _network_values(REPO_ROOT / path): if value.endswith(".container"): target = (host, value[: -len(".container")]) if target not in graph.units: problems.append(f"{path}: Network={value} names no unit on {host}") elif value.endswith(".network"): target = (host, value[: -len(".network")]) if target not in graph.networks: problems.append(f"{path}: Network={value} names no network on {host}") return problems def check_config(inv: Inventory) -> list[str]: problems: list[str] = [] for svc in inv.services: label = f"{'+'.join(svc['hosts'])}/{svc['name']}" for key in REQUIRED_KEYS: if key not in svc: problems.append(f"{label}: missing required key {key}") if svc.get("kind") not in KINDS: problems.append(f"{label}: kind {svc.get('kind')!r} not one of {sorted(KINDS)}") unknown = set(svc.get("hosts", [])) - set(inv.hosts) if unknown: problems.append(f"{label}: undeclared hosts: {sorted(unknown)}") if svc.get("units") and not svc.get("unit_dir"): problems.append(f"{label}: units declared without unit_dir") if svc.get("deployed_by") and not inv.any_named(svc["deployed_by"]): problems.append(f"{label}: deployed_by names no service: {svc['deployed_by']}") if (svc.get("health") and svc.get("wait") is None) or \ (svc.get("wait") is not None and not svc.get("health")): problems.append(f"{label}: health and wait are a pair") for seam in svc.get("tests", []): if seam not in tracked_files(): problems.append(f"{label}: seam does not exist: {seam}") for name in svc.get("dns", []): if not str(name).replace(".", "").replace("-", "").isalnum(): problems.append(f"{label}: dns name is not a plain hostname: {name}") if svc.get("kind") == "root-unit" and svc.get("unit_dir"): problems.append(f"{label}: a root unit lives in scripts/, not unit_dir") seen: set[str] = set() for svc in inv.services: key = (svc["name"], tuple(svc["hosts"])) if key in seen: problems.append(f"duplicate inventory entry: {svc['name']} on {svc['hosts']}") seen.add(key) return problems def check_drift(inv: Inventory) -> list[str]: problems: list[str] = [] rendered = render_pihole_hosts(inv) hosts_file = REPO_ROOT / "dankclaw" / "pihole" / "addn-hosts" if hosts_file.read_text() != rendered: problems.append("dankclaw/pihole/addn-hosts is not the render of the inventory " "(scripts/service-map.py render pihole-hosts > dankclaw/pihole/addn-hosts)") want = sorted(json.dumps(f, sort_keys=True) for f in scrape_facts(inv)) have = sorted(json.dumps(f, sort_keys=True) for f in prometheus_facts(REPO_ROOT / "shared" / "prometheus" / "prometheus.yml")) for extra in [f for f in have if f not in want]: problems.append(f"prometheus scrapes a target the inventory does not declare: {extra}") for missing in [f for f in want if f not in have]: problems.append(f"inventory declares a scrape prometheus.yml lacks: {missing}") probes_want = sorted(set(probe_facts(inv))) probes_have = homepage_probes(REPO_ROOT / "shared" / "homepage" / "services.json") for extra in [p for p in probes_have if p not in probes_want]: problems.append(f"homepage probes a target the inventory does not declare: {extra}") for missing in [p for p in probes_want if p not in probes_have]: problems.append(f"inventory declares a probe homepage does not list: {missing}") return problems CHECKS = { "config": lambda inv, graph: check_config(inv), "units": check_units, "ownership": check_ownership, "drift": lambda inv, graph: check_drift(inv), } def check(inv: Inventory, names: list[str]) -> list[str]: graph = graph_for(tuple(inv.hosts)) problems: list[str] = [] for name in names or list(CHECKS): problems.extend(CHECKS[name](inv, graph)) return problems # -------------------------------------------------------------------------- # CLI # -------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--inventory", type=Path, default=DEFAULT_INVENTORY) sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("hosts") p = sub.add_parser("services") p.add_argument("--host") p = sub.add_parser("allows") p.add_argument("--host", required=True) p.add_argument("--service", required=True) p = sub.add_parser("plan") p.add_argument("--host", required=True) p.add_argument("paths", nargs="+") p = sub.add_parser("seams") p.add_argument("--host") p.add_argument("paths", nargs="+") p = sub.add_parser("owner") p.add_argument("--host", required=True) p.add_argument("path") p = sub.add_parser("units") p.add_argument("--host", required=True) p.add_argument("--service", required=True) p = sub.add_parser("stack") p.add_argument("--host", required=True) p.add_argument("--service", required=True) p = sub.add_parser("get") p.add_argument("--host") p.add_argument("--service", required=True) p.add_argument("field") p = sub.add_parser("render") p.add_argument("artifact", choices=("pihole-hosts",)) p = sub.add_parser("check") p.add_argument("names", nargs="*") args = parser.parse_args(argv) inv = Inventory.load(args.inventory) if args.command == "hosts": print("\n".join(inv.hosts)) elif args.command == "services": services = inv.on(args.host) if args.host else inv.services seen: list[str] = [] for svc in services: if svc["name"] not in seen: seen.append(svc["name"]) print("\n".join(seen)) elif args.command == "allows": return 0 if inv.named(args.host, args.service) else 1 elif args.command == "plan": print("\n".join(select(inv, args.host, args.paths))) elif args.command == "seams": if args.host is not None: seams = seams_for(inv, args.host, args.paths) else: # No host: the union across every host (the dev machine's # pre-commit mapper has no deployment target). seams = [] for host in inv.hosts: for seam in seams_for(inv, host, args.paths): if seam not in seams: seams.append(seam) print("\n".join(seams)) elif args.command == "owner": found = select(inv, args.host, [args.path]) print(found[-1] if found else "") elif args.command == "units": matches = inv.named(args.host, args.service) if not matches: print(f"no service {args.service} on {args.host}", file=sys.stderr) return 2 for svc in matches: for directory in inv.unit_dirs(svc, args.host): for unit in svc.get("units", []): print(f"{unit} {directory}") elif args.command == "stack": matches = inv.named(args.host, args.service) if not matches: print(f"no service {args.service} on {args.host}", file=sys.stderr) return 2 owner = matches[0].get("deployed_by") or args.service print(owner) for svc in inv.on(args.host): if svc.get("deployed_by") == owner and svc["name"] != owner: print(svc["name"]) elif args.command == "get": matches = inv.on(args.host) if args.host else inv.services matches = [s for s in matches if s["name"] == args.service] if not matches: print(f"no service {args.service}" + (f" on {args.host}" if args.host else ""), file=sys.stderr) return 2 values = [s.get(args.field) for s in matches if args.field in s] print(json.dumps(values[0] if len(values) == 1 else values)) elif args.command == "render": sys.stdout.write(render_pihole_hosts(inv)) elif args.command == "check": problems = check(inv, args.names) if problems: for problem in problems: print(f"FAIL: {problem}", file=sys.stderr) return 1 print("OK: the service inventory is sound (schema, units, ownership, drift)") return 0 if __name__ == "__main__": sys.exit(main())