Merge pull request #351 from bootjp/feature/redis-proxy

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

Author: bootjp

adapter/add_voter_join_test.go

@@ -182,7 +182,7 @@ func startAddVoterJoinNode(
 		return err
 	})
 
-	rd := NewRedisServer(lis.redis, port.redisAddress, st, coordinator, leaderRedisMap, relay)
+	rd := NewRedisServer(lis.redis, port.redisAddress, routedStore, coordinator, leaderRedisMap, relay)
 	workers.Go(func() error {
 		err := rd.Run()
 		if errors.Is(err, net.ErrClosed) {

adapter/redis_server_setup_test.go

@@ -0,0 +1,19 @@
+package adapter
+
+import (
+	"testing"
+
+	"github.com/bootjp/elastickv/kv"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRedisServerUsesLeaderRoutedStoreInSingleGroupCluster(t *testing.T) {
+	t.Parallel()
+
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	require.IsType(t, &kv.LeaderRoutedStore{}, nodes[0].redisServer.store)
+	require.IsType(t, &kv.LeaderRoutedStore{}, nodes[1].redisServer.store)
+	require.IsType(t, &kv.LeaderRoutedStore{}, nodes[2].redisServer.store)
+}

adapter/test_util.go

@@ -370,7 +370,7 @@ func setupNodes(t *testing.T, ctx context.Context, n int, ports []portsAdress) (
 			assert.NoError(t, srv.Serve(lis))
 		}(s, grpcSock)
 
-		rd := NewRedisServer(redisSock, port.redisAddress, st, coordinator, leaderRedisMap, relay)
+		rd := NewRedisServer(redisSock, port.redisAddress, routedStore, coordinator, leaderRedisMap, relay)
 		go func(server *RedisServer) {
 			assert.NoError(t, server.Run())
 		}(rd)

cmd/redis-proxy/main.go

@@ -0,0 +1,126 @@
+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
+	metricsShutdownTimeout = 5 * 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)
+	defer dual.Close() // wait for in-flight async goroutines
+	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()
+			shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), metricsShutdownTimeout)
+			defer shutdownCancel()
+			if err := metricsSrv.Shutdown(shutdownCtx); 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
+}

cmd/server/demo.go

@@ -354,7 +354,8 @@ func setupRedis(ctx context.Context, lc net.ListenConfig, st store.MVCCStore, co
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
-	return adapter.NewRedisServer(l, redisAddr, st, coordinator, leaderRedis, relay), nil
+	routedStore := kv.NewLeaderRoutedStore(st, coordinator)
+	return adapter.NewRedisServer(l, redisAddr, routedStore, coordinator, leaderRedis, relay), nil
 }
 
 func run(ctx context.Context, eg *errgroup.Group, cfg config) error {

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/cockroachdb/errors v1.12.0
 	github.com/cockroachdb/pebble v1.1.5
 	github.com/emirpasic/gods v1.18.1
+	github.com/getsentry/sentry-go v0.27.0
 	github.com/hashicorp/go-hclog v1.6.3
 	github.com/hashicorp/raft v1.7.3
 	github.com/pkg/errors v0.9.1
@@ -58,7 +59,6 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/fatih/color v1.15.0 // indirect
-	github.com/getsentry/sentry-go v0.27.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/golang/snappy v0.0.5-0.20231225225746-43d5d4cd4e0e // indirect

jepsen/src/elastickv/redis_workload.clj

@@ -175,6 +175,13 @@
     :parse-fn #(Integer/parseInt %)]
    ["-h" "--help"]])
 
+(defn fail-on-invalid!
+  "Raises when Jepsen completed analysis and found the history invalid."
+  [result]
+  (when (false? (:valid? result))
+    (throw (ex-info "Jepsen analysis invalid" {:result result})))
+  result)
+
 (defn -main
   [& args]
   (let [{:keys [options errors summary]} (tools.cli/parse-opts args cli-opts)
@@ -211,5 +218,5 @@
       (seq errors) (binding [*out* *err*]
                      (println "Error parsing options:" (str/join "; " errors)))
       (:local options) (binding [control/*dummy* true]
-                         (jepsen/run! (elastickv-redis-test options)))
-      :else (jepsen/run! (elastickv-redis-test options)))))
+                         (fail-on-invalid! (jepsen/run! (elastickv-redis-test options))))
+      :else (fail-on-invalid! (jepsen/run! (elastickv-redis-test options))))))

jepsen/test/elastickv/redis_workload_test.clj

@@ -0,0 +1,13 @@
+(ns elastickv.redis-workload-test
+  (:require [clojure.test :refer :all]
+            [elastickv.redis-workload :as workload]))
+
+(deftest fail-on-invalid-passes-through-valid-results
+  (let [result {:valid? true}]
+    (is (= result (workload/fail-on-invalid! result)))))
+
+(deftest fail-on-invalid-throws-for-invalid-results
+  (is (thrown-with-msg?
+       clojure.lang.ExceptionInfo
+       #"Jepsen analysis invalid"
+       (workload/fail-on-invalid! {:valid? false}))))

proxy/backend.go

@@ -0,0 +1,122 @@
+package proxy
+
+import (
+	"context"
+	"errors"
+	"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 {
+		// go-redis pipelines return redis.Error for Redis reply errors (e.g., EXECABORT).
+		// Return results with nil error so callers can read per-command results (especially EXEC).
+		// Only propagate true transport/context errors.
+		var redisErr redis.Error
+		if errors.As(err, &redisErr) || errors.Is(err, redis.Nil) {
+			return results, 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)
+}