Adding Support for Redis Cluster Client

build.settings

@@ -1,4 +1,4 @@
 # Build configuration
 
-version = 0.6.0
+version = 0.6.1
 

cmd/main.go

@@ -76,7 +76,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)")
@@ -108,7 +112,7 @@ func main() {
     } else {
         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)
+        store = storage.NewRedisStore(*redisUrl, *cluster, 1, *timeout, *maxRetries)
     }
     server.SetStore(store)
 

docker/entrypoint.sh

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

pubsub/listener.go

@@ -90,6 +90,7 @@ func (listener *EventsListener) ListenForMessages() {
                 fmt.Sprintf("configuration [%s] could not be found", fsm.ConfigId)))
             continue
         }
+        previousState := fsm.State
         cfgFsm := ConfiguredStateMachine{
             Config: cfg,
             FSM:    fsm,
@@ -101,8 +102,8 @@ func (listener *EventsListener) ListenForMessages() {
                     request.GetEvent().GetTransition().GetEvent(), err)))
             continue
         }
-        listener.logger.Info("Event `%s` transitioned FSM [%s] to state `%s` - updating store",
-            request.Event.Transition.Event, smId, fsm.State)
+        listener.logger.Info("Event `%s` transitioned FSM [%s] to state `%s` from state `%s` - updating store",
+            request.Event.Transition.Event, smId, fsm.State, previousState)
         err := listener.store.PutStateMachine(smId, fsm)
         if err != nil {
             listener.PostNotificationAndReportOutcome(makeResponse(&request,

storage/redis_store.go

@@ -44,7 +44,7 @@ const (
 
 type RedisStore struct {
     logger     *slf4go.Log
-    client     *redis.Client
+    client     redis.Cmdable
     Timeout    time.Duration
     MaxRetries int
 }
@@ -141,42 +141,36 @@ func (csm *RedisStore) GetTimeout() time.Duration {
 }
 
 func NewRedisStoreWithDefaults(address string) StoreManager {
-    return NewRedisStore(address, DefaultRedisDb, DefaultTimeout, DefaultMaxRetries)
+    return NewRedisStore(address, false, DefaultRedisDb, DefaultTimeout, DefaultMaxRetries)
 }
 
-func NewRedisStore(address string, db int, timeout time.Duration, maxRetries int) StoreManager {
-
+func NewRedisStore(address string, isCluster bool, db int, timeout time.Duration, maxRetries int) StoreManager {
     logger := slf4go.NewLog(fmt.Sprintf("redis://%s/%d", address, db))
+
     var tlsConfig *tls.Config
+    var client redis.Cmdable
+
     if os.Getenv("REDIS_TLS") != "" {
         logger.Info("Using TLS for Redis connection")
         tlsConfig = &tls.Config{MinVersion: tls.VersionTLS12}
     }
-    return &RedisStore{
-        logger: logger,
-        client: redis.NewClient(&redis.Options{
+
+    if isCluster {
+        client = redis.NewClusterClient(&redis.ClusterOptions{
+            TLSConfig: tlsConfig,
+            Addrs:     strings.Split(address, ","),
+        })
+    } else {
+        client = redis.NewClient(&redis.Options{
             TLSConfig: tlsConfig,
             Addr:      address,
             DB:        db, // 0 means default DB
-        }),
-        Timeout:    timeout,
-        MaxRetries: maxRetries,
+        })
     }
-}
-
-// FIXME: the "constructor" functions are very similar, the creation pattern will need to be
-//  refactored to avoid code duplication.
 
-func NewRedisStoreWithCreds(address string, db int, timeout time.Duration, maxRetries int,
-    username string, password string) StoreManager {
     return &RedisStore{
-        logger: slf4go.NewLog(fmt.Sprintf("redis:%s", address)),
-        client: redis.NewClient(&redis.Options{
-            Addr:     address,
-            Username: username,
-            Password: password,
-            DB:       db,
-        }),
+        logger:     slf4go.NewLog(fmt.Sprintf("redis://%s/%d", address, db)),
+        client:     client,
         Timeout:    timeout,
         MaxRetries: maxRetries,
     }