Redis universal client (flyteorg#162)

* Move to Redis Universal Client to enable different Redis Configurations

* generate flags

* fix logs

* fix comment

* Comments

* Comments

pkg/controller/nodes/handler/state_test.go

@@ -0,0 +1,25 @@
+package handler
+
+import (
+	"bytes"
+	"encoding/base64"
+	"encoding/gob"
+	"testing"
+
+	"github.com/lyft/flytepropeller/pkg/controller/nodes/task/k8s"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// A test to demonstrate how to unmarshal a serialized state from a workflow CRD.
+func TestDecodeTaskState(t *testing.T) {
+	str := `I/+DAwEBC1BsdWdpblN0YXRlAf+EAAEBAQVQaGFzZQEGAAAABf+EAQIA`
+	reader := base64.NewDecoder(base64.RawStdEncoding, bytes.NewReader([]byte(str)))
+	dec := gob.NewDecoder(reader)
+	st := &k8s.PluginState{}
+	err := dec.Decode(st)
+	if assert.NoError(t, err) {
+		t.Logf("Deserialized State: [%+v]", st)
+		assert.Equal(t, k8s.PluginPhaseStarted, st.Phase)
+	}
+}

pkg/controller/nodes/task/resourcemanager/config/config.go

@@ -33,7 +33,11 @@ type Config struct {
 }
 
 // Specific configs for Redis resource manager
+// Ref: https://redis.io/topics/sentinel for information on how to fill in these fields.
 type RedisConfig struct {
+	HostPaths   []string `json:"hostPaths" pflag:",Redis hosts locations."`
+	PrimaryName string   `json:"primaryName" pflag:",Redis primary name, fill in only if you are connecting to a redis sentinel cluster."`
+	// deprecated: Please use HostPaths instead
 	HostPath   string `json:"hostPath" pflag:",Redis host location"`
 	HostKey    string `json:"hostKey" pflag:",Key for local Redis access"`
 	MaxRetries int    `json:"maxRetries" pflag:",See Redis client options for more info"`

pkg/controller/nodes/task/resourcemanager/redis_client.go

@@ -27,7 +27,7 @@ type RedisClient interface {
 }
 
 type Redis struct {
-	c *redis.Client
+	c redis.UniversalClient
 }
 
 func (r *Redis) SCard(key string) (int64, error) {
@@ -55,9 +55,15 @@ func (r *Redis) Ping() (string, error) {
 }
 
 func NewRedisClient(ctx context.Context, config config.RedisConfig) (RedisClient, error) {
+	// Backward compatibility
+	if len(config.HostPaths) == 0 && len(config.HostPath) > 0 {
+		config.HostPaths = []string{config.HostPath}
+	}
+
 	client := &Redis{
-		c: redis.NewClient(&redis.Options{
-			Addr:       config.HostPath,
+		c: redis.NewUniversalClient(&redis.UniversalOptions{
+			Addrs:      config.HostPaths,
+			MasterName: config.PrimaryName,
 			Password:   config.HostKey,
 			DB:         0, // use default DB
 			MaxRetries: config.MaxRetries,
@@ -66,10 +72,10 @@ func NewRedisClient(ctx context.Context, config config.RedisConfig) (RedisClient
 
 	_, err := client.Ping()
 	if err != nil {
-		logger.Errorf(ctx, "Error creating Redis client at [%s]. Error: %v", config.HostPath, err)
+		logger.Errorf(ctx, "Error creating Redis client at [%+v]. Error: %v", config.HostPaths, err)
 		return nil, err
 	}
 
-	logger.Infof(ctx, "Created Redis client with host [%s]...", config.HostPath)
+	logger.Infof(ctx, "Created Redis client with host [%+v]...", config.HostPaths)
 	return client, nil
 }