From 4ff8ab74bdf4424e07fecd8f8a2bb931c66c28d4 Mon Sep 17 00:00:00 2001 From: Richard Date: Sat, 25 Jul 2026 16:51:46 +0200 Subject: [PATCH] disk health report --- disk-health-report.py | 626 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 626 insertions(+) create mode 100755 disk-health-report.py diff --git a/disk-health-report.py b/disk-health-report.py new file mode 100755 index 0000000..9bbde55 --- /dev/null +++ b/disk-health-report.py @@ -0,0 +1,626 @@ +#!/usr/bin/env python3 +"""Disk-health / node-removal-safety report for a Gallium cluster. + +The question this answers: **is it safe to remove a node** (permanently, or +temporarily for maintenance)? A node is unsafe to remove when some volume's +last healthy copy of its data lives only on that node -- removing it would take +that data offline (permanently, or for the duration of the maintenance). + +Data model (all read live via `kubectl`): + + * KubeVirt VirtualMachines / VirtualMachineInstances -- the VMs and whether + they're running, and which node each runs on. + * The PVCs each VM references (directly, via DataVolume, or hot-plugged). + * Longhorn Volumes + Replicas -- for every PVC's underlying volume, where its + replicas live (which node) and whether each replica is healthy. + * Pods -- to attribute PVCs that aren't owned by a VM to a non-VM workload + vs. nothing at all. + +The report has three parts: + + 1. A per-node removal-safety summary (the headline -- SAFE / UNSAFE). + 2. Per-VM detail: the VM, its running state, and for each of its PVCs a + table of (replica, node, health). + 3. PVCs not attached to any VM, split into "attached to a non-VM pod" and + "completely detached". + +Self-contained: standard library only; the sole dependency is a `kubectl` +binary that can reach the cluster. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from collections import defaultdict +from dataclasses import dataclass, field + +LONGHORN_NS = "longhorn-system" +LONGHORN_CSI_DRIVER = "driver.longhorn.io" + + +# --------------------------------------------------------------------------- # +# kubectl plumbing +# --------------------------------------------------------------------------- # +class Kubectl: + """Thin wrapper that shells out to kubectl and parses JSON.""" + + def __init__(self, binary: str, kubeconfig: str | None, context: str | None): + self.base = [binary] + if kubeconfig: + self.base += ["--kubeconfig", kubeconfig] + if context: + self.base += ["--context", context] + + def get(self, resource: str, namespace: str | None = None) -> list[dict]: + """`kubectl get -o json` -> list of items. + + Returns [] (with a warning) when the resource type isn't present -- e.g. + a cluster without KubeVirt or Longhorn installed -- rather than aborting + the whole report. + """ + cmd = list(self.base) + ["get", resource, "-o", "json"] + cmd += ["-n", namespace] if namespace else ["-A"] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + err = proc.stderr.strip() + low = err.lower() + if "the server doesn't have a resource type" in low or "could not find" in low: + warn(f"resource {resource!r} not found on the cluster; skipping ({err})") + return [] + die(f"kubectl get {resource} failed: {err}") + try: + return json.loads(proc.stdout).get("items", []) + except json.JSONDecodeError as exc: + die(f"could not parse kubectl output for {resource}: {exc}") + return [] + + +# --------------------------------------------------------------------------- # +# colour / output helpers +# --------------------------------------------------------------------------- # +class C: + """ANSI colour codes; blanked out when colour is disabled.""" + + RESET = BOLD = DIM = RED = GREEN = YELLOW = BLUE = CYAN = "" + + @classmethod + def enable(cls): + cls.RESET = "\033[0m" + cls.BOLD = "\033[1m" + cls.DIM = "\033[2m" + cls.RED = "\033[31m" + cls.GREEN = "\033[32m" + cls.YELLOW = "\033[33m" + cls.BLUE = "\033[34m" + cls.CYAN = "\033[36m" + + +def warn(msg: str): + print(f"{C.YELLOW}warning:{C.RESET} {msg}", file=sys.stderr) + + +def die(msg: str): + print(f"{C.RED}error:{C.RESET} {msg}", file=sys.stderr) + sys.exit(1) + + +def render_table(headers: list[str], rows: list[list[str]], indent: str = " ") -> str: + """Render a simple fixed-width table. Colour codes are ignored for width.""" + import re + + ansi = re.compile(r"\033\[[0-9;]*m") + + def vlen(s: str) -> int: + return len(ansi.sub("", s)) + + widths = [vlen(h) for h in headers] + for row in rows: + for i, cell in enumerate(row): + widths[i] = max(widths[i], vlen(cell)) + + def fmt(row: list[str]) -> str: + cells = [] + for i, cell in enumerate(row): + cells.append(cell + " " * (widths[i] - vlen(cell))) + return indent + " ".join(cells).rstrip() + + sep = indent + " ".join("-" * w for w in widths) + out = [fmt([f"{C.BOLD}{h}{C.RESET}" for h in headers]), sep] + out += [fmt(r) for r in rows] + return "\n".join(out) + + +# --------------------------------------------------------------------------- # +# domain types +# --------------------------------------------------------------------------- # +HEALTHY = "healthy" + + +@dataclass +class Replica: + name: str + node: str + health: str # display: healthy | stopped | rebuilding | failed | error | unknown + has_data: bool # holds a usable copy of the data (healthyAt set, not failed) + + +@dataclass +class Volume: + """A Longhorn volume backing one PVC (or a non-Longhorn PVC's stand-in).""" + + pvc_ns: str + pvc_name: str + lh_name: str | None # Longhorn volume name (== bound PV name) + storage_class: str + robustness: str # healthy | degraded | faulted | unknown | - + state: str # attached | detached | - + desired_replicas: int | None + replicas: list[Replica] = field(default_factory=list) + note: str = "" # e.g. non-Longhorn, unbound, missing + + @property + def data_nodes(self) -> set[str]: + """Nodes that hold a usable copy of this volume's data. + + Uses data validity (healthyAt && !failedAt), NOT engine-running state: + a cleanly-detached volume (stopped VM) still has good copies on its + replica nodes, so losing one of those nodes still loses data. + """ + return {r.node for r in self.replicas if r.has_data and r.node} + + @property + def all_nodes(self) -> set[str]: + return {r.node for r in self.replicas if r.node} + + +@dataclass +class VM: + ns: str + name: str + status: str # Running | Stopped | ... + node: str | None # node the VMI runs on + pvcs: list[tuple[str, str]] # (ns, name) referenced by this VM + + @property + def running(self) -> bool: + return self.status == "Running" + + +# --------------------------------------------------------------------------- # +# building the model +# --------------------------------------------------------------------------- # +def classify_replica(rep: dict) -> tuple[str, bool]: + """Return (display-health, has_data) for a Longhorn Replica CR. + + has_data is the data-safety signal: the replica holds a usable copy iff it + has been healthy (`healthyAt`) and hasn't failed (`failedAt`) -- true even + when the engine is stopped because the volume is detached. + """ + spec = rep.get("spec", {}) + status = rep.get("status", {}) + failed = bool(spec.get("failedAt")) + healthy_at = bool(spec.get("healthyAt")) + cur = status.get("currentState", "") + has_data = healthy_at and not failed + if failed: + disp = "failed" + elif cur == "running": + disp = HEALTHY if healthy_at else "rebuilding" + elif healthy_at: + disp = "stopped" # good copy, engine down (volume detached) + elif cur in ("stopped", "error"): + disp = cur + else: + disp = cur or "unknown" + return disp, has_data + + +def vm_referenced_pvcs(vm_spec_volumes: list[dict], ns: str) -> list[tuple[str, str]]: + """PVC (ns, name) pairs a set of KubeVirt `volumes[]` entries reference.""" + out: list[tuple[str, str]] = [] + for v in vm_spec_volumes or []: + pvc = v.get("persistentVolumeClaim") + if pvc and pvc.get("claimName"): + out.append((ns, pvc["claimName"])) + continue + # A DataVolume materialises a PVC with the same name in the same ns. + dv = v.get("dataVolume") + if dv and dv.get("name"): + out.append((ns, dv["name"])) + return out + + +def is_virt_launcher(pod: dict) -> bool: + labels = pod.get("metadata", {}).get("labels", {}) or {} + if labels.get("kubevirt.io") == "virt-launcher": + return True + for ref in pod.get("metadata", {}).get("ownerReferences", []) or []: + if ref.get("kind") == "VirtualMachineInstance": + return True + return False + + +def virt_launcher_vm_name(pod: dict) -> str: + """The VM/VMI name a virt-launcher pod belongs to (VMI name == VM name).""" + for ref in pod.get("metadata", {}).get("ownerReferences", []) or []: + if ref.get("kind") == "VirtualMachineInstance" and ref.get("name"): + return ref["name"] + return (pod.get("metadata", {}).get("labels", {}) or {}).get("vm.kubevirt.io/name", "") + + +def build_volume(pvc_ns: str, pvc_name: str, pvc_index, pv_index, + lh_vol_index, replicas_by_vol) -> Volume: + pvc = pvc_index.get((pvc_ns, pvc_name)) + if pvc is None: + return Volume(pvc_ns, pvc_name, None, "-", "-", "-", None, + note="PVC not found") + sc = pvc.get("spec", {}).get("storageClassName", "") or "-" + pv_name = pvc.get("spec", {}).get("volumeName") or "" + phase = pvc.get("status", {}).get("phase", "") + + if not pv_name: + return Volume(pvc_ns, pvc_name, None, sc, "-", "-", None, + note=f"unbound PVC (phase={phase})") + + pv = pv_index.get(pv_name) + driver = "" + if pv: + driver = pv.get("spec", {}).get("csi", {}).get("driver", "") + if pv and driver and driver != LONGHORN_CSI_DRIVER: + # Non-Longhorn PV: its data lives wherever the PV binds (often one node). + node = _pv_pinned_node(pv) + note = f"non-Longhorn ({driver})" + vol = Volume(pvc_ns, pvc_name, pv_name, sc, "-", "-", None, note=note) + if node: + vol.replicas.append(Replica(pv_name, node, HEALTHY, has_data=True)) + return vol + + # Longhorn: the volume CR name equals the bound PV name. + lh = lh_vol_index.get(pv_name) + if lh is None: + return Volume(pvc_ns, pvc_name, pv_name, sc, "-", "-", None, + note="no Longhorn volume CR") + robustness = lh.get("status", {}).get("robustness", "unknown") or "unknown" + state = lh.get("status", {}).get("state", "unknown") or "unknown" + desired = lh.get("spec", {}).get("numberOfReplicas") + vol = Volume(pvc_ns, pvc_name, pv_name, sc, robustness, state, desired) + for rep in replicas_by_vol.get(pv_name, []): + # An empty nodeID means Longhorn couldn't schedule this replica (e.g. a + # 3rd replica on a 2-node cluster). Keep it "" so it never counts as a + # node that holds data, and render it as "(unscheduled)". + node = rep.get("spec", {}).get("nodeID", "") or "" + disp, has_data = classify_replica(rep) + vol.replicas.append(Replica( + name=rep.get("metadata", {}).get("name", "?"), + node=node, + health=disp, + has_data=has_data, + )) + vol.replicas.sort(key=lambda r: (r.node, r.name)) + return vol + + +def _pv_pinned_node(pv: dict) -> str | None: + """Best-effort node a non-Longhorn PV is pinned to (nodeAffinity).""" + terms = (pv.get("spec", {}).get("nodeAffinity", {}) + .get("required", {}).get("nodeSelectorTerms", [])) + for term in terms: + for expr in term.get("matchExpressions", []): + if expr.get("key") == "kubernetes.io/hostname" and expr.get("values"): + return expr["values"][0] + return None + + +# --------------------------------------------------------------------------- # +# health formatting +# --------------------------------------------------------------------------- # +def colour_health(health: str) -> str: + good = {HEALTHY, "attached"} + warnish = {"rebuilding", "degraded", "stopped", "detached"} + bad = {"failed", "faulted", "error", "unknown"} + if health in good: + return f"{C.GREEN}{health}{C.RESET}" + if health in warnish: + return f"{C.YELLOW}{health}{C.RESET}" + if health in bad: + return f"{C.RED}{health}{C.RESET}" + return health + + +def colour_status(status: str) -> str: + if status == "Running": + return f"{C.GREEN}{status}{C.RESET}" + if status in ("Stopped", "Paused"): + return f"{C.YELLOW}{status}{C.RESET}" + return f"{C.CYAN}{status}{C.RESET}" + + +# --------------------------------------------------------------------------- # +# main report +# --------------------------------------------------------------------------- # +def main(): + ap = argparse.ArgumentParser( + description="Report Longhorn disk health and assess node-removal safety.") + ap.add_argument("--kubectl", default="kubectl", + help="kubectl binary to use (default: kubectl on PATH)") + ap.add_argument("--kubeconfig", help="path to kubeconfig (else KUBECONFIG/default)") + ap.add_argument("--context", help="kube context to use") + ap.add_argument("--json", action="store_true", + help="emit the structured model as JSON instead of a report") + ap.add_argument("--no-color", action="store_true", help="disable ANSI colour") + args = ap.parse_args() + + if not args.no_color and sys.stdout.isatty(): + C.enable() + + kc = Kubectl(args.kubectl, args.kubeconfig, args.context) + + # --- fetch everything up front ---------------------------------------- # + vms_raw = kc.get("virtualmachines.kubevirt.io") + vmis_raw = kc.get("virtualmachineinstances.kubevirt.io") + pvcs_raw = kc.get("persistentvolumeclaims") + pvs_raw = kc.get("persistentvolumes") + pods_raw = kc.get("pods") + lh_vols_raw = kc.get("volumes.longhorn.io", LONGHORN_NS) + lh_reps_raw = kc.get("replicas.longhorn.io", LONGHORN_NS) + + pvc_index = {(p["metadata"]["namespace"], p["metadata"]["name"]): p for p in pvcs_raw} + pv_index = {p["metadata"]["name"]: p for p in pvs_raw} + lh_vol_index = {v["metadata"]["name"]: v for v in lh_vols_raw} + replicas_by_vol: dict[str, list[dict]] = defaultdict(list) + for rep in lh_reps_raw: + vol_name = rep.get("spec", {}).get("volumeName") + if vol_name: + replicas_by_vol[vol_name].append(rep) + + # VMI node + status lookup, keyed by (ns, name). + vmi_index = {(v["metadata"]["namespace"], v["metadata"]["name"]): v for v in vmis_raw} + + # --- assemble VMs ------------------------------------------------------ # + vms: list[VM] = [] + for raw in vms_raw: + ns = raw["metadata"]["namespace"] + name = raw["metadata"]["name"] + status = raw.get("status", {}).get("printableStatus", "") or "Unknown" + spec_vols = (raw.get("spec", {}).get("template", {}) + .get("spec", {}).get("volumes", [])) + pvcs = vm_referenced_pvcs(spec_vols, ns) + node = None + vmi = vmi_index.get((ns, name)) + if vmi: + node = vmi.get("status", {}).get("nodeName") + # Include hot-plugged / VMI-level volumes too. + for extra in vm_referenced_pvcs(vmi.get("spec", {}).get("volumes", []), ns): + if extra not in pvcs: + pvcs.append(extra) + vms.append(VM(ns, name, status, node, pvcs)) + vms.sort(key=lambda v: (v.ns, v.name)) + vm_by_key = {(vm.ns, vm.name): vm for vm in vms} + + # --- pods that mount PVCs --------------------------------------------- # + # A running VM's virt-launcher pod mounts every volume the VM actually uses + # -- including ones KubeVirt manages outside spec.template.spec.volumes (the + # persistent-state/EFI-vars PVC, hot-plugged disks). Attribute those to the + # owning VM so they show under the VM, not as "detached". Other pods are + # real non-VM consumers. + pods_by_pvc: dict[tuple[str, str], list[tuple[str, str]]] = defaultdict(list) + for pod in pods_raw: + pns = pod["metadata"]["namespace"] + pnode = pod.get("spec", {}).get("nodeName", "") or "?" + pname = pod["metadata"]["name"] + claims = [v["persistentVolumeClaim"]["claimName"] + for v in pod.get("spec", {}).get("volumes", []) or [] + if v.get("persistentVolumeClaim", {}).get("claimName")] + if is_virt_launcher(pod): + vm = vm_by_key.get((pns, virt_launcher_vm_name(pod))) + if vm is not None: + for c in claims: + if (pns, c) not in vm.pvcs: + vm.pvcs.append((pns, c)) + continue # a VM's own pod; its PVCs belong to the VM section + for c in claims: + pods_by_pvc[(pns, c)].append((pname, pnode)) + + # Set of PVCs owned by a VM (after folding in virt-launcher-mounted ones). + vm_pvc_set: set[tuple[str, str]] = {p for vm in vms for p in vm.pvcs} + + # --- build volumes for every PVC -------------------------------------- # + def vol_for(ns_name: tuple[str, str]) -> Volume: + return build_volume(ns_name[0], ns_name[1], pvc_index, pv_index, + lh_vol_index, replicas_by_vol) + + all_pvc_keys = set(pvc_index.keys()) | vm_pvc_set + volumes = {k: vol_for(k) for k in all_pvc_keys} + + # --- node inventory ---------------------------------------------------- # + nodes = sorted({r.node for v in volumes.values() for r in v.replicas if r.node}) + + if args.json: + emit_json(vms, volumes, pods_by_pvc, vm_pvc_set, nodes) + return + + print_node_safety(nodes, vms, volumes) + print_vm_detail(vms, volumes) + print_unattached(volumes, vm_pvc_set, pods_by_pvc) + + +def _risk_line(vol: Volume) -> str: + lh = vol.lh_name or "-" + hn = ", ".join(sorted(vol.data_nodes)) or "(none)" + return (f"{vol.pvc_ns}/{vol.pvc_name} [{lh}] " + f"robustness={vol.robustness} usable-copies-on: {hn}") + + +def print_node_safety(nodes, vms: list[VM], volumes: dict): + hdr = f"{C.BOLD}=== NODE REMOVAL SAFETY ==={C.RESET}" + print(hdr) + if not nodes: + print(" no replica-bearing nodes found (no Longhorn volumes?).\n") + return + + # Which VM (running?) owns each PVC, for severity. + running_pvcs = {p for vm in vms if vm.running for p in vm.pvcs} + + # Volumes already unavailable (no usable copy on any node) -- independent of + # which node you remove. Reported once, globally, not per node. + broken = [vol for vol in volumes.values() if vol.replicas and not vol.data_nodes] + + for node in nodes: + crit, warns = [], [] + for key, vol in volumes.items(): + if not vol.replicas or not vol.data_nodes: + continue + if vol.data_nodes == {node}: + # Removing this node drops the last usable copy. + (crit if key in running_pvcs else warns).append(vol) + + if not crit and not warns: + print(f" {C.GREEN}SAFE{C.RESET} {C.BOLD}{node}{C.RESET}: " + f"every volume keeps a usable copy on another node.") + continue + + verdict = f"{C.RED}UNSAFE{C.RESET}" if crit else f"{C.YELLOW}CAUTION{C.RESET}" + print(f" {verdict} {C.BOLD}{node}{C.RESET}:") + if crit: + print(f" {C.RED}running-VM data lost if removed ({len(crit)}):{C.RESET}") + for vol in crit: + print(f" - {_risk_line(vol)}") + if warns: + print(f" {C.YELLOW}idle/non-VM data lost if removed ({len(warns)}):{C.RESET}") + for vol in warns: + print(f" - {_risk_line(vol)}") + + if broken: + print(f"\n {C.RED}Already unavailable{C.RESET} " + f"(no usable copy on ANY node -- fix before draining anything):") + for vol in broken: + print(f" - {_risk_line(vol)}") + print() + + +def _pvc_block(vol: Volume) -> str: + head = (f" PVC {C.CYAN}{vol.pvc_ns}/{vol.pvc_name}{C.RESET}" + f" vol={vol.lh_name or '-'} sc={vol.storage_class}") + meta = [] + if vol.robustness != "-": + meta.append(f"robustness={colour_health(vol.robustness)}") + if vol.state != "-": + meta.append(f"state={colour_health(vol.state)}") + if vol.desired_replicas is not None: + meta.append(f"desired-replicas={vol.desired_replicas}") + if meta: + head += " (" + " ".join(meta) + ")" + if vol.note: + head += f" {C.DIM}[{vol.note}]{C.RESET}" + lines = [head] + if vol.replicas: + rows = [[r.name, r.node or "(unscheduled)", colour_health(r.health)] + for r in vol.replicas] + lines.append(render_table(["REPLICA", "NODE", "HEALTH"], rows, indent=" ")) + elif not vol.note: + lines.append(f" {C.DIM}(no replicas reported){C.RESET}") + return "\n".join(lines) + + +def print_vm_detail(vms: list[VM], volumes: dict): + print(f"{C.BOLD}=== VIRTUAL MACHINES ==={C.RESET}") + if not vms: + print(" no VirtualMachines found.\n") + return + for vm in vms: + node = f" node={vm.node}" if vm.node else "" + print(f"\n{C.BOLD}VM {vm.ns}/{vm.name}{C.RESET} " + f"[{colour_status(vm.status)}]{node}") + if not vm.pvcs: + print(f" {C.DIM}(no PVC-backed volumes){C.RESET}") + continue + for key in vm.pvcs: + print(_pvc_block(volumes[key])) + print() + + +def print_unattached(volumes: dict, vm_pvc_set, pods_by_pvc): + non_vm = {k: v for k, v in volumes.items() if k not in vm_pvc_set} + pod_attached = {k: v for k, v in non_vm.items() if pods_by_pvc.get(k)} + detached = {k: v for k, v in non_vm.items() if not pods_by_pvc.get(k)} + + print(f"{C.BOLD}=== PVCs NOT ATTACHED TO A VM ==={C.RESET}") + if not non_vm: + print(" none: every PVC is attached to a VM.\n") + return + + if pod_attached: + print(f"\n {C.BOLD}Attached to non-VM pod(s):{C.RESET}") + for key in sorted(pod_attached): + consumers = ", ".join(f"{n} (node={nd})" for n, nd in pods_by_pvc[key]) + print(f" used by: {consumers}") + print(_pvc_block(pod_attached[key])) + + if detached: + print(f"\n {C.BOLD}Completely detached (no consumer):{C.RESET}") + for key in sorted(detached): + print(_pvc_block(detached[key])) + print() + + +def emit_json(vms, volumes, pods_by_pvc, vm_pvc_set, nodes): + def vol_dict(v: Volume): + return { + "pvc_namespace": v.pvc_ns, + "pvc_name": v.pvc_name, + "longhorn_volume": v.lh_name, + "storage_class": v.storage_class, + "robustness": v.robustness, + "state": v.state, + "desired_replicas": v.desired_replicas, + "note": v.note or None, + "replicas": [ + {"name": r.name, "node": r.node, "health": r.health, + "has_data": r.has_data} + for r in v.replicas + ], + "usable_copy_nodes": sorted(v.data_nodes), + } + + running_pvcs = {p for vm in vms if vm.running for p in vm.pvcs} + node_safety = {} + for node in nodes: + crit, warns = [], [] + for key, vol in volumes.items(): + if vol.replicas and vol.data_nodes == {node}: + (crit if key in running_pvcs else warns).append( + f"{vol.pvc_ns}/{vol.pvc_name}") + node_safety[node] = { + "safe": not crit and not warns, + "running_vm_data_only_here": crit, + "other_data_only_here": warns, + } + + out = { + "nodes": nodes, + "node_safety": node_safety, + "vms": [ + { + "namespace": vm.ns, "name": vm.name, "status": vm.status, + "running": vm.running, "node": vm.node, + "pvcs": [f"{ns}/{n}" for ns, n in vm.pvcs], + } + for vm in vms + ], + "volumes": [vol_dict(v) for v in volumes.values()], + "unattached": { + f"{k[0]}/{k[1]}": { + "pods": pods_by_pvc.get(k, []), + "detached": not pods_by_pvc.get(k), + } + for k in volumes if k not in vm_pvc_set + }, + } + print(json.dumps(out, indent=2)) + + +if __name__ == "__main__": + main()