Move the common Redis code to internal (#859)

* Move pubsub redis client to internal

* Add failover for common redis. Refactor state redis.

* refactor binding redis

* fix lint

* Refactor to setting partten

* fix db and lint

* fix lint

* for backward compatibility

* Using a type alias to handle decoding duration values of -1 for Redis.

* Linter fixes

Co-authored-by: Phil Kedy <phil.kedy@gmail.com>

Author: Taction

bindings/redis/metadata.go

@@ -8,9 +8,6 @@ package redis
 import "time"
 
 type metadata struct {
-	host            string
-	password        string
-	enableTLS       bool
 	maxRetries      int
 	maxRetryBackoff time.Duration
 }

bindings/redis/redis.go

@@ -7,35 +7,32 @@ package redis
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"strconv"
 	"time"
 
+	"github.com/go-redis/redis/v8"
+
 	"github.com/dapr/components-contrib/bindings"
+	rediscomponent "github.com/dapr/components-contrib/internal/component/redis"
 	"github.com/dapr/kit/logger"
-	redis "github.com/go-redis/redis/v8"
 )
 
 const (
-	host                   = "redisHost"
-	password               = "redisPassword"
-	enableTLS              = "enableTLS"
 	maxRetries             = "maxRetries"
 	maxRetryBackoff        = "maxRetryBackoff"
 	defaultBase            = 10
 	defaultBitSize         = 0
-	defaultDB              = 0
 	defaultMaxRetries      = 3
 	defaultMaxRetryBackoff = time.Second * 2
-	defaultEnableTLS       = false
 )
 
 // Redis is a redis output binding
 type Redis struct {
-	client *redis.Client
-	logger logger.Logger
+	client         redis.UniversalClient
+	clientSettings *rediscomponent.Settings
+	logger         logger.Logger
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -48,32 +45,21 @@ func NewRedis(logger logger.Logger) *Redis {
 
 // Init performs metadata parsing and connection creation
 func (r *Redis) Init(meta bindings.Metadata) error {
-	m, err := r.parseMetadata(meta)
+	_, err := r.parseMetadata(meta)
 	if err != nil {
 		return err
 	}
 
-	opts := &redis.Options{
-		Addr:            m.host,
-		Password:        m.password,
-		DB:              defaultDB,
-		MaxRetries:      m.maxRetries,
-		MaxRetryBackoff: m.maxRetryBackoff,
-	}
-
-	/* #nosec */
-	if m.enableTLS {
-		opts.TLSConfig = &tls.Config{
-			InsecureSkipVerify: m.enableTLS,
-		}
+	r.client, r.clientSettings, err = rediscomponent.ParseClientFromProperties(meta.Properties, nil)
+	if err != nil {
+		return err
 	}
 
-	r.client = redis.NewClient(opts)
 	r.ctx, r.cancel = context.WithCancel(context.Background())
 
 	_, err = r.client.Ping(r.ctx).Result()
 	if err != nil {
-		return fmt.Errorf("redis binding: error connecting to redis at %s: %s", m.host, err)
+		return fmt.Errorf("redis binding: error connecting to redis at %s: %s", r.clientSettings.Host, err)
 	}
 
 	return err
@@ -82,25 +68,6 @@ func (r *Redis) Init(meta bindings.Metadata) error {
 func (r *Redis) parseMetadata(meta bindings.Metadata) (metadata, error) {
 	m := metadata{}
 
-	if val, ok := meta.Properties[host]; ok && val != "" {
-		m.host = val
-	} else {
-		return m, errors.New("redis binding error: missing host address")
-	}
-
-	if val, ok := meta.Properties[password]; ok && val != "" {
-		m.password = val
-	}
-
-	m.enableTLS = defaultEnableTLS
-	if val, ok := meta.Properties[enableTLS]; ok && val != "" {
-		tls, err := strconv.ParseBool(val)
-		if err != nil {
-			return m, fmt.Errorf("redis binding error: can't parse enableTLS field: %s", err)
-		}
-		m.enableTLS = tls
-	}
-
 	m.maxRetries = defaultMaxRetries
 	if val, ok := meta.Properties[maxRetries]; ok && val != "" {
 		parsedVal, err := strconv.ParseInt(val, defaultBase, defaultBitSize)

bindings/redis/redis_test.go

@@ -20,9 +20,6 @@ func TestParseMetadata(t *testing.T) {
 	r := Redis{logger: logger.NewLogger("test")}
 	redisM, err := r.parseMetadata(m)
 	assert.Nil(t, err)
-	assert.Equal(t, "host", redisM.host)
-	assert.Equal(t, "password", redisM.password)
-	assert.Equal(t, true, redisM.enableTLS)
 	assert.Equal(t, 3, redisM.maxRetries)
 	assert.Equal(t, time.Duration(10000), redisM.maxRetryBackoff)
 }

internal/component/redis/redis.go

@@ -0,0 +1,135 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation and Dapr Contributors.
+// Licensed under the MIT License.
+// ------------------------------------------------------------
+
+package redis
+
+import (
+	"crypto/tls"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+)
+
+const (
+	ClusterType = "cluster"
+	NodeType    = "node"
+)
+
+func ParseClientFromProperties(properties map[string]string, defaultSettings *Settings) (client redis.UniversalClient, settings *Settings, err error) {
+	if defaultSettings == nil {
+		settings = &Settings{}
+	} else {
+		settings = defaultSettings
+	}
+	err = settings.Decode(properties)
+	if err != nil {
+		return nil, nil, fmt.Errorf("redis client configuration error: %w", err)
+	}
+	if settings.Failover {
+		return newFailoverClient(settings), settings, nil
+	}
+
+	return newClient(settings), settings, nil
+}
+
+func newFailoverClient(s *Settings) redis.UniversalClient {
+	if s == nil {
+		return nil
+	}
+	opts := &redis.FailoverOptions{
+		DB:                 s.DB,
+		MasterName:         s.SentinelMasterName,
+		SentinelAddrs:      []string{s.Host},
+		Password:           s.Password,
+		MaxRetries:         s.RedisMaxRetries,
+		MaxRetryBackoff:    time.Duration(s.RedisMaxRetryInterval),
+		MinRetryBackoff:    time.Duration(s.RedisMinRetryInterval),
+		DialTimeout:        time.Duration(s.DialTimeout),
+		ReadTimeout:        time.Duration(s.ReadTimeout),
+		WriteTimeout:       time.Duration(s.WriteTimeout),
+		PoolSize:           s.PoolSize,
+		MaxConnAge:         time.Duration(s.MaxConnAge),
+		MinIdleConns:       s.MinIdleConns,
+		PoolTimeout:        time.Duration(s.PoolTimeout),
+		IdleCheckFrequency: time.Duration(s.IdleCheckFrequency),
+		IdleTimeout:        time.Duration(s.IdleTimeout),
+	}
+
+	/* #nosec */
+	if s.EnableTLS {
+		opts.TLSConfig = &tls.Config{
+			InsecureSkipVerify: s.EnableTLS,
+		}
+	}
+
+	if s.RedisType == ClusterType {
+		opts.SentinelAddrs = strings.Split(s.Host, ",")
+
+		return redis.NewFailoverClusterClient(opts)
+	}
+
+	return redis.NewFailoverClient(opts)
+}
+
+func newClient(s *Settings) redis.UniversalClient {
+	if s == nil {
+		return nil
+	}
+	if s.RedisType == ClusterType {
+		options := &redis.ClusterOptions{
+			Addrs:              strings.Split(s.Host, ","),
+			Password:           s.Password,
+			MaxRetries:         s.RedisMaxRetries,
+			MaxRetryBackoff:    time.Duration(s.RedisMaxRetryInterval),
+			MinRetryBackoff:    time.Duration(s.RedisMinRetryInterval),
+			DialTimeout:        time.Duration(s.DialTimeout),
+			ReadTimeout:        time.Duration(s.ReadTimeout),
+			WriteTimeout:       time.Duration(s.WriteTimeout),
+			PoolSize:           s.PoolSize,
+			MaxConnAge:         time.Duration(s.MaxConnAge),
+			MinIdleConns:       s.MinIdleConns,
+			PoolTimeout:        time.Duration(s.PoolTimeout),
+			IdleCheckFrequency: time.Duration(s.IdleCheckFrequency),
+			IdleTimeout:        time.Duration(s.IdleTimeout),
+		}
+		/* #nosec */
+		if s.EnableTLS {
+			options.TLSConfig = &tls.Config{
+				InsecureSkipVerify: s.EnableTLS,
+			}
+		}
+
+		return redis.NewClusterClient(options)
+	}
+
+	options := &redis.Options{
+		Addr:               s.Host,
+		Password:           s.Password,
+		DB:                 s.DB,
+		MaxRetries:         s.RedisMaxRetries,
+		MaxRetryBackoff:    time.Duration(s.RedisMaxRetryInterval),
+		MinRetryBackoff:    time.Duration(s.RedisMinRetryInterval),
+		DialTimeout:        time.Duration(s.DialTimeout),
+		ReadTimeout:        time.Duration(s.ReadTimeout),
+		WriteTimeout:       time.Duration(s.WriteTimeout),
+		PoolSize:           s.PoolSize,
+		MaxConnAge:         time.Duration(s.MaxConnAge),
+		MinIdleConns:       s.MinIdleConns,
+		PoolTimeout:        time.Duration(s.PoolTimeout),
+		IdleCheckFrequency: time.Duration(s.IdleCheckFrequency),
+		IdleTimeout:        time.Duration(s.IdleTimeout),
+	}
+
+	/* #nosec */
+	if s.EnableTLS {
+		options.TLSConfig = &tls.Config{
+			InsecureSkipVerify: s.EnableTLS,
+		}
+	}
+
+	return redis.NewClient(options)
+}

internal/component/redis/redis_test.go

@@ -0,0 +1,127 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation and Dapr Contributors.
+// Licensed under the MIT License.
+// ------------------------------------------------------------
+
+package redis
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	host                  = "redisHost"
+	password              = "redisPassword"
+	db                    = "redisDB"
+	redisType             = "redisType"
+	redisMaxRetries       = "redisMaxRetries"
+	redisMinRetryInterval = "redisMinRetryInterval"
+	redisMaxRetryInterval = "redisMaxRetryInterval"
+	dialTimeout           = "dialTimeout"
+	readTimeout           = "readTimeout"
+	writeTimeout          = "writeTimeout"
+	poolSize              = "poolSize"
+	minIdleConns          = "minIdleConns"
+	poolTimeout           = "poolTimeout"
+	idleTimeout           = "idleTimeout"
+	idleCheckFrequency    = "idleCheckFrequency"
+	maxConnAge            = "maxConnAge"
+	enableTLS             = "enableTLS"
+	failover              = "failover"
+	sentinelMasterName    = "sentinelMasterName"
+)
+
+func getFakeProperties() map[string]string {
+	return map[string]string{
+		host:                  "fake.redis.com",
+		password:              "fakePassword",
+		redisType:             "node",
+		enableTLS:             "true",
+		dialTimeout:           "5s",
+		readTimeout:           "5s",
+		writeTimeout:          "50000",
+		poolSize:              "20",
+		maxConnAge:            "200s",
+		db:                    "1",
+		redisMaxRetries:       "1",
+		redisMinRetryInterval: "8ms",
+		redisMaxRetryInterval: "1s",
+		minIdleConns:          "1",
+		poolTimeout:           "1s",
+		idleTimeout:           "1s",
+		idleCheckFrequency:    "1s",
+		failover:              "true",
+		sentinelMasterName:    "master",
+	}
+}
+
+func TestParseRedisMetadata(t *testing.T) {
+	t.Run("ClientMetadata is correct", func(t *testing.T) {
+		fakeProperties := getFakeProperties()
+
+		// act
+		m := &Settings{}
+		err := m.Decode(fakeProperties)
+
+		// assert
+		assert.NoError(t, err)
+		assert.Equal(t, fakeProperties[host], m.Host)
+		assert.Equal(t, fakeProperties[password], m.Password)
+		assert.Equal(t, fakeProperties[redisType], m.RedisType)
+		assert.Equal(t, true, m.EnableTLS)
+		assert.Equal(t, 5*time.Second, time.Duration(m.DialTimeout))
+		assert.Equal(t, 5*time.Second, time.Duration(m.ReadTimeout))
+		assert.Equal(t, 50000*time.Millisecond, time.Duration(m.WriteTimeout))
+		assert.Equal(t, 20, m.PoolSize)
+		assert.Equal(t, 200*time.Second, time.Duration(m.MaxConnAge))
+		assert.Equal(t, 1, m.DB)
+		assert.Equal(t, 1, m.RedisMaxRetries)
+		assert.Equal(t, 8*time.Millisecond, time.Duration(m.RedisMinRetryInterval))
+		assert.Equal(t, 1*time.Second, time.Duration(m.RedisMaxRetryInterval))
+		assert.Equal(t, 1, m.MinIdleConns)
+		assert.Equal(t, 1*time.Second, time.Duration(m.PoolTimeout))
+		assert.Equal(t, 1*time.Second, time.Duration(m.IdleTimeout))
+		assert.Equal(t, 1*time.Second, time.Duration(m.IdleCheckFrequency))
+		assert.Equal(t, true, m.Failover)
+		assert.Equal(t, "master", m.SentinelMasterName)
+	})
+
+	t.Run("host is not given", func(t *testing.T) {
+		fakeProperties := getFakeProperties()
+
+		fakeProperties[host] = ""
+
+		// act
+		m := &Settings{}
+		err := m.Decode(fakeProperties)
+
+		// assert
+		assert.Error(t, errors.New("redis streams error: missing host address"), err)
+		assert.Empty(t, m.Host)
+	})
+
+	t.Run("check values can be set as -1", func(t *testing.T) {
+		fakeProperties := getFakeProperties()
+
+		fakeProperties[readTimeout] = "-1"
+		fakeProperties[idleTimeout] = "-1"
+		fakeProperties[idleCheckFrequency] = "-1"
+		fakeProperties[redisMaxRetryInterval] = "-1"
+		fakeProperties[redisMinRetryInterval] = "-1"
+
+		// act
+		m := &Settings{}
+		err := m.Decode(fakeProperties)
+		// assert
+		assert.NoError(t, err)
+		assert.True(t, m.ReadTimeout == -1)
+		assert.True(t, m.IdleTimeout == -1)
+		assert.True(t, m.IdleCheckFrequency == -1)
+		assert.True(t, m.RedisMaxRetryInterval == -1)
+		assert.True(t, m.RedisMinRetryInterval == -1)
+	})
+}