Interpreting nodetool compactionstats Output: Production Automation & Threshold Validation for Cassandra 4.x/5.x

#!/usr/bin/env python3
"""
Deterministic compactionstats extractor for Cassandra 4.x/5.x.
Safety: Node state validation, disk pressure guardrails, atomic state files.
Expected Output: Parsed compaction-state dict or None on failure.
Rollback Path: Clears corrupted state, falls back to safe exit, logs diagnostic trace.
"""
import json
import re
import subprocess
import sys
import os
import hashlib
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"  # local 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:
    """Safety: 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
    # Scope the UN check to the local node's row, not any node in the cluster.
    for line in proc.stdout.splitlines():
        cols = line.split()
        if len(cols) >= 2 and cols[1] == LOCAL_ADDR:
            return cols[0] == "UN"
    return False

def verify_disk_pressure() -> bool:
    """Safety: Halt polling if data directory exceeds 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 `nodetool compactionstats` text columns:
    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
            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 of the text output with structural validation."""
    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:
        return None
    except (subprocess.SubprocessError, OSError):
        return None
    if proc.returncode != 0:
        return None
    if "pending tasks:" not in proc.stdout:
        return None
    return parse_compactionstats(proc.stdout)

def enforce_idempotency(payload: dict) -> bool:
    """Persist a snapshot digest. Returns True if this snapshot was already
    seen (no-op), False if it is new and was recorded. Raises IdempotencyError
    if the state could not be written, so callers do not confuse a write
    failure with 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
    # Atomic write with rollback on failure
    tmp = STATE_FILE.with_suffix(".tmp")
    try:
        tmp.write_text(digest)
        tmp.replace(STATE_FILE)
    except OSError as exc:
        # Rollback: remove partial state to force fresh evaluation next cycle
        if tmp.exists():
            tmp.unlink()
        raise IdempotencyError("failed to persist compaction state") from exc
    return False

if __name__ == "__main__":
    data = fetch_compactionstats()
    if data is None:
        sys.exit(1)
    try:
        already_seen = enforce_idempotency(data)
    except IdempotencyError:
        sys.exit(1)
    if already_seen:
        sys.exit(0)
    print(json.dumps(data, indent=2))

def validate_compaction_health(payload: dict) -> dict:
    """Returns health verdict with actionable thresholds.
    Consumes the parsed dict from parse_compactionstats(): a `pending`
    integer and an `active` list of compaction rows."""
    pending = payload.get("pending", 0)
    active = payload.get("active", [])

    # Lowest progress among active compactions; compactionstats does not
    # report per-task throughput, so use progress to spot stalled work.
    min_progress = min(
        (row["completed"] / row["total"]
         for row in active if row.get("total", 0) > 0),
        default=1.0,
    )

    verdict = {"status": "healthy", "actions": []}
    if pending > 20:
        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

# Safety: Throttle only during active repair windows.
# setcompactionthroughput is silent on success; verify via getcompactionthroughput.
nodetool setcompactionthroughput 50
nodetool getcompactionthroughput
# Rollback: nodetool setcompactionthroughput 0 (unlimited)

  rm -rf /var/lib/cassandra/automation/compaction_state/*
  # Safety: Only remove automation artifacts, never data directories
  # Expected Output: Clean slate for next polling cycle
  # Rollback: N/A (state is ephemeral)

  nodetool stop COMPACTION
  # Safety: Halts active compaction operations of this type
  # Output: none on success (silent); check the exit code
  # Rollback: nodetool enableautocompaction && nodetool repair -pr

  systemctl restart cassandra
  # Safety: Verify node drains connections before restart
  # Expected Output: Clean JMX reinitialization
  # Rollback: Restore from snapshot if data directory corruption detected