Interpreting nodetool compactionstats Output for Cassandra 4.x/5.x Automation

# Gate 1: the LOCAL node must be Up/Normal (UN). Scope the check to this node's
# address — a cluster-wide grep for "UN" passes even when the local node is down.
nodetool status | awk -v ip="127.0.0.1" '$2==ip {print $1}'
# Expected output: UN

# Gate 2: data volume must have headroom. Refuse to poll above ~85% used, since
# compaction needs scratch space and a stall here is a disk-pressure symptom.
df --output=pcent /var/lib/cassandra | tail -1 | tr -d ' %'
# Expected output: an integer < 85

# Gate 3: capture the current snapshot once, cheaply, to confirm JMX responds.
# --human-readable is safe to parse; the machine form omits the unit suffix.
nodetool compactionstats --human-readable
# Expected output begins with a "pending tasks: N" line.

#!/usr/bin/env python3
# requirements: Python 3.10+, Cassandra 4.0/4.1/5.0 with nodetool on PATH.
"""
Deterministic compactionstats extractor for Cassandra 4.x/5.x.
Safety: node-state validation, disk-pressure guardrails, atomic state files.
Output: parsed compaction-state dict as JSON, or a non-zero exit on failure.
Rollback: clears partial state and raises rather than reporting a false no-op.
"""
import hashlib
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Optional

STATE_DIR = Path("/var/lib/cassandra/automation/compaction_state")
STATE_FILE = STATE_DIR / "compactionstats_last_run.sha256"
DISK_THRESHOLD = 0.85
NODETOOL_TIMEOUT = 30
LOCAL_ADDR = "127.0.0.1"  # this node's listen/broadcast address

PENDING_RE = re.compile(r"^pending tasks:\s*(\d+)", re.MULTILINE)
ACTIVE_TYPES = {"Compaction", "Validation", "Cleanup", "Scrub",
                "Upgradesstables", "Index build"}


class IdempotencyError(RuntimeError):
    """Raised when state could not be persisted (distinct from a no-op match)."""


def verify_node_state() -> bool:
    """Ensure the LOCAL node is Up/Normal (UN) before invocation."""
    try:
        proc = subprocess.run(["nodetool", "status"],
                              capture_output=True, text=True, timeout=15)
    except (subprocess.SubprocessError, OSError):
        return False
    if proc.returncode != 0:
        return False
    for line in proc.stdout.splitlines():
        cols = line.split()
        # Scope the UN check to the local node's row, not any node in the ring.
        if len(cols) >= 2 and cols[1] == LOCAL_ADDR:
            return cols[0] == "UN"
    return False


def verify_disk_pressure() -> bool:
    """Halt polling if the data directory exceeds the used-space threshold."""
    try:
        stat = os.statvfs("/var/lib/cassandra")
        used = 1.0 - (stat.f_bavail / stat.f_blocks)
        return used < DISK_THRESHOLD
    except OSError:
        return False


def parse_compactionstats(text: str) -> dict:
    """Parse the real column layout:
    id  compaction type  keyspace  table  completed  total  unit  progress."""
    match = PENDING_RE.search(text)
    pending = int(match.group(1)) if match else 0

    active = []
    for line in text.splitlines():
        cols = line.split()
        # Skip header/blank/non-row lines; the type token follows the id.
        if len(cols) >= 8 and cols[1] in ACTIVE_TYPES:
            try:
                completed = int(cols[-4])
                total = int(cols[-3])
            except ValueError:
                continue  # human-readable form carries unit suffixes; skip here
            active.append({
                "id": cols[0],
                "compaction_type": cols[1],
                "keyspace": cols[2],
                "table": cols[3],
                "completed": completed,
                "total": total,
                "unit": cols[-2],
                "progress": cols[-1],
            })
    return {"pending": pending, "active": active}


def fetch_compactionstats() -> Optional[dict]:
    """Idempotent fetch with the pre-flight gates applied as guard clauses."""
    if not verify_node_state() or not verify_disk_pressure():
        return None
    try:
        proc = subprocess.run(["nodetool", "compactionstats"],
                              capture_output=True, text=True,
                              timeout=NODETOOL_TIMEOUT)
    except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError):
        return None
    if proc.returncode != 0 or "pending tasks:" not in proc.stdout:
        return None
    return parse_compactionstats(proc.stdout)


def enforce_idempotency(payload: dict) -> bool:
    """Persist a snapshot digest. Return True if already seen (no-op), False if
    new and recorded. Raise IdempotencyError if the write fails, so a disk error
    is never mistaken for a successful no-op."""
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    digest = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()).hexdigest()
    if STATE_FILE.exists() and STATE_FILE.read_text().strip() == digest:
        return True
    tmp = STATE_FILE.with_suffix(".tmp")
    try:
        tmp.write_text(digest)
        tmp.replace(STATE_FILE)  # atomic on POSIX
    except OSError as exc:
        if tmp.exists():
            tmp.unlink()  # rollback the partial write
        raise IdempotencyError("failed to persist compaction state") from exc
    return False


if __name__ == "__main__":
    data = fetch_compactionstats()
    if data is None:
        sys.exit(1)  # a gate failed or JMX did not respond
    try:
        already_seen = enforce_idempotency(data)
    except IdempotencyError:
        sys.exit(1)
    if already_seen:
        sys.exit(0)  # nothing changed since the last poll
    print(json.dumps(data, indent=2))

#!/usr/bin/env python3
# requirements: Python 3.10+. Pure function over the parsed dict; no I/O.
def validate_compaction_health(payload: dict, concurrent_compactors: int = 2) -> dict:
    """Return a health verdict with actionable thresholds. Uses progress to spot
    stalls because compactionstats does not report per-task throughput."""
    pending: int = payload.get("pending", 0)
    active: list[dict] = payload.get("active", [])

    min_progress = min(
        (row["completed"] / row["total"]
         for row in active if row.get("total", 0) > 0),
        default=1.0,
    )

    verdict: dict = {"status": "healthy", "actions": []}
    if pending > 2 * concurrent_compactors:
        verdict["status"] = "backpressure"
        verdict["actions"].append("throttle_throughput")
    if active and min_progress < 0.05 and pending > 5:
        verdict["status"] = "possible_stall"
        verdict["actions"].append("investigate_disk_io")
    return verdict

# setcompactionthroughput is silent on success; always read it back.
nodetool setcompactionthroughput 50   # MiB/s; 0 = unlimited (the default)
nodetool getcompactionthroughput
# Rollback: nodetool setcompactionthroughput 0

# 1. The extractor prints valid JSON with pending + active keys on a healthy node.
python3 compactionstats_extractor.py | python3 -m json.tool
# Expected: {"pending": <int>, "active": [ ... ]}

# 2. Running it twice with no cluster change is a no-op (exit 0, no output).
python3 compactionstats_extractor.py; echo "exit=$?"
# Expected on the second run: exit=0 with no stdout

# 3. The digest file exists and is a 64-char hex sha256.
wc -c /var/lib/cassandra/automation/compaction_state/compactionstats_last_run.sha256
# Expected: 64 (plus a trailing newline if your editor added one)

-- Cassandra 5.0+ only; absent on 4.x.
SELECT keyspace_name, table_name, kind, progress, total
FROM system_views.sstable_tasks;