Merge pull request 'feat: the configuration that says which services run together (4.17.0)' (#47) from feat/service-clusters into main
CI / test (3.10) (push) Successful in 27s
CI / test (3.11) (push) Successful in 24s
CI / test (3.12) (push) Successful in 27s
CI / test (3.13) (push) Successful in 26s

This commit was merged in pull request #47.
This commit is contained in:
christianmanivong committed 2026-10-18 13:32:33 +00:00
commit a0e70d1686
5 files changed
+1108

No files matched your search

+11
View File
@@ -51,6 +51,7 @@ instead of being restated on every role that happens to need it:
* :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
* :class:`~napalm_device_types.packages.PackageManagementMixin`
* :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
* :class:`~napalm_device_types.service_clusters.ServiceClustersMixin`
* :class:`~napalm_device_types.services.ServiceControlMixin`
* :class:`~napalm_device_types.systemd.SystemdServicesMixin`
* :class:`~napalm_device_types.updates.UpdateMixin`
@@ -161,6 +162,12 @@ from napalm_device_types.system_identity import (
is_placeholder_identity,
parse_system_identity,
)
from napalm_device_types.service_clusters import (
SERVICE_CLUSTERS_COMMAND,
ServiceClustersMixin,
parse_service_clusters,
service_clusters_command,
)
from napalm_device_types.virtual_ips import (
VIRTUAL_IPS_COMMAND,
VirtualIpsMixin,
@@ -268,6 +275,10 @@ __all__ = [
"UpdateMixin",
"VIRTUAL_IPS_COMMAND",
"VirtualIpsMixin",
"SERVICE_CLUSTERS_COMMAND",
"ServiceClustersMixin",
"parse_service_clusters",
"service_clusters_command",
"SystemIdentityMixin",
"SYSTEM_IDENTITY_COMMAND",
"is_placeholder_identity",
+62
View File
@@ -425,6 +425,68 @@ class VirtualIpDict(TypedDict):
holds: NotRequired[Optional[bool]]
class ServiceClusterMemberDict(TypedDict):
"""One member of a service cluster, as the configuration names it.
A member is named by what the configuration has: an address (a URL's host, a
whitelisted address -- possibly a host name), a port where it gives one, a
name where the mechanism has its own (Patroni's). ``this_host`` marks the
member the reading host is when the configuration does not name it by
address. Only these fields leave the host -- never a password or a DCS
endpoint.
"""
name: Optional[str]
address: Optional[str]
port: Optional[int]
#: as the mechanism words it: Patroni's "leader"/"sync_standby"/"replica",
#: adguardhome-sync's "origin"/"replica", BunkerWeb's "manager"/"worker"
role: Optional[str]
#: the member's state where the mechanism reports one (Patroni's "running",
#: "streaming"); configuration alone has none
state: Optional[str]
this_host: bool
#: the index in ``members`` of the member whose configuration this one takes:
#: a replica's origin, a worker's manager. None where it takes none.
config_from: Optional[int]
class ServiceClusterDict(TypedDict):
"""A cluster the host's configuration says it belongs to (netOrk #1125).
The evidence that two hosts run one service together, not that they merely
answer at one address: Patroni's scope and the members its DCS lists, the
origin and replicas adguardhome-sync keeps in step, a BunkerWeb worker's
manager.
"""
mechanism: str # "patroni" | "adguardhome-sync" | "bunkerweb"
#: what the members run, as a host reports it: "patroni", "adguardhome", "bunkerweb"
service: str
#: what the evidence was read from on this host: the unit without ".service"
#: ("patroni", "patroni@16-main") or the container's name ("adguardhome-sync")
source: str
#: what the members share where the reading host can name it: Patroni's scope,
#: the sync's origin ("10.9.100.5:3000"), a worker's manager address. None for a
#: manager, which cannot name its own address.
cluster: Optional[str]
running: Optional[bool] # the unit is active; a listed container runs
members: List[ServiceClusterMemberDict]
class ServiceClustersDict(TypedDict):
"""What a driver's ``get_service_clusters`` read, per mechanism.
A list is what was looked at -- ``[]`` a host without that mechanism. None is
"did not look": a configuration it could not read, a host without systemd, a
container engine it may not ask, a sync configured by file.
"""
patroni: Optional[List[ServiceClusterDict]]
adguardhome_sync: Optional[List[ServiceClusterDict]]
bunkerweb: Optional[List[ServiceClusterDict]]
class VirtualIpsDict(TypedDict):
"""What a driver's ``get_virtual_ips`` read, per mechanism.
+528
View File
@@ -0,0 +1,528 @@
# -*- coding: utf-8 -*-
"""Service clusters: the configuration that says which services run together.
Two hosts that answer at one virtual IP may run the same service by accident.
What proves a cluster is the configuration its members share. That is read the
same way on every Linux host, so the command and its parse live here once, and
a driver only carries the command across (netOrk #1125).
**What is read.**
- *Patroni*: every ``patroni.service`` and ``patroni@*.service`` unit systemd
knows. Its configuration file -- the unit's first non-flag argument, else
``/etc/patroni/config.yml`` -- gives ``scope``, ``name`` and
``restapi.connect_address``. The REST API there (``/cluster``, which Patroni
answers without a login) lists the members its DCS knows, each with its role:
``leader``, ``sync_standby``, ``replica``. When it does not answer, the host
is a member of its scope on its own.
- *adguardhome-sync*: a running container whose image is adguardhome-sync.
``ORIGIN_URL`` and ``REPLICA*_URL`` name the AdGuard Home instances it keeps in
step: the origin pushes to every replica. A sync configured by a YAML file
instead of its environment is not read.
- *BunkerWeb*: a running instance container (``bunkerity/bunkerweb``) admits a
manager's configuration pushes from the addresses in ``API_WHITELIST_IP``; a
manager (``bunkerweb-all-in-one``, ``bunkerweb-scheduler``) names its
instances in ``BUNKERWEB_INSTANCES``. BunkerWeb's own default whitelist,
loopback and the docker network, names no one. A manager that keeps its
instances in its database names none here: its workers name it.
**Filtered on the host.** The same configuration holds secrets -- Patroni's
superuser and REST passwords, the sync's AdGuard logins, BunkerWeb's database
URI. The command never prints a configuration: awk programs print allowlisted
fields only, and a URL loses its ``user:password@``. The DCS endpoints stay on
the host too; the members come from ``/cluster``.
**None is "did not look".** A Patroni configuration that exists but cannot be
read, a host without systemd, a docker the login may not ask, a sync configured
by file: that mechanism is None, and the caller keeps what it knew. ``[]`` is a
host that has none.
**Root, and without it.** Patroni's configuration is the postgres user's alone.
The whole script goes to the host as one ``sh -c`` argument; when the privileged
call brings no report back, it runs again without privilege.
"""
from __future__ import annotations
import ipaddress
import json
import re
from shlex import quote
from typing import Callable, Dict, List, Optional, Tuple, TYPE_CHECKING
from urllib.parse import urlsplit
from napalm_device_types.models import (
ServiceClusterDict,
ServiceClusterMemberDict,
ServiceClustersDict,
)
from napalm_device_types.terminal import strip_terminal_codes
_BEGIN = "SVCC_BEGIN"
_END = "SVCC_END"
#: Exists only while systemd runs as PID 1; a container may carry systemctl without.
_SYSTEMD_RUN = "/run/systemd/system"
_PATRONI_CONFIG = "/etc/patroni/config.yml"
#: The container environment that may leave the host.
_ENV_KEYS = "ORIGIN_URL|REPLICA[0-9]*_URL|API_WHITELIST_IP|BUNKERWEB_INSTANCES|API_HTTP_PORT"
#: BunkerWeb's instance API, where a manager pushes its configuration.
_BUNKERWEB_API_PORT = 5000
#: Fed a unit's ``systemctl show`` output: ``active <state>`` and ``config <path>``,
#: the first argument of the command line that is no flag.
_UNIT_AWK = " ".join(
(
r'/^ActiveState=/ { print "active " substr($0, 13); next }',
r'/^ExecStart=/ { s = $0; i = index(s, "argv[]="); if (!i) next; s = substr(s, i + 7);'
r' j = index(s, " ;"); if (j) s = substr(s, 1, j - 1); n = split(s, t, " ");'
r' for (k = 2; k <= n; k++) if (!c && t[k] !~ /^-/) { print "config " t[k]; c = 1 } }',
)
)
#: Fed Patroni's YAML: ``scope``, ``name`` (top level only -- ``tags`` may carry a
#: ``name`` too), ``restapi`` (its connect_address) and ``tls yes`` when the REST
#: API has a certificate. Nothing else leaves the file.
_PATRONI_AWK = " ".join(
(
r'function v(l) { sub(/^[^:]*:[ \t]*/, "", l); return l }',
r'/^[A-Za-z_][A-Za-z0-9_]*:/ { sec = $0; sub(/:.*/, "", sec);'
r' if (sec == "scope" || sec == "name") print sec " " v($0); next }',
r'sec == "restapi" && /^[ \t]+connect_address:/ { print "restapi " v($0) }',
r'sec == "restapi" && /^[ \t]+certfile:/ { print "tls yes" }',
)
)
#: Fed a container's environment, one ``KEY=value`` a line: the allowlisted keys,
#: each URL without its ``user:password@``.
_ENV_AWK = " ".join(
(
r'{ x = index($0, "="); if (!x) next; k = substr($0, 1, x - 1); v = substr($0, x + 1);',
rf" if (k !~ /^({_ENV_KEYS})$/) next;",
r' gsub(/:\/\/[^@\/ ]*@/, "://", v); print "env " k " " v }',
)
)
_PATRONI_SH = (
f"if command -v systemctl >/dev/null 2>&1 && [ -d {_SYSTEMD_RUN} ]; then echo '[systemd]'; "
"for u in $(systemctl list-units --all --plain --no-legend --type=service "
"patroni.service 'patroni@*' 2>/dev/null | awk '{print $1}'); do "
'echo "[patroni $u]"; '
'o=$(systemctl show -p ActiveState -p ExecStart "$u" 2>/dev/null | '
f"awk {quote(_UNIT_AWK)}); "
'echo "$o"; '
"f=$(echo \"$o\" | sed -n 's/^config //p' | head -n 1); "
f'[ -n "$f" ] || f={_PATRONI_CONFIG}; '
'if [ -r "$f" ]; then '
f'p=$(awk {quote(_PATRONI_AWK)} "$f"); echo "$p"; '
"r=$(echo \"$p\" | sed -n 's/^restapi //p' | head -n 1 | tr -d '\\042\\047 '); "
's=http; case "$p" in *"tls yes"*) s=https;; esac; '
'if [ -n "$r" ]; then '
'b=$(curl -s -k -m 3 "$s://$r/cluster" 2>/dev/null || '
'wget -q -T 3 --no-check-certificate -O - "$s://$r/cluster" 2>/dev/null); '
"if [ -n \"$b\" ]; then echo \"rest $(echo \"$b\" | tr -d '\\n')\"; "
"else echo rest-unreachable; fi; fi; "
'elif [ -e "$f" ]; then echo "unreadable $f"; '
'else echo "missing $f"; fi; '
"done; "
"else echo '[no-systemd]'; fi; "
)
def _docker_sh(binary: Optional[str]) -> str:
"""The containers part: running containers of the clustered images, their
allowlisted environment. ``[no-docker]`` for a host without the engine."""
if not binary:
return "echo '[no-docker]'; "
return (
f"D={quote(binary)}; "
'if command -v "$D" >/dev/null 2>&1; then '
"if c=$(\"$D\" ps --format '{{.Names}}' 2>/dev/null); then echo '[docker]'; "
"for n in $c; do "
"i=$(\"$D\" inspect --format '{{.Config.Image}}' \"$n\" 2>/dev/null); "
'case "$i" in *adguardhome-sync*|*bunkerweb*) '
'echo "[container $n]"; echo "image $i"; '
"\"$D\" inspect --format '{{range .Config.Env}}{{println .}}{{end}}' \"$n\" "
f"2>/dev/null | awk {quote(_ENV_AWK)};; esac; done; "
"else echo '[docker-unreadable]'; fi; "
"else echo '[no-docker]'; fi; "
)
def service_clusters_command(docker: Optional[str] = "docker") -> str:
"""One line of POSIX ``sh``, read-only. *docker* is the engine's binary as the
driver reaches it (QNAP's lives elsewhere), None for a host without one. The
frame markers are printed in two halves so that a transport which echoes the
command does not show them early."""
return (
"PATH=$PATH:/usr/sbin:/sbin; "
"printf '%s%s\\n' SVCC_ BEGIN; "
+ _PATRONI_SH
+ _docker_sh(docker)
+ "printf '%s%s\\n' SVCC_ END"
)
SERVICE_CLUSTERS_COMMAND = service_clusters_command()
_COMMENT_RE = re.compile(r"\s+#.*$")
_Record = Tuple[str, str]
_Section = Tuple[str, str, List[_Record]]
def _frame(output: str) -> List[str]:
lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
try:
start = lines.index(_BEGIN)
end = lines.index(_END, start)
except ValueError:
raise ValueError("no intact service cluster report in the output") from None
return lines[start + 1 : end]
def _sections(lines: List[str]) -> List[_Section]:
"""``[<kind> <name>]`` headers, each with its records as ``(kind, rest)``."""
sections: List[_Section] = []
for line in lines:
if line.startswith("[") and line.endswith("]"):
kind, _, name = line[1:-1].partition(" ")
sections.append((kind, name, []))
elif line and sections:
kind, _, rest = line.partition(" ")
sections[-1][2].append((kind, rest.strip()))
return sections
def _clean(value: str) -> str:
"""Without a trailing ``# comment`` and the quotes around it."""
value = _COMMENT_RE.sub("", value.strip())
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
value = value[1:-1]
return value.strip()
def _member(
*,
address: Optional[str] = None,
port: Optional[int] = None,
name: Optional[str] = None,
role: Optional[str] = None,
state: Optional[str] = None,
this_host: bool = False,
config_from: Optional[int] = None,
) -> ServiceClusterMemberDict:
return {
"name": name,
"address": address,
"port": port,
"role": role,
"state": state,
"this_host": this_host,
"config_from": config_from,
}
def _cluster(
mechanism: str,
service: str,
source: str,
cluster: Optional[str],
members: List[ServiceClusterMemberDict],
running: Optional[bool] = True,
) -> ServiceClusterDict:
return {
"mechanism": mechanism,
"service": service,
"source": source,
"cluster": cluster,
"running": running,
"members": members,
}
def _host_port(text: str) -> Tuple[Optional[str], Optional[int]]:
"""``"10.0.0.5:8008"`` or ``"[fd00::5]:8008"`` -> host and port."""
try:
split = urlsplit(f"//{text.strip()}")
return split.hostname, split.port
except ValueError:
return None, None
def _url(text: str) -> Tuple[Optional[str], Optional[int]]:
"""A URL's host and port, the scheme's default where it names none."""
value = text.strip()
if "://" not in value:
value = f"http://{value}"
try:
split = urlsplit(value)
port = split.port or {"http": 80, "https": 443}.get(split.scheme)
except ValueError:
return None, None
return split.hostname, port
# --- Patroni ------------------------------------------------------------------
def _patroni_members(rest: str, own: Optional[str]) -> Optional[List[ServiceClusterMemberDict]]:
try:
members = json.loads(rest).get("members")
except (ValueError, AttributeError):
return None
if not isinstance(members, list):
return None
found = []
for entry in members:
if not isinstance(entry, dict) or not entry.get("name"):
continue
port = entry.get("port")
found.append(
_member(
name=str(entry["name"]),
address=str(entry["host"]) if entry.get("host") else None,
port=port if isinstance(port, int) else None,
role=str(entry["role"]) if entry.get("role") else None,
state=str(entry["state"]) if entry.get("state") else None,
this_host=entry["name"] == own,
)
)
return sorted(found, key=lambda m: m["name"] or "")
def _patroni(unit: str, records: List[_Record]) -> List[ServiceClusterDict]:
values: Dict[str, str] = {}
rest: Optional[str] = None
for kind, text in records:
if kind in ("active", "scope", "name", "restapi") and kind not in values:
values[kind] = _clean(text)
elif kind == "rest":
rest = text
scope = values.get("scope")
if not scope:
return []
own = values.get("name") or None
members = _patroni_members(rest, own) if rest else None
if not members:
address, _ = _host_port(values.get("restapi", ""))
members = [_member(name=own, address=address, this_host=True)]
running = values["active"] == "active" if "active" in values else None
source = unit[: -len(".service")] if unit.endswith(".service") else unit
return [_cluster("patroni", "patroni", source, scope, members, running)]
# --- Containers ----------------------------------------------------------------
def _env(records: List[_Record]) -> Dict[str, str]:
env: Dict[str, str] = {}
for kind, text in records:
if kind == "env":
key, _, value = text.partition(" ")
env[key] = value.strip()
return env
def _image_name(records: List[_Record]) -> str:
"""``bunkerity/bunkerweb@sha256:...`` -> ``bunkerweb``."""
image = next((text for kind, text in records if kind == "image"), "")
return image.split("@")[0].rsplit("/", 1)[-1].split(":")[0]
def _adguardhome_sync(name: str, records: List[_Record]) -> Optional[List[ServiceClusterDict]]:
env = _env(records)
origin = _url(env["ORIGIN_URL"]) if env.get("ORIGIN_URL") else (None, None)
if origin[0] is None:
return None # configured by its YAML file, which is not read
replicas = sorted(
(key for key in env if re.fullmatch(r"REPLICA[0-9]*_URL", key)),
key=lambda key: int(key[7:-4] or 0),
)
members = [_member(address=origin[0], port=origin[1], role="origin")]
for key in replicas:
host, port = _url(env[key])
if host:
members.append(_member(address=host, port=port, role="replica", config_from=0))
cluster = f"{origin[0]}:{origin[1]}" if origin[1] else origin[0]
return [_cluster("adguardhome-sync", "adguardhome", name, cluster, members)]
def _peer_address(entry: str) -> Optional[str]:
"""An entry of a whitelist that names one peer: a single address that is no
loopback, link-local or unspecified one -- not a network."""
text = entry.strip()
if "/" in text:
try:
network = ipaddress.ip_network(text, strict=False)
except ValueError:
return None
if network.num_addresses != 1:
return None
text = str(network.network_address)
try:
address = ipaddress.ip_address(text)
except ValueError:
return None
if address.is_loopback or address.is_link_local or address.is_unspecified:
return None
return str(address)
def _api_port(env: Dict[str, str]) -> int:
port = env.get("API_HTTP_PORT", "")
return int(port) if port.isdigit() else _BUNKERWEB_API_PORT
def _bunkerweb_worker(name: str, env: Dict[str, str]) -> List[ServiceClusterDict]:
managers = [
address
for entry in re.split(r"[\s,]+", env.get("API_WHITELIST_IP", ""))
if entry and (address := _peer_address(entry))
]
if not managers:
return []
members = [_member(port=_api_port(env), role="worker", this_host=True, config_from=1)]
members += [_member(address=address, role="manager") for address in managers]
return [_cluster("bunkerweb", "bunkerweb", name, managers[0], members)]
def _bunkerweb_manager(name: str, env: Dict[str, str]) -> List[ServiceClusterDict]:
instances = [entry for entry in env.get("BUNKERWEB_INSTANCES", "").split() if entry]
if not instances:
return []
members = [_member(role="manager", this_host=True)]
for entry in instances:
host, port = _host_port(entry)
if host:
members.append(
_member(address=host, port=port or _BUNKERWEB_API_PORT, role="worker", config_from=0)
)
return [_cluster("bunkerweb", "bunkerweb", name, None, members)] if len(members) > 1 else []
def _bunkerweb(name: str, records: List[_Record]) -> List[ServiceClusterDict]:
image = _image_name(records)
if image == "bunkerweb":
return _bunkerweb_worker(name, _env(records))
if image in ("bunkerweb-all-in-one", "bunkerweb-scheduler"):
return _bunkerweb_manager(name, _env(records))
return [] # the UI, the API, autoconf: no instance and no manager
_CONTAINER_READERS: Dict[str, Callable[[str, List[_Record]], Optional[List[ServiceClusterDict]]]] = {
"adguardhome_sync": _adguardhome_sync,
"bunkerweb": _bunkerweb,
}
def _reader_for(records: List[_Record]) -> Optional[str]:
image = next((text for kind, text in records if kind == "image"), "")
if "adguardhome-sync" in image:
return "adguardhome_sync"
if "bunkerweb" in image:
return "bunkerweb"
return None
def _containers(sections: List[_Section]) -> Dict[str, Optional[List[ServiceClusterDict]]]:
kinds = {kind for kind, _, _ in sections}
if "no-docker" in kinds:
return {"adguardhome_sync": [], "bunkerweb": []}
if "docker" not in kinds:
return {"adguardhome_sync": None, "bunkerweb": None}
found: Dict[str, Optional[List[ServiceClusterDict]]] = {"adguardhome_sync": [], "bunkerweb": []}
for kind, name, records in sections:
key = _reader_for(records) if kind == "container" else None
if key is None or found[key] is None:
continue
clusters = _CONTAINER_READERS[key](name, records)
if clusters is None:
found[key] = None
else:
found[key].extend(clusters) # type: ignore[union-attr]
return found
def parse_service_clusters(output: str) -> ServiceClustersDict:
"""Parse what :func:`service_clusters_command` printed.
:raises ValueError: when the output carries no intact report.
"""
sections = _sections(_frame(output))
patroni: Optional[List[ServiceClusterDict]]
if any(kind == "no-systemd" for kind, _, _ in sections):
patroni = None
elif any(kind == "patroni" and any(r[0] == "unreadable" for r in records) for kind, _, records in sections):
patroni = None
else:
patroni = [
c for kind, unit, records in sections if kind == "patroni" for c in _patroni(unit, records)
]
containers = _containers(sections)
return {
"patroni": patroni,
"adguardhome_sync": containers["adguardhome_sync"],
"bunkerweb": containers["bunkerweb"],
}
class ServiceClustersMixin:
"""Adds :meth:`get_service_clusters` to a driver that can run a command on a Linux host.
The template form (README, "Function classes"): the command and its parse are
the same everywhere, so they are concrete here, and a driver supplies only
:meth:`_run_service_clusters_command` -- how a command reaches its host, and
how it gains root there. A driver with a container engine of its own path
(``ContainerEngineMixin._container_engine_binary``) has it asked with that.
Mixed in by the drivers that can, so ``hasattr(driver,
"get_service_clusters")`` stays a truthful answer.
"""
if TYPE_CHECKING: # pragma: no cover - declared for type checkers only
def _run_service_clusters_command(self, command: str, *, privileged: bool) -> str:
"""Run *command* on the host and return what it printed; as root when
*privileged*. The command is a single ``sh -c`` invocation of a few kB:
send it on an exec channel, not typed into a shell."""
...
def _service_clusters_docker(self) -> str:
binary = getattr(self, "_container_engine_binary", None)
return binary("docker") if callable(binary) else "docker"
def get_service_clusters(self) -> ServiceClustersDict:
"""
Returns the clusters the host's configuration says it belongs to, per
mechanism: Patroni (its scope and the members its DCS lists),
adguardhome-sync (its origin and replicas), BunkerWeb (a worker's manager,
a manager's instances). See :class:`~napalm_device_types.models.ServiceClusterDict`.
None for a mechanism means it could not be read; ``[]`` that the host has none.
Example::
{
"patroni": [
{"mechanism": "patroni", "service": "patroni", "source": "patroni",
"cluster": "netork-db",
"running": True,
"members": [
{"name": "db-01", "address": "10.7.234.12", "port": 5432,
"role": "leader", "state": "running", "this_host": True,
"config_from": None},
{"name": "db-02", "address": "10.7.234.11", "port": 5432,
"role": "sync_standby", "state": "streaming", "this_host": False,
"config_from": None},
]},
],
"adguardhome_sync": [],
"bunkerweb": [],
}
:raises ValueError: if neither reading carried an intact report.
"""
command = f"sh -c {quote(service_clusters_command(self._service_clusters_docker()))}"
try:
return parse_service_clusters(self._run_service_clusters_command(command, privileged=True))
except ValueError:
pass
return parse_service_clusters(self._run_service_clusters_command(command, privileged=False))