from __future__ import annotations import json import re import secrets import subprocess import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Callable, Sequence import xml.etree.ElementTree as ET TOPOLOGY_SCHEMA_VERSION = "1.0" INVENTORY_BEGIN_MARKER = "PROLE_NODE_INVENTORY_BEGIN" INVENTORY_END_MARKER = "PROLE_NODE_INVENTORY_END" TOPOLOGY_NAMESPACE = "kube-system" DEFAULT_REQUIRED_MONITORING_MOUNTPOINTS: tuple[str, ...] = ("/synology/d001",) DEFAULT_MONITORING_MIN_FREE_BYTES = 200 * 1024 * 1024 * 1024 _SYNOLOGY_DATA_MOUNT_RE = re.compile(r"^/synology/d\d{3}$") _NETWORK_FILESYSTEM_TYPES = { "nfs", "nfs4", "cifs", "smb", "glusterfs", "ceph", "cephfs", "fuse.sshfs", } @dataclass(slots=True, frozen=True) class FilesystemTopology: device: str mountpoint: str fstype: str size_bytes: int | None used_bytes: int | None available_bytes: int | None transport: str rotational: bool | None storage_class: str @dataclass(slots=True, frozen=True) class CapabilityFlags: has_synology_root: bool | None matching_mountpoints: tuple[str, ...] monitoring_eligible: bool | None cnpg_eligible: bool | None reasons: tuple[str, ...] has_synology_root_state: str monitoring_eligible_state: str cnpg_eligible_state: str @dataclass(slots=True, frozen=True) class NodeTopology: name: str roles: tuple[str, ...] labels: dict[str, str] taints: tuple[str, ...] ready: bool internal_ip: str architecture: str operating_system: str allocatable_cpu: str allocatable_memory_bytes: int | None allocatable_ephemeral_storage_bytes: int | None cpu_model: str | None physical_cores: int | None logical_cores: int | None memory_bytes: int | None filesystems: tuple[FilesystemTopology, ...] host_inventory_available: bool host_inventory_error: str | None capabilities: CapabilityFlags @dataclass(slots=True, frozen=True) class ClusterTopology: version: str generated_at: str generated_epoch_ms: int collection_status: str required_monitoring_mountpoints: tuple[str, ...] monitoring_min_free_bytes: int nodes: tuple[NodeTopology, ...] topology_root: str messages: tuple[str, ...] @property def discovered_nodes(self) -> int: return len(self.nodes) @property def ready_nodes(self) -> int: return sum(1 for node in self.nodes if node.ready) @property def monitoring_eligible_nodes(self) -> int: return sum(1 for node in self.nodes if node.capabilities.monitoring_eligible is True) @property def cnpg_eligible_nodes(self) -> int: return sum(1 for node in self.nodes if node.capabilities.cnpg_eligible is True) @dataclass(slots=True, frozen=True) class TopologyDiscoveryConfig: namespace: str = TOPOLOGY_NAMESPACE timeout_seconds: int = 120 poll_interval_seconds: float = 2.0 required_monitoring_mountpoints: tuple[str, ...] = DEFAULT_REQUIRED_MONITORING_MOUNTPOINTS monitoring_min_free_bytes: int = DEFAULT_MONITORING_MIN_FREE_BYTES collector_image: str = "python:3.11-slim" @dataclass(slots=True, frozen=True) class TopologyDiscoveryResult: topology: ClusterTopology collector_applied: bool expected_ready_nodes: tuple[str, ...] reported_nodes: tuple[str, ...] missing_nodes: tuple[str, ...] def _run_kubectl( base_cmd: Sequence[str], args: Sequence[str], *, timeout: int = 20, env: dict[str, str] | None = None, input_text: str | None = None, ) -> subprocess.CompletedProcess[str]: return subprocess.run( [*base_cmd, *args], capture_output=True, text=True, timeout=timeout, env=env, input=input_text, ) def _kubectl_json( base_cmd: Sequence[str], args: Sequence[str], *, timeout: int = 20, env: dict[str, str] | None = None, ) -> dict: res = _run_kubectl(base_cmd, [*args, "-o", "json"], timeout=timeout, env=env) if res.returncode != 0: stderr = (res.stderr or "").strip() raise RuntimeError(f"kubectl command failed: {' '.join(args)}: {stderr}") try: return json.loads(res.stdout or "{}") except json.JSONDecodeError as exc: raise RuntimeError(f"Invalid kubectl JSON payload for command: {' '.join(args)}") from exc def _parse_k8s_quantity_bytes(raw: str | None) -> int | None: text = str(raw or "").strip() if not text: return None m = re.match(r"^([0-9]+(?:\.[0-9]+)?)([A-Za-z]+)?$", text) if not m: return None num = float(m.group(1)) unit = (m.group(2) or "").strip() binary_units = { "Ki": 1024, "Mi": 1024**2, "Gi": 1024**3, "Ti": 1024**4, "Pi": 1024**5, "Ei": 1024**6, } decimal_units = { "K": 1000, "M": 1000**2, "G": 1000**3, "T": 1000**4, "P": 1000**5, "E": 1000**6, } if not unit: return int(num) if unit in binary_units: return int(num * binary_units[unit]) if unit in decimal_units: return int(num * decimal_units[unit]) return None def _node_roles(labels: dict[str, str]) -> tuple[str, ...]: roles: list[str] = [] for key, val in labels.items(): if key.startswith("node-role.kubernetes.io/"): suffix = key.split("/", 1)[1].strip() if suffix: roles.append(suffix) elif val: roles.append(str(val).strip()) return tuple(sorted({r for r in roles if r})) def _node_ready(status: dict) -> bool: for cond in status.get("conditions") or []: if str(cond.get("type") or "") == "Ready": return str(cond.get("status") or "").lower() == "true" return False def _node_internal_ip(status: dict) -> str: for addr in status.get("addresses") or []: if str(addr.get("type") or "") == "InternalIP": return str(addr.get("address") or "").strip() return "" def _normalize_taints(spec: dict) -> tuple[str, ...]: out: list[str] = [] for taint in spec.get("taints") or []: key = str(taint.get("key") or "").strip() value = str(taint.get("value") or "").strip() effect = str(taint.get("effect") or "").strip() if not key: continue token = key if value: token += f"={value}" if effect: token += f":{effect}" out.append(token) return tuple(sorted(out)) def _parse_k8s_nodes(nodes_payload: dict) -> dict[str, dict]: out: dict[str, dict] = {} for item in nodes_payload.get("items") or []: metadata = item.get("metadata") or {} status = item.get("status") or {} labels = { str(k): str(v) for k, v in (metadata.get("labels") or {}).items() if str(k).strip() } name = str(metadata.get("name") or "").strip() if not name: continue alloc = status.get("allocatable") or {} out[name] = { "name": name, "roles": _node_roles(labels), "labels": labels, "taints": _normalize_taints(item.get("spec") or {}), "ready": _node_ready(status), "internal_ip": _node_internal_ip(status), "architecture": str( labels.get("kubernetes.io/arch") or labels.get("beta.kubernetes.io/arch") or "" ).strip(), "operating_system": str( labels.get("kubernetes.io/os") or labels.get("beta.kubernetes.io/os") or "" ).strip(), "allocatable_cpu": str(alloc.get("cpu") or "").strip(), "allocatable_memory_bytes": _parse_k8s_quantity_bytes( str(alloc.get("memory") or "") ), "allocatable_ephemeral_storage_bytes": _parse_k8s_quantity_bytes( str(alloc.get("ephemeral-storage") or "") ), } return out def _collector_python_script() -> str: return r''' import json import os import platform import re import time from pathlib import Path BEGIN = "PROLE_NODE_INVENTORY_BEGIN" END = "PROLE_NODE_INVENTORY_END" _PSEUDO_FS = { "proc", "sysfs", "tmpfs", "cgroup", "cgroup2", "devpts", "devtmpfs", "mqueue", "tracefs", "pstore", "securityfs", "debugfs", "ramfs", "overlay", "squashfs", "autofs" } _NETWORK_FS = {"nfs", "nfs4", "cifs", "smb", "glusterfs", "ceph", "cephfs", "fuse.sshfs"} def _read_text(path: Path) -> str: try: return path.read_text(encoding="utf-8", errors="replace") except Exception: return "" def _decode_mount_path(path: str) -> str: def repl(match): try: return chr(int(match.group(1), 8)) except Exception: return match.group(0) return re.sub(r"\\([0-7]{3})", repl, path) def _device_block_name(device: str) -> str: if not device.startswith("/dev/"): return "" name = device.split("/")[-1] if name.startswith("nvme") and "p" in name: return name.split("p", 1)[0] m = re.match(r"^(.*?)(\d+)$", name) if m: return m.group(1) return name def _read_rotational(block_name: str): if not block_name: return None p = Path("/host/sys/block") / block_name / "queue/rotational" text = _read_text(p).strip() if text == "0": return False if text == "1": return True return None def _read_transport(device: str, fstype: str, block_name: str) -> str: if fstype in _NETWORK_FS: return "network" if "nvme" in device or block_name.startswith("nvme"): return "nvme" if not block_name: return "unknown" if block_name.startswith("sd"): return "sata" if block_name.startswith("vd"): return "virtio" if block_name.startswith("xvd"): return "virtio" return "unknown" def _infer_storage_class(device: str, fstype: str, transport: str, rotational): if fstype in _NETWORK_FS: return "network" if transport == "nvme" or "nvme" in device: return "nvme" if rotational is False: return "ssd" if rotational is True: return "hdd" return "unknown" def _cpu_facts(): cpuinfo = _read_text(Path("/host/proc/cpuinfo")) model = "" logical = 0 phys_cores = set() current_phys = "" current_core = "" for raw_line in cpuinfo.splitlines(): line = raw_line.strip() if not line: if current_phys or current_core: phys_cores.add((current_phys, current_core)) current_phys = "" current_core = "" continue if line.startswith("model name") and not model: parts = line.split(":", 1) if len(parts) == 2: model = parts[1].strip() if line.startswith("processor"): logical += 1 if line.startswith("physical id"): parts = line.split(":", 1) if len(parts) == 2: current_phys = parts[1].strip() if line.startswith("core id"): parts = line.split(":", 1) if len(parts) == 2: current_core = parts[1].strip() if current_phys or current_core: phys_cores.add((current_phys, current_core)) physical = len([x for x in phys_cores if x != ("", "")]) if physical <= 0 and logical > 0: physical = logical return model or None, physical or None, logical or None def _memory_bytes(): meminfo = _read_text(Path("/host/proc/meminfo")) for line in meminfo.splitlines(): if not line.startswith("MemTotal:"): continue parts = line.split() if len(parts) < 2: continue try: return int(parts[1]) * 1024 except Exception: return None return None def _mount_rows(): mounts_text = _read_text(Path("/host/proc/1/mounts")) or _read_text(Path("/host/proc/mounts")) rows = [] seen = set() for line in mounts_text.splitlines(): parts = line.split() if len(parts) < 3: continue device = parts[0] mountpoint = _decode_mount_path(parts[1]) fstype = parts[2] if fstype in _PSEUDO_FS: continue key = (mountpoint, device, fstype) if key in seen: continue seen.add(key) rows.append((device, mountpoint, fstype)) rows.sort(key=lambda x: x[1]) return rows def _fs_stats(mountpoint: str): host_mount = Path("/host") / mountpoint.lstrip("/") if not host_mount.exists(): return None, None, None try: st = os.statvfs(str(host_mount)) size = int(st.f_frsize * st.f_blocks) avail = int(st.f_frsize * st.f_bavail) used = max(0, size - int(st.f_frsize * st.f_bfree)) return size, used, avail except Exception: return None, None, None def _collect_filesystems(): out = [] for device, mountpoint, fstype in _mount_rows(): size, used, available = _fs_stats(mountpoint) block_name = _device_block_name(device) rotational = _read_rotational(block_name) transport = _read_transport(device, fstype, block_name) storage_class = _infer_storage_class(device, fstype, transport, rotational) out.append({ "device": device, "mountpoint": mountpoint, "fstype": fstype, "sizeBytes": size, "usedBytes": used, "availableBytes": available, "transport": transport, "rotational": rotational, "storageClass": storage_class, }) return out def main(): cpu_model, physical_cores, logical_cores = _cpu_facts() payload = { "nodeName": os.environ.get("NODE_NAME", ""), "reportedAt": int(time.time()), "architecture": platform.machine(), "cpuModel": cpu_model, "physicalCores": physical_cores, "logicalCores": logical_cores, "memoryBytes": _memory_bytes(), "filesystems": _collect_filesystems(), } print(BEGIN, flush=True) print(json.dumps(payload, separators=(",", ":"), sort_keys=True), flush=True) print(END, flush=True) # Keep pod alive so logs can be collected exactly once without restart loops. time.sleep(3600) if __name__ == "__main__": main() '''.strip() def _collector_manifest(name: str, *, image: str) -> dict: labels = { "app.kubernetes.io/name": "prole-topology-collector", "prole.io/topology-collector": name, } return { "apiVersion": "apps/v1", "kind": "DaemonSet", "metadata": { "name": name, "namespace": TOPOLOGY_NAMESPACE, "labels": labels, }, "spec": { "selector": {"matchLabels": labels}, "template": { "metadata": {"labels": labels}, "spec": { "serviceAccountName": "default", "terminationGracePeriodSeconds": 0, "tolerations": [{"operator": "Exists"}], "containers": [ { "name": "collector", "image": image, "imagePullPolicy": "IfNotPresent", "command": ["python", "-c", _collector_python_script()], "env": [ { "name": "NODE_NAME", "valueFrom": { "fieldRef": {"fieldPath": "spec.nodeName"} }, } ], "securityContext": {"privileged": True}, "volumeMounts": [ { "name": "host-root", "mountPath": "/host", "readOnly": True, } ], } ], "volumes": [{"name": "host-root", "hostPath": {"path": "/"}}], }, }, "updateStrategy": {"type": "RollingUpdate"}, }, } def _extract_inventory_from_log(log_text: str) -> dict | None: text = str(log_text or "") begin = text.find(INVENTORY_BEGIN_MARKER) end = text.find(INVENTORY_END_MARKER) if begin < 0 or end < 0 or end <= begin: return None payload = text[begin + len(INVENTORY_BEGIN_MARKER) : end].strip() if not payload: return None try: data = json.loads(payload) except json.JSONDecodeError: return None return data if isinstance(data, dict) else None def _merge_node( k8s_node: dict, inventory: dict | None, *, required_monitoring_mountpoints: tuple[str, ...], monitoring_min_free_bytes: int, inventory_error: str | None = None, ) -> NodeTopology: filesystems: list[FilesystemTopology] = [] if isinstance(inventory, dict): for item in inventory.get("filesystems") or []: if not isinstance(item, dict): continue filesystems.append( FilesystemTopology( device=str(item.get("device") or "").strip(), mountpoint=str(item.get("mountpoint") or "").strip(), fstype=str(item.get("fstype") or "").strip(), size_bytes=item.get("sizeBytes") if isinstance(item.get("sizeBytes"), int) else None, used_bytes=item.get("usedBytes") if isinstance(item.get("usedBytes"), int) else None, available_bytes=item.get("availableBytes") if isinstance(item.get("availableBytes"), int) else None, transport=str(item.get("transport") or "unknown").strip() or "unknown", rotational=item.get("rotational") if isinstance(item.get("rotational"), bool) else None, storage_class=str(item.get("storageClass") or "unknown").strip() or "unknown", ) ) filesystems.sort(key=lambda fs: fs.mountpoint) host_inventory_available = bool(inventory) mountpoints = {fs.mountpoint for fs in filesystems if fs.mountpoint} matching_mountpoints_set = { mp for mp in required_monitoring_mountpoints if mp in mountpoints } if not matching_mountpoints_set and any( _SYNOLOGY_DATA_MOUNT_RE.fullmatch(mp) for mp in required_monitoring_mountpoints ): matching_mountpoints_set = { mountpoint for mountpoint in mountpoints if _SYNOLOGY_DATA_MOUNT_RE.fullmatch(mountpoint) } matching_mountpoints = tuple(sorted(matching_mountpoints_set)) reasons: set[str] = set() has_synology_root: bool | None has_synology_root_state: str monitoring_eligible: bool | None monitoring_eligible_state: str cnpg_eligible: bool | None cnpg_eligible_state: str if not host_inventory_available: has_synology_root = None has_synology_root_state = "unavailable" monitoring_eligible = None monitoring_eligible_state = "unavailable" cnpg_eligible = None cnpg_eligible_state = "unavailable" reasons.add("topology_unavailable") else: has_synology_root = bool( any(fs.mountpoint.startswith("/synology/") for fs in filesystems if fs.mountpoint) ) has_synology_root_state = "discovered" if not k8s_node["ready"]: monitoring_eligible = False monitoring_eligible_state = "inferred" reasons.add("node_not_ready") elif not matching_mountpoints: monitoring_eligible = False monitoring_eligible_state = "inferred" reasons.add("missing_required_mountpoint") else: max_available = max( ( fs.available_bytes for fs in filesystems if fs.mountpoint in matching_mountpoints and fs.available_bytes is not None ), default=None, ) if max_available is None: monitoring_eligible = None monitoring_eligible_state = "unavailable" reasons.add("topology_unavailable") elif max_available < monitoring_min_free_bytes: monitoring_eligible = False monitoring_eligible_state = "inferred" reasons.add("insufficient_free_space") else: monitoring_eligible = True monitoring_eligible_state = "inferred" if not k8s_node["ready"]: cnpg_eligible = False cnpg_eligible_state = "inferred" reasons.add("node_not_ready") elif has_synology_root is True: cnpg_eligible = True cnpg_eligible_state = "inferred" elif has_synology_root is False: cnpg_eligible = False cnpg_eligible_state = "inferred" reasons.add("missing_required_mountpoint") else: cnpg_eligible = None cnpg_eligible_state = "unavailable" reasons.add("topology_unavailable") if inventory_error: reasons.add("inventory_parse_error") capabilities = CapabilityFlags( has_synology_root=has_synology_root, matching_mountpoints=matching_mountpoints, monitoring_eligible=monitoring_eligible, cnpg_eligible=cnpg_eligible, reasons=tuple(sorted(reasons)), has_synology_root_state=has_synology_root_state, monitoring_eligible_state=monitoring_eligible_state, cnpg_eligible_state=cnpg_eligible_state, ) return NodeTopology( name=str(k8s_node["name"]), roles=tuple(k8s_node["roles"]), labels=dict(sorted(k8s_node["labels"].items())), taints=tuple(k8s_node["taints"]), ready=bool(k8s_node["ready"]), internal_ip=str(k8s_node["internal_ip"]), architecture=str(k8s_node["architecture"]), operating_system=str(k8s_node["operating_system"]), allocatable_cpu=str(k8s_node["allocatable_cpu"]), allocatable_memory_bytes=k8s_node["allocatable_memory_bytes"], allocatable_ephemeral_storage_bytes=k8s_node["allocatable_ephemeral_storage_bytes"], cpu_model=(str(inventory.get("cpuModel") or "").strip() if inventory else None) or None, physical_cores=(inventory.get("physicalCores") if inventory else None) if isinstance((inventory or {}).get("physicalCores"), int) else None, logical_cores=(inventory.get("logicalCores") if inventory else None) if isinstance((inventory or {}).get("logicalCores"), int) else None, memory_bytes=(inventory.get("memoryBytes") if inventory else None) if isinstance((inventory or {}).get("memoryBytes"), int) else None, filesystems=tuple(filesystems), host_inventory_available=host_inventory_available, host_inventory_error=inventory_error, capabilities=capabilities, ) def _render_xml_bytes(value: int | None) -> str: return "" if value is None else str(int(value)) def _set_text(parent: ET.Element, tag: str, text: str) -> ET.Element: elem = ET.SubElement(parent, tag) elem.text = text return elem def _append_bool(parent: ET.Element, tag: str, value: bool | None) -> None: _set_text(parent, tag, "unknown" if value is None else ("true" if value else "false")) def _node_to_xml(node: NodeTopology) -> ET.Element: root = ET.Element("nodeTopology", {"version": TOPOLOGY_SCHEMA_VERSION}) identity = ET.SubElement(root, "identity") _set_text(identity, "name", node.name) _set_text(identity, "internalIp", node.internal_ip) kubernetes = ET.SubElement(root, "kubernetes") _set_text(kubernetes, "ready", "true" if node.ready else "false") _set_text(kubernetes, "roles", ",".join(node.roles)) _set_text(kubernetes, "architecture", node.architecture) _set_text(kubernetes, "operatingSystem", node.operating_system) _set_text(kubernetes, "allocatableCpu", node.allocatable_cpu) _set_text(kubernetes, "allocatableMemoryBytes", _render_xml_bytes(node.allocatable_memory_bytes)) _set_text( kubernetes, "allocatableEphemeralStorageBytes", _render_xml_bytes(node.allocatable_ephemeral_storage_bytes), ) labels = ET.SubElement(kubernetes, "labels") for key, val in sorted(node.labels.items()): label = ET.SubElement(labels, "label", {"key": key}) label.text = val taints = ET.SubElement(kubernetes, "taints") for taint in node.taints: _set_text(taints, "taint", taint) hardware = ET.SubElement(root, "hardware") _set_text(hardware, "cpuModel", node.cpu_model or "") _set_text(hardware, "physicalCores", _render_xml_bytes(node.physical_cores)) _set_text(hardware, "logicalCores", _render_xml_bytes(node.logical_cores)) _set_text(hardware, "memoryBytes", _render_xml_bytes(node.memory_bytes)) filesystems = ET.SubElement(root, "filesystems") for fs in node.filesystems: elem = ET.SubElement(filesystems, "filesystem") _set_text(elem, "device", fs.device) _set_text(elem, "mountpoint", fs.mountpoint) _set_text(elem, "fstype", fs.fstype) _set_text(elem, "sizeBytes", _render_xml_bytes(fs.size_bytes)) _set_text(elem, "usedBytes", _render_xml_bytes(fs.used_bytes)) _set_text(elem, "availableBytes", _render_xml_bytes(fs.available_bytes)) _set_text(elem, "transport", fs.transport) _set_text( elem, "rotational", "unknown" if fs.rotational is None else ("true" if fs.rotational else "false"), ) _set_text(elem, "storageClass", fs.storage_class) capabilities = ET.SubElement(root, "capabilities") _append_bool(capabilities, "hasSynologyRoot", node.capabilities.has_synology_root) _set_text(capabilities, "hasSynologyRootState", node.capabilities.has_synology_root_state) _set_text( capabilities, "matchingMountpoints", ",".join(node.capabilities.matching_mountpoints), ) _append_bool(capabilities, "monitoringEligible", node.capabilities.monitoring_eligible) _set_text( capabilities, "monitoringEligibleState", node.capabilities.monitoring_eligible_state, ) _append_bool(capabilities, "cnpgEligible", node.capabilities.cnpg_eligible) _set_text(capabilities, "cnpgEligibleState", node.capabilities.cnpg_eligible_state) reasons = ET.SubElement(root, "reasons") for reason in node.capabilities.reasons: _set_text(reasons, "reason", reason) if node.host_inventory_error: _set_text(reasons, "reason", node.host_inventory_error) inventory = ET.SubElement(root, "hostInventory") _set_text(inventory, "available", "true" if node.host_inventory_available else "false") _set_text(inventory, "error", node.host_inventory_error or "") return root def _cluster_to_xml(cluster: ClusterTopology) -> ET.Element: root = ET.Element("clusterTopology", {"version": TOPOLOGY_SCHEMA_VERSION}) identity = ET.SubElement(root, "identity") _set_text(identity, "generatedAt", cluster.generated_at) _set_text(identity, "generatedEpochMs", str(cluster.generated_epoch_ms)) _set_text(identity, "collectionStatus", cluster.collection_status) _set_text(identity, "topologyRoot", cluster.topology_root) summary = ET.SubElement(root, "summary") _set_text(summary, "discoveredNodes", str(cluster.discovered_nodes)) _set_text(summary, "readyNodes", str(cluster.ready_nodes)) _set_text(summary, "monitoringEligibleNodes", str(cluster.monitoring_eligible_nodes)) _set_text(summary, "cnpgEligibleNodes", str(cluster.cnpg_eligible_nodes)) required = ET.SubElement(root, "requiredMountpoints") for mp in cluster.required_monitoring_mountpoints: _set_text(required, "mountpoint", mp) _set_text(required, "monitoringMinFreeBytes", str(cluster.monitoring_min_free_bytes)) nodes = ET.SubElement(root, "nodes") for node in cluster.nodes: ref = ET.SubElement(nodes, "nodeRef") _set_text(ref, "name", node.name) _set_text(ref, "path", f"nodes/{node.name}.xml") messages = ET.SubElement(root, "messages") for msg in cluster.messages: _set_text(messages, "message", msg) return root def _write_xml(path: Path, element: ET.Element) -> None: path.parent.mkdir(parents=True, exist_ok=True) tree = ET.ElementTree(element) ET.indent(tree, space=" ") tree.write(path, encoding="utf-8", xml_declaration=True) def save_topology_xml(cluster: ClusterTopology, output_root: Path) -> tuple[Path, dict[str, Path]]: root = Path(output_root) nodes_dir = root / "nodes" cluster_path = root / "cluster.xml" written_nodes: dict[str, Path] = {} for node in cluster.nodes: node_path = nodes_dir / f"{node.name}.xml" _write_xml(node_path, _node_to_xml(node)) written_nodes[node.name] = node_path _write_xml(cluster_path, _cluster_to_xml(cluster)) return cluster_path, written_nodes def discover_cluster_topology( *, kubectl_base_cmd: Sequence[str], topology_root: Path, config: TopologyDiscoveryConfig | None = None, progress_cb: Callable[[str], None] | None = None, env: dict[str, str] | None = None, ) -> TopologyDiscoveryResult: cfg = config or TopologyDiscoveryConfig() required_mountpoints = tuple(cfg.required_monitoring_mountpoints) monitoring_min_free_bytes = int(cfg.monitoring_min_free_bytes) messages: list[str] = [] def _progress(message: str) -> None: messages.append(message) if progress_cb: try: progress_cb(message) except Exception: pass _progress("Starting topology discovery") _progress("Discovering Kubernetes nodes...") try: nodes_payload = _kubectl_json(kubectl_base_cmd, ["get", "nodes"], env=env) k8s_nodes = _parse_k8s_nodes(nodes_payload) except Exception as exc: now = datetime.now(timezone.utc) cluster = ClusterTopology( version=TOPOLOGY_SCHEMA_VERSION, generated_at=now.isoformat(), generated_epoch_ms=int(now.timestamp() * 1000), collection_status="failed", required_monitoring_mountpoints=required_mountpoints, monitoring_min_free_bytes=monitoring_min_free_bytes, nodes=tuple(), topology_root=str(topology_root), messages=tuple(messages + [f"Topology discovery failed to query nodes: {exc}"]), ) return TopologyDiscoveryResult( topology=cluster, collector_applied=False, expected_ready_nodes=tuple(), reported_nodes=tuple(), missing_nodes=tuple(), ) discovered = len(k8s_nodes) ready_nodes = sorted(name for name, data in k8s_nodes.items() if data["ready"]) _progress(f"Discovered {discovered} Kubernetes nodes ({len(ready_nodes)} Ready)") collector_name = f"prole-topology-{secrets.token_hex(4)}" inventory_payloads: dict[str, dict] = {} inventory_errors: dict[str, str] = {} collector_applied = False if ready_nodes: _progress("Creating temporary in-cluster topology collector") manifest = _collector_manifest(collector_name, image=cfg.collector_image) res_apply = _run_kubectl( kubectl_base_cmd, ["-n", cfg.namespace, "apply", "-f", "-"], timeout=20, env=env, input_text=json.dumps(manifest, separators=(",", ":"), sort_keys=True), ) collector_applied = res_apply.returncode == 0 if not collector_applied: stderr = (res_apply.stderr or "").strip() or "unknown_error" _progress(f"Topology collector deployment failed: {stderr}") else: _progress(f"Waiting for inventory reports from Ready nodes ({len(ready_nodes)} total)") deadline = time.monotonic() + max(5, int(cfg.timeout_seconds)) collected_for: set[str] = set() while time.monotonic() < deadline and len(collected_for) < len(ready_nodes): try: pods_payload = _kubectl_json( kubectl_base_cmd, [ "-n", cfg.namespace, "get", "pods", "-l", f"prole.io/topology-collector={collector_name}", ], timeout=15, env=env, ) except Exception: time.sleep(max(0.5, cfg.poll_interval_seconds)) continue pod_by_node: dict[str, str] = {} for item in pods_payload.get("items") or []: spec = item.get("spec") or {} metadata = item.get("metadata") or {} pod_name = str(metadata.get("name") or "").strip() node_name = str(spec.get("nodeName") or "").strip() if pod_name and node_name: pod_by_node[node_name] = pod_name for node_name in ready_nodes: if node_name in collected_for: continue pod_name = pod_by_node.get(node_name) if not pod_name: continue res_logs = _run_kubectl( kubectl_base_cmd, ["-n", cfg.namespace, "logs", pod_name, "--tail=-1"], timeout=15, env=env, ) if res_logs.returncode != 0: continue payload = _extract_inventory_from_log(res_logs.stdout or "") if not payload: continue inventory_payloads[node_name] = payload collected_for.add(node_name) _progress(f"Inventory received from node {node_name}") if len(collected_for) >= len(ready_nodes): break parsed = len(collected_for) _progress(f"Parsed topology from {parsed} of {len(ready_nodes)} Ready nodes") time.sleep(max(0.5, cfg.poll_interval_seconds)) missing = sorted(set(ready_nodes) - set(inventory_payloads.keys())) for node_name in missing: inventory_errors[node_name] = "timed_out" _progress(f"Timed out waiting for node {node_name}") if collector_applied: _run_kubectl( kubectl_base_cmd, [ "-n", cfg.namespace, "delete", "daemonset", collector_name, "--ignore-not-found=true", "--wait=false", ], timeout=15, env=env, ) merged_nodes: list[NodeTopology] = [] for node_name in sorted(k8s_nodes.keys()): inv = inventory_payloads.get(node_name) error = inventory_errors.get(node_name) merged_nodes.append( _merge_node( k8s_nodes[node_name], inv, required_monitoring_mountpoints=required_mountpoints, monitoring_min_free_bytes=monitoring_min_free_bytes, inventory_error=error, ) ) if merged_nodes and all(not node.host_inventory_available for node in merged_nodes): status = "failed" elif any(not node.host_inventory_available for node in merged_nodes): status = "partial" else: status = "complete" if status == "partial": missing = [node.name for node in merged_nodes if not node.host_inventory_available and node.ready] if missing: _progress( "Continuing with partial topology because " f"{len(missing)} node(s) did not report in time" ) elif status == "failed": _progress("Topology collection unavailable, continuing with Kubernetes-only node facts") _progress("Writing topology XML files") now = datetime.now(timezone.utc) cluster = ClusterTopology( version=TOPOLOGY_SCHEMA_VERSION, generated_at=now.isoformat(), generated_epoch_ms=int(now.timestamp() * 1000), collection_status=status, required_monitoring_mountpoints=required_mountpoints, monitoring_min_free_bytes=monitoring_min_free_bytes, nodes=tuple(merged_nodes), topology_root=str(topology_root), messages=tuple(messages), ) save_topology_xml(cluster, topology_root) _progress(f"Topology snapshot saved to {topology_root}") cluster = ClusterTopology( version=cluster.version, generated_at=cluster.generated_at, generated_epoch_ms=cluster.generated_epoch_ms, collection_status=cluster.collection_status, required_monitoring_mountpoints=cluster.required_monitoring_mountpoints, monitoring_min_free_bytes=cluster.monitoring_min_free_bytes, nodes=cluster.nodes, topology_root=cluster.topology_root, messages=tuple(messages), ) return TopologyDiscoveryResult( topology=cluster, collector_applied=collector_applied, expected_ready_nodes=tuple(ready_nodes), reported_nodes=tuple(sorted(inventory_payloads.keys())), missing_nodes=tuple(sorted(set(ready_nodes) - set(inventory_payloads.keys()))), )