diff --git a/disk-health-report.py b/disk-health-report.py index 9bbde55..1aa0162 100755 --- a/disk-health-report.py +++ b/disk-health-report.py @@ -3,7 +3,7 @@ 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 +last usable 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`): @@ -24,37 +24,37 @@ The report has three parts: 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. +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. """ -from __future__ import annotations - import argparse import json +import re import subprocess import sys from collections import defaultdict -from dataclasses import dataclass, field LONGHORN_NS = "longhorn-system" LONGHORN_CSI_DRIVER = "driver.longhorn.io" +HEALTHY = "healthy" # --------------------------------------------------------------------------- # # kubectl plumbing # --------------------------------------------------------------------------- # -class Kubectl: +class Kubectl(object): """Thin wrapper that shells out to kubectl and parses JSON.""" - def __init__(self, binary: str, kubeconfig: str | None, context: str | None): + 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: str, namespace: str | None = None) -> list[dict]: + def get(self, resource, namespace=None): """`kubectl get -o json` -> list of items. Returns [] (with a warning) when the resource type isn't present -- e.g. @@ -63,25 +63,26 @@ class Kubectl: """ cmd = list(self.base) + ["get", resource, "-o", "json"] cmd += ["-n", namespace] if namespace else ["-A"] - proc = subprocess.run(cmd, capture_output=True, text=True) + proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True) if proc.returncode != 0: - err = proc.stderr.strip() + 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(f"resource {resource!r} not found on the cluster; skipping ({err})") + warn("resource %r not found on the cluster; skipping (%s)" % (resource, err)) return [] - die(f"kubectl get {resource} failed: {err}") + die("kubectl get %s failed: %s" % (resource, err)) try: return json.loads(proc.stdout).get("items", []) - except json.JSONDecodeError as exc: - die(f"could not parse kubectl output for {resource}: {exc}") + except ValueError as exc: + die("could not parse kubectl output for %s: %s" % (resource, exc)) return [] # --------------------------------------------------------------------------- # # colour / output helpers # --------------------------------------------------------------------------- # -class C: +class C(object): """ANSI colour codes; blanked out when colour is disabled.""" RESET = BOLD = DIM = RED = GREEN = YELLOW = BLUE = CYAN = "" @@ -98,37 +99,36 @@ class C: cls.CYAN = "\033[36m" -def warn(msg: str): - print(f"{C.YELLOW}warning:{C.RESET} {msg}", file=sys.stderr) +def warn(msg): + print("%swarning:%s %s" % (C.YELLOW, C.RESET, msg), file=sys.stderr) -def die(msg: str): - print(f"{C.RED}error:{C.RESET} {msg}", file=sys.stderr) +def die(msg): + print("%serror:%s %s" % (C.RED, C.RESET, msg), file=sys.stderr) sys.exit(1) -def render_table(headers: list[str], rows: list[list[str]], indent: str = " ") -> str: +_ANSI = re.compile(r"\033\[[0-9;]*m") + + +def render_table(headers, rows, indent=" "): """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)) + def vlen(s): + 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: + def fmt(row): 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(["%s%s%s" % (C.BOLD, h, C.RESET) for h in headers]), sep] out += [fmt(r) for r in rows] return "\n".join(out) @@ -136,63 +136,61 @@ def render_table(headers: list[str], rows: list[list[str]], indent: str = " " # --------------------------------------------------------------------------- # # domain types # --------------------------------------------------------------------------- # -HEALTHY = "healthy" +class Replica(object): + def __init__(self, name, node, health, has_data): + self.name = name # replica CR name + self.node = node # "" when Longhorn couldn't schedule it + self.health = health # display: healthy|stopped|rebuilding|failed|error|unknown + self.has_data = has_data # holds a usable copy (healthyAt set, not failed) -@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: +class Volume(object): """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 + 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 # Longhorn volume name (== bound PV name) + self.storage_class = storage_class + self.robustness = robustness # healthy|degraded|faulted|unknown|- + self.state = state # attached|detached|- + self.desired_replicas = desired_replicas + self.replicas = replicas if replicas is not None else [] + self.note = note # e.g. non-Longhorn, unbound, missing @property - def data_nodes(self) -> set[str]: + def data_nodes(self): """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} + return set(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} + def all_nodes(self): + return set(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 +class VM(object): + def __init__(self, ns, name, status, node, pvcs): + self.ns = ns + self.name = name + self.status = status # Running | Stopped | ... + self.node = node # node the VMI runs on (or None) + self.pvcs = pvcs # list of (ns, name) referenced by this VM @property - def running(self) -> bool: + def running(self): return self.status == "Running" # --------------------------------------------------------------------------- # # building the model # --------------------------------------------------------------------------- # -def classify_replica(rep: dict) -> tuple[str, bool]: +def classify_replica(rep): """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 @@ -218,9 +216,9 @@ def classify_replica(rep: dict) -> tuple[str, bool]: return disp, has_data -def vm_referenced_pvcs(vm_spec_volumes: list[dict], ns: str) -> list[tuple[str, str]]: +def vm_referenced_pvcs(vm_spec_volumes, ns): """PVC (ns, name) pairs a set of KubeVirt `volumes[]` entries reference.""" - out: list[tuple[str, str]] = [] + out = [] for v in vm_spec_volumes or []: pvc = v.get("persistentVolumeClaim") if pvc and pvc.get("claimName"): @@ -233,7 +231,7 @@ def vm_referenced_pvcs(vm_spec_volumes: list[dict], ns: str) -> list[tuple[str, return out -def is_virt_launcher(pod: dict) -> bool: +def is_virt_launcher(pod): labels = pod.get("metadata", {}).get("labels", {}) or {} if labels.get("kubevirt.io") == "virt-launcher": return True @@ -243,7 +241,7 @@ def is_virt_launcher(pod: dict) -> bool: return False -def virt_launcher_vm_name(pod: dict) -> str: +def virt_launcher_vm_name(pod): """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"): @@ -251,19 +249,17 @@ def virt_launcher_vm_name(pod: dict) -> str: 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: +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") + 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})") + note="unbound PVC (phase=%s)" % phase) pv = pv_index.get(pv_name) driver = "" @@ -272,10 +268,10 @@ def build_volume(pvc_ns: str, pvc_name: str, pvc_index, pv_index, 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) + 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, has_data=True)) + vol.replicas.append(Replica(pv_name, node, HEALTHY, True)) return vol # Longhorn: the volume CR name equals the bound PV name. @@ -294,16 +290,12 @@ def build_volume(pvc_ns: str, pvc_name: str, pvc_index, pv_index, 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, - )) + 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: dict) -> str | None: +def _pv_pinned_node(pv): """Best-effort node a non-Longhorn PV is pinned to (nodeAffinity).""" terms = (pv.get("spec", {}).get("nodeAffinity", {}) .get("required", {}).get("nodeSelectorTerms", [])) @@ -317,25 +309,25 @@ def _pv_pinned_node(pv: dict) -> str | None: # --------------------------------------------------------------------------- # # health formatting # --------------------------------------------------------------------------- # -def colour_health(health: str) -> str: - good = {HEALTHY, "attached"} - warnish = {"rebuilding", "degraded", "stopped", "detached"} - bad = {"failed", "faulted", "error", "unknown"} +def colour_health(health): + good = (HEALTHY, "attached") + warnish = ("rebuilding", "degraded", "stopped", "detached") + bad = ("failed", "faulted", "error", "unknown") if health in good: - return f"{C.GREEN}{health}{C.RESET}" + return "%s%s%s" % (C.GREEN, health, C.RESET) if health in warnish: - return f"{C.YELLOW}{health}{C.RESET}" + return "%s%s%s" % (C.YELLOW, health, C.RESET) if health in bad: - return f"{C.RED}{health}{C.RESET}" + return "%s%s%s" % (C.RED, health, C.RESET) return health -def colour_status(status: str) -> str: +def colour_status(status): if status == "Running": - return f"{C.GREEN}{status}{C.RESET}" + return "%s%s%s" % (C.GREEN, status, C.RESET) if status in ("Stopped", "Paused"): - return f"{C.YELLOW}{status}{C.RESET}" - return f"{C.CYAN}{status}{C.RESET}" + return "%s%s%s" % (C.YELLOW, status, C.RESET) + return "%s%s%s" % (C.CYAN, status, C.RESET) # --------------------------------------------------------------------------- # @@ -348,7 +340,7 @@ def main(): 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", + ap.add_argument("--json", action="store_true", dest="as_json", 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() @@ -367,20 +359,19 @@ def main(): 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) + 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: 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} + vmi_index = dict(((v["metadata"]["namespace"], v["metadata"]["name"]), v) for v in vmis_raw) # --- assemble VMs ------------------------------------------------------ # - vms: list[VM] = [] + vms = [] for raw in vms_raw: ns = raw["metadata"]["namespace"] name = raw["metadata"]["name"] @@ -398,7 +389,7 @@ def main(): 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} + vm_by_key = dict(((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 @@ -406,7 +397,7 @@ def main(): # 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) + pods_by_pvc = defaultdict(list) for pod in pods_raw: pns = pod["metadata"]["namespace"] pnode = pod.get("spec", {}).get("nodeName", "") or "?" @@ -425,20 +416,20 @@ def main(): 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} + vm_pvc_set = set(p for vm in vms for p in vm.pvcs) # --- build volumes for every PVC -------------------------------------- # - def vol_for(ns_name: tuple[str, str]) -> Volume: + def vol_for(ns_name): 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} + volumes = dict((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}) + nodes = sorted(set(r.node for v in volumes.values() for r in v.replicas if r.node)) - if args.json: + if args.as_json: emit_json(vms, volumes, pods_by_pvc, vm_pvc_set, nodes) return @@ -447,22 +438,21 @@ def main(): print_unattached(volumes, vm_pvc_set, pods_by_pvc) -def _risk_line(vol: Volume) -> str: +def _risk_line(vol): 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}") + return ("%s/%s [%s] robustness=%s usable-copies-on: %s" + % (vol.pvc_ns, vol.pvc_name, lh, vol.robustness, hn)) -def print_node_safety(nodes, vms: list[VM], volumes: dict): - hdr = f"{C.BOLD}=== NODE REMOVAL SAFETY ==={C.RESET}" - print(hdr) +def print_node_safety(nodes, vms, volumes): + print("%s=== NODE REMOVAL SAFETY ===%s" % (C.BOLD, C.RESET)) 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} + running_pvcs = set(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. @@ -473,101 +463,101 @@ def print_node_safety(nodes, vms: list[VM], volumes: dict): for key, vol in volumes.items(): if not vol.replicas or not vol.data_nodes: continue - if vol.data_nodes == {node}: + if vol.data_nodes == set([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.") + print(" %sSAFE%s %s%s%s: every volume keeps a usable copy on another node." + % (C.GREEN, C.RESET, C.BOLD, node, C.RESET)) 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}:") + verdict = ("%sUNSAFE%s" % (C.RED, C.RESET)) if crit else ("%sCAUTION%s" % (C.YELLOW, C.RESET)) + print(" %s %s%s%s:" % (verdict, C.BOLD, node, C.RESET)) if crit: - print(f" {C.RED}running-VM data lost if removed ({len(crit)}):{C.RESET}") + print(" %srunning-VM data lost if removed (%d):%s" % (C.RED, len(crit), C.RESET)) for vol in crit: - print(f" - {_risk_line(vol)}") + print(" - %s" % _risk_line(vol)) if warns: - print(f" {C.YELLOW}idle/non-VM data lost if removed ({len(warns)}):{C.RESET}") + print(" %sidle/non-VM data lost if removed (%d):%s" % (C.YELLOW, len(warns), C.RESET)) for vol in warns: - print(f" - {_risk_line(vol)}") + print(" - %s" % _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):") + print("\n %sAlready unavailable%s (no usable copy on ANY node -- fix before draining anything):" + % (C.RED, C.RESET)) for vol in broken: - print(f" - {_risk_line(vol)}") + print(" - %s" % _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}") +def _pvc_block(vol): + head = (" PVC %s%s/%s%s vol=%s sc=%s" + % (C.CYAN, vol.pvc_ns, vol.pvc_name, C.RESET, vol.lh_name or "-", vol.storage_class)) meta = [] if vol.robustness != "-": - meta.append(f"robustness={colour_health(vol.robustness)}") + meta.append("robustness=%s" % colour_health(vol.robustness)) if vol.state != "-": - meta.append(f"state={colour_health(vol.state)}") + meta.append("state=%s" % colour_health(vol.state)) if vol.desired_replicas is not None: - meta.append(f"desired-replicas={vol.desired_replicas}") + meta.append("desired-replicas=%s" % vol.desired_replicas) if meta: head += " (" + " ".join(meta) + ")" if vol.note: - head += f" {C.DIM}[{vol.note}]{C.RESET}" + head += " %s[%s]%s" % (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}") + lines.append(" %s(no replicas reported)%s" % (C.DIM, C.RESET)) return "\n".join(lines) -def print_vm_detail(vms: list[VM], volumes: dict): - print(f"{C.BOLD}=== VIRTUAL MACHINES ==={C.RESET}") +def print_vm_detail(vms, volumes): + print("%s=== VIRTUAL MACHINES ===%s" % (C.BOLD, 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}") + node = (" node=%s" % vm.node) if vm.node else "" + print("\n%sVM %s/%s%s [%s]%s" + % (C.BOLD, vm.ns, vm.name, C.RESET, colour_status(vm.status), node)) if not vm.pvcs: - print(f" {C.DIM}(no PVC-backed volumes){C.RESET}") + print(" %s(no PVC-backed volumes)%s" % (C.DIM, 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)} +def print_unattached(volumes, vm_pvc_set, pods_by_pvc): + non_vm = dict((k, v) for k, v in volumes.items() if k not in vm_pvc_set) + pod_attached = dict((k, v) for k, v in non_vm.items() if pods_by_pvc.get(k)) + detached = dict((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}") + print("%s=== PVCs NOT ATTACHED TO A VM ===%s" % (C.BOLD, 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}") + print("\n %sAttached to non-VM pod(s):%s" % (C.BOLD, 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}") + consumers = ", ".join("%s (node=%s)" % (n, nd) for n, nd in pods_by_pvc[key]) + print(" used by: %s" % consumers) print(_pvc_block(pod_attached[key])) if detached: - print(f"\n {C.BOLD}Completely detached (no consumer):{C.RESET}") + print("\n %sCompletely detached (no consumer):%s" % (C.BOLD, 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): + def vol_dict(v): return { "pvc_namespace": v.pvc_ns, "pvc_name": v.pvc_name, @@ -578,21 +568,20 @@ def emit_json(vms, volumes, pods_by_pvc, vm_pvc_set, nodes): "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} + {"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} + running_pvcs = set(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}: + if vol.replicas and vol.data_nodes == set([node]): (crit if key in running_pvcs else warns).append( - f"{vol.pvc_ns}/{vol.pvc_name}") + "%s/%s" % (vol.pvc_ns, vol.pvc_name)) node_safety[node] = { "safe": not crit and not warns, "running_vm_data_only_here": crit, @@ -606,18 +595,16 @@ def emit_json(vms, volumes, pods_by_pvc, vm_pvc_set, nodes): { "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], + "pvcs": ["%s/%s" % (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), - } + "unattached": dict( + ("%s/%s" % (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))