Adding Support for Redis Cluster Client

cmd/main.go

@@ -32,6 +32,7 @@ import (
     log "github.com/massenz/slf4go/logging"
     protos "github.com/massenz/statemachine-proto/golang/api"
     "net"
+    "strings"
     "sync"
 )
 
@@ -76,7 +77,11 @@ func main() {
     var localOnly = flag.Bool("local", false,
         "If set, it only listens to incoming requests from the local host")
     var port = flag.Int("http-port", 7399, "HTTP Server port for the REST API")
-    var redisUrl = flag.String("redis", "", "URI for the Redis cluster (host:port)")
+    var redisUrl = flag.String("redis", "", "For single node redis instances: URI "+
+        "for the Redis instance (host:port). For redis clusters: a comma-separated list of redis nodes. "+
+        "If using an ElastiCache redis cluster with cluster mode enabled, you can supply the configuration endpoint.")
+    var cluster = flag.Bool("cluster", false,
+        "Needs to be set if connecting to a Redis instance with cluster mode enabled")
     var awsEndpoint = flag.String("endpoint-url", "",
         "HTTP URL for AWS SQS to connect to; usually best left undefined, "+
             "unless required for local testing purposes (LocalStack uses http://localhost:4566)")
@@ -102,13 +107,20 @@ func main() {
     }
     addr := fmt.Sprintf("%s:%d", host, *port)
 
-    if *redisUrl == "" {
-        logger.Warn("in-memory storage configured, all data will NOT survive a server restart")
-        store = storage.NewInMemoryStore()
-    } else {
+    if *cluster {
+        redisNodes := strings.Split(*redisUrl, ",")
+        logger.Info("Connecting to Redis cluster at with the following nodes:")
+        for _, n := range redisNodes {
+            logger.Info(n)
+        }
+        store = storage.NewRedisClusterStore(redisNodes, *timeout, *maxRetries)
+    } else if *redisUrl != "" {
         logger.Info("Connecting to Redis server at %s", *redisUrl)
         logger.Info("with timeout: %s, max-retries: %d", *timeout, *maxRetries)
         store = storage.NewRedisStore(*redisUrl, 1, *timeout, *maxRetries)
+    } else {
+        logger.Warn("in-memory storage configured, all data will NOT survive a server restart")
+        store = storage.NewInMemoryStore()
     }
     server.SetStore(store)
 

docker/entrypoint.sh

@@ -25,12 +25,12 @@ then
   endpoint="--endpoint-url ${AWS_ENDPOINT}"
 fi
 
-echo -e "./sm-server -port ${SERVER_PORT}  ${endpoint:-} ${DEBUG}\n\
+echo -e "./sm-server -port ${SERVER_PORT}  ${endpoint:-} ${CLUSTER} ${DEBUG}\n\
     -redis ${REDIS}:${REDIS_PORT} -timeout ${TIMEOUT:-25ms} -max-retries ${RETRIES:-3}\n\
     -events ${EVENTS_Q} -errors ${ERRORS_Q}\n\
     $@"
 
-./sm-server -http-port "${SERVER_PORT}"  ${endpoint:-} ${DEBUG} \
+./sm-server -http-port "${SERVER_PORT}"  ${endpoint:-} ${CLUSTER} ${DEBUG} \
     -redis ${REDIS}:${REDIS_PORT} -timeout ${TIMEOUT:-25ms} -max-retries ${RETRIES:-3} \
     -events "${EVENTS_Q}" -errors "${ERRORS_Q}" \
     "$@"

storage/redis_store.go

@@ -20,6 +20,7 @@ package storage
 
 import (
     "context"
+    "crypto/tls"
     "fmt"
     "github.com/go-redis/redis/v8"
     "github.com/golang/protobuf/proto"
@@ -42,7 +43,7 @@ const (
 
 type RedisStore struct {
     logger     *slf4go.Log
-    client     *redis.Client
+    client     RedisClient
     Timeout    time.Duration
     MaxRetries int
 }
@@ -142,11 +143,30 @@ func NewRedisStoreWithDefaults(address string) StoreManager {
     return NewRedisStore(address, DefaultRedisDb, DefaultTimeout, DefaultMaxRetries)
 }
 
+func NewRedisClusterStore(addresses []string, timeout time.Duration, maxRetries int) StoreManager {
+    logger := slf4go.NewLog(fmt.Sprintf("redis cluster: %v nodes", len(addresses)))
+
+    return &RedisStore{
+        logger: logger,
+        client: redis.NewClusterClient(&redis.ClusterOptions{
+            TLSConfig: &tls.Config{
+                MinVersion: tls.VersionTLS12,
+            },
+            Addrs: addresses,
+        }),
+        Timeout:    timeout,
+        MaxRetries: maxRetries,
+    }
+}
+
 func NewRedisStore(address string, db int, timeout time.Duration, maxRetries int) StoreManager {
     logger := slf4go.NewLog(fmt.Sprintf("redis://%s/%d", address, db))
     return &RedisStore{
         logger: logger,
         client: redis.NewClient(&redis.Options{
+            TLSConfig: &tls.Config{
+                MinVersion: tls.VersionTLS12,
+            },
             Addr: address,
             DB:   db, // 0 means default DB
         }),

storage/types.go

@@ -19,7 +19,9 @@
 package storage
 
 import (
+    "context"
     "fmt"
+    "github.com/go-redis/redis/v8"
     log "github.com/massenz/slf4go/logging"
     protos "github.com/massenz/statemachine-proto/golang/api"
     "time"
@@ -69,3 +71,9 @@ type StoreManager interface {
     GetTimeout() time.Duration
     Health() error
 }
+
+type RedisClient interface {
+    Get(ctx context.Context, id string) *redis.StringCmd
+    Set(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.StatusCmd
+    Ping(ctx context.Context) *redis.StatusCmd
+}