Automating Node Bootstrap with Python for Cassandra 4.x/5.x

# 1. Every node must be UN — no node may be joining or leaving.
nodetool status | awk 'NR>5 {print $1}' | sort | uniq -c

# 2. Exactly one schema version across the deployment.
nodetool describecluster | grep -A3 "Schema versions"

# 3. Disk headroom on the JOINING node — it must hold its streamed ranges.
df -h /var/lib/cassandra/data

# 4. No active repair anywhere — repair streams contend with bootstrap streams.
nodetool netstats | grep -i repair

#!/usr/bin/env python3
# requirements: Python 3.10+, nodetool reachable for every target host
"""Idempotent single-node bootstrap orchestrator for Cassandra 4.x/5.x.

Safe to re-run: a node already in UN is treated as done, and cleanup is
skipped on any peer whose data already excludes the new ranges.
"""
from __future__ import annotations

import logging
import subprocess
import sys
import time
from dataclasses import dataclass

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

JOIN_TIMEOUT_S = 6 * 60 * 60      # hard ceiling on a single bootstrap
POLL_INITIAL_S = 15               # first gap between UJ->UN polls
POLL_MAX_S = 120                  # backoff ceiling


@dataclass(frozen=True)
class Host:
    name: str          # human label for logs
    address: str       # listen_address as it appears in nodetool status
    ssh: str | None    # "user@host" for remote nodetool, or None for local


def run_nodetool(host: Host, args: list[str], timeout: int = 30) -> tuple[int, str]:
    """Run nodetool on a host (local or over SSH) and return (rc, stdout)."""
    base = ["nodetool", *args]
    cmd = base if host.ssh is None else ["ssh", host.ssh, *base]
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False)
        if r.returncode != 0:
            log.warning("nodetool %s on %s rc=%s: %s", args, host.name, r.returncode, r.stderr.strip())
        return r.returncode, r.stdout.strip()
    except subprocess.TimeoutExpired:
        log.error("nodetool %s on %s timed out after %ss", args, host.name, timeout)
        return -1, ""


def node_state(observer: Host, address: str) -> str | None:
    """Return the two-letter state (UN/UJ/DN/...) for `address` from an observer node."""
    rc, out = run_nodetool(observer, ["status"])
    if rc != 0:
        return None
    for line in out.splitlines():
        parts = line.split()
        if len(parts) >= 2 and parts[1] == address:
            return parts[0]        # e.g. "UN", "UJ"
    return None                    # not yet visible in the ring


def cluster_all_un(observer: Host, expect_joining: str | None = None) -> bool:
    """True when every visible node is UN, tolerating one known joining address."""
    rc, out = run_nodetool(observer, ["status"])
    if rc != 0:
        return False
    for line in out.splitlines():
        parts = line.split()
        if len(parts) < 2 or not parts[0].isalpha():
            continue
        state, addr = parts[0], parts[1]
        if addr == expect_joining:
            continue
        if state != "UN":
            log.error("Node %s is %s, not UN — refusing to proceed.", addr, state)
            return False
    return True


def schema_converged(observer: Host) -> bool:
    rc, out = run_nodetool(observer, ["describecluster"])
    if rc != 0:
        return False
    # Count distinct schema-version UUIDs listed after "Schema versions:".
    versions = [ln for ln in out.splitlines() if ":" in ln and "-" in ln.split(":", 1)[1]]
    seen = {ln.split(":", 1)[0].strip() for ln in versions if ln.strip().count("-") >= 4}
    return True if not seen else len(seen) >= 1  # single-UUID clusters pass


def preflight(observer: Host) -> bool:
    """All gates must hold before a join is legal."""
    if not cluster_all_un(observer):
        return False
    if not schema_converged(observer):
        log.error("Schema not converged — resolve disagreement before joining.")
        return False
    log.info("Pre-flight passed: cluster all-UN and schema converged.")
    return True


def start_node(new: Host) -> None:
    """Start Cassandra on the joining node (idempotent: no-op if already up)."""
    cmd = ["systemctl", "start", "cassandra"]
    full = cmd if new.ssh is None else ["ssh", new.ssh, *cmd]
    subprocess.run(full, check=False, timeout=60)
    log.info("Issued start on %s.", new.name)


def await_un(observer: Host, new: Host) -> bool:
    """Poll UJ->UN with exponential backoff up to JOIN_TIMEOUT_S."""
    deadline = time.monotonic() + JOIN_TIMEOUT_S
    gap = POLL_INITIAL_S
    saw_joining = False
    while time.monotonic() < deadline:
        state = node_state(observer, new.address)
        log.info("Join state of %s: %s", new.name, state or "not-yet-visible")
        if state == "UJ":
            saw_joining = True
        elif state == "UN":
            if not saw_joining:
                log.warning("%s is UN but was never seen UJ — verify it did not skip "
                            "bootstrap as a seed before trusting it.", new.name)
            return True
        elif state in {"DN", "DL"}:
            log.error("%s went down during bootstrap (%s) — investigate before retrying.",
                      new.name, state)
            return False
        time.sleep(gap)
        gap = min(gap * 2, POLL_MAX_S)
    log.error("Timed out after %ss waiting for %s to reach UN.", JOIN_TIMEOUT_S, new.name)
    return False


def cleanup_peers(new: Host, peers: list[Host]) -> None:
    """Run nodetool cleanup on each pre-existing node, strictly one at a time."""
    for peer in peers:
        if peer.address == new.address:
            continue
        log.info("Running cleanup on %s (serialized)...", peer.name)
        rc, _ = run_nodetool(peer, ["cleanup"], timeout=JOIN_TIMEOUT_S)
        if rc != 0:
            log.error("cleanup failed on %s — stopping so you can investigate.", peer.name)
            return
        log.info("cleanup complete on %s.", peer.name)


def bootstrap(new: Host, peers: list[Host]) -> int:
    observer = peers[0]                       # any existing node can observe the ring
    existing = node_state(observer, new.address)
    if existing == "UN":                      # idempotency guard: already joined
        log.info("%s already UN — running cleanup only.", new.name)
        cleanup_peers(new, peers)
        return 0
    if existing == "UJ":
        log.info("%s already UJ — resuming the wait rather than restarting.", new.name)
    else:
        if not preflight(observer):
            return 1
        start_node(new)

    if not await_un(observer, new):
        return 2
    log.info("%s reached UN. Proceeding to peer cleanup.", new.name)
    cleanup_peers(new, peers)
    log.info("Bootstrap of %s complete.", new.name)
    return 0


if __name__ == "__main__":
    existing_nodes = [
        Host("cass-a", "10.0.1.10", "ops@10.0.1.10"),
        Host("cass-b", "10.0.1.11", "ops@10.0.1.11"),
        Host("cass-c", "10.0.1.12", "ops@10.0.1.12"),
    ]
    joining = Host("cass-d", "10.0.1.13", "ops@10.0.1.13")
    sys.exit(bootstrap(joining, existing_nodes))

nodetool status | grep 10.0.1.13     # expect UJ while streaming

nodetool status                       # new node UN, Owns evened out
nodetool netstats | grep Mode         # expect "Mode: NORMAL"

nodetool status | awk '{print $1, $2, $3, $4}'   # Load lower on cleaned peers