Automating Tombstone Threshold Alerts with Python

#!/usr/bin/env python3
"""
Cassandra Tombstone Threshold Alerting Engine (v4.x/v5.x compatible)
Idempotent, stateful, and safe for production execution.
"""
import argparse
import json
import logging
import os
import re
import subprocess
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import requests

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

@dataclass
class AlertState:
    """Tracks last alert timestamp per table to enforce cooldown deduplication."""
    last_alerted: Dict[str, float] = field(default_factory=dict)
    cooldown_seconds: int = 3600

    def should_alert(self, table_key: str) -> bool:
        now = time.time()
        last = self.last_alerted.get(table_key, 0.0)
        return (now - last) >= self.cooldown_seconds

    def record_alert(self, table_key: str) -> None:
        self.last_alerted[table_key] = time.time()

    def save(self, path: Path) -> None:
        path.write_text(json.dumps(self.last_alerted, indent=2))

    @classmethod
    def load(cls, path: Path) -> "AlertState":
        if path.exists():
            data = json.loads(path.read_text())
            state = cls()
            state.last_alerted = data
            return state
        return cls()

def run_nodetool(command: List[str], timeout_sec: int = 30) -> str:
    """Execute nodetool with strict safety boundaries."""
    logger.info("Executing: %s", " ".join(command))
    try:
        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            timeout=timeout_sec,
            check=True
        )
        return result.stdout
    except subprocess.TimeoutExpired:
        logger.error("nodetool command timed out after %ds", timeout_sec)
        sys.exit(1)
    except subprocess.CalledProcessError as e:
        logger.error("nodetool failed with exit code %d: %s", e.returncode, e.stderr)
        sys.exit(1)

def parse_tablestats(output: str) -> Dict[str, int]:
    """Extract max tombstones-per-slice per keyspace.table from nodetool tablestats output."""
    tombstone_map: Dict[str, int] = {}
    current_keyspace: Optional[str] = None
    current_table: Optional[str] = None
    
    # Regex matches standard nodetool tablestats format. The "Keyspace:" line
    # precedes the indented "Table:" line; tombstone data is reported as the
    # per-slice histogram, not a single "Tombstone count" field.
    keyspace_pattern = re.compile(r"^\s*Keyspace\s*:\s*(.+)$")
    table_pattern = re.compile(r"^\s*Table:\s*(.+)$")
    tombstone_pattern = re.compile(
        r"^\s*Maximum tombstones per slice \(last five minutes\):\s*(\d+)$"
    )
    
    for line in output.splitlines():
        keyspace_match = keyspace_pattern.match(line)
        if keyspace_match:
            current_keyspace = keyspace_match.group(1).strip()
            current_table = None
            continue

        table_match = table_pattern.match(line)
        if table_match:
            table_name = table_match.group(1).strip()
            current_table = (
                f"{current_keyspace}.{table_name}" if current_keyspace else table_name
            )
            continue
            
        if current_table:
            tomb_match = tombstone_pattern.match(line)
            if tomb_match:
                tombstone_map[current_table] = int(tomb_match.group(1))
                
    return tombstone_map

def send_webhook_alert(url: str, payload: Dict, retries: int = 3) -> bool:
    """Route alert to incident management stack with exponential backoff."""
    for attempt in range(retries):
        try:
            response = requests.post(
                url,
                json=payload,
                headers={"Content-Type": "application/json"},
                timeout=10
            )
            response.raise_for_status()
            logger.info("Alert dispatched successfully (HTTP %d)", response.status_code)
            return True
        except requests.RequestException as e:
            logger.warning("Webhook attempt %d failed: %s", attempt + 1, e)
            time.sleep(2 ** attempt)
    logger.error("Failed to dispatch alert after %d retries", retries)
    return False

def main() -> None:
    parser = argparse.ArgumentParser(description="Cassandra Tombstone Threshold Monitor")
    parser.add_argument("--warn-threshold", type=int, default=5000, help="Warning threshold")
    parser.add_argument("--fail-threshold", type=int, default=50000, help="Failure threshold")
    parser.add_argument("--webhook-url", required=True, help="Incident management webhook URL")
    parser.add_argument("--state-file", default="/var/lib/cassandra-monitor/tombstone_state.json")
    parser.add_argument("--dry-run", action="store_true", help="Evaluate thresholds without sending alerts")
    args = parser.parse_args()

    state_path = Path(args.state_file)
    state_path.parent.mkdir(parents=True, exist_ok=True)
    state = AlertState.load(state_path)

    raw_output = run_nodetool(["nodetool", "tablestats"])
    tombstones = parse_tablestats(raw_output)

    triggered = False
    for table_key, count in tombstones.items():
        severity = None
        if count >= args.fail_threshold:
            severity = "CRITICAL"
        elif count >= args.warn_threshold:
            severity = "WARNING"
            
        if severity and state.should_alert(table_key):
            payload = {
                "service": "cassandra-tombstone-monitor",
                "table": table_key,
                "severity": severity,
                "tombstone_count": count,
                "threshold": args.fail_threshold if severity == "CRITICAL" else args.warn_threshold,
                "timestamp": time.time()
            }
            logger.info("Threshold breached: %s (%s) - %d tombstones", table_key, severity, count)
            
            if not args.dry_run:
                if send_webhook_alert(args.webhook_url, payload):
                    state.record_alert(table_key)
                    triggered = True
            else:
                logger.info("[DRY-RUN] Would alert for %s with severity %s", table_key, severity)
                triggered = True

    if triggered:
        state.save(state_path)
    else:
        logger.info("No thresholds breached. State preserved.")

if __name__ == "__main__":
    main()

[Unit]
Description=Cassandra Tombstone Threshold Monitor
After=cassandra.service
ConditionPathExists=/usr/local/bin/cassandra_tombstone_monitor.py

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /usr/local/bin/cassandra_tombstone_monitor.py \
  --warn-threshold 5000 \
  --fail-threshold 50000 \
  --webhook-url "https://hooks.pagerduty.com/services/XXXXXXX/enqueue" \
  --state-file /var/lib/cassandra-monitor/tombstone_state.json
User=cassandra
Group=cassandra
StandardOutput=journal
StandardError=journal
TimeoutStartSec=45

*/15 * * * * /usr/bin/python3 /usr/local/bin/cassandra_tombstone_monitor.py --webhook-url "https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX" --dry-run 2>&1 | logger -t cassandra-tombstone-monitor

python3 cassandra_tombstone_monitor.py --dry-run --warn-threshold 100 --fail-threshold 1000

{
  "keyspace1.users": 1715511303.42,
  "analytics.events": 1715511600.18
}