Add redis-proxy: dual-write Redis proxy with pub/sub forwarding

cmd/redis-proxy/main.go

@@ -0,0 +1,120 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/bootjp/elastickv/proxy"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+const sentryFlushTimeout = 2 * time.Second
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintf(os.Stderr, "error: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	cfg := proxy.DefaultConfig()
+	var modeStr string
+
+	flag.StringVar(&cfg.ListenAddr, "listen", cfg.ListenAddr, "Proxy listen address")
+	flag.StringVar(&cfg.PrimaryAddr, "primary", cfg.PrimaryAddr, "Primary (Redis) address")
+	flag.IntVar(&cfg.PrimaryDB, "primary-db", cfg.PrimaryDB, "Primary Redis DB number")
+	flag.StringVar(&cfg.PrimaryPassword, "primary-password", cfg.PrimaryPassword, "Primary Redis password")
+	flag.StringVar(&cfg.SecondaryAddr, "secondary", cfg.SecondaryAddr, "Secondary (ElasticKV) address")
+	flag.IntVar(&cfg.SecondaryDB, "secondary-db", cfg.SecondaryDB, "Secondary Redis DB number")
+	flag.StringVar(&cfg.SecondaryPassword, "secondary-password", cfg.SecondaryPassword, "Secondary Redis password")
+	flag.StringVar(&modeStr, "mode", "dual-write", "Proxy mode: redis-only, dual-write, dual-write-shadow, elastickv-primary, elastickv-only")
+	flag.DurationVar(&cfg.SecondaryTimeout, "secondary-timeout", cfg.SecondaryTimeout, "Secondary write timeout")
+	flag.DurationVar(&cfg.ShadowTimeout, "shadow-timeout", cfg.ShadowTimeout, "Shadow read timeout")
+	flag.StringVar(&cfg.SentryDSN, "sentry-dsn", cfg.SentryDSN, "Sentry DSN (empty = disabled)")
+	flag.StringVar(&cfg.SentryEnv, "sentry-env", cfg.SentryEnv, "Sentry environment")
+	flag.Float64Var(&cfg.SentrySampleRate, "sentry-sample", cfg.SentrySampleRate, "Sentry sample rate")
+	flag.StringVar(&cfg.MetricsAddr, "metrics", cfg.MetricsAddr, "Prometheus metrics address")
+	flag.Parse()
+
+	mode, ok := proxy.ParseProxyMode(modeStr)
+	if !ok {
+		return fmt.Errorf("unknown mode: %s", modeStr)
+	}
+	cfg.Mode = mode
+
+	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
+
+	// Sentry
+	sentryReporter := proxy.NewSentryReporter(cfg.SentryDSN, cfg.SentryEnv, cfg.SentrySampleRate, logger)
+	defer sentryReporter.Flush(sentryFlushTimeout)
+
+	// Prometheus
+	reg := prometheus.NewRegistry()
+	metrics := proxy.NewProxyMetrics(reg)
+
+	// Backends
+	primaryOpts := proxy.DefaultBackendOptions()
+	primaryOpts.DB = cfg.PrimaryDB
+	primaryOpts.Password = cfg.PrimaryPassword
+	secondaryOpts := proxy.DefaultBackendOptions()
+	secondaryOpts.DB = cfg.SecondaryDB
+	secondaryOpts.Password = cfg.SecondaryPassword
+
+	var primary, secondary proxy.Backend
+	switch cfg.Mode {
+	case proxy.ModeElasticKVPrimary, proxy.ModeElasticKVOnly:
+		primary = proxy.NewRedisBackendWithOptions(cfg.SecondaryAddr, "elastickv", secondaryOpts)
+		secondary = proxy.NewRedisBackendWithOptions(cfg.PrimaryAddr, "redis", primaryOpts)
+	case proxy.ModeRedisOnly, proxy.ModeDualWrite, proxy.ModeDualWriteShadow:
+		primary = proxy.NewRedisBackendWithOptions(cfg.PrimaryAddr, "redis", primaryOpts)
+		secondary = proxy.NewRedisBackendWithOptions(cfg.SecondaryAddr, "elastickv", secondaryOpts)
+	}
+	defer primary.Close()
+	defer secondary.Close()
+
+	dual := proxy.NewDualWriter(primary, secondary, cfg, metrics, sentryReporter, logger)
+	srv := proxy.NewProxyServer(cfg, dual, metrics, sentryReporter, logger)
+
+	// Context for graceful shutdown
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+
+	// Start metrics server
+	go func() {
+		mux := http.NewServeMux()
+		mux.Handle("/metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
+		var lc net.ListenConfig
+		ln, err := lc.Listen(ctx, "tcp", cfg.MetricsAddr)
+		if err != nil {
+			logger.Error("metrics listen failed", "addr", cfg.MetricsAddr, "err", err)
+			return
+		}
+		metricsSrv := &http.Server{Handler: mux, ReadHeaderTimeout: time.Second}
+		go func() {
+			<-ctx.Done()
+			if err := metricsSrv.Shutdown(context.Background()); err != nil {
+				logger.Warn("metrics server shutdown error", "err", err)
+			}
+		}()
+		logger.Info("metrics server starting", "addr", cfg.MetricsAddr)
+		if err := metricsSrv.Serve(ln); err != nil && err != http.ErrServerClosed {
+			logger.Error("metrics server error", "err", err)
+		}
+	}()
+
+	// Start proxy
+	if err := srv.ListenAndServe(ctx); err != nil {
+		return fmt.Errorf("proxy server: %w", err)
+	}
+	return nil
+}

proxy/backend.go

@@ -0,0 +1,114 @@
+package proxy
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	defaultPoolSize     = 128
+	defaultDialTimeout  = 5 * time.Second
+	defaultReadTimeout  = 3 * time.Second
+	defaultWriteTimeout = 3 * time.Second
+)
+
+// Backend abstracts a Redis-protocol endpoint (real Redis or ElasticKV).
+type Backend interface {
+	// Do sends a single command and returns its result.
+	Do(ctx context.Context, args ...any) *redis.Cmd
+	// Pipeline sends multiple commands in a pipeline.
+	Pipeline(ctx context.Context, cmds [][]any) ([]*redis.Cmd, error)
+	// Close releases the underlying connection.
+	Close() error
+	// Name identifies this backend for logging and metrics.
+	Name() string
+}
+
+// BackendOptions configures the underlying go-redis connection pool.
+type BackendOptions struct {
+	DB           int
+	Password     string
+	PoolSize     int
+	DialTimeout  time.Duration
+	ReadTimeout  time.Duration
+	WriteTimeout time.Duration
+}
+
+// DefaultBackendOptions returns reasonable defaults for a proxy backend.
+func DefaultBackendOptions() BackendOptions {
+	return BackendOptions{
+		PoolSize:     defaultPoolSize,
+		DialTimeout:  defaultDialTimeout,
+		ReadTimeout:  defaultReadTimeout,
+		WriteTimeout: defaultWriteTimeout,
+	}
+}
+
+// PubSubBackend is an optional interface for backends that support
+// creating dedicated PubSub connections.
+type PubSubBackend interface {
+	NewPubSub(ctx context.Context) *redis.PubSub
+}
+
+// RedisBackend connects to an upstream Redis instance via go-redis.
+type RedisBackend struct {
+	client *redis.Client
+	name   string
+}
+
+// NewRedisBackend creates a Backend targeting a Redis server with default pool options.
+func NewRedisBackend(addr string, name string) *RedisBackend {
+	return NewRedisBackendWithOptions(addr, name, DefaultBackendOptions())
+}
+
+// NewRedisBackendWithOptions creates a Backend with explicit pool configuration.
+func NewRedisBackendWithOptions(addr string, name string, opts BackendOptions) *RedisBackend {
+	return &RedisBackend{
+		client: redis.NewClient(&redis.Options{
+			Addr:         addr,
+			DB:           opts.DB,
+			Password:     opts.Password,
+			PoolSize:     opts.PoolSize,
+			DialTimeout:  opts.DialTimeout,
+			ReadTimeout:  opts.ReadTimeout,
+			WriteTimeout: opts.WriteTimeout,
+		}),
+		name: name,
+	}
+}
+
+func (b *RedisBackend) Do(ctx context.Context, args ...any) *redis.Cmd {
+	return b.client.Do(ctx, args...)
+}
+
+func (b *RedisBackend) Pipeline(ctx context.Context, cmds [][]any) ([]*redis.Cmd, error) {
+	pipe := b.client.Pipeline()
+	results := make([]*redis.Cmd, len(cmds))
+	for i, args := range cmds {
+		results[i] = pipe.Do(ctx, args...)
+	}
+	_, err := pipe.Exec(ctx)
+	if err != nil {
+		return results, fmt.Errorf("pipeline exec: %w", err)
+	}
+	return results, nil
+}
+
+func (b *RedisBackend) Close() error {
+	if err := b.client.Close(); err != nil {
+		return fmt.Errorf("close %s backend: %w", b.name, err)
+	}
+	return nil
+}
+
+func (b *RedisBackend) Name() string {
+	return b.name
+}
+
+// NewPubSub creates a dedicated PubSub connection (not from the pool).
+func (b *RedisBackend) NewPubSub(ctx context.Context) *redis.PubSub {
+	return b.client.Subscribe(ctx)
+}

proxy/command.go

@@ -0,0 +1,146 @@
+package proxy
+
+import "strings"
+
+// CommandCategory classifies a Redis command for routing purposes.
+type CommandCategory int
+
+const (
+	CmdRead     CommandCategory = iota // GET, HGET, LRANGE, ZRANGE, etc.
+	CmdWrite                           // SET, DEL, HSET, LPUSH, ZADD, etc.
+	CmdBlocking                        // BZPOPMIN, XREAD (with BLOCK)
+	CmdPubSub                          // SUBSCRIBE, PUBLISH, PUBSUB
+	CmdAdmin                           // PING, INFO, CLIENT, SELECT, QUIT, DBSIZE, SCAN, AUTH
+	CmdTxn                             // MULTI, EXEC, DISCARD
+	CmdScript                          // EVAL, EVALSHA
+)
+
+var commandTable = map[string]CommandCategory{
+	// Read commands
+	"GET":              CmdRead,
+	"GETDEL":           CmdWrite, // read+write → write
+	"HGET":             CmdRead,
+	"HGETALL":          CmdRead,
+	"HEXISTS":          CmdRead,
+	"HLEN":             CmdRead,
+	"HMGET":            CmdRead,
+	"EXISTS":           CmdRead,
+	"KEYS":             CmdRead,
+	"LINDEX":           CmdRead,
+	"LLEN":             CmdRead,
+	"LPOS":             CmdRead,
+	"LRANGE":           CmdRead,
+	"PTTL":             CmdRead,
+	"TTL":              CmdRead,
+	"TYPE":             CmdRead,
+	"SCARD":            CmdRead,
+	"SISMEMBER":        CmdRead,
+	"SMEMBERS":         CmdRead,
+	"XLEN":             CmdRead,
+	"XRANGE":           CmdRead,
+	"XREVRANGE":        CmdRead,
+	"ZCARD":            CmdRead,
+	"ZCOUNT":           CmdRead,
+	"ZRANGE":           CmdRead,
+	"ZRANGEBYSCORE":    CmdRead,
+	"ZREVRANGE":        CmdRead,
+	"ZREVRANGEBYSCORE": CmdRead,
+	"ZSCORE":           CmdRead,
+	"PFCOUNT":          CmdRead,
+
+	// Write commands
+	"SET":              CmdWrite,
+	"SETEX":            CmdWrite,
+	"SETNX":            CmdWrite,
+	"DEL":              CmdWrite,
+	"HSET":             CmdWrite,
+	"HMSET":            CmdWrite,
+	"HDEL":             CmdWrite,
+	"HINCRBY":          CmdWrite,
+	"INCR":             CmdWrite,
+	"LPUSH":            CmdWrite,
+	"LPOP":             CmdWrite,
+	"RPUSH":            CmdWrite,
+	"RPOP":             CmdWrite,
+	"RPOPLPUSH":        CmdWrite,
+	"LREM":             CmdWrite,
+	"LSET":             CmdWrite,
+	"LTRIM":            CmdWrite,
+	"SADD":             CmdWrite,
+	"SREM":             CmdWrite,
+	"EXPIRE":           CmdWrite,
+	"PEXPIRE":          CmdWrite,
+	"RENAME":           CmdWrite,
+	"XADD":             CmdWrite,
+	"XTRIM":            CmdWrite,
+	"ZADD":             CmdWrite,
+	"ZINCRBY":          CmdWrite,
+	"ZREM":             CmdWrite,
+	"ZREMRANGEBYSCORE": CmdWrite,
+	"ZREMRANGEBYRANK":  CmdWrite,
+	"ZPOPMIN":          CmdWrite,
+	"PFADD":            CmdWrite,
+	"FLUSHALL":         CmdWrite,
+	"FLUSHDB":          CmdWrite,
+	"PUBLISH":          CmdWrite, // write to both backends
+
+	// Blocking commands
+	"BZPOPMIN": CmdBlocking,
+	// XREAD is handled specially in ClassifyCommand (BLOCK arg check)
+
+	// PubSub commands
+	"SUBSCRIBE":    CmdPubSub,
+	"UNSUBSCRIBE":  CmdPubSub,
+	"PSUBSCRIBE":   CmdPubSub,
+	"PUNSUBSCRIBE": CmdPubSub,
+	"PUBSUB":       CmdPubSub,
+
+	// Admin commands — forwarded to primary only
+	"PING":    CmdAdmin,
+	"INFO":    CmdAdmin,
+	"CLIENT":  CmdAdmin,
+	"SELECT":  CmdAdmin,
+	"QUIT":    CmdAdmin,
+	"DBSIZE":  CmdAdmin,
+	"SCAN":    CmdAdmin,
+	"AUTH":    CmdAdmin,
+	"HELLO":   CmdAdmin,
+	"WAIT":    CmdAdmin,
+	"CONFIG":  CmdAdmin,
+	"OBJECT":  CmdAdmin,
+	"DEBUG":   CmdAdmin,
+	"CLUSTER": CmdAdmin,
+	"COMMAND": CmdAdmin,
+
+	// Transaction commands
+	"MULTI":   CmdTxn,
+	"EXEC":    CmdTxn,
+	"DISCARD": CmdTxn,
+
+	// Script commands
+	"EVAL":    CmdScript,
+	"EVALSHA": CmdScript,
+}
+
+// ClassifyCommand returns the category for a Redis command name.
+// XREAD is classified as CmdBlocking if args contain BLOCK, otherwise CmdRead.
+// Unknown commands default to CmdWrite (sent to both backends).
+func ClassifyCommand(name string, args [][]byte) CommandCategory {
+	upper := strings.ToUpper(name)
+
+	// Special case: XREAD with BLOCK
+	if upper == "XREAD" {
+		for _, arg := range args {
+			if strings.ToUpper(string(arg)) == "BLOCK" {
+				return CmdBlocking
+			}
+		}
+		return CmdRead
+	}
+
+	if cat, ok := commandTable[upper]; ok {
+		return cat
+	}
+	// Unknown commands → treat as write (safe default, sent to both backends)
+	return CmdWrite
+}