node maint report

node-maintenance-report.py

@@ -0,0 +1,491 @@
+#!/usr/bin/env python3
+"""Node maintenance / shutdown plan for a Gallium cluster.
+
+You want to take a node down -- for temporary/emergency maintenance, or to
+remove it for good. This answers: **what must I shut down (or migrate) first,
+and what data lives only on that node?**
+
+For each node it reports, in priority order:
+
+  1. Shut down cleanly FIRST -- VMs that have a volume whose only usable copy is
+     on this node. Taking the node down pulls that data out from under them: the
+     data goes offline for the maintenance window, and is lost for good if the
+     node is removed permanently. Stop these VMs cleanly (so the volume detaches
+     safely) -- and replicate/move the data first if the removal is permanent.
+  2. Running on this node -- VMs whose compute is here but whose data is safe on
+     other nodes. Taking the node down stops them ungracefully; stop or migrate
+     them for clean maintenance. (They can restart elsewhere; no data loss.)
+  3. Other data only on this node -- PVCs with their only copy here that aren't
+     owned by a VM (a non-VM pod's data, or a detached volume). Nothing to "shut
+     down", but this data is offline during maintenance / lost on removal.
+
+A node with none of the above is safe to take down as-is.
+
+Self-contained: standard library only (Python 3.6+); the sole dependency is a
+`kubectl` binary that can reach the cluster -- so it runs unchanged on a cluster
+node (SLES / Python 3.6) or a workstation.
+"""
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+from collections import defaultdict
+
+LONGHORN_NS = "longhorn-system"
+LONGHORN_CSI_DRIVER = "driver.longhorn.io"
+HEALTHY = "healthy"
+
+
+# --------------------------------------------------------------------------- #
+# kubectl plumbing
+# --------------------------------------------------------------------------- #
+class Kubectl(object):
+    """Thin wrapper that shells out to kubectl and parses JSON."""
+
+    def __init__(self, binary, kubeconfig, context):
+        self.base = [binary]
+        if kubeconfig:
+            self.base += ["--kubeconfig", kubeconfig]
+        if context:
+            self.base += ["--context", context]
+
+    def get(self, resource, namespace=None):
+        """`kubectl get <resource> -o json` -> list of items (or [] if absent)."""
+        cmd = list(self.base) + ["get", resource, "-o", "json"]
+        cmd += ["-n", namespace] if namespace else ["-A"]
+        proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                              universal_newlines=True)
+        if proc.returncode != 0:
+            err = (proc.stderr or "").strip()
+            low = err.lower()
+            if "the server doesn't have a resource type" in low or "could not find" in low:
+                warn("resource %r not found on the cluster; skipping (%s)" % (resource, err))
+                return []
+            die("kubectl get %s failed: %s" % (resource, err))
+        try:
+            return json.loads(proc.stdout).get("items", [])
+        except ValueError as exc:
+            die("could not parse kubectl output for %s: %s" % (resource, exc))
+        return []
+
+
+# --------------------------------------------------------------------------- #
+# colour / output helpers
+# --------------------------------------------------------------------------- #
+class C(object):
+    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):
+    print("%swarning:%s %s" % (C.YELLOW, C.RESET, msg), file=sys.stderr)
+
+
+def die(msg):
+    print("%serror:%s %s" % (C.RED, C.RESET, msg), file=sys.stderr)
+    sys.exit(1)
+
+
+# --------------------------------------------------------------------------- #
+# domain types
+# --------------------------------------------------------------------------- #
+class Replica(object):
+    def __init__(self, name, node, health, has_data):
+        self.name = name
+        self.node = node          # "" when Longhorn couldn't schedule it
+        self.health = health
+        self.has_data = has_data  # holds a usable copy (healthyAt set, not failed)
+
+
+class Volume(object):
+    def __init__(self, pvc_ns, pvc_name, lh_name, storage_class, robustness,
+                 state, desired_replicas, replicas=None, note=""):
+        self.pvc_ns = pvc_ns
+        self.pvc_name = pvc_name
+        self.lh_name = lh_name
+        self.storage_class = storage_class
+        self.robustness = robustness
+        self.state = state
+        self.desired_replicas = desired_replicas
+        self.replicas = replicas if replicas is not None else []
+        self.note = note
+
+    @property
+    def data_nodes(self):
+        """Nodes holding a usable copy (healthyAt && !failedAt), regardless of
+        whether the engine is currently running."""
+        return set(r.node for r in self.replicas if r.has_data and r.node)
+
+    @property
+    def all_nodes(self):
+        return set(r.node for r in self.replicas if r.node)
+
+    @property
+    def label(self):
+        return "%s/%s" % (self.pvc_ns, self.pvc_name)
+
+
+class VM(object):
+    def __init__(self, ns, name, status, node, pvcs):
+        self.ns = ns
+        self.name = name
+        self.status = status
+        self.node = node        # node the VMI runs on (or None)
+        self.pvcs = pvcs
+
+    @property
+    def running(self):
+        return self.status == "Running"
+
+    @property
+    def label(self):
+        return "%s/%s" % (self.ns, self.name)
+
+
+# --------------------------------------------------------------------------- #
+# building the model (identical semantics to disk-health-report.py)
+# --------------------------------------------------------------------------- #
+def classify_replica(rep):
+    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"
+    elif cur in ("stopped", "error"):
+        disp = cur
+    else:
+        disp = cur or "unknown"
+    return disp, has_data
+
+
+def vm_referenced_pvcs(vm_spec_volumes, ns):
+    out = []
+    for v in vm_spec_volumes or []:
+        pvc = v.get("persistentVolumeClaim")
+        if pvc and pvc.get("claimName"):
+            out.append((ns, pvc["claimName"]))
+            continue
+        dv = v.get("dataVolume")
+        if dv and dv.get("name"):
+            out.append((ns, dv["name"]))
+    return out
+
+
+def is_virt_launcher(pod):
+    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):
+    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, pvc_name, pvc_index, pv_index, lh_vol_index, replicas_by_vol):
+    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="unbound PVC (phase=%s)" % 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:
+        node = _pv_pinned_node(pv)
+        vol = Volume(pvc_ns, pvc_name, pv_name, sc, "-", "-", None,
+                     note="non-Longhorn (%s)" % driver)
+        if node:
+            vol.replicas.append(Replica(pv_name, node, HEALTHY, True))
+        return vol
+
+    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, []):
+        node = rep.get("spec", {}).get("nodeID", "") or ""
+        disp, has_data = classify_replica(rep)
+        vol.replicas.append(Replica(
+            rep.get("metadata", {}).get("name", "?"), node, disp, has_data))
+    vol.replicas.sort(key=lambda r: (r.node, r.name))
+    return vol
+
+
+def _pv_pinned_node(pv):
+    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
+
+
+def build_model(kc):
+    """Fetch + assemble the shared model. Returns (vms, volumes, vm_by_pvc,
+    pods_by_pvc, nodes)."""
+    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 = dict(((p["metadata"]["namespace"], p["metadata"]["name"]), p) for p in pvcs_raw)
+    pv_index = dict((p["metadata"]["name"], p) for p in pvs_raw)
+    lh_vol_index = dict((v["metadata"]["name"], v) for v in lh_vols_raw)
+    replicas_by_vol = defaultdict(list)
+    for rep in lh_reps_raw:
+        vn = rep.get("spec", {}).get("volumeName")
+        if vn:
+            replicas_by_vol[vn].append(rep)
+    vmi_index = dict(((v["metadata"]["namespace"], v["metadata"]["name"]), v) for v in vmis_raw)
+
+    vms = []
+    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")
+            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 = dict(((vm.ns, vm.name), vm) for vm in vms)
+
+    # A running VM's virt-launcher pod mounts everything the VM uses, including
+    # the persistent-state/EFI-vars and hot-plug PVCs not in the VM spec.
+    pods_by_pvc = 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
+        for c in claims:
+            pods_by_pvc[(pns, c)].append((pname, pnode))
+
+    vm_pvc_set = set(p for vm in vms for p in vm.pvcs)
+    all_pvc_keys = set(pvc_index.keys()) | vm_pvc_set
+    volumes = dict((k, build_volume(k[0], k[1], pvc_index, pv_index,
+                                     lh_vol_index, replicas_by_vol)) for k in all_pvc_keys)
+
+    vm_by_pvc = {}
+    for vm in vms:
+        for key in vm.pvcs:
+            vm_by_pvc[key] = vm
+
+    nodes = sorted(set(r.node for v in volumes.values() for r in v.replicas if r.node))
+    return vms, volumes, vm_by_pvc, pods_by_pvc, nodes
+
+
+# --------------------------------------------------------------------------- #
+# node plan
+# --------------------------------------------------------------------------- #
+def plan_for_node(node, vms, volumes, vm_by_pvc, pods_by_pvc):
+    """Return the per-node plan dict for `node`."""
+    # Volumes whose ONLY usable copy is on this node, grouped by owner.
+    vm_only = defaultdict(list)      # VM -> [Volume only here]
+    pod_only = []                    # [(Volume, [(pod, node), ...])]
+    detached_only = []               # [Volume]
+    for key, vol in volumes.items():
+        if not vol.replicas or vol.data_nodes != set([node]):
+            continue
+        owner = vm_by_pvc.get(key)
+        if owner is not None:
+            vm_only[owner].append(vol)
+        elif pods_by_pvc.get(key):
+            pod_only.append((vol, pods_by_pvc[key]))
+        else:
+            detached_only.append(vol)
+
+    running_here = [vm for vm in vms if vm.node == node]
+    # VMs running here whose data is all safe elsewhere (not already in vm_only).
+    running_safe = [vm for vm in running_here if vm not in vm_only]
+    return {
+        "vm_only": vm_only,          # data-loss risk
+        "running_safe": running_safe,  # compute-only risk
+        "pod_only": pod_only,
+        "detached_only": detached_only,
+        "runs": len(running_here),
+    }
+
+
+def print_report(nodes, vms, volumes, vm_by_pvc, pods_by_pvc):
+    print("%s=== NODE SHUTDOWN / MAINTENANCE PLAN ===%s" % (C.BOLD, C.RESET))
+    if not nodes:
+        print("  no replica-bearing nodes found (no Longhorn volumes?).\n")
+        return
+    for node in nodes:
+        p = plan_for_node(node, vms, volumes, vm_by_pvc, pods_by_pvc)
+        vm_only, running_safe = p["vm_only"], p["running_safe"]
+        pod_only, detached_only = p["pod_only"], p["detached_only"]
+
+        header = "\n%sNODE %s%s  %s(runs %d VM%s)%s" % (
+            C.BOLD, node, C.RESET, C.DIM, p["runs"], "" if p["runs"] == 1 else "s", C.RESET)
+        print(header)
+
+        if not vm_only and not running_safe and not pod_only and not detached_only:
+            print("  %sSAFE to take down%s: nothing lives only here and no VMs run here."
+                  % (C.GREEN, C.RESET))
+            continue
+
+        running_risk = [vm for vm in vm_only if vm.running]
+        stopped_risk = [vm for vm in vm_only if not vm.running]
+
+        def only_here(vm):
+            return ", ".join(v.pvc_name for v in sorted(vm_only[vm], key=lambda v: v.pvc_name))
+
+        if running_risk:
+            print("  %sShut down cleanly FIRST -- running VMs whose only data copy is here:%s"
+                  % (C.RED, C.RESET))
+            for vm in sorted(running_risk, key=lambda v: v.label):
+                run = "runs here" if vm.node == node else ("runs on %s" % vm.node)
+                print("      - %s%s%s [%s]  (%s)  only-here: %s"
+                      % (C.BOLD, vm.label, C.RESET, colour_status(vm.status), run, only_here(vm)))
+
+        if running_safe:
+            print("  %sRunning on this node -- stop or migrate (data is safe elsewhere):%s"
+                  % (C.YELLOW, C.RESET))
+            for vm in sorted(running_safe, key=lambda v: v.label):
+                print("      - %s [%s]" % (vm.label, colour_status(vm.status)))
+
+        if stopped_risk or pod_only or detached_only:
+            print("  %sOnly here, no shutdown needed -- offline during maintenance, "
+                  "lost if removed permanently:%s" % (C.DIM, C.RESET))
+            for vm in sorted(stopped_risk, key=lambda v: v.label):
+                print("      - VM %s [%s]  only-here: %s"
+                      % (vm.label, colour_status(vm.status), only_here(vm)))
+            for vol, consumers in sorted(pod_only, key=lambda x: x[0].label):
+                who = ", ".join("%s" % n for n, _ in consumers)
+                print("      - %s  (used by pod: %s)" % (vol.label, who))
+            for vol in sorted(detached_only, key=lambda v: v.label):
+                print("      - %s  (detached)" % vol.label)
+    print()
+
+
+def colour_status(status):
+    if status == "Running":
+        return "%s%s%s" % (C.GREEN, status, C.RESET)
+    if status in ("Stopped", "Paused"):
+        return "%s%s%s" % (C.YELLOW, status, C.RESET)
+    return "%s%s%s" % (C.CYAN, status, C.RESET)
+
+
+def emit_json(nodes, vms, volumes, vm_by_pvc, pods_by_pvc):
+    out = {}
+    for node in nodes:
+        p = plan_for_node(node, vms, volumes, vm_by_pvc, pods_by_pvc)
+        vm_only = p["vm_only"]
+        running_risk = [vm for vm in vm_only if vm.running]
+        stopped_risk = [vm for vm in vm_only if not vm.running]
+        out[node] = {
+            "runs_vms": p["runs"],
+            "shut_down_first": [
+                {"vm": vm.label, "status": vm.status, "runs_on": vm.node,
+                 "volumes_only_here": sorted(v.pvc_name for v in vm_only[vm])}
+                for vm in sorted(running_risk, key=lambda v: v.label)
+            ],
+            "running_here_data_safe": [
+                {"vm": vm.label, "status": vm.status}
+                for vm in sorted(p["running_safe"], key=lambda v: v.label)
+            ],
+            "only_here_no_shutdown": (
+                [{"kind": "stopped-vm", "name": vm.label,
+                  "volumes_only_here": sorted(v.pvc_name for v in vm_only[vm])}
+                 for vm in sorted(stopped_risk, key=lambda v: v.label)]
+                + [{"kind": "pod", "name": vol.label, "consumer_pods": [n for n, _ in cons]}
+                   for vol, cons in p["pod_only"]]
+                + [{"kind": "detached", "name": vol.label} for vol in p["detached_only"]]
+            ),
+            "safe": not (vm_only or p["running_safe"] or p["pod_only"] or p["detached_only"]),
+        }
+    print(json.dumps(out, indent=2))
+
+
+def main():
+    ap = argparse.ArgumentParser(
+        description="Per-node maintenance plan: what lives only on a node and "
+                    "which VMs to shut down before taking it down.")
+    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("--node", help="report only this node (default: all nodes)")
+    ap.add_argument("--json", action="store_true", dest="as_json",
+                    help="emit the structured plan as JSON")
+    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)
+    vms, volumes, vm_by_pvc, pods_by_pvc, nodes = build_model(kc)
+
+    if args.node:
+        if args.node not in nodes:
+            die("node %r not found; known nodes: %s" % (args.node, ", ".join(nodes) or "(none)"))
+        nodes = [args.node]
+
+    if args.as_json:
+        emit_json(nodes, vms, volumes, vm_by_pvc, pods_by_pvc)
+    else:
+        print_report(nodes, vms, volumes, vm_by_pvc, pods_by_pvc)
+
+
+if __name__ == "__main__":
+    main()