Opinion: Why Redis 8 Cluster Beats Memcached 1.6 for Scalable Apps

package main

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

    "github.com/bradfitz/gomemcache/memcache"
    "github.com/redis/go-redis/v9"
)

// BenchmarkConfig holds configuration for cache benchmark runs
type BenchmarkConfig struct {
    KeyPrefix    string
    ValueSize    int // bytes
    TotalOps     int
    Concurrency  int
    Value        []byte
}

// BenchmarkResult stores throughput and latency metrics
type BenchmarkResult struct {
    CacheType    string
    TotalOps     int
    Duration     time.Duration
    OpsPerSec    float64
    AvgLatency   time.Duration
    P99Latency   time.Duration
    Errors       int
}

func runMemcachedBenchmark(cfg BenchmarkConfig, client *memcache.Client) BenchmarkResult {
    var (
        wg       sync.WaitGroup
        opsDone  int
        errors   int
        latencies []time.Duration
        mu        sync.Mutex
    )

    // Pre-generate 1KB value if not provided
    if cfg.Value == nil {
        cfg.Value = make([]byte, cfg.ValueSize)
        for i := range cfg.Value {
            cfg.Value[i] = byte(i % 256)
        }
    }

    opsPerWorker := cfg.TotalOps / cfg.Concurrency
    wg.Add(cfg.Concurrency)

    start := time.Now()
    for i := 0; i < cfg.Concurrency; i++ {
        go func(workerID int) {
            defer wg.Done()
            for j := 0; j < opsPerWorker; j++ {
                key := fmt.Sprintf("%s:%d:%d", cfg.KeyPrefix, workerID, j)
                opStart := time.Now()
                // Set value
                err := client.Set(&memcache.Item{
                    Key:   key,
                    Value: cfg.Value,
                })
                if err != nil {
                    mu.Lock()
                    errors++
                    mu.Unlock()
                    continue
                }
                // Get value to validate
                _, err = client.Get(key)
                if err != nil {
                    mu.Lock()
                    errors++
                    mu.Unlock()
                    continue
                }
                latency := time.Since(opStart)
                mu.Lock()
                latencies = append(latencies, latency)
                opsDone++
                mu.Unlock()
            }
        }(i)
    }
    wg.Wait()
    dur := time.Since(start)

    // Calculate p99 latency
    var p99 time.Duration
    if len(latencies) > 0 {
        mu.Lock()
        defer mu.Unlock()
        sortLatencies(latencies)
        p99Idx := int(float64(len(latencies)) * 0.99)
        if p99Idx >= len(latencies) {
            p99Idx = len(latencies) - 1
        }
        p99 = latencies[p99Idx]
    }

    return BenchmarkResult{
        CacheType:  "Memcached 1.6",
        TotalOps:   opsDone,
        Duration:   dur,
        OpsPerSec:  float64(opsDone) / dur.Seconds(),
        AvgLatency: dur / time.Duration(opsDone),
        P99Latency: p99,
        Errors:     errors,
    }
}

// sortLatencies is a helper to sort latency durations ascending
func sortLatencies(latencies []time.Duration) {
    for i := 0; i < len(latencies); i++ {
        for j := i + 1; j < len(latencies); j++ {
            if latencies[i] > latencies[j] {
                latencies[i], latencies[j] = latencies[j], latencies[i]
            }
        }
    }
}

func runRedisBenchmark(cfg BenchmarkConfig, client *redis.ClusterClient) BenchmarkResult {
    var (
        wg       sync.WaitGroup
        opsDone  int
        errors   int
        latencies []time.Duration
        mu        sync.Mutex
        ctx       = context.Background()
    )

    if cfg.Value == nil {
        cfg.Value = make([]byte, cfg.ValueSize)
        for i := range cfg.Value {
            cfg.Value[i] = byte(i % 256)
        }
    }

    opsPerWorker := cfg.TotalOps / cfg.Concurrency
    wg.Add(cfg.Concurrency)

    start := time.Now()
    for i := 0; i < cfg.Concurrency; i++ {
        go func(workerID int) {
            defer wg.Done()
            for j := 0; j < opsPerWorker; j++ {
                key := fmt.Sprintf("%s:%d:%d", cfg.KeyPrefix, workerID, j)
                opStart := time.Now()
                // Set value
                err := client.Set(ctx, key, cfg.Value, 0).Err()
                if err != nil {
                    mu.Lock()
                    errors++
                    mu.Unlock()
                    continue
                }
                // Get value to validate
                _, err = client.Get(ctx, key).Result()
                if err != nil {
                    mu.Lock()
                    errors++
                    mu.Unlock()
                    continue
                }
                latency := time.Since(opStart)
                mu.Lock()
                latencies = append(latencies, latency)
                opsDone++
                mu.Unlock()
            }
        }(i)
    }
    wg.Wait()
    dur := time.Since(start)

    // Calculate p99 latency
    var p99 time.Duration
    if len(latencies) > 0 {
        mu.Lock()
        defer mu.Unlock()
        sortLatencies(latencies)
        p99Idx := int(float64(len(latencies)) * 0.99)
        if p99Idx >= len(latencies) {
            p99Idx = len(latencies) - 1
        }
        p99 = latencies[p99Idx]
    }

    return BenchmarkResult{
        CacheType:  "Redis 8 Cluster",
        TotalOps:   opsDone,
        Duration:   dur,
        OpsPerSec:  float64(opsDone) / dur.Seconds(),
        AvgLatency: dur / time.Duration(opsDone),
        P99Latency: p99,
        Errors:     errors,
    }
}

func main() {
    // Initialize Memcached client
    memcachedClient := memcache.New("cache1:11211", "cache2:11211", "cache3:11211")
    // Initialize Redis Cluster client
    redisClient := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs: []string{"redis1:6379", "redis2:6379", "redis3:6379"},
        PoolSize: 100,
    })

    cfg := BenchmarkConfig{
        KeyPrefix:   "bench",
        ValueSize:   1024, // 1KB
        TotalOps:    100000,
        Concurrency: 10,
    }

    memcachedRes := runMemcachedBenchmark(cfg, memcachedClient)
    redisRes := runRedisBenchmark(cfg, redisClient)

    fmt.Printf("Memcached 1.6 Results:\n")
    fmt.Printf("  Ops/sec: %.2f\n", memcachedRes.OpsPerSec)
    fmt.Printf("  Avg Latency: %v\n", memcachedRes.AvgLatency)
    fmt.Printf("  P99 Latency: %v\n", memcachedRes.P99Latency)
    fmt.Printf("  Errors: %d\n\n", memcachedRes.Errors)

    fmt.Printf("Redis 8 Cluster Results:\n")
    fmt.Printf("  Ops/sec: %.2f\n", redisRes.OpsPerSec)
    fmt.Printf("  Avg Latency: %v\n", redisRes.AvgLatency)
    fmt.Printf("  P99 Latency: %v\n", redisRes.P99Latency)
    fmt.Printf("  Errors: %d\n", redisRes.Errors)
}

package main

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

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

// RedisClusterConfig holds configuration for Redis 8 Cluster client
type RedisClusterConfig struct {
    Addrs        []string
    PoolSize     int
    MinIdleConns int
    DialTimeout  time.Duration
    ReadTimeout  time.Duration
    WriteTimeout time.Duration
}

// NewRedisClusterClient initializes a production-grade Redis 8 Cluster client with error handling
func NewRedisClusterClient(cfg RedisClusterConfig) (*redis.ClusterClient, error) {
    if len(cfg.Addrs) == 0 {
        return nil, fmt.Errorf("no Redis cluster addresses provided")
    }
    if cfg.PoolSize <= 0 {
        cfg.PoolSize = 50 // default pool size
    }
    if cfg.DialTimeout <= 0 {
        cfg.DialTimeout = 5 * time.Second
    }

    client := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs:        cfg.Addrs,
        PoolSize:      cfg.PoolSize,
        MinIdleConns: cfg.MinIdleConns,
        DialTimeout:  cfg.DialTimeout,
        ReadTimeout:  cfg.ReadTimeout,
        WriteTimeout: cfg.WriteTimeout,
        // Enable retry for transient errors
        MaxRetries:      3,
        MinRetryBackoff: 100 * time.Millisecond,
        MaxRetryBackoff: 2 * time.Second,
    })

    // Verify cluster connectivity
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // Ping all cluster nodes
    err := client.ForEachNode(ctx, func(ctx context.Context, node *redis.Client) error {
        return node.Ping(ctx).Err()
    })
    if err != nil {
        return nil, fmt.Errorf("failed to ping cluster nodes: %w", err)
    }

    // Verify cluster state
    info, err := client.ClusterInfo(ctx).Result()
    if err != nil {
        return nil, fmt.Errorf("failed to get cluster info: %w", err)
    }
    if !contains(info, "cluster_state:ok") {
        return nil, fmt.Errorf("cluster is not in healthy state: %s", info)
    }

    log.Println("Successfully initialized Redis 8 Cluster client")
    return client, nil
}

// contains checks if a string contains a substring (simplified helper)
func contains(s, substr string) bool {
    for i := 0; i <= len(s)-len(substr); i++ {
        if s[i:i+len(substr)] == substr {
            return true
        }
    }
    return false
}

func main() {
    cfg := RedisClusterConfig{
        Addrs:        []string{"redis-node1:6379", "redis-node2:6379", "redis-node3:6379"},
        PoolSize:     100,
        MinIdleConns: 20,
        DialTimeout:  3 * time.Second,
        ReadTimeout:  1 * time.Second,
        WriteTimeout: 1 * time.Second,
    }

    client, err := NewRedisClusterClient(cfg)
    if err != nil {
        log.Fatalf("Failed to create Redis client: %v", err)
    }
    defer client.Close()

    // Example operation: store and retrieve a product hash
    ctx := context.Background()
    productKey := "product:12345"
    err = client.HSet(ctx, productKey, map[string]interface{}{
        "name":  "Wireless Headphones",
        "price": 129.99,
        "stock": 450,
    }).Err()
    if err != nil {
        log.Fatalf("Failed to HSet product: %v", err)
    }

    // Retrieve partial product data (only name and price)
    productData, err := client.HMGet(ctx, productKey, "name", "price").Result()
    if err != nil {
        log.Fatalf("Failed to HMGet product: %v", err)
    }

    fmt.Printf("Product data: %v\n", productData)
}

package main

import (
    "context"
    "fmt"
    "log"
    "hash/fnv"
    "sync"
    "time"

    "github.com/bradfitz/gomemcache/memcache"
)

// MemcachedShard is a single Memcached node client
type MemcachedShard struct {
    Client *memcache.Client
    Addr   string
}

// ManualShardingClient implements client-side sharding for Memcached 1.6
type ManualShardingClient struct {
    shards []*MemcachedShard
    mu     sync.RWMutex
}

// NewManualShardingClient initializes a sharded Memcached client
func NewManualShardingClient(addrs []string) (*ManualShardingClient, error) {
    if len(addrs) == 0 {
        return nil, fmt.Errorf("no Memcached addresses provided")
    }

    shards := make([]*MemcachedShard, len(addrs))
    for i, addr := range addrs {
        shards[i] = &MemcachedShard{
            Client: memcache.New(addr),
            Addr:   addr,
        }
        // Test connectivity to each shard
        err := shards[i].Client.Ping()
        if err != nil {
            return nil, fmt.Errorf("failed to ping Memcached shard %s: %w", addr, err)
        }
    }

    return &ManualShardingClient{shards: shards}, nil
}

// getShard returns the shard for a given key using FNV-1a hash
func (c *ManualShardingClient) getShard(key string) *MemcachedShard {
    h := fnv.New32a()
    h.Write([]byte(key))
    hash := h.Sum32()
    shardIdx := hash % uint32(len(c.shards))
    return c.shards[shardIdx]
}

// Set stores a key-value pair in the appropriate shard
func (c *ManualShardingClient) Set(ctx context.Context, key string, value []byte, expiration time.Duration) error {
    shard := c.getShard(key)
    item := &memcache.Item{
        Key:        key,
        Value:      value,
        Expiration: int32(expiration.Seconds()),
    }

    err := shard.Client.Set(item)
    if err != nil {
        return fmt.Errorf("failed to set key %s on shard %s: %w", key, shard.Addr, err)
    }
    return nil
}

// Get retrieves a key-value pair from the appropriate shard
func (c *ManualShardingClient) Get(ctx context.Context, key string) ([]byte, error) {
    shard := c.getShard(key)
    item, err := shard.Client.Get(key)
    if err != nil {
        return nil, fmt.Errorf("failed to get key %s from shard %s: %w", key, shard.Addr, err)
    }
    return item.Value, nil
}

// AddShard adds a new Memcached node to the cluster (requires manual rebalancing)
func (c *ManualShardingClient) AddShard(addr string) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    // Check if shard already exists
    for _, shard := range c.shards {
        if shard.Addr == addr {
            return fmt.Errorf("shard %s already exists", addr)
        }
    }

    // Initialize new shard
    newShard := &MemcachedShard{
        Client: memcache.New(addr),
        Addr:   addr,
    }
    err := newShard.Client.Ping()
    if err != nil {
        return fmt.Errorf("failed to ping new shard %s: %w", addr, err)
    }

    // Add to shards (note: this does not rebalance existing keys, requires manual key migration)
    c.shards = append(c.shards, newShard)
    log.Printf("Added new shard %s, total shards: %d (manual rebalancing required)", addr, len(c.shards))
    return nil
}

func main() {
    shards := []string{"memcached1:11211", "memcached2:11211", "memcached3:11211"}
    client, err := NewManualShardingClient(shards)
    if err != nil {
        log.Fatalf("Failed to create sharded Memcached client: %v", err)
    }

    ctx := context.Background()
    // Set a value
    err = client.Set(ctx, "user:123", []byte("John Doe"), 1*time.Hour)
    if err != nil {
        log.Fatalf("Failed to set user: %v", err)
    }

    // Get the value
    val, err := client.Get(ctx, "user:123")
    if err != nil {
        log.Fatalf("Failed to get user: %v", err)
    }

    fmt.Printf("Retrieved user: %s\n", val)

    // Add a new shard (manual rebalancing needed)
    err = client.AddShard("memcached4:11211")
    if err != nil {
        log.Fatalf("Failed to add shard: %v", err)
    }
}

// Update only the price field of a product hash
err := redisClient.HSet(ctx, "product:12345", "price", 139.99).Err()
if err != nil {
    log.Fatalf("Failed to update product price: %v", err)
}

// Trigger cluster rebalancing via go-redis
err := redisClient.ClusterRebalance(ctx, &redis.ClusterRebalanceOptions{
    UseEmptyMasters: true,
}).Err()
if err != nil {
    log.Fatalf("Failed to rebalance cluster: %v", err)
}

// Configure Redis client with optimized pooling and pipelining
redisClient := redis.NewClusterClient(&redis.ClusterOptions{
    Addrs:        []string{"redis1:6379", "redis2:6379"},
    PoolSize:     100, // 100 connections per node
    MinIdleConns: 20,
    // Enable pipelining for batch operations
    MaxPipeline: 100,
})