Redis Clustering and Scaling

package main

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

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

// ClusterManager handles Redis Cluster operations
type ClusterManager struct {
    client *redis.ClusterClient
    ctx    context.Context
}

// ClusterConfig holds cluster configuration
type ClusterConfig struct {
    Addrs              []string
    Password           string
    ReadTimeout        time.Duration
    WriteTimeout       time.Duration
    PoolSize           int
    MinIdleConns       int
    MaxRetries         int
    RouteByLatency     bool
    RouteRandomly      bool
}

func NewClusterManager(config ClusterConfig) *ClusterManager {
    clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs:        config.Addrs,
        Password:     config.Password,
        ReadTimeout:  config.ReadTimeout,
        WriteTimeout: config.WriteTimeout,
        PoolSize:     config.PoolSize,
        MinIdleConns: config.MinIdleConns,
        MaxRetries:   config.MaxRetries,
        
        // Cluster-specific optimizations
        RouteByLatency: config.RouteByLatency,
        RouteRandomly:  config.RouteRandomly,
        MaxRedirects:   8,
        ReadOnly:       false,
    })

    return &ClusterManager{
        client: clusterClient,
        ctx:    context.Background(),
    }
}

// GetClusterInfo retrieves cluster topology information
func (cm *ClusterManager) GetClusterInfo() (*ClusterInfo, error) {
    // Get cluster nodes information
    nodesCmd := cm.client.ClusterNodes(cm.ctx)
    if nodesCmd.Err() != nil {
        return nil, fmt.Errorf("failed to get cluster nodes: %w", nodesCmd.Err())
    }

    // Get cluster slots information
    slotsCmd := cm.client.ClusterSlots(cm.ctx)
    if slotsCmd.Err() != nil {
        return nil, fmt.Errorf("failed to get cluster slots: %w", slotsCmd.Err())
    }

    info := &ClusterInfo{
        Nodes:     parseNodes(nodesCmd.Val()),
        Slots:     parseSlots(slotsCmd.Val()),
        Timestamp: time.Now(),
    }

    return info, nil
}

type ClusterInfo struct {
    Nodes     []NodeInfo    `json:"nodes"`
    Slots     []SlotRange   `json:"slots"`
    Timestamp time.Time     `json:"timestamp"`
}

type NodeInfo struct {
    ID       string   `json:"id"`
    Address  string   `json:"address"`
    Role     string   `json:"role"` // master/replica
    Status   string   `json:"status"` // connected/disconnected
    Slots    []int    `json:"slots,omitempty"`
    MasterID string   `json:"master_id,omitempty"`
}

type SlotRange struct {
    Start  int      `json:"start"`
    End    int      `json:"end"`
    Master NodeInfo `json:"master"`
    Replicas []NodeInfo `json:"replicas"`
}

// DistributedCache implements cache operations across cluster
type DistributedCache struct {
    cluster *ClusterManager
}

func NewDistributedCache(cluster *ClusterManager) *DistributedCache {
    return &DistributedCache{cluster: cluster}
}

// Set stores a key-value pair in the cluster
func (dc *DistributedCache) Set(key string, value interface{}, ttl time.Duration) error {
    return dc.cluster.client.Set(dc.cluster.ctx, key, value, ttl).Err()
}

// Get retrieves a value from the cluster
func (dc *DistributedCache) Get(key string) (string, error) {
    return dc.cluster.client.Get(dc.cluster.ctx, key).Result()
}

// MGet retrieves multiple values efficiently across cluster
func (dc *DistributedCache) MGet(keys []string) (map[string]string, error) {
    // Group keys by node to minimize cross-slot operations
    nodeKeys := dc.groupKeysByNode(keys)
    
    results := make(map[string]string)
    var mu sync.Mutex
    var wg sync.WaitGroup
    errChan := make(chan error, len(nodeKeys))

    // Execute gets in parallel per node
    for _, keyGroup := range nodeKeys {
        wg.Add(1)
        go func(keys []string) {
            defer wg.Done()
            
            values, err := dc.cluster.client.MGet(dc.cluster.ctx, keys...).Result()
            if err != nil {
                errChan <- err
                return
            }

            mu.Lock()
            for i, key := range keys {
                if values[i] != nil {
                    results[key] = values[i].(string)
                }
            }
            mu.Unlock()
        }(keyGroup)
    }

    wg.Wait()
    close(errChan)

    // Check for errors
    if len(errChan) > 0 {
        return nil, <-errChan
    }

    return results, nil
}

// groupKeysByNode groups keys by their cluster node (simplified implementation)
func (dc *DistributedCache) groupKeysByNode(keys []string) [][]string {
    // In a real implementation, this would use the cluster's hash slot mapping
    // For demo purposes, we'll just group sequentially
    groupSize := 10
    var groups [][]string
    
    for i := 0; i < len(keys); i += groupSize {
        end := i + groupSize
        if end > len(keys) {
            end = len(keys)
        }
        groups = append(groups, keys[i:end])
    }
    
    return groups
}

// Pipeline operations across cluster nodes
func (dc *DistributedCache) BatchSet(items map[string]interface{}, ttl time.Duration) error {
    // Group operations by hash slot to minimize cross-slot operations
    pipe := dc.cluster.client.Pipeline()
    
    for key, value := range items {
        pipe.Set(dc.cluster.ctx, key, value, ttl)
    }
    
    _, err := pipe.Exec(dc.cluster.ctx)
    return err
}

// Cross-slot operations (requires special handling)
func (dc *DistributedCache) AtomicIncrement(keys []string) error {
    // For cross-slot atomic operations, we need to use hash tags
    // or handle them specially
    
    if len(keys) == 0 {
        return nil
    }

    // Check if all keys are in the same slot
    if dc.sameSlot(keys) {
        // Can use transaction
        return dc.cluster.client.Watch(dc.cluster.ctx, func(tx *redis.Tx) error {
            _, err := tx.TxPipelined(dc.cluster.ctx, func(pipe redis.Pipeliner) error {
                for _, key := range keys {
                    pipe.Incr(dc.cluster.ctx, key)
                }
                return nil
            })
            return err
        }, keys...)
    } else {
        // Must handle individually (not atomic across cluster)
        pipe := dc.cluster.client.Pipeline()
        for _, key := range keys {
            pipe.Incr(dc.cluster.ctx, key)
        }
        _, err := pipe.Exec(dc.cluster.ctx)
        return err
    }
}

// sameSlot checks if all keys map to the same hash slot
func (dc *DistributedCache) sameSlot(keys []string) bool {
    if len(keys) <= 1 {
        return true
    }
    
    // Use Redis's hash tag extraction logic
    firstSlot := dc.keySlot(keys[0])
    for _, key := range keys[1:] {
        if dc.keySlot(key) != firstSlot {
            return false
        }
    }
    return true
}

// keySlot calculates the hash slot for a key (simplified)
func (dc *DistributedCache) keySlot(key string) int {
    // Simplified - real implementation would use CRC16
    hash := 0
    for _, c := range key {
        hash = hash*31 + int(c)
    }
    return hash % 16384
}

// Health check for cluster nodes
func (cm *ClusterManager) HealthCheck() (*HealthStatus, error) {
    status := &HealthStatus{
        Healthy:   true,
        Nodes:     make(map[string]NodeHealth),
        Timestamp: time.Now(),
    }

    // Check each node's health
    info, err := cm.GetClusterInfo()
    if err != nil {
        status.Healthy = false
        status.Error = err.Error()
        return status, err
    }

    for _, node := range info.Nodes {
        nodeHealth := NodeHealth{
            Address: node.Address,
            Role:    node.Role,
            Status:  node.Status,
        }

        // Ping individual node
        nodeClient := redis.NewClient(&redis.Options{
            Addr: node.Address,
        })
        
        pingStart := time.Now()
        _, err := nodeClient.Ping(cm.ctx).Result()
        pingDuration := time.Since(pingStart)
        
        nodeClient.Close()

        if err != nil {
            nodeHealth.Healthy = false
            nodeHealth.Error = err.Error()
            status.Healthy = false
        } else {
            nodeHealth.Healthy = true
            nodeHealth.Latency = pingDuration
        }

        status.Nodes[node.ID] = nodeHealth
    }

    return status, nil
}

type HealthStatus struct {
    Healthy   bool                   `json:"healthy"`
    Nodes     map[string]NodeHealth  `json:"nodes"`
    Error     string                 `json:"error,omitempty"`
    Timestamp time.Time              `json:"timestamp"`
}

type NodeHealth struct {
    Address string        `json:"address"`
    Role    string        `json:"role"`
    Status  string        `json:"status"`
    Healthy bool          `json:"healthy"`
    Latency time.Duration `json:"latency"`
    Error   string        `json:"error,omitempty"`
}

// Parse helper functions (simplified implementations)
func parseNodes(nodesStr string) []NodeInfo {
    // In real implementation, parse the CLUSTER NODES output
    return []NodeInfo{
        {ID: "node1", Address: "127.0.0.1:7001", Role: "master", Status: "connected"},
        {ID: "node2", Address: "127.0.0.1:7002", Role: "master", Status: "connected"},
        {ID: "node3", Address: "127.0.0.1:7003", Role: "master", Status: "connected"},
    }
}

func parseSlots(slots []redis.ClusterSlot) []SlotRange {
    var ranges []SlotRange
    for _, slot := range slots {
        ranges = append(ranges, SlotRange{
            Start: slot.Start,
            End:   slot.End,
            Master: NodeInfo{
                Address: fmt.Sprintf("%s:%d", slot.Nodes[0].Addr, slot.Nodes[0].Port),
                Role:    "master",
            },
        })
    }
    return ranges
}

// Demo application
func main() {
    // Configure cluster connection
    config := ClusterConfig{
        Addrs: []string{
            "127.0.0.1:7001",
            "127.0.0.1:7002", 
            "127.0.0.1:7003",
        },
        ReadTimeout:    3 * time.Second,
        WriteTimeout:   3 * time.Second,
        PoolSize:       10,
        MinIdleConns:   5,
        MaxRetries:     3,
        RouteByLatency: true,
        RouteRandomly:  false,
    }

    cluster := NewClusterManager(config)
    defer cluster.client.Close()

    cache := NewDistributedCache(cluster)

    // Test cluster operations
    log.Println("=== Cluster Operations Demo ===")

    // Set some values
    items := map[string]interface{}{
        "user:1":    `{"name":"Alice","age":30}`,
        "user:2":    `{"name":"Bob","age":25}`,
        "user:3":    `{"name":"Charlie","age":35}`,
        "session:1": "session_data_1",
        "session:2": "session_data_2",
    }

    err := cache.BatchSet(items, time.Hour)
    if err != nil {
        log.Printf("Batch set failed: %v", err)
    } else {
        log.Printf("Batch set completed successfully")
    }

    // Get multiple values
    keys := []string{"user:1", "user:2", "user:3", "session:1", "session:2"}
    results, err := cache.MGet(keys)
    if err != nil {
        log.Printf("MGet failed: %v", err)
    } else {
        log.Printf("Retrieved %d values", len(results))
        for key, value := range results {
            log.Printf("%s: %s", key, value)
        }
    }

    // Health check
    log.Println("\n=== Cluster Health Check ===")
    health, err := cluster.HealthCheck()
    if err != nil {
        log.Printf("Health check failed: %v", err)
    } else {
        log.Printf("Cluster healthy: %t", health.Healthy)
        for nodeID, nodeHealth := range health.Nodes {
            log.Printf("Node %s (%s): %s, Latency: %v", 
                      nodeID, nodeHealth.Address, 
                      map[bool]string{true: "Healthy", false: "Unhealthy"}[nodeHealth.Healthy],
                      nodeHealth.Latency)
        }
    }

    // Cluster info
    log.Println("\n=== Cluster Information ===")
    info, err := cluster.GetClusterInfo()
    if err != nil {
        log.Printf("Failed to get cluster info: %v", err)
    } else {
        log.Printf("Cluster has %d nodes and %d slot ranges", 
                  len(info.Nodes), len(info.Slots))
        
        for _, node := range info.Nodes {
            log.Printf("Node: %s (%s) - Role: %s, Status: %s", 
                      node.ID, node.Address, node.Role, node.Status)
        }
    }
}

package main

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

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

// SentinelManager handles Redis Sentinel operations
type SentinelManager struct {
    sentinel *redis.SentinelClient
    client   *redis.Client
    ctx      context.Context
    config   SentinelConfig
    mu       sync.RWMutex
}

type SentinelConfig struct {
    SentinelAddrs []string
    MasterName    string
    Password      string
    DB            int
    PoolSize      int
    MinIdleConns  int
}

func NewSentinelManager(config SentinelConfig) *SentinelManager {
    // Create sentinel client for service discovery
    sentinelClient := redis.NewSentinelClient(&redis.Options{
        Addr: config.SentinelAddrs[0], // Primary sentinel
    })

    // Create failover client that automatically handles master changes
    failoverClient := redis.NewFailoverClient(&redis.FailoverOptions{
        MasterName:    config.MasterName,
        SentinelAddrs: config.SentinelAddrs,
        Password:      config.Password,
        DB:            config.DB,
        PoolSize:      config.PoolSize,
        MinIdleConns:  config.MinIdleConns,
        
        // Failover-specific settings
        MaxRetries:      3,
        MaxRetryBackoff: 512 * time.Millisecond,
        
        // Connection timeouts
        DialTimeout:  5 * time.Second,
        ReadTimeout:  3 * time.Second,
        WriteTimeout: 3 * time.Second,
    })

    sm := &SentinelManager{
        sentinel: sentinelClient,
        client:   failoverClient,
        ctx:      context.Background(),
        config:   config,
    }

    return sm
}

// GetMasterInfo retrieves current master information
func (sm *SentinelManager) GetMasterInfo() (*MasterInfo, error) {
    result, err := sm.sentinel.Master(sm.ctx, sm.config.MasterName).Result()
    if err != nil {
        return nil, fmt.Errorf("failed to get master info: %w", err)
    }

    return parseMasterInfo(result), nil
}

// GetReplicaInfo retrieves replica information
func (sm *SentinelManager) GetReplicaInfo() ([]ReplicaInfo, error) {
    result, err := sm.sentinel.Slaves(sm.ctx, sm.config.MasterName).Result()
    if err != nil {
        return nil, fmt.Errorf("failed to get replica info: %w", err)
    }

    var replicas []ReplicaInfo
    for _, replica := range result {
        replicas = append(replicas, parseReplicaInfo(replica))
    }

    return replicas, nil
}

// GetSentinelInfo retrieves sentinel information
func (sm *SentinelManager) GetSentinelInfo() ([]SentinelInfo, error) {
    result, err := sm.sentinel.Sentinels(sm.ctx, sm.config.MasterName).Result()
    if err != nil {
        return nil, fmt.Errorf("failed to get sentinel info: %w", err)
    }

    var sentinels []SentinelInfo
    for _, sentinel := range result {
        sentinels = append(sentinels, parseSentinelInfo(sentinel))
    }

    return sentinels, nil
}

type MasterInfo struct {
    Name               string        `json:"name"`
    IP                 string        `json:"ip"`
    Port               string        `json:"port"`
    RunID              string        `json:"runid"`
    Flags              []string      `json:"flags"`
    PendingCommands    int           `json:"pending_commands"`
    LastPingSent       time.Duration `json:"last_ping_sent"`
    LastOkPingReply    time.Duration `json:"last_ok_ping_reply"`
    LastPingReply      time.Duration `json:"last_ping_reply"`
    DownAfterPeriod    time.Duration `json:"down_after_period"`
    InfoRefresh        time.Duration `json:"info_refresh"`
    RoleReported       string        `json:"role_reported"`
    RoleReportedTime   time.Duration `json:"role_reported_time"`
    ConfigEpoch        int           `json:"config_epoch"`
    NumSlaves          int           `json:"num_slaves"`
    NumOtherSentinels  int           `json:"num_other_sentinels"`
    Quorum             int           `json:"quorum"`
    FailoverTimeout    time.Duration `json:"failover_timeout"`
    ParallelSyncs      int           `json:"parallel_syncs"`
}

type ReplicaInfo struct {
    Name               string        `json:"name"`
    IP                 string        `json:"ip"`
    Port               string        `json:"port"`
    RunID              string        `json:"runid"`
    Flags              []string      `json:"flags"`
    PendingCommands    int           `json:"pending_commands"`
    LastPingSent       time.Duration `json:"last_ping_sent"`
    LastOkPingReply    time.Duration `json:"last_ok_ping_reply"`
    LastPingReply      time.Duration `json:"last_ping_reply"`
    DownAfterPeriod    time.Duration `json:"down_after_period"`
    InfoRefresh        time.Duration `json:"info_refresh"`
    RoleReported       string        `json:"role_reported"`
    RoleReportedTime   time.Duration `json:"role_reported_time"`
    MasterLinkDownTime time.Duration `json:"master_link_down_time"`
    MasterLinkStatus   string        `json:"master_link_status"`
    MasterHost         string        `json:"master_host"`
    MasterPort         string        `json:"master_port"`
    SlavePriority      int           `json:"slave_priority"`
    SlaveReplOffset    int           `json:"slave_repl_offset"`
}

type SentinelInfo struct {
    Name               string        `json:"name"`
    IP                 string        `json:"ip"`
    Port               string        `json:"port"`
    RunID              string        `json:"runid"`
    Flags              []string      `json:"flags"`
    PendingCommands    int           `json:"pending_commands"`
    LastPingSent       time.Duration `json:"last_ping_sent"`
    LastOkPingReply    time.Duration `json:"last_ok_ping_reply"`
    LastPingReply      time.Duration `json:"last_ping_reply"`
    DownAfterPeriod    time.Duration `json:"down_after_period"`
    LastHelloMessage   time.Duration `json:"last_hello_message"`
}

// HighAvailabilityCache implements cache with automatic failover
type HighAvailabilityCache struct {
    sentinel *SentinelManager
}

func NewHighAvailabilityCache(sentinel *SentinelManager) *HighAvailabilityCache {
    return &HighAvailabilityCache{sentinel: sentinel}
}

// Set with automatic failover handling
func (hac *HighAvailabilityCache) Set(key string, value interface{}, ttl time.Duration) error {
    return hac.sentinel.client.Set(hac.sentinel.ctx, key, value, ttl).Err()
}

// Get with automatic failover handling
func (hac *HighAvailabilityCache) Get(key string) (string, error) {
    return hac.sentinel.client.Get(hac.sentinel.ctx, key).Result()
}

// SetWithRetry implements retry logic for failover scenarios
func (hac *HighAvailabilityCache) SetWithRetry(key string, value interface{}, ttl time.Duration, maxRetries int) error {
    var lastErr error
    
    for attempt := 0; attempt <= maxRetries; attempt++ {
        err := hac.Set(key, value, ttl)
        if err == nil {
            return nil
        }
        
        lastErr = err
        
        // Check if this is a failover-related error
        if isFailoverError(err) {
            log.Printf("Failover detected on attempt %d, retrying...", attempt+1)
            time.Sleep(time.Duration(attempt*100) * time.Millisecond)
            continue
        }
        
        // Non-failover error, don't retry
        return err
    }
    
    return fmt.Errorf("failed after %d retries: %w", maxRetries, lastErr)
}

// isFailoverError checks if an error is related to failover
func isFailoverError(err error) bool {
    if err == nil {
        return false
    }
    
    errStr := strings.ToLower(err.Error())
    failoverKeywords := []string{
        "connection refused",
        "connection reset",
        "no route to host",
        "timeout",
        "broken pipe",
        "readonly",
    }
    
    for _, keyword := range failoverKeywords {
        if strings.Contains(errStr, keyword) {
            return true
        }
    }
    
    return false
}

// MonitorFailover monitors for failover events
func (sm *SentinelManager) MonitorFailover() {
    // Subscribe to sentinel events
    pubsub := sm.sentinel.Subscribe(sm.ctx, "__sentinel__:hello")
    defer pubsub.Close()

    log.Println("Starting failover monitoring...")

    for {
        select {
        case msg := <-pubsub.Channel():
            log.Printf("Sentinel event: %s", msg.Payload)
            
            // Parse and handle different event types
            if strings.Contains(msg.Payload, "+switch-master") {
                log.Println("Master switch detected!")
                sm.handleMasterSwitch()
            }
            
        case <-sm.ctx.Done():
            log.Println("Failover monitoring stopped")
            return
        }
    }
}

func (sm *SentinelManager) handleMasterSwitch() {
    // Get new master info
    masterInfo, err := sm.GetMasterInfo()
    if err != nil {
        log.Printf("Failed to get new master info: %v", err)
        return
    }
    
    log.Printf("New master: %s:%s", masterInfo.IP, masterInfo.Port)
    
    // Trigger application-specific failover handling
    sm.notifyApplications(masterInfo)
}

func (sm *SentinelManager) notifyApplications(masterInfo *MasterInfo) {
    // In a real application, this would notify other components
    // about the master change
    log.Printf("Notifying applications about master change to %s:%s", 
              masterInfo.IP, masterInfo.Port)
}

// Parse helper functions
func parseMasterInfo(data []interface{}) *MasterInfo {
    info := &MasterInfo{}
    
    for i := 0; i < len(data); i += 2 {
        if i+1 >= len(data) {
            break
        }
        
        key := data[i].(string)
        value := data[i+1].(string)
        
        switch key {
        case "name":
            info.Name = value
        case "ip":
            info.IP = value
        case "port":
            info.Port = value
        case "runid":
            info.RunID = value
        case "flags":
            info.Flags = strings.Split(value, ",")
        case "role-reported":
            info.RoleReported = value
        case "num-slaves":
            fmt.Sscanf(value, "%d", &info.NumSlaves)
        case "num-other-sentinels":
            fmt.Sscanf(value, "%d", &info.NumOtherSentinels)
        case "quorum":
            fmt.Sscanf(value, "%d", &info.Quorum)
        case "parallel-syncs":
            fmt.Sscanf(value, "%d", &info.ParallelSyncs)
        }
    }
    
    return info
}

func parseReplicaInfo(data []interface{}) ReplicaInfo {
    // Similar parsing logic for replica info
    return ReplicaInfo{}
}

func parseSentinelInfo(data []interface{}) SentinelInfo {
    // Similar parsing logic for sentinel info
    return SentinelInfo{}
}

// Demo application
func main() {
    config := SentinelConfig{
        SentinelAddrs: []string{
            "127.0.0.1:26379",
            "127.0.0.1:26380",
            "127.0.0.1:26381",
        },
        MasterName:   "mymaster",
        Password:     "",
        DB:           0,
        PoolSize:     10,
        MinIdleConns: 5,
    }

    sentinel := NewSentinelManager(config)
    defer sentinel.client.Close()
    defer sentinel.sentinel.Close()

    cache := NewHighAvailabilityCache(sentinel)

    // Start failover monitoring in background
    go sentinel.MonitorFailover()

    // Test high availability operations
    log.Println("=== High Availability Demo ===")

    // Set some values with retry logic
    for i := 0; i < 10; i++ {
        key := fmt.Sprintf("ha:test:%d", i)
        value := fmt.Sprintf("value_%d", i)
        
        err := cache.SetWithRetry(key, value, time.Hour, 3)
        if err != nil {
            log.Printf("Failed to set %s: %v", key, err)
        } else {
            log.Printf("Set %s successfully", key)
        }
        
        time.Sleep(500 * time.Millisecond)
    }

    // Get master info
    log.Println("\n=== Sentinel Information ===")
    masterInfo, err := sentinel.GetMasterInfo()
    if err != nil {
        log.Printf("Failed to get master info: %v", err)
    } else {
        log.Printf("Master: %s:%s (RunID: %s)", 
                  masterInfo.IP, masterInfo.Port, masterInfo.RunID)
        log.Printf("Slaves: %d, Sentinels: %d, Quorum: %d", 
                  masterInfo.NumSlaves, masterInfo.NumOtherSentinels, masterInfo.Quorum)
    }

    // Get replica info
    replicas, err := sentinel.GetReplicaInfo()
    if err != nil {
        log.Printf("Failed to get replica info: %v", err)
    } else {
        log.Printf("Found %d replicas", len(replicas))
        for i, replica := range replicas {
            log.Printf("Replica %d: %s:%s", i+1, replica.IP, replica.Port)
        }
    }

    // Keep running to monitor failover events
    log.Println("\nMonitoring for failover events... (Press Ctrl+C to stop)")
    select {}
}

# Create cluster nodes
redis-server --port 7001 --cluster-enabled yes --cluster-config-file nodes-7001.conf
redis-server --port 7002 --cluster-enabled yes --cluster-config-file nodes-7002.conf
# ... repeat for ports 7003-7006

# Initialize cluster
redis-cli --cluster create 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 \
  127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 --cluster-replicas 1

# Add new node
redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7001

# Reshard data
redis-cli --cluster reshard 127.0.0.1:7001 --cluster-slots 1000 \
  --cluster-from <source-node-id> --cluster-to <target-node-id>

# Cluster diagnosis
redis-cli --cluster check 127.0.0.1:7001
redis-cli --cluster info 127.0.0.1:7001
redis-cli --cluster nodes 127.0.0.1:7001

# Sentinel monitoring  
redis-cli -p 26379 sentinel masters
redis-cli -p 26379 sentinel slaves mymaster
redis-cli -p 26379 sentinel sentinels mymaster

# Performance monitoring
redis-cli --cluster call 127.0.0.1:7001 info stats
redis-cli --latency -i 1 -h 127.0.0.1 -p 7001

# Docker Compose for Redis Cluster
version: '3.8'
services:
  redis-node1:
    image: redis:7-alpine
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf
    volumes:
      - redis1-data:/data
    ports:
      - "7001:6379"
      - "17001:16379"
    
  redis-node2:
    image: redis:7-alpine  
    command: redis-server --cluster-enabled yes --cluster-config-file nodes.conf
    volumes:
      - redis2-data:/data
    ports:
      - "7002:6379"
      - "17002:16379"

  # ... additional nodes