SELECT temperature, humidity, timestamp
FROM sensor_metrics
WHERE sensor_id = 'sensor-12345'
  AND timestamp > now() - 1h;

SELECT sensor_id, date, avg_temperature, max_temperature, min_temperature
FROM sensor_daily_aggregates
WHERE sensor_id = 'sensor-12345'
  AND date >= '2023-11-01'
  AND date <= '2023-11-15';

SELECT sensor_id, latest_temperature, latest_timestamp
FROM sensors_by_building
WHERE building_id = 'building-789';

SELECT sensor_id, temperature, timestamp
FROM sensor_metrics
WHERE sensor_id = 'sensor-12345'
  AND timestamp > now() - 10m
  AND temperature > 80  -- Filtering (requires ALLOW FILTERING or index)

┌─────────────────────────────────────────────────────────────┐
│                     IoT Devices (100K+)                      │
└──────────────┬──────────────────────────────────────────────┘
               │ HTTP POST /metrics
               ▼
┌─────────────────────────────────────────────────────────────┐
│              Load Balancer (HAProxy / AWS ALB)               │
└──────────────┬──────────────────────────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────────────────────────┐
│           Ingestion API (Python Flask / FastAPI)             │
│  • Validation                                                 │
│  • Deduplication (bloom filter)                              │
│  • Batching (100 writes → 1 batch)                           │
└──────────────┬──────────────────────────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────────────────────────┐
│                    Cassandra Cluster                         │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  Datacenter: us-east (3 nodes)                       │   │
│  │  Datacenter: eu-west (3 nodes)                       │   │
│  │  RF=3 per DC, CL=LOCAL_QUORUM                        │   │
│  └─────────────────────────────────────────────────────┘   │
└──────────────┬──────────────────────────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────────────────────────┐
│         Background Processing (Python / Spark)               │
│  • Compute daily aggregates                                  │
│  • Anomaly detection                                         │
│  • Data quality monitoring                                   │
└──────────────┬──────────────────────────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────────────────────────┐
│           Dashboard API (Python FastAPI)                     │
│  • Real-time queries                                         │
│  • Aggregated queries                                        │
│  • Caching (Redis)                                           │
└──────────────┬──────────────────────────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────────────────────────┐
│              Frontend (React / Grafana)                      │
│  • Real-time charts                                          │
│  • Alert dashboard                                           │
└─────────────────────────────────────────────────────────────┘

Datacenter: us-east
├── Node 1: 10.0.1.10 (64GB RAM, 8 cores, 4TB SSD)
├── Node 2: 10.0.1.11 (64GB RAM, 8 cores, 4TB SSD)
└── Node 3: 10.0.1.12 (64GB RAM, 8 cores, 4TB SSD)

Datacenter: eu-west
├── Node 4: 10.1.1.10 (64GB RAM, 8 cores, 4TB SSD)
├── Node 5: 10.1.1.11 (64GB RAM, 8 cores, 4TB SSD)
└── Node 6: 10.1.1.12 (64GB RAM, 8 cores, 4TB SSD)

Replication Factor: 3 per DC
Total Capacity: 6 nodes × 4TB = 24TB raw (8TB effective with RF=3)

Data volume calculation:
• 100,000 sensors × 1 reading/10s = 10,000 writes/sec
• Each reading: ~200 bytes (sensor_id, timestamp, 4 metrics, metadata)
• Daily: 10,000 writes/s × 86,400s × 200 bytes = 172.8 GB/day
• 365 days: 172.8 GB × 365 = 63 TB/year

With RF=3: 63 TB × 3 = 189 TB/year total cluster data

Compression (TWCS): ~50% compression ratio
Actual: 189 TB × 0.5 = 94.5 TB/year

With 6 nodes: 94.5 TB / 6 ≈ 16 TB/node/year
Current storage: 4 TB/node → Need expansion within 3 months!

Recommendation: Start with 12 nodes (2TB/node/year = 2 years runway)

SELECT temperature, humidity, pressure, battery, timestamp
FROM sensor_metrics
WHERE sensor_id = ?
  AND timestamp > ?
  AND timestamp < ?;

CREATE TABLE sensor_metrics (
    sensor_id text,           -- Partition key
    timestamp timestamp,      -- Clustering key (time-series)
    temperature decimal,
    humidity decimal,
    pressure decimal,
    battery_level int,        -- 0-100
    metadata map<text, text>, -- Extensible metadata
    PRIMARY KEY (sensor_id, timestamp)
) WITH CLUSTERING ORDER BY (timestamp DESC)  -- Newest first
  AND compaction = {
    'class': 'TimeWindowCompactionStrategy',
    'compaction_window_size': '1',
    'compaction_window_unit': 'DAYS'
  }
  AND default_time_to_live = 31536000;  -- 365 days

Max partition size = 200 bytes × (365 days × 8,640 readings/day)
                   = 200 bytes × 3,153,600 readings
                   = 630 MB (acceptable, < 100MB warning threshold per year)

SELECT date, avg_temperature, max_temperature, min_temperature, sample_count
FROM sensor_daily_aggregates
WHERE sensor_id = ?
  AND date >= ?
  AND date <= ?;

CREATE TABLE sensor_daily_aggregates (
    sensor_id text,
    date date,               -- Clustering key (YYYY-MM-DD)
    avg_temperature decimal,
    max_temperature decimal,
    min_temperature decimal,
    avg_humidity decimal,
    max_humidity decimal,
    min_humidity decimal,
    avg_pressure decimal,
    max_pressure decimal,
    min_pressure decimal,
    sample_count bigint,     -- Number of readings aggregated
    PRIMARY KEY (sensor_id, date)
) WITH CLUSTERING ORDER BY (date DESC)
  AND compaction = {
    'class': 'LeveledCompactionStrategy',
    'sstable_size_in_mb': 160
  }
  AND default_time_to_live = 94608000;  -- 3 years (longer retention for aggregates)

SELECT sensor_id, latest_temperature, latest_humidity, latest_timestamp
FROM sensors_by_building
WHERE building_id = ?;

CREATE TABLE sensors_by_building (
    building_id text,
    sensor_id text,          -- Clustering key (allows filtering by sensor)
    latest_temperature decimal,
    latest_humidity decimal,
    latest_pressure decimal,
    latest_battery_level int,
    latest_timestamp timestamp,
    sensor_location text,    -- e.g., "Floor 3, Room 301"
    PRIMARY KEY (building_id, sensor_id)
) WITH compaction = {
    'class': 'LeveledCompactionStrategy',
    'sstable_size_in_mb': 160
  };

SELECT sensor_id, event_time, event_type, metric_name, metric_value, threshold
FROM sensor_anomalies
WHERE sensor_id = ?
  AND event_time > ?;

CREATE TABLE sensor_anomalies (
    sensor_id text,
    event_time timestamp,    -- When anomaly detected
    event_type text,         -- 'HIGH_TEMP', 'LOW_BATTERY', etc.
    metric_name text,
    metric_value decimal,
    threshold decimal,
    severity text,           -- 'WARNING', 'CRITICAL'
    PRIMARY KEY (sensor_id, event_time, event_type)
) WITH CLUSTERING ORDER BY (event_time DESC, event_type ASC)
  AND compaction = {
    'class': 'TimeWindowCompactionStrategy',
    'compaction_window_size': '7',
    'compaction_window_unit': 'DAYS'
  }
  AND default_time_to_live = 7776000;  -- 90 days

CREATE KEYSPACE sensor_platform WITH replication = {
  'class': 'NetworkTopologyStrategy',
  'us-east': 3,
  'eu-west': 3
};

USE sensor_platform;

-- Tables defined above
CREATE TABLE sensor_metrics (...);
CREATE TABLE sensor_daily_aggregates (...);
CREATE TABLE sensors_by_building (...);
CREATE TABLE sensor_anomalies (...);

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from cassandra.cluster import Cluster
from cassandra.policies import DCAwareRoundRobinPolicy, TokenAwarePolicy
from cassandra.query import BatchStatement, SimpleStatement
from datetime import datetime, timedelta
from typing import List, Dict
import hashlib

app = FastAPI()

# Cassandra connection
cluster = Cluster(
    contact_points=['10.0.1.10', '10.0.1.11', '10.0.1.12'],
    port=9042,
    load_balancing_policy=TokenAwarePolicy(
        DCAwareRoundRobinPolicy(local_dc='us-east')
    ),
    protocol_version=4,
)
session = cluster.connect('sensor_platform')

# Prepared statements (performance optimization)
insert_metric_stmt = session.prepare("""
    INSERT INTO sensor_metrics
    (sensor_id, timestamp, temperature, humidity, pressure, battery_level)
    VALUES (?, ?, ?, ?, ?, ?)
    USING TTL 31536000
""")

update_building_stmt = session.prepare("""
    UPDATE sensors_by_building
    SET latest_temperature = ?,
        latest_humidity = ?,
        latest_pressure = ?,
        latest_battery_level = ?,
        latest_timestamp = ?
    WHERE building_id = ? AND sensor_id = ?
""")

# Deduplication bloom filter (in-memory, simple implementation)
# Production: Use Redis or Cassandra-based deduplication
recent_hashes = set()

class SensorReading(BaseModel):
    sensor_id: str
    building_id: str
    timestamp: datetime
    temperature: float
    humidity: float
    pressure: float
    battery_level: int

    @validator('timestamp')
    def timestamp_not_future(cls, v):
        if v > datetime.utcnow() + timedelta(minutes=5):
            raise ValueError('Timestamp cannot be > 5 minutes in future')
        return v

    @validator('timestamp')
    def timestamp_not_too_old(cls, v):
        if v < datetime.utcnow() - timedelta(days=7):
            raise ValueError('Timestamp too old (> 7 days)')
        return v

    @validator('battery_level')
    def battery_in_range(cls, v):
        if not 0 <= v <= 100:
            raise ValueError('Battery level must be 0-100')
        return v

@app.post("/metrics")
async def ingest_metrics(readings: List[SensorReading]):
    """
    Ingest batch of sensor readings.
    Expected batch size: 100-1000 readings per request.
    """
    if len(readings) > 1000:
        raise HTTPException(status_code=400, detail="Batch too large (max 1000)")

    # Deduplication check
    unique_readings = []
    for reading in readings:
        # Hash: sensor_id + timestamp (dedup key)
        hash_key = hashlib.md5(
            f"{reading.sensor_id}_{reading.timestamp.isoformat()}".encode()
        ).hexdigest()

        if hash_key not in recent_hashes:
            unique_readings.append(reading)
            recent_hashes.add(hash_key)

            # Limit in-memory set size (rolling window)
            if len(recent_hashes) > 100000:
                recent_hashes.pop()  # Simple FIFO (production: use TTL)

    if not unique_readings:
        return {"status": "ok", "inserted": 0, "duplicates": len(readings)}

    # Batch insert (performance optimization)
    batch = BatchStatement()

    for reading in unique_readings:
        # Insert raw metric
        batch.add(insert_metric_stmt, (
            reading.sensor_id,
            reading.timestamp,
            reading.temperature,
            reading.humidity,
            reading.pressure,
            reading.battery_level
        ))

        # Update building latest values (denormalization)
        batch.add(update_building_stmt, (
            reading.temperature,
            reading.humidity,
            reading.pressure,
            reading.battery_level,
            reading.timestamp,
            reading.building_id,
            reading.sensor_id
        ))

    # Execute batch with LOCAL_QUORUM (fast, consistent)
    session.execute(batch, timeout=10.0)

    # Async anomaly detection (fire-and-forget)
    # Production: Use message queue (Kafka/RabbitMQ)
    for reading in unique_readings:
        if reading.temperature > 80:
            _trigger_anomaly_alert(reading, 'HIGH_TEMP', 'temperature', 80)
        if reading.battery_level < 10:
            _trigger_anomaly_alert(reading, 'LOW_BATTERY', 'battery_level', 10)

    return {
        "status": "ok",
        "inserted": len(unique_readings),
        "duplicates": len(readings) - len(unique_readings)
    }

def _trigger_anomaly_alert(reading: SensorReading, event_type: str, metric_name: str, threshold: float):
    """Insert anomaly event (async background task in production)"""
    insert_anomaly = session.prepare("""
        INSERT INTO sensor_anomalies
        (sensor_id, event_time, event_type, metric_name, metric_value, threshold, severity)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    """)

    severity = 'CRITICAL' if event_type == 'LOW_BATTERY' else 'WARNING'
    metric_value = getattr(reading, metric_name.replace('battery_level', 'battery_level'))

    session.execute(insert_anomaly, (
        reading.sensor_id,
        reading.timestamp,
        event_type,
        metric_name,
        metric_value,
        threshold,
        severity
    ))

@app.get("/health")
async def health_check():
    """Health check for load balancer"""
    try:
        session.execute("SELECT now() FROM system.local", timeout=2.0)
        return {"status": "healthy"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Cassandra unavailable: {e}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, workers=4)

from cassandra.cluster import Cluster
from cassandra.query import SimpleStatement
from datetime import datetime, timedelta
from statistics import mean
import schedule
import time

cluster = Cluster(['10.0.1.10'])
session = cluster.connect('sensor_platform')

def compute_daily_aggregates(date: datetime.date):
    """
    Compute daily aggregates for all sensors.
    Runs once per day at 00:05 (after day completes).
    """
    print(f"Computing aggregates for {date}...")

    # Get all unique sensors (production: maintain sensor registry)
    sensors_query = "SELECT DISTINCT sensor_id FROM sensor_metrics LIMIT 100000"
    sensors = session.execute(sensors_query)

    insert_aggregate = session.prepare("""
        INSERT INTO sensor_daily_aggregates
        (sensor_id, date, avg_temperature, max_temperature, min_temperature,
         avg_humidity, max_humidity, min_humidity,
         avg_pressure, max_pressure, min_pressure, sample_count)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """)

    for sensor_row in sensors:
        sensor_id = sensor_row.sensor_id

        # Query all readings for this sensor on this date
        start_ts = datetime.combine(date, datetime.min.time())
        end_ts = datetime.combine(date, datetime.max.time())

        readings_query = session.prepare("""
            SELECT temperature, humidity, pressure
            FROM sensor_metrics
            WHERE sensor_id = ? AND timestamp >= ? AND timestamp <= ?
        """)

        readings = list(session.execute(readings_query, (sensor_id, start_ts, end_ts)))

        if not readings:
            continue  # No data for this sensor today

        # Compute aggregates
        temperatures = [r.temperature for r in readings]
        humidities = [r.humidity for r in readings]
        pressures = [r.pressure for r in readings]

        session.execute(insert_aggregate, (
            sensor_id,
            date,
            mean(temperatures), max(temperatures), min(temperatures),
            mean(humidities), max(humidities), min(humidities),
            mean(pressures), max(pressures), min(pressures),
            len(readings)
        ))

    print(f"Aggregates computed for {date}: {len(list(sensors))} sensors")

def daily_job():
    """Run daily at 00:05"""
    yesterday = (datetime.utcnow() - timedelta(days=1)).date()
    compute_daily_aggregates(yesterday)

# Schedule job
schedule.every().day.at("00:05").do(daily_job)

if __name__ == "__main__":
    print("Daily aggregator started...")
    while True:
        schedule.run_pending()
        time.sleep(60)

# spark_aggregator.py (production version)
from pyspark.sql import SparkSession
from pyspark.sql.functions import mean, max, min, count

spark = SparkSession.builder \
    .appName("SensorAggregator") \
    .config("spark.cassandra.connection.host", "10.0.1.10") \
    .getOrCreate()

date = "2023-11-15"

# Read from Cassandra
df = spark.read \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="sensor_metrics", keyspace="sensor_platform") \
    .load() \
    .filter(f"timestamp >= '{date} 00:00:00' AND timestamp < '{date} 23:59:59'")

# Compute aggregates
aggregates = df.groupBy("sensor_id").agg(
    mean("temperature").alias("avg_temperature"),
    max("temperature").alias("max_temperature"),
    min("temperature").alias("min_temperature"),
    mean("humidity").alias("avg_humidity"),
    max("humidity").alias("max_humidity"),
    min("humidity").alias("min_humidity"),
    mean("pressure").alias("avg_pressure"),
    max("pressure").alias("max_pressure"),
    min("pressure").alias("min_pressure"),
    count("*").alias("sample_count")
).withColumn("date", lit(date))

# Write back to Cassandra
aggregates.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="sensor_daily_aggregates", keyspace="sensor_platform") \
    .mode("append") \
    .save()

from fastapi import FastAPI, HTTPException
from cassandra.cluster import Cluster
from datetime import datetime, timedelta
from typing import List, Optional
import redis

app = FastAPI()

# Cassandra
cluster = Cluster(['10.0.1.10'])
session = cluster.connect('sensor_platform')

# Redis cache (for expensive queries)
cache = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
CACHE_TTL = 300  # 5 minutes

@app.get("/sensors/{sensor_id}/recent")
async def get_recent_metrics(sensor_id: str, hours: int = 1):
    """Get recent metrics for a sensor (last N hours)"""
    if hours > 24:
        raise HTTPException(status_code=400, detail="Max 24 hours")

    # Check cache
    cache_key = f"recent:{sensor_id}:{hours}"
    cached = cache.get(cache_key)
    if cached:
        return {"source": "cache", "data": eval(cached)}

    # Query Cassandra
    query = session.prepare("""
        SELECT timestamp, temperature, humidity, pressure, battery_level
        FROM sensor_metrics
        WHERE sensor_id = ?
          AND timestamp >= ?
        LIMIT 1000
    """)

    start_time = datetime.utcnow() - timedelta(hours=hours)
    rows = session.execute(query, (sensor_id, start_time))

    results = [{
        "timestamp": row.timestamp.isoformat(),
        "temperature": float(row.temperature),
        "humidity": float(row.humidity),
        "pressure": float(row.pressure),
        "battery_level": row.battery_level
    } for row in rows]

    # Cache result
    cache.setex(cache_key, CACHE_TTL, str(results))

    return {"source": "cassandra", "data": results}

@app.get("/sensors/{sensor_id}/aggregates")
async def get_daily_aggregates(
    sensor_id: str,
    start_date: str,  # YYYY-MM-DD
    end_date: str
):
    """Get daily aggregates for a sensor (date range)"""
    query = session.prepare("""
        SELECT date, avg_temperature, max_temperature, min_temperature,
               avg_humidity, max_humidity, min_humidity,
               avg_pressure, max_pressure, min_pressure, sample_count
        FROM sensor_daily_aggregates
        WHERE sensor_id = ?
          AND date >= ?
          AND date <= ?
    """)

    rows = session.execute(query, (sensor_id, start_date, end_date))

    return [
        {
            "date": row.date.isoformat(),
            "temperature": {
                "avg": float(row.avg_temperature),
                "max": float(row.max_temperature),
                "min": float(row.min_temperature)
            },
            "humidity": {
                "avg": float(row.avg_humidity),
                "max": float(row.max_humidity),
                "min": float(row.min_humidity)
            },
            "pressure": {
                "avg": float(row.avg_pressure),
                "max": float(row.max_pressure),
                "min": float(row.min_pressure)
            },
            "sample_count": row.sample_count
        }
        for row in rows
    ]

@app.get("/buildings/{building_id}/sensors")
async def get_building_sensors(building_id: str):
    """Get all sensors in a building with latest values"""
    query = session.prepare("""
        SELECT sensor_id, sensor_location,
               latest_temperature, latest_humidity, latest_pressure,
               latest_battery_level, latest_timestamp
        FROM sensors_by_building
        WHERE building_id = ?
    """)

    rows = session.execute(query, (building_id,))

    return [
        {
            "sensor_id": row.sensor_id,
            "location": row.sensor_location,
            "latest": {
                "temperature": float(row.latest_temperature),
                "humidity": float(row.latest_humidity),
                "pressure": float(row.latest_pressure),
                "battery_level": row.latest_battery_level,
                "timestamp": row.latest_timestamp.isoformat()
            }
        }
        for row in rows
    ]

@app.get("/sensors/{sensor_id}/anomalies")
async def get_sensor_anomalies(sensor_id: str, hours: int = 24):
    """Get recent anomalies for a sensor"""
    query = session.prepare("""
        SELECT event_time, event_type, metric_name, metric_value, threshold, severity
        FROM sensor_anomalies
        WHERE sensor_id = ?
          AND event_time >= ?
    """)

    start_time = datetime.utcnow() - timedelta(hours=hours)
    rows = session.execute(query, (sensor_id, start_time))

    return [
        {
            "event_time": row.event_time.isoformat(),
            "event_type": row.event_type,
            "metric_name": row.metric_name,
            "metric_value": float(row.metric_value),
            "threshold": float(row.threshold),
            "severity": row.severity
        }
        for row in rows
    ]

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

cluster_name: 'SensorPlatform'

# Seeds (2 per DC)
seed_provider:
  - class_name: org.apache.cassandra.locator.SimpleSeedProvider
    parameters:
      - seeds: "10.0.1.10,10.0.1.11,10.1.1.10,10.1.1.11"

# Listen addresses
listen_address: 10.0.1.10  # Node-specific
rpc_address: 10.0.1.10

# Data directories (separate disks)
data_file_directories:
  - /mnt/data1/cassandra/data
  - /mnt/data2/cassandra/data
commitlog_directory: /mnt/ssd1/cassandra/commitlog
saved_caches_directory: /var/lib/cassandra/saved_caches
hints_directory: /var/lib/cassandra/hints

# Snitch
endpoint_snitch: GossipingPropertyFileSnitch

# Num tokens
num_tokens: 256

# Compaction
compaction_throughput_mb_per_sec: 64
concurrent_compactors: 4

# MemTable
memtable_heap_space_in_mb: 2048
memtable_flush_writers: 4

# CommitLog
commitlog_sync: periodic
commitlog_sync_period_in_ms: 10000
commitlog_total_space_in_mb: 8192

# Caching
key_cache_size_in_mb: 100
row_cache_size_in_mb: 0  # Disabled

# Thread pools
concurrent_reads: 32
concurrent_writes: 32
concurrent_counter_writes: 32

# Timeouts
read_request_timeout_in_ms: 5000
write_request_timeout_in_ms: 2000
range_request_timeout_in_ms: 10000

# Failure detection
phi_convict_threshold: 8

# Repair
repair_session_max_tree_depth: 15

dc=us-east
rack=rack1

# Heap size
-Xms8G
-Xmx8G

# Young gen
-Xmn1600M

# GC
-XX:+UseG1GC
-XX:G1RSetUpdatingPauseTimePercent=5
-XX:MaxGCPauseMillis=200
-XX:InitiatingHeapOccupancyPercent=70
-XX:ParallelGCThreads=16

# GC logging
-Xlog:gc*,gc+age=trace,safepoint:file=/var/log/cassandra/gc.log:time,uptime:filecount=10,filesize=10m

# Heap dump on OOM
-XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=/var/log/cassandra/heap_dump.hprof

# JMX
-Dcom.sun.management.jmxremote.port=7199
-Dcom.sun.management.jmxremote.ssl=false
-Dcom.sun.management.jmxremote.authenticate=true

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'cassandra'
    static_configs:
      - targets:
        - '10.0.1.10:7070'
        - '10.0.1.11:7070'
        - '10.0.1.12:7070'
        - '10.1.1.10:7070'
        - '10.1.1.11:7070'
        - '10.1.1.12:7070'
        labels:
          cluster: 'sensor-platform'

  - job_name: 'ingestion-api'
    static_configs:
      - targets: ['10.0.1.20:8000', '10.1.1.20:8000']

  - job_name: 'dashboard-api'
    static_configs:
      - targets: ['10.0.1.21:8001', '10.1.1.21:8001']

groups:
  - name: cassandra_alerts
    interval: 30s
    rules:
      # Critical: Dropped mutations (data loss!)
      - alert: DroppedMutations
        expr: cassandra_dropped_message_total{message_type="MUTATION"} > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Cassandra node {{ $labels.instance }} dropping writes"
          description: "{{ $value }} mutations dropped in last minute"

      # Critical: GC pause too long
      - alert: HighGCPause
        expr: cassandra_jvm_gc_pause_seconds{quantile="0.99"} > 1.0
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Cassandra node {{ $labels.instance }} has high GC pauses"
          description: "p99 GC pause: {{ $value }}s"

      # Warning: Pending compactions
      - alert: PendingCompactions
        expr: cassandra_table_pending_compactions > 30
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Node {{ $labels.instance }} has {{ $value }} pending compactions"

      # Warning: High read latency
      - alert: HighReadLatency
        expr: cassandra_table_read_latency_seconds{quantile="0.95"} > 0.05
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "High read latency on {{ $labels.keyspace }}.{{ $labels.table }}"
          description: "p95 latency: {{ $value }}s"

      # Critical: Node down
      - alert: NodeDown
        expr: up{job="cassandra"} == 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Cassandra node {{ $labels.instance }} is down"

#!/bin/bash
# Daily snapshots uploaded to S3

KEYSPACE="sensor_platform"
SNAPSHOT_NAME="daily_$(date +%Y%m%d)"
S3_BUCKET="s3://sensor-platform-backups"
HOSTNAME=$(hostname)

# Create snapshot
nodetool snapshot -t $SNAPSHOT_NAME $KEYSPACE

# Find snapshot directories
SNAPSHOT_DIRS=$(find /var/lib/cassandra/data/$KEYSPACE -type d -name $SNAPSHOT_NAME)

# Upload each table's snapshot to S3
for DIR in $SNAPSHOT_DIRS; do
  TABLE=$(echo $DIR | awk -F'/' '{print $(NF-2)}')
  aws s3 sync $DIR $S3_BUCKET/$HOSTNAME/$KEYSPACE/$TABLE/$SNAPSHOT_NAME/
done

# Clean up local snapshot
nodetool clearsnapshot -t $SNAPSHOT_NAME

# Delete snapshots older than 7 days from S3
aws s3 ls $S3_BUCKET/$HOSTNAME/ | awk '{print $2}' | while read PREFIX; do
  SNAPSHOT_DATE=$(echo $PREFIX | grep -oP '\d{8}')
  if [ ! -z "$SNAPSHOT_DATE" ]; then
    DAYS_OLD=$(( ($(date +%s) - $(date -d $SNAPSHOT_DATE +%s)) / 86400 ))
    if [ $DAYS_OLD -gt 7 ]; then
      aws s3 rm --recursive $S3_BUCKET/$HOSTNAME/$PREFIX
    fi
  fi
done

echo "Backup complete: $S3_BUCKET/$HOSTNAME/$KEYSPACE/$SNAPSHOT_NAME/"

# Daily backup at 3 AM
0 3 * * * /usr/local/bin/backup.sh >> /var/log/cassandra/backup.log 2>&1

# Install Reaper
docker run -d \
  --name reaper \
  -p 8080:8080 \
  -e REAPER_STORAGE_TYPE=cassandra \
  -e REAPER_CASS_CONTACT_POINTS=["10.0.1.10"] \
  thelastpickle/cassandra-reaper:latest

# Create repair schedule (via API)
curl -X POST http://localhost:8080/repair_schedule \
  -d "clusterName=SensorPlatform" \
  -d "keyspace=sensor_platform" \
  -d "owner=admin" \
  -d "intensity=0.9" \
  -d "scheduleDaysBetween=7" \
  -d "segmentCountPerNode=64"

#!/bin/bash
# repair.sh - Weekly incremental repair

nodetool repair -inc sensor_platform sensor_metrics
nodetool repair -inc sensor_platform sensor_daily_aggregates
nodetool repair -inc sensor_platform sensors_by_building
nodetool repair -inc sensor_platform sensor_anomalies

echo "Repair complete: $(date)"

# Weekly repair on Sunday at 2 AM
0 2 * * 0 /usr/local/bin/repair.sh >> /var/log/cassandra/repair.log 2>&1

# sensor_metrics_stress.yaml
keyspace: sensor_platform

table: sensor_metrics

columnspec:
  - name: sensor_id
    size: fixed(20)
    population: uniform(1..100000)
  - name: timestamp
    cluster: fixed(1000)
  - name: temperature
    size: fixed(8)
  - name: humidity
    size: fixed(8)
  - name: pressure
    size: fixed(8)
  - name: battery_level
    size: fixed(4)

insert:
  partitions: fixed(1)
  batchtype: UNLOGGED

queries:
  recent_metrics:
    cql: SELECT * FROM sensor_metrics WHERE sensor_id = ? AND timestamp > ?
    fields: samerow

# Write test (10K writes/sec for 1 hour)
cassandra-stress user profile=sensor_metrics_stress.yaml \
  ops\(insert=1\) \
  n=36000000 \
  -rate threads=100 throttle=10000/s \
  -node 10.0.1.10,10.0.1.11,10.0.1.12 \
  -log file=write_test.log

# Read test (5K reads/sec for 1 hour)
cassandra-stress user profile=sensor_metrics_stress.yaml \
  ops\(recent_metrics=1\) \
  n=18000000 \
  -rate threads=50 throttle=5000/s \
  -node 10.0.1.10,10.0.1.11,10.0.1.12 \
  -log file=read_test.log

# Mixed workload (70% read, 30% write)
cassandra-stress user profile=sensor_metrics_stress.yaml \
  ops\(insert=3,recent_metrics=7\) \
  n=36000000 \
  -rate threads=100 \
  -node 10.0.1.10,10.0.1.11,10.0.1.12 \
  -log file=mixed_test.log

# Simulate node failure
systemctl stop cassandra  # On node 10.0.1.12

# Verify writes still succeed (RF=3, CL=QUORUM)
# Expected: No errors, hints accumulate

# Bring node back
systemctl start cassandra

# Verify hint replay
nodetool statushandoff

# Expected: Hints delivered, data consistent

# Simulate network partition (us-east isolated from eu-west)
iptables -A INPUT -s 10.1.0.0/16 -j DROP
iptables -A OUTPUT -d 10.1.0.0/16 -j DROP

# Verify LOCAL_QUORUM writes succeed
# Verify EACH_QUORUM writes fail (expected)

# Heal partition
iptables -F

# Verify cross-DC replication resumes
# Run repair to ensure consistency

# Send 100K writes/sec (10x normal load)
cassandra-stress user profile=sensor_metrics_stress.yaml \
  ops\(insert=1\) \
  n=360000000 \
  -rate threads=500 throttle=100000/s \
  -node 10.0.1.10

# Monitor for:
# - Dropped mutations (should be 0)
# - GC pauses (should stay < 500ms)
# - Pending compactions (monitor, should recover)

# Expected: Some increased latency, but no data loss

CREATE MATERIALIZED VIEW sensors_by_temperature AS
  SELECT sensor_id, latest_temperature, latest_timestamp
  FROM sensors_by_building
  WHERE building_id IS NOT NULL
    AND sensor_id IS NOT NULL
    AND latest_temperature IS NOT NULL
  PRIMARY KEY (latest_temperature, building_id, sensor_id)
  WITH CLUSTERING ORDER BY (latest_temperature DESC);

# cassandra.yaml
authenticator: PasswordAuthenticator
authorizer: CassandraAuthorizer

# Enable SSL/TLS
client_encryption_options:
  enabled: true
  keystore: /etc/cassandra/keystore.jks
  keystore_password: changeit