mirror of
https://github.com/dredx/prole.git
synced 2026-09-27 21:34:30 +00:00
cf51335ced
- detect local k3s node kubeconfig and skip kubectx/use-context mutation when already targeting local API\n- add configurable KUBE_CONTEXT_NAME resolution with compatibility fallbacks and switch only when required\n- update init scripts to use ensure_kube_context helper naming\n- broaden monitoring eligibility to discovered /synology/d### mounts so /synology/d004 qualifies\n- add focused kube-context and topology tests covering local/remote and read-only kubeconfig cases Co-authored-by: Junie <junie@jetbrains.com>
1112 lines
38 KiB
Python
1112 lines
38 KiB
Python
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:
|
|
min_available = min(
|
|
(
|
|
fs.available_bytes
|
|
for fs in filesystems
|
|
if fs.mountpoint in matching_mountpoints and fs.available_bytes is not None
|
|
),
|
|
default=None,
|
|
)
|
|
if min_available is None:
|
|
monitoring_eligible = None
|
|
monitoring_eligible_state = "unavailable"
|
|
reasons.add("topology_unavailable")
|
|
elif min_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()))),
|
|
)
|