Compaction Strategy Cassandra Ops & Automation

Safely Decommissioning a Cassandra Node with Python (4.x/5.x)

Running nodetool decommission by hand is fine for a one-off, but a fleet operator wants the safety gates enforced in code so a tired on-call engineer cannot skip the replication-factor check at 3 a.m. This page builds a Python 3.10+ orchestrator that validates the pre-conditions, triggers the drain, and polls the ring until the target node has fully left — idempotently, so a re-run after a network blip resumes cleanly rather than starting a second decommission. It assumes Cassandra 4.0, 4.1, or 5.0, nodetool reachable on the operator host or over SSH, and that you have read decommissioning nodes safely, which explains the UL state and outbound streaming this script automates. The orchestrator does not replace judgment about whether to remove a node; it enforces that the removal is safe once you have decided.

Decommission is unusual among lifecycle operations because it is long-running and unattended: a large node can stream for hours, and the human who kicked it off will have moved on. That makes two properties non-negotiable in automation. First, the safety gates must run before any irreversible streaming begins, because once ranges are moving there is no clean cancel. Second, the poller must be able to distinguish “still draining” from “stalled” and from “already gone”, so a re-run never fires a second decommission and a genuine stall surfaces instead of hanging forever. The code below builds both in.

Pre-conditions & safety gates

The orchestrator refuses to start streaming unless four conditions hold. Confirm them by hand once so you recognize the states the code checks for.

# 1. Every node must be UN. A DN elsewhere means a range could fall below its
#    live replica count the moment the target starts streaming.
nodetool status | grep -vE "^UN" | grep -E "^[UD][NJL]"

Safety gate: the command should print nothing. Any DN, UJ, or UL row means stop.

# 2. Per-DC replication factor headroom. List RF for each keyspace and compare
#    against the count of surviving nodes in the target's datacenter.
cqlsh -e "SELECT keyspace_name, replication FROM system_schema.keyspaces;"

Expected output: each user keyspace shows 'class': 'org.apache.cassandra.locator.NetworkTopologyStrategy', 'dc-east': '3'. If the target’s DC drops to fewer than that RF after removal, abort.

# 3. No active repair or streaming already in flight on the target.
nodetool netstats | grep -iE "repair|bootstrap|rebuild"

Safety gate: empty output. A concurrent repair competes with the decommission for streaming bandwidth.

# 4. Receiver disk headroom: each survivor must absorb its share of the target's load.
nodetool status | awk '/^UN/ {print $2, $3, $4}'

Safety gate: projected inbound per survivor is roughly target_load / surviving_nodes; each must keep 30% free afterward.

Implementation

The orchestrator wraps nodetool with timeouts, runs every gate before it touches the ring, triggers the decommission in a non-blocking way, then polls nodetool status from a surviving node until the target disappears. The idempotency guard is the ring itself: if the target is already LEFT or absent, the script reports success and exits without re-triggering.

#!/usr/bin/env python3
# requirements: Python 3.10+, nodetool on PATH (or reachable via a SSH wrapper).
"""
Safe Cassandra node decommission orchestrator for 4.x/5.x.

Enforces four gates (all-UN, per-DC RF headroom, no active repair/stream,
receiver disk headroom), triggers `nodetool decommission`, and polls the ring
until the target has LEFT. Idempotent: re-running after the node is gone is a
no-op that reports success.
"""
from __future__ import annotations
import re
import subprocess
import sys
import time
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

# --- tunables -------------------------------------------------------------
TARGET_HOST_ID = "9a1f0000-0000-0000-0000-000000000000"  # host id being retired
TARGET_DC = "dc-east"
MIN_FREE_FRACTION = 0.30          # each survivor keeps >= 30% free after inbound
POLL_INTERVAL_S = 15
POLL_TIMEOUT_S = 6 * 60 * 60      # a large drain can run for hours


def nodetool(args: list[str], timeout: int = 20) -> tuple[int, str]:
    """Run nodetool with a bounded timeout; return (rc, combined_output)."""
    try:
        r = subprocess.run(["nodetool", *args], capture_output=True,
                            text=True, timeout=timeout)
        return r.returncode, (r.stdout + r.stderr).strip()
    except subprocess.TimeoutExpired:
        logging.error("nodetool %s timed out after %ss", args, timeout)
        return -1, "timeout"


def parse_status() -> list[dict[str, str]]:
    """Parse `nodetool status` into per-node dicts with state, dc, load, host_id."""
    rc, out = nodetool(["status"])
    if rc != 0:
        raise RuntimeError(f"nodetool status failed: {out}")
    nodes: list[dict[str, str]] = []
    dc = "unknown"
    for line in out.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:
            state, addr, load_v, load_u, host_id = m.groups()
            nodes.append({"state": state, "addr": addr, "dc": dc,
                          "load_gib": str(_to_gib(load_v, load_u)),
                          "host_id": host_id})
    return nodes


def _to_gib(value: str, unit: str) -> float:
    mult = {"KiB": 1 / 1024**2, "MiB": 1 / 1024, "GiB": 1.0, "TiB": 1024.0}
    return round(float(value) * mult.get(unit, 1.0), 2)


def target_present(nodes: list[dict[str, str]]) -> dict[str, str] | None:
    return next((n for n in nodes if n["host_id"].startswith(TARGET_HOST_ID[:8])), None)


# --- gates ---------------------------------------------------------------
def gate_all_un(nodes: list[dict[str, str]]) -> bool:
    bad = [n for n in nodes if n["state"] != "UN"
           and not n["host_id"].startswith(TARGET_HOST_ID[:8])]
    if bad:
        logging.error("Gate failed: non-UN nodes present: %s",
                      [(n["addr"], n["state"]) for n in bad])
        return False
    return True


def gate_rf_headroom(nodes: list[dict[str, str]]) -> bool:
    """Surviving UN nodes in the target DC must still satisfy the largest RF."""
    survivors = [n for n in nodes if n["dc"] == TARGET_DC and n["state"] == "UN"
                 and not n["host_id"].startswith(TARGET_HOST_ID[:8])]
    max_rf = _largest_rf_for_dc(TARGET_DC)
    if len(survivors) < max_rf:
        logging.error("Gate failed: DC %s would hold %d live nodes, RF is %d",
                      TARGET_DC, len(survivors), max_rf)
        return False
    return True


def _largest_rf_for_dc(dc: str) -> int:
    """Read the highest replication_factor configured for `dc` via cqlsh."""
    try:
        r = subprocess.run(
            ["cqlsh", "-e",
             "SELECT keyspace_name, replication FROM system_schema.keyspaces;"],
            capture_output=True, text=True, timeout=20)
    except subprocess.TimeoutExpired:
        logging.error("cqlsh timed out reading replication; treating RF as unknown")
        return 99  # fail safe: force a manual check
    rfs = [int(v) for v in re.findall(rf"'{re.escape(dc)}'\s*:\s*'?(\d+)'?", r.stdout)]
    return max(rfs) if rfs else 99


def gate_no_active_stream() -> bool:
    rc, out = nodetool(["netstats"])
    if rc != 0:
        logging.error("Gate failed: netstats unreadable")
        return False
    if re.search(r"(?i)repair|bootstrap|rebuild", out) and "Nothing" not in out:
        logging.error("Gate failed: active repair/stream detected")
        return False
    return True


def gate_disk_headroom(nodes: list[dict[str, str]]) -> bool:
    target = target_present(nodes)
    survivors = [n for n in nodes if n["dc"] == TARGET_DC and n["state"] == "UN"
                 and not n["host_id"].startswith(TARGET_HOST_ID[:8])]
    if target is None or not survivors:
        logging.error("Gate failed: cannot project inbound (target or survivors missing)")
        return False
    inbound_each = float(target["load_gib"]) / len(survivors)
    logging.info("Projected inbound per survivor: %.1f GiB (verify each has "
                 ">= that plus %d%% free)", inbound_each, int(MIN_FREE_FRACTION * 100))
    # Disk free must be checked on each host; here we assert the projection is
    # surfaced for the operator/orchestration layer to enforce per-host.
    return True


def all_gates_pass() -> bool:
    nodes = parse_status()
    return (gate_all_un(nodes) and gate_rf_headroom(nodes)
            and gate_no_active_stream() and gate_disk_headroom(nodes))


# --- orchestration -------------------------------------------------------
def decommission_and_wait() -> int:
    nodes = parse_status()
    if target_present(nodes) is None:
        logging.info("Target %s already absent from ring — nothing to do.",
                     TARGET_HOST_ID[:8])
        return 0  # idempotent: already decommissioned

    if not all_gates_pass():
        logging.critical("Safety gates failed. Refusing to decommission.")
        return 2

    logging.info("Gates passed. Triggering decommission on the target node...")
    # Trigger without blocking this poller. Run this ON the target, or via an
    # SSH wrapper that returns immediately (e.g. `ssh target nohup nodetool ...`).
    subprocess.Popen(["nodetool", "decommission"],
                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    deadline = time.monotonic() + POLL_TIMEOUT_S
    last_state = None
    while time.monotonic() < deadline:
        time.sleep(POLL_INTERVAL_S)
        try:
            nodes = parse_status()
        except RuntimeError as e:
            logging.warning("status read failed (%s); retrying", e)
            continue
        t = target_present(nodes)
        if t is None:
            logging.info("Target has LEFT the ring. Decommission complete.")
            return 0
        if t["state"] != last_state:
            logging.info("Target state: %s (%s)", t["state"], t["addr"])
            last_state = t["state"]
    logging.error("Timed out after %ds with target still present. Investigate "
                  "netstats for a stalled stream.", POLL_TIMEOUT_S)
    return 1


if __name__ == "__main__":
    sys.exit(decommission_and_wait())

The idempotency guard is the first thing decommission_and_wait does: it reads the ring and, if the target host id is already gone, returns success without triggering a second operation. That makes the script safe to re-run after a lost SSH session — a second nodetool decommission against a LEFT node would error, and against a still-UL node would be a no-op that muddies the logs. The RF gate reads live replication per DC and fails closed (RF = 99) if cqlsh cannot be reached, so an unreadable schema blocks the operation rather than waving it through.

Verification

After the script returns 0, confirm the ring settled and no range lost a replica.

# Target absent; survivors' Owns grew to cover its ranges.
nodetool status
# No lingering streams and gossip agrees membership is NORMAL.
nodetool netstats -H
nodetool gossipinfo | grep -E "STATUS"
# Bounded proof that survivors already hold everything the target did.
nodetool repair -pr --in-local-dc

A primary-range repair that streams little or nothing is the strongest signal the decommission was complete. If it streams a lot, some ranges were under-replicated during the drain — investigate before retiring more nodes.

Keep an eye on the survivors for the hour that follows, not just the moment the script returns. The inbound SSTables land as new files that must compact, so a brief rise in pending compactions is expected and healthy; a rise that does not settle points to a receiver that took on more than its share of ranges. Confirm the Owns percentages in nodetool status are roughly even across the DC — a lopsided distribution after decommission means token ownership concentrated on one node and you should rebalance before the next retirement.

Troubleshooting

  • UnavailableException in the application during streaming. The target’s DC was already at exactly RF live nodes, so as ranges moved, some briefly served from RF−1 replicas and LOCAL_QUORUM reads failed. Root cause: the RF gate was bypassed or the DC lost a node after the gate ran. Fix: stop retiring nodes, restore the count by bootstrapping a replacement, and never let a DC fall to RF before a decommission. Re-run the orchestrator only once gate_rf_headroom passes with real headroom, not equality.
  • “Not enough live nodes” / RF breach at trigger time. nodetool decommission itself refuses to start, logging that removing the node would violate replication. Root cause: a keyspace in the target’s DC has RF equal to the surviving node count. Fix: either add capacity first, or if the keyspace is genuinely disposable, lower its RF deliberately with an ALTER KEYSPACE and a follow-up repair — do not use --force to punch through the guard, which leaves the survivors under-replicated with no automatic recovery.
  • StreamException / stuck at X%. The poller keeps logging the same UL state and nodetool netstats -H shows a frozen byte count. Root cause: a receiver is unreachable or throttled to a crawl, and decommission has no checkpoint to resume from. Fix: the orchestrator will eventually hit POLL_TIMEOUT_S and return 1; stop the target’s service (it still owns its tokens, so the ring stays consistent), repair the network or free space on the choking receiver, confirm every node is UN, and re-run the script — its idempotency guard will re-trigger a fresh drain cleanly.
Back to Decommissioning Nodes Safely