Diagnosing Gossip Protocol Failures in Multi-DC Clusters: Production-Ready Diagnostics & Automation

#!/usr/bin/env python3
"""
Idempotent gossip baseline validator for Cassandra v4.x/v5.x.
Reads thread-pool saturation from system_views.thread_pools and per-endpoint
phi values from `nodetool failuredetector`, without mutating state.
"""
import re
import subprocess
import sys
import json
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider

SAFETY_CHECKS = {
    "min_version": 4,
    "max_phi_threshold": 12.0,
    "read_only": True
}

def validate_cluster_version(session):
    row = session.execute("SELECT release_version FROM system.local LIMIT 1").one()
    major = int(row.release_version.split('.')[0])
    if major < SAFETY_CHECKS["min_version"]:
        raise RuntimeError(f"Unsupported version {row.release_version}. system_views requires Cassandra 4.0+")
    return major

def collect_gossip_baseline(contact_points, username=None, password=None):
    auth = PlainTextAuthProvider(username, password) if username else None
    cluster = Cluster(contact_points, auth_provider=auth, connect_timeout=10)
    session = cluster.connect()
    validate_cluster_version(session)

    baseline = {"peers": [], "thread_pools": []}

    # Per-endpoint phi values are not exposed via CQL; parse `nodetool failuredetector`.
    # Endpoint header lines look like "/10.0.0.1"; strip the leading "/".
    for ep, phi in parse_failuredetector().items():
        baseline["peers"].append({"peer": ep, "phi": phi})

    # Thread pool saturation check via the real system_views.thread_pools table.
    monitored = ("GossipStage", "CompactionExecutor", "RepairSession")
    tp_rows = session.execute("""
        SELECT name, active_tasks, pending_tasks, completed_tasks, blocked_tasks
        FROM system_views.thread_pools
    """)
    for row in tp_rows:
        if row.name not in monitored:
            continue
        baseline["thread_pools"].append({
            "pool": row.name,
            "active": row.active_tasks,
            "pending": row.pending_tasks,
            "completed": row.completed_tasks,
            "blocked": row.blocked_tasks,
        })

    cluster.shutdown()
    return baseline


def parse_failuredetector():
    """Return {endpoint: phi} parsed from `nodetool failuredetector` output."""
    result = subprocess.run(
        ["nodetool", "failuredetector"],
        capture_output=True, text=True, timeout=15
    )
    if result.returncode != 0:
        raise RuntimeError(f"nodetool failuredetector failed: {result.stderr.strip()}")

    phi_map = {}
    # Rows look like: "/10.0.0.1            8.0          true"
    row_re = re.compile(r"^/?(\S+?):?\d*\s+([0-9.]+|Infinity)\s+\S+")
    for line in result.stdout.splitlines():
        line = line.strip()
        m = row_re.match(line)
        if not m:
            continue
        endpoint = m.group(1)
        raw_phi = m.group(2)
        phi_map[endpoint] = float("inf") if raw_phi == "Infinity" else float(raw_phi)
    return phi_map

if __name__ == "__main__":
    try:
        result = collect_gossip_baseline(["127.0.0.1"])
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"[FATAL] Baseline collection failed: {e}", file=sys.stderr)
        sys.exit(1)

#!/usr/bin/env bash
# compaction_gossip_correlation.sh
# Correlates high phi values with compaction/repair backlog

set -euo pipefail

SAFETY_GATE() {
    local disk_util
    # Match sd*, nvme*, vd*, xvd*, and dm-* device names; %util is the last column.
    disk_util=$(iostat -x 1 1 | awk '/^(sd|nvme|vd|xvd|dm-)/{print $NF}' | sort -rn | head -1)
    if [ -z "$disk_util" ]; then
        echo "[WARN] Could not read disk utilization from iostat. Skipping I/O gate."
        return 0
    fi
    if (( $(echo "$disk_util > 90" | bc -l) )); then
        echo "[WARN] Disk I/O utilization >90%. Deferring diagnostics to prevent further contention."
        exit 0
    fi
}

EXPECTED_OUTPUT="Tabular output mapping nodes with phi > 7.0 to pending compaction tasks."
ROLLBACK_PATH="If compaction was manually paused via 'nodetool disableautocompaction', re-enable with: nodetool enableautocompaction"

SAFETY_GATE

echo "[INFO] Extracting high-phi nodes..."
# nodetool gossipinfo has no phi field; per-endpoint phi comes from
# nodetool failuredetector. Rows look like: "/10.0.0.1   8.0   true".
# Strip the leading "/" from the endpoint and compare the phi column.
HIGH_PHI_NODES=$(nodetool failuredetector | awk '$2+0 > 7.0 && $2 != "" {sub(/^\//,"",$1); print $1}')

if [ -z "$HIGH_PHI_NODES" ]; then
    echo "[OK] No nodes exceed phi threshold. Gossip healthy."
    exit 0
fi

echo "[INFO] Checking compaction/repair backlog for affected nodes..."
for node in $HIGH_PHI_NODES; do
    echo "--- Node: $node ---"
    nodetool compactionstats | grep -E "pending|active" || echo "No compaction backlog"
    nodetool tpstats | grep -E "RepairSession|GossipStage" | awk '{print $1, $2, $3}'
done

echo "[INFO] Correlation complete."

#!/usr/bin/env python3
"""
Safe phi_convict_threshold adjustment with automatic rollback.
There is no nodetool command for this value; we use the JMX FailureDetector
MBean (org.apache.cassandra.net:type=FailureDetector) via jmxterm.
"""
import subprocess
import sys
import time

TARGET_THRESHOLD = 10.0
DEFAULT_THRESHOLD = 8.0
RESTORE_DELAY = 300  # seconds
JMX_HOST = "127.0.0.1:7199"
FD_MBEAN = "org.apache.cassandra.net:type=FailureDetector"

def jmxterm(script, timeout=30):
    """Run a jmxterm command script against the local Cassandra JMX port."""
    result = subprocess.run(
        ["jmxterm", "-l", JMX_HOST, "-n", "-v", "silent"],
        input=script, capture_output=True, text=True, timeout=timeout
    )
    if result.returncode != 0:
        raise RuntimeError(f"jmxterm failed: {result.stderr.strip()}")
    return result.stdout.strip()

def read_phi_threshold():
    """Read the current PhiConvictThreshold attribute from the FailureDetector MBean."""
    return jmxterm(f"get -b {FD_MBEAN} PhiConvictThreshold\n")

def set_phi_threshold(value):
    """Invoke setPhiConvictThreshold(double) on the FailureDetector MBean."""
    jmxterm(f"run -b {FD_MBEAN} setPhiConvictThreshold {value}\n")

def adjust_phi_threshold():
    print("[SAFETY] Reading current phi_convict_threshold via JMX...")
    current = read_phi_threshold()
    print(f"[INFO] Current PhiConvictThreshold: {current}")

    print(f"[INFO] Applying temporary threshold: {TARGET_THRESHOLD}")
    set_phi_threshold(TARGET_THRESHOLD)

    print(f"[INFO] Monitoring gossip stabilization for {RESTORE_DELAY}s...")
    time.sleep(RESTORE_DELAY)

    print("[ROLLBACK] Restoring original threshold...")
    set_phi_threshold(DEFAULT_THRESHOLD)
    print("[OK] Threshold restored. Verify with nodetool failuredetector.")

if __name__ == "__main__":
    try:
        adjust_phi_threshold()
    except Exception as e:
        print(f"[FATAL] Adjustment failed. Manual rollback required: {e}", file=sys.stderr)
        sys.exit(1)