War Story: We Survived a 2-Hour Outage with Redis 8.0 Cluster and Sentinel

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
)

// ClusterHealthConfig holds configuration for Redis cluster health checks
type ClusterHealthConfig struct {
    Addrs        []string      // List of cluster node addresses (host:port)
    SentinelAddrs []string     // List of Sentinel addresses
    CheckInterval time.Duration // Time between health checks
    FailoverTimeout time.Duration // Timeout for failover detection
}

// ClusterHealthChecker monitors Redis 8.0 Cluster and Sentinel health
type ClusterHealthChecker struct {
    config *ClusterHealthConfig
    clusterClient *redis.ClusterClient
    sentinelClient *redis.SentinelClient
    mu sync.RWMutex
    lastFailover time.Time
    alerts chan string
}

// NewClusterHealthChecker initializes a new health checker
func NewClusterHealthChecker(ctx context.Context, config *ClusterHealthConfig) (*ClusterHealthChecker, error) {
    // Initialize cluster client
    clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs:    config.Addrs,
        PoolSize: 10,
    })
    // Ping cluster to verify connectivity
    if err := clusterClient.Ping(ctx).Err(); err != nil {
        return nil, fmt.Errorf("failed to ping cluster: %w", err)
    }

    // Initialize sentinel client (connected to first sentinel for simplicity)
    sentinelClient := redis.NewSentinelClient(&redis.Options{
        Addr: config.SentinelAddrs[0],
    })
    if err := sentinelClient.Ping(ctx).Err(); err != nil {
        return nil, fmt.Errorf("failed to ping sentinel: %w", err)
    }

    return &ClusterHealthChecker{
        config:        config,
        clusterClient: clusterClient,
        sentinelClient: sentinelClient,
        alerts: make(chan string, 10),
    }, nil
}

// Run starts the health check loop
func (c *ClusterHealthChecker) Run(ctx context.Context) {
    ticker := time.NewTicker(c.config.CheckInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            c.checkHealth(ctx)
        case <-ctx.Done():
            log.Println("Health checker stopped")
            return
        }
    }
}

// checkHealth performs a single health check iteration
func (c *ClusterHealthChecker) checkHealth(ctx context.Context) {
    c.mu.Lock()
    defer c.mu.Unlock()

    // Get cluster slots to verify shard distribution
    slots, err := c.clusterClient.ClusterSlots(ctx).Result()
    if err != nil {
        c.sendAlert(fmt.Sprintf("Cluster slots check failed: %v", err))
        return
    }

    // Verify all 16384 slots are covered (broken in our initial implementation: we only checked 10 shards)
    // BUG: Initial implementation only checked len(slots) == 10, missing partial slot coverage
    if len(slots) != 12 { // We have 12 shards, this check is insufficient
        c.sendAlert(fmt.Sprintf("Unexpected number of slot groups: got %d, expected 12", len(slots)))
    }

    // Get sentinel master status for our cluster
    masters, err := c.sentinelClient.SentinelMasters(ctx).Result()
    if err != nil {
        c.sendAlert(fmt.Sprintf("Sentinel masters check failed: %v", err))
        return
    }

    // BUG: Initial implementation didn't check for +failover-state-reconf-slaves master state
    for _, master := range masters {
        if master.Name == "mymaster" {
            // Check if master is in failover state (missed in initial code)
            if master.Flags["failover_state_reconf_slaves"] {
                c.lastFailover = time.Now()
                c.sendAlert("Active Sentinel failover detected for mymaster")
            }
        }
    }
}

// sendAlert sends an alert to the alerts channel
func (c *ClusterHealthChecker) sendAlert(msg string) {
    select {
    case c.alerts <- msg:
    default:
        log.Printf("Alert channel full, dropping: %s", msg)
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    config := &ClusterHealthConfig{
        Addrs: []string{
            "redis-cluster-1:6379",
            "redis-cluster-2:6379",
            "redis-cluster-3:6379",
            "redis-cluster-4:6379",
            "redis-cluster-5:6379",
            "redis-cluster-6:6379",
            "redis-cluster-7:6379",
            "redis-cluster-8:6379",
            "redis-cluster-9:6379",
            "redis-cluster-10:6379",
            "redis-cluster-11:6379",
            "redis-cluster-12:6379",
        },
        SentinelAddrs: []string{
            "sentinel-1:26379",
            "sentinel-2:26379",
            "sentinel-3:26379",
        },
        CheckInterval: 5 * time.Second,
        FailoverTimeout: 30 * time.Second,
    }

    checker, err := NewClusterHealthChecker(ctx, config)
    if err != nil {
        log.Fatalf("Failed to initialize health checker: %v", err)
    }

    go checker.Run(ctx)

    // Consume alerts
    for alert := range checker.alerts {
        log.Printf("ALERT: %s", alert)
    }
}

package main

import (
    "context"
    "fmt"
    "log"
    "strings"
    "time"

    "github.com/redis/go-redis/v9"
)

// SentinelConfigUpdater automates safe configuration updates for Redis 8.0 Sentinels
type SentinelConfigUpdater struct {
    sentinelAddrs []string
    clusterName   string // Name of the Redis cluster monitored by Sentinel (e.g., "mymaster")
    client        *redis.SentinelClient
}

// NewSentinelConfigUpdater initializes a new updater connected to the first available Sentinel
func NewSentinelConfigUpdater(ctx context.Context, sentinelAddrs []string, clusterName string) (*SentinelConfigUpdater, error) {
    var lastErr error
    // Try connecting to each Sentinel until one succeeds
    for _, addr := range sentinelAddrs {
        client := redis.NewSentinelClient(&redis.Options{
            Addr:        addr,
            DialTimeout: 5 * time.Second,
        })
        // Verify connection with ping
        if err := client.Ping(ctx).Err(); err != nil {
            lastErr = err
            client.Close()
            continue
        }
        // Verify the cluster is monitored by this Sentinel
        masters, err := client.SentinelMasters(ctx).Result()
        if err != nil {
            lastErr = err
            client.Close()
            continue
        }
        found := false
        for _, master := range masters {
            if master.Name == clusterName {
                found = true
                break
            }
        }
        if !found {
            lastErr = fmt.Errorf("sentinel %s does not monitor cluster %s", addr, clusterName)
            client.Close()
            continue
        }
        return &SentinelConfigUpdater{
            sentinelAddrs: sentinelAddrs,
            clusterName:   clusterName,
            client:        client,
        }, nil
    }
    return nil, fmt.Errorf("failed to connect to any sentinel: %w", lastErr)
}

// GetCurrentElectionTimeout retrieves the current Sentinel leader election timeout in milliseconds
func (u *SentinelConfigUpdater) GetCurrentElectionTimeout(ctx context.Context) (int64, error) {
    // Sentinel 8.0 stores election timeout in the "sentinel election-timeout" config key
    val, err := u.client.SentinelConfigGet(ctx, "election-timeout").Result()
    if err != nil {
        return 0, fmt.Errorf("failed to get election-timeout: %w", err)
    }
    if len(val) != 2 {
        return 0, fmt.Errorf("unexpected config get response: %v", val)
    }
    var timeout int64
    if _, err := fmt.Sscanf(val[1], "%d", &timeout); err != nil {
        return 0, fmt.Errorf("failed to parse election timeout: %w", err)
    }
    return timeout, nil
}

// UpdateElectionTimeout updates the Sentinel leader election timeout to the specified value (ms)
// It propagates the change to all Sentinels in the cluster to avoid split brain
func (u *SentinelConfigUpdater) UpdateElectionTimeout(ctx context.Context, newTimeoutMs int64) error {
    // Validate timeout: Redis 8.0 requires election timeout >= 100ms and <= 10000ms
    if newTimeoutMs < 100 || newTimeoutMs > 10000 {
        return fmt.Errorf("invalid election timeout: %dms (must be 100-10000ms)", newTimeoutMs)
    }

    // Update all Sentinels sequentially to avoid config drift
    for _, addr := range u.sentinelAddrs {
        // Create a temporary client for each Sentinel to avoid stale connections
        tempClient := redis.NewSentinelClient(&redis.Options{
            Addr:        addr,
            DialTimeout: 5 * time.Second,
        })
        defer tempClient.Close()

        // Set the election timeout config
        if err := tempClient.SentinelConfigSet(ctx, "election-timeout", fmt.Sprintf("%d", newTimeoutMs)).Err(); err != nil {
            return fmt.Errorf("failed to update election-timeout on %s: %w", addr, err)
        }

        // Verify the change was applied
        val, err := tempClient.SentinelConfigGet(ctx, "election-timeout").Result()
        if err != nil {
            return fmt.Errorf("failed to verify election-timeout on %s: %w", addr, err)
        }
        if len(val) != 2 || val[1] != fmt.Sprintf("%d", newTimeoutMs) {
            return fmt.Errorf("election timeout not updated on %s: got %v, expected %d", addr, val, newTimeoutMs)
        }
        log.Printf("Updated election-timeout to %dms on Sentinel %s", newTimeoutMs, addr)
    }

    // Force Sentinel to rewrite config to disk to persist across restarts
    if err := u.client.SentinelConfigRewrite(ctx).Err(); err != nil {
        return fmt.Errorf("failed to rewrite sentinel config: %w", err)
    }
    return nil
}

// EnableClusterSlaveNoEvict enables the cluster-slave-no-evict config to prevent slaves from evicting keys during failover
func (u *SentinelConfigUpdater) EnableClusterSlaveNoEvict(ctx context.Context) error {
    // Check current status
    val, err := u.client.SentinelConfigGet(ctx, "cluster-slave-no-evict").Result()
    if err != nil {
        return fmt.Errorf("failed to get cluster-slave-no-evict: %w", err)
    }
    if len(val) == 2 && strings.ToLower(val[1]) == "yes" {
        log.Println("cluster-slave-no-evict already enabled")
        return nil
    }

    // Enable for all Sentinels
    for _, addr := range u.sentinelAddrs {
        tempClient := redis.NewSentinelClient(&redis.Options{
            Addr:        addr,
            DialTimeout: 5 * time.Second,
        })
        defer tempClient.Close()

        if err := tempClient.SentinelConfigSet(ctx, "cluster-slave-no-evict", "yes").Err(); err != nil {
            return fmt.Errorf("failed to enable cluster-slave-no-evict on %s: %w", addr, err)
        }
    }

    // Rewrite config to disk
    return u.client.SentinelConfigRewrite(ctx).Err()
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // Configuration matching our production setup
    sentinelAddrs := []string{
        "sentinel-1.prod.internal:26379",
        "sentinel-2.prod.internal:26379",
        "sentinel-3.prod.internal:26379",
    }
    clusterName := "mymaster"

    updater, err := NewSentinelConfigUpdater(ctx, sentinelAddrs, clusterName)
    if err != nil {
        log.Fatalf("Failed to initialize updater: %v", err)
    }

    // Get current election timeout (was 500ms by default, causing false failovers)
    currentTimeout, err := updater.GetCurrentElectionTimeout(ctx)
    if err != nil {
        log.Fatalf("Failed to get current timeout: %v", err)
    }
    log.Printf("Current Sentinel election timeout: %dms", currentTimeout)

    // Update to 2000ms to account for cross-region latency (us-east-1 to us-west-2: 68ms RTT)
    if err := updater.UpdateElectionTimeout(ctx, 2000); err != nil {
        log.Fatalf("Failed to update election timeout: %v", err)
    }

    // Enable cluster-slave-no-evict to prevent data loss during failover
    if err := updater.EnableClusterSlaveNoEvict(ctx); err != nil {
        log.Fatalf("Failed to enable cluster-slave-no-evict: %v", err)
    }

    log.Println("All Sentinel config updates applied successfully")
}

package main

import (
    "context"
    "encoding/csv"
    "fmt"
    "log"
    "os"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
)

// BenchmarkConfig holds configuration for Redis Sentinel failover benchmarking
type BenchmarkConfig struct {
    ClusterAddrs     []string
    SentinelAddrs    []string
    ClusterName      string
    ElectionTimeoutMs int64
    NumIterations    int
    FailoverDelay    time.Duration // Time to wait between failover triggers
}

// BenchmarkResult stores metrics from a single benchmark iteration
type BenchmarkResult struct {
    Iteration       int
    FailoverDuration time.Duration
    WriteDropCount  int
    ReadDropCount   int
    ElectionTimeoutMs int64
}

// SentinelFailoverBenchmark runs controlled failover tests against a Redis 8.0 Cluster/Sentinel setup
type SentinelFailoverBenchmark struct {
    config *BenchmarkConfig
    results []BenchmarkResult
    mu sync.Mutex
}

// NewSentinelFailoverBenchmark initializes a new benchmark runner
func NewSentinelFailoverBenchmark(config *BenchmarkConfig) *SentinelFailoverBenchmark {
    return &SentinelFailoverBenchmark{
        config:  config,
        results: make([]BenchmarkResult, 0, config.NumIterations),
    }
}

// Run executes the full benchmark suite
func (b *SentinelFailoverBenchmark) Run(ctx context.Context) error {
    // Initialize cluster client for writing test data
    clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs:    b.config.ClusterAddrs,
        PoolSize: 50,
    })
    defer clusterClient.Close()

    // Verify cluster is healthy before starting
    if err := clusterClient.Ping(ctx).Err(); err != nil {
        return fmt.Errorf("cluster unhealthy before benchmark: %w", err)
    }

    // Run iterations sequentially to avoid overlapping failovers
    for i := 0; i < b.config.NumIterations; i++ {
        log.Printf("Starting iteration %d/%d", i+1, b.config.NumIterations)
        result, err := b.runIteration(ctx, clusterClient, i)
        if err != nil {
            return fmt.Errorf("iteration %d failed: %w", i, err)
        }
        b.mu.Lock()
        b.results = append(b.results, result)
        b.mu.Unlock()
        // Wait between iterations to let cluster stabilize
        time.Sleep(b.config.FailoverDelay)
    }
    return nil
}

// runIteration executes a single failover test iteration
func (b *SentinelFailoverBenchmark) runIteration(ctx context.Context, clusterClient *redis.ClusterClient, iteration int) (BenchmarkResult, error) {
    sentinelClient := redis.NewSentinelClient(&redis.Options{
        Addr: b.config.SentinelAddrs[0],
    })
    defer sentinelClient.Close()

    // Start a background writer to simulate production load (142k writes/sec)
    writeCtx, writeCancel := context.WithCancel(ctx)
    defer writeCancel()
    var writeDropCount, readDropCount int
    var writeMu, readMu sync.Mutex

    // Start 10 concurrent writers
    var wg sync.WaitGroup
    for w := 0; w < 10; w++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            localDrops := 0
            for {
                select {
                case <-writeCtx.Done():
                    writeMu.Lock()
                    writeDropCount += localDrops
                    writeMu.Unlock()
                    return
                default:
                    key := fmt.Sprintf("benchmark:write:%d:%d", iteration, workerID)
                    if err := clusterClient.Set(ctx, key, "test-value", 1*time.Minute).Err(); err != nil {
                        localDrops++
                    }
                }
            }
        }(w)
    }

    // Trigger a manual failover via Sentinel
    failoverStart := time.Now()
    if err := sentinelClient.SentinelFailover(ctx, b.config.ClusterName).Err(); err != nil {
        writeCancel()
        wg.Wait()
        return BenchmarkResult{}, fmt.Errorf("failed to trigger failover: %w", err)
    }

    // Wait for failover to complete by polling master status
    var failoverEnd time.Time
    for {
        masters, err := sentinelClient.SentinelMasters(ctx).Result()
        if err != nil {
            writeCancel()
            wg.Wait()
            return BenchmarkResult{}, fmt.Errorf("failed to get master status: %w", err)
        }
        for _, master := range masters {
            if master.Name == b.config.ClusterName {
                // Check if failover is complete (master flags no longer have failover state)
                if !master.Flags["failover_state_reconf_slaves"] && !master.Flags["failover_state_select_slave"] {
                    failoverEnd = time.Now()
                    break
                }
            }
        }
        if !failoverEnd.IsZero() {
            break
        }
        time.Sleep(100 * time.Millisecond)
    }

    // Stop writers and get drop counts
    writeCancel()
    wg.Wait()

    // Run read checks during failover (simplified for brevity, full impl would check all shards)
    readClient := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs: b.config.ClusterAddrs,
    })
    defer readClient.Close()
    for r := 0; r < 1000; r++ {
        key := fmt.Sprintf("benchmark:write:%d:0", iteration)
        if err := readClient.Get(ctx, key).Err(); err != nil {
            readMu.Lock()
            readDropCount++
            readMu.Unlock()
        }
    }

    return BenchmarkResult{
        Iteration:        iteration + 1,
        FailoverDuration: failoverEnd.Sub(failoverStart),
        WriteDropCount:   writeDropCount,
        ReadDropCount:    readDropCount,
        ElectionTimeoutMs: b.config.ElectionTimeoutMs,
    }, nil
}

// ExportResults writes benchmark results to a CSV file
func (b *SentinelFailoverBenchmark) ExportResults(path string) error {
    b.mu.Lock()
    defer b.mu.Unlock()

    file, err := os.Create(path)
    if err != nil {
        return fmt.Errorf("failed to create results file: %w", err)
    }
    defer file.Close()

    writer := csv.NewWriter(file)
    defer writer.Flush()

    // Write header
    if err := writer.Write([]string{
        "iteration",
        "election_timeout_ms",
        "failover_duration_ms",
        "write_drop_count",
        "read_drop_count",
    }); err != nil {
        return fmt.Errorf("failed to write CSV header: %w", err)
    }

    // Write rows
    for _, res := range b.results {
        if err := writer.Write([]string{
            fmt.Sprintf("%d", res.Iteration),
            fmt.Sprintf("%d", res.ElectionTimeoutMs),
            fmt.Sprintf("%d", res.FailoverDuration.Milliseconds()),
            fmt.Sprintf("%d", res.WriteDropCount),
            fmt.Sprintf("%d", res.ReadDropCount),
        }); err != nil {
            return fmt.Errorf("failed to write CSV row: %w", err)
        }
    }
    return nil
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
    defer cancel()

    // Benchmark config comparing default 500ms vs our fixed 2000ms election timeout
    config := &BenchmarkConfig{
        ClusterAddrs: []string{
            "redis-cluster-1:6379",
            "redis-cluster-2:6379",
            "redis-cluster-3:6379",
            "redis-cluster-4:6379",
            "redis-cluster-5:6379",
            "redis-cluster-6:6379",
            "redis-cluster-7:6379",
            "redis-cluster-8:6379",
            "redis-cluster-9:6379",
            "redis-cluster-10:6379",
            "redis-cluster-11:6379",
            "redis-cluster-12:6379",
        },
        SentinelAddrs: []string{
            "sentinel-1:26379",
            "sentinel-2:26379",
            "sentinel-3:26379",
        },
        ClusterName:      "mymaster",
        ElectionTimeoutMs: 2000, // Change to 500 to test default config
        NumIterations:    10,
        FailoverDelay:    30 * time.Second,
    }

    benchmark := NewSentinelFailoverBenchmark(config)
    if err := benchmark.Run(ctx); err != nil {
        log.Fatalf("Benchmark failed: %v", err)
    }

    if err := benchmark.ExportResults("failover_benchmark_results.csv"); err != nil {
        log.Fatalf("Failed to export results: %v", err)
    }
    log.Println("Benchmark completed successfully")
}

func measureSentinelRTT(ctx context.Context, addr string) (time.Duration, error) {
    client := redis.NewSentinelClient(&redis.Options{Addr: addr})
    defer client.Close()
    start := time.Now()
    if err := client.Ping(ctx).Err(); err != nil {
        return 0, err
    }
    return time.Since(start), nil
}

func checkFailoverState(ctx context.Context, client *redis.SentinelClient, clusterName string) (bool, error) {
    masters, err := client.SentinelMasters(ctx).Result()
    if err != nil {
        return false, err
    }
    for _, m := range masters {
        if m.Name == clusterName {
            return m.Flags["failover_state_select_slave"] || m.Flags["failover_state_reconf_slaves"], nil
        }
    }
    return false, fmt.Errorf("cluster %s not found", clusterName)
}

func triggerControlledFailover(ctx context.Context, client *redis.SentinelClient, clusterName string) error {
    return client.SentinelFailover(ctx, clusterName).Err()
}