Pre-flight Checks Before Decommissioning a Cassandra Node (4.x/5.x)

nodetool status                                  # states + per-node load + DC
cqlsh -e "SELECT keyspace_name, replication FROM system_schema.keyspaces;"
nodetool netstats | grep -iE "repair|bootstrap"  # active anti-entropy / joins
nodetool compactionstats | grep "pending tasks"  # backlog per node

#!/usr/bin/env python3
# requirements: Python 3.10+, read-only nodetool + cqlsh access. Mutates nothing.
"""
Decommission pre-flight validator for Cassandra 4.x/5.x.

Runs six read-only gates and prints a GO/NO-GO verdict with reasons.
Exit code 0 = GO, 1 = NO-GO, 2 = could not evaluate (probe error).
Safe to run repeatedly; it changes no cluster state.
"""
from __future__ import annotations
import re
import subprocess
import sys
from dataclasses import dataclass

TARGET_HOST_ID_PREFIX = "9a1f"     # first octet-group of the host id to retire
TARGET_DC = "dc-east"
PENDING_COMPACTION_CEILING = 25    # per-node pending tasks tolerated pre-drain


@dataclass
class GateResult:
    name: str
    ok: bool
    reason: str


def probe(cmd: list[str], timeout: int = 20) -> str:
    """Run a read-only command; return stdout, or raise on failure."""
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if r.returncode != 0:
        raise RuntimeError(f"{cmd[:2]} failed: {r.stderr.strip()}")
    return r.stdout


def parse_status(raw: str) -> list[dict[str, str]]:
    nodes, dc = [], "unknown"
    for line in raw.splitlines():
        s = line.strip()
        if s.startswith("Datacenter:"):
            dc = s.split(":", 1)[1].strip()
            continue
        m = re.match(r"^([UD][NJL])\s+(\S+)\s+([\d.]+)\s*(\w+)\s+\d+\s+\S+\s+(\S+)", s)
        if m:
            st, addr, lv, lu, hid = m.groups()
            nodes.append({"state": st, "addr": addr, "dc": dc,
                          "load_gib": f"{_gib(lv, lu):.2f}", "host_id": hid})
    return nodes


def _gib(v: str, u: str) -> float:
    return float(v) * {"KiB": 1/1024**2, "MiB": 1/1024, "GiB": 1.0, "TiB": 1024.0}.get(u, 1.0)


def _is_target(node: dict[str, str]) -> bool:
    return node["host_id"].startswith(TARGET_HOST_ID_PREFIX)


# --- gates ---------------------------------------------------------------
def gate_all_un(nodes: list[dict[str, str]]) -> GateResult:
    bad = [(n["addr"], n["state"]) for n in nodes if n["state"] != "UN"]
    return GateResult("all-UN", not bad,
                      "all nodes UN" if not bad else f"non-UN nodes: {bad}")


def gate_rf_headroom(nodes: list[dict[str, str]], repl: str) -> GateResult:
    survivors = sum(1 for n in nodes
                    if n["dc"] == TARGET_DC and n["state"] == "UN" and not _is_target(n))
    rfs = [int(x) for x in re.findall(rf"'{re.escape(TARGET_DC)}'\s*:\s*'?(\d+)'?", repl)]
    max_rf = max(rfs) if rfs else 99
    ok = survivors > max_rf   # strictly greater keeps quorum after removal
    return GateResult("rf-headroom", ok,
                      f"{survivors} survivors vs RF {max_rf} in {TARGET_DC}")


def gate_no_repair(netstats: str) -> GateResult:
    active = bool(re.search(r"(?i)repair", netstats)) and "Nothing" not in netstats
    return GateResult("no-active-repair", not active,
                      "no repair streams" if not active else "active repair detected")


def gate_no_bootstrap(nodes: list[dict[str, str]], netstats: str) -> GateResult:
    joining = [n["addr"] for n in nodes if n["state"] in ("UJ", "UL") and not _is_target(n)]
    streaming = bool(re.search(r"(?i)bootstrap|rebuild", netstats)) and "Nothing" not in netstats
    ok = not joining and not streaming
    return GateResult("no-lifecycle-op", ok,
                      "no bootstrap/leave in flight" if ok else f"in-flight: {joining or 'stream'}")


def gate_pending_compactions(compactionstats: str) -> GateResult:
    # `nodetool compactionstats` here is run per node; sum the pending line.
    pend = [int(x) for x in re.findall(r"pending tasks:\s*(\d+)", compactionstats)]
    worst = max(pend) if pend else 0
    ok = worst <= PENDING_COMPACTION_CEILING
    return GateResult("compaction-backlog", ok,
                      f"worst pending tasks {worst} (ceiling {PENDING_COMPACTION_CEILING})")


def gate_disk_headroom(nodes: list[dict[str, str]]) -> GateResult:
    target = next((n for n in nodes if _is_target(n)), None)
    survivors = [n for n in nodes if n["dc"] == TARGET_DC and n["state"] == "UN" and not _is_target(n)]
    if not target or not survivors:
        return GateResult("disk-headroom", False, "cannot project inbound load")
    inbound = float(target["load_gib"]) / len(survivors)
    # Read free space per host (df over the data dir); here we require the caller
    # to supply it. We surface the projection so a per-host check can enforce it.
    return GateResult("disk-headroom", True,
                      f"projected inbound per survivor {inbound:.1f} GiB "
                      f"(verify each keeps >= 30% free)")


def evaluate() -> tuple[bool, list[GateResult]]:
    try:
        status = probe(["nodetool", "status"])
        repl = probe(["cqlsh", "-e",
                      "SELECT keyspace_name, replication FROM system_schema.keyspaces;"])
        netstats = probe(["nodetool", "netstats"])
        comp = probe(["nodetool", "compactionstats"])
    except (subprocess.TimeoutExpired, RuntimeError) as e:
        print(f"NO-GO (could not evaluate): {e}", file=sys.stderr)
        sys.exit(2)

    nodes = parse_status(status)
    results = [
        gate_all_un(nodes),
        gate_rf_headroom(nodes, repl),
        gate_no_repair(netstats),
        gate_no_bootstrap(nodes, netstats),
        gate_pending_compactions(comp),
        gate_disk_headroom(nodes),
    ]
    return all(g.ok for g in results), results


if __name__ == "__main__":
    go, results = evaluate()
    for g in results:
        print(f"[{'PASS' if g.ok else 'FAIL'}] {g.name}: {g.reason}")
    print("\nVERDICT:", "GO — clear to decommission" if go else "NO-GO — do not proceed")
    sys.exit(0 if go else 1)

# Run it; every gate prints PASS/FAIL and the verdict follows.
python3 preflight.py; echo "exit=$?"