Adds support for Redis 7 via redisVersion metadata in all redis components

bindings/redis/redis.go

@@ -18,7 +18,8 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/go-redis/redis/v8"
+	v8 "github.com/go-redis/redis/v8"
+	v9 "github.com/go-redis/redis/v9"
 
 	"github.com/dapr/components-contrib/bindings"
 	rediscomponent "github.com/dapr/components-contrib/internal/component/redis"
@@ -27,9 +28,11 @@ import (
 
 // Redis is a redis output binding.
 type Redis struct {
-	client         redis.UniversalClient
+	clientv8       v8.UniversalClient
+	clientv9       v9.UniversalClient
 	clientSettings *rediscomponent.Settings
 	logger         logger.Logger
+	legacyRedis    bool
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -42,14 +45,25 @@ func NewRedis(logger logger.Logger) bindings.OutputBinding {
 
 // Init performs metadata parsing and connection creation.
 func (r *Redis) Init(meta bindings.Metadata) (err error) {
-	r.client, r.clientSettings, err = rediscomponent.ParseClientFromProperties(meta.Properties, nil)
+	if rediscomponent.IsLegacyRedisVersion(meta.Properties) {
+		r.legacyRedis = true
+		r.clientv8, r.clientSettings, err = rediscomponent.ParseClientv8FromProperties(meta.Properties, nil)
+	} else {
+		r.legacyRedis = false
+		r.clientv9, r.clientSettings, err = rediscomponent.ParseClientv9FromProperties(meta.Properties, nil)
+		r.logger.Warnf("Redis version 7 and above uses a Beta SDK at this time. Please use caution.")
+	}
 	if err != nil {
 		return err
 	}
 
 	r.ctx, r.cancel = context.WithCancel(context.Background())
 
-	_, err = r.client.Ping(r.ctx).Result()
+	if r.legacyRedis {
+		_, err = r.clientv8.Ping(r.ctx).Result()
+	} else {
+		_, err = r.clientv9.Ping(r.ctx).Result()
+	}
 	if err != nil {
 		return fmt.Errorf("redis binding: error connecting to redis at %s: %s", r.clientSettings.Host, err)
 	}
@@ -58,8 +72,14 @@ func (r *Redis) Init(meta bindings.Metadata) (err error) {
 }
 
 func (r *Redis) Ping() error {
-	if _, err := r.client.Ping(r.ctx).Result(); err != nil {
-		return fmt.Errorf("redis binding: error connecting to redis at %s: %s", r.clientSettings.Host, err)
+	if r.legacyRedis {
+		if _, err := r.clientv8.Ping(r.ctx).Result(); err != nil {
+			return fmt.Errorf("redis binding: error connecting to redis at %s: %s", r.clientSettings.Host, err)
+		}
+	} else {
+		if _, err := r.clientv9.Ping(r.ctx).Result(); err != nil {
+			return fmt.Errorf("redis binding: error connecting to redis at %s: %s", r.clientSettings.Host, err)
+		}
 	}
 
 	return nil
@@ -72,7 +92,12 @@ func (r *Redis) Operations() []bindings.OperationKind {
 func (r *Redis) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
 	if val, ok := req.Metadata["key"]; ok && val != "" {
 		key := val
-		_, err := r.client.Do(ctx, "SET", key, req.Data).Result()
+		var err error
+		if r.legacyRedis {
+			_, err = r.clientv8.Do(ctx, "SET", key, req.Data).Result()
+		} else {
+			_, err = r.clientv9.Do(ctx, "SET", key, req.Data).Result()
+		}
 		if err != nil {
 			return nil, err
 		}
@@ -86,5 +111,9 @@ func (r *Redis) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 func (r *Redis) Close() error {
 	r.cancel()
 
-	return r.client.Close()
+	if r.legacyRedis {
+		return r.clientv8.Close()
+	} else {
+		return r.clientv9.Close()
+	}
 }

bindings/redis/redis_test.go

@@ -18,7 +18,8 @@ import (
 	"testing"
 
 	miniredis "github.com/alicebob/miniredis/v2"
-	"github.com/go-redis/redis/v8"
+	v8 "github.com/go-redis/redis/v8"
+	v9 "github.com/go-redis/redis/v9"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/dapr/components-contrib/bindings"
@@ -30,18 +31,19 @@ const (
 	testKey  = "test"
 )
 
-func TestInvoke(t *testing.T) {
-	s, c := setupMiniredis()
+func TestInvokev8(t *testing.T) {
+	s, c := setupMiniredisv8()
 	defer s.Close()
 
 	bind := &Redis{
-		client: c,
-		logger: logger.NewLogger("test"),
+		legacyRedis: true,
+		clientv8:    c,
+		logger:      logger.NewLogger("test"),
 	}
 	bind.ctx, bind.cancel = context.WithCancel(context.Background())
 
 	_, err := c.Do(context.Background(), "GET", testKey).Result()
-	assert.Equal(t, redis.Nil, err)
+	assert.Equal(t, v8.Nil, err)
 
 	bindingRes, err := bind.Invoke(context.TODO(), &bindings.InvokeRequest{
 		Data:     []byte(testData),
@@ -55,15 +57,53 @@ func TestInvoke(t *testing.T) {
 	assert.Equal(t, true, getRes == testData)
 }
 
-func setupMiniredis() (*miniredis.Miniredis, *redis.Client) {
+func TestInvokev9(t *testing.T) {
+	s, c := setupMiniredisv9()
+	defer s.Close()
+
+	bind := &Redis{
+		clientv9: c,
+		logger:   logger.NewLogger("test"),
+	}
+	bind.ctx, bind.cancel = context.WithCancel(context.Background())
+
+	_, err := c.Do(context.Background(), "GET", testKey).Result()
+	assert.Equal(t, v9.Nil, err)
+
+	bindingRes, err := bind.Invoke(context.TODO(), &bindings.InvokeRequest{
+		Data:     []byte(testData),
+		Metadata: map[string]string{"key": testKey},
+	})
+	assert.Equal(t, nil, err)
+	assert.Equal(t, true, bindingRes == nil)
+
+	getRes, err := c.Do(context.Background(), "GET", testKey).Result()
+	assert.Equal(t, nil, err)
+	assert.Equal(t, true, getRes == testData)
+}
+
+func setupMiniredisv8() (*miniredis.Miniredis, *v8.Client) {
+	s, err := miniredis.Run()
+	if err != nil {
+		panic(err)
+	}
+	opts := &v8.Options{
+		Addr: s.Addr(),
+		DB:   0,
+	}
+
+	return s, v8.NewClient(opts)
+}
+
+func setupMiniredisv9() (*miniredis.Miniredis, *v9.Client) {
 	s, err := miniredis.Run()
 	if err != nil {
 		panic(err)
 	}
-	opts := &redis.Options{
+	opts := &v9.Options{
 		Addr: s.Addr(),
 		DB:   0,
 	}
 
-	return s, redis.NewClient(opts)
+	return s, v9.NewClient(opts)
 }

go.mod

@@ -200,6 +200,8 @@ require (
 	github.com/go-openapi/jsonreference v0.19.5 // indirect
 	github.com/go-openapi/swag v0.19.14 // indirect
 	github.com/go-ozzo/ozzo-validation/v4 v4.3.0 // indirect
+	github.com/go-redis/redis/v9 v9.0.0-beta.2
+	github.com/go-stack/stack v1.8.0 // indirect
 	github.com/go-playground/locales v0.14.0 // indirect
 	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/go-playground/validator/v10 v10.11.0 // indirect

go.sum

@@ -567,6 +567,8 @@ github.com/go-playground/validator/v10 v10.11.0 h1:0W+xRM511GY47Yy3bZUbJVitCNg2B
 github.com/go-playground/validator/v10 v10.11.0/go.mod h1:i+3WkQ1FvaUjjxh1kSvIA4dMGDBiPU55YFDl0WbKdWU=
 github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
 github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+github.com/go-redis/redis/v9 v9.0.0-beta.2 h1:ZSr84TsnQyKMAg8gnV+oawuQezeJR11/09THcWCQzr4=
+github.com/go-redis/redis/v9 v9.0.0-beta.2/go.mod h1:Bldcd/M/bm9HbnNPi/LUtYBSD8ttcZYBMupwMXhdU0o=
 github.com/go-resty/resty/v2 v2.7.0/go.mod h1:9PWDzw47qPphMRFfhsyk0NnSgvluHcljSMVIq3w7q0I=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
@@ -1126,6 +1128,7 @@ github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1Cpa
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.17.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=
+github.com/onsi/gomega v1.20.0 h1:8W0cWlwFkflGPLltQvLRB7ZVD5HuP6ng320w2IS245Q=
 github.com/onsi/gomega v1.19.0/go.mod h1:LY+I3pBVzYsTBU1AnDwOSxaYi9WoWiqgwooUqq9yPro=
 github.com/onsi/gomega v1.20.1 h1:PA/3qinGoukvymdIDV8pii6tiZgC8kbmJO6Z5+b002Q=
 github.com/op/go-logging v0.0.0-20160315200505-970db520ece7/go.mod h1:HzydrMdWErDVzsI23lYNej1Htcns9BCg93Dk0bBINWk=

internal/component/redis/metadata.go

@@ -14,63 +14,30 @@ limitations under the License.
 package redis
 
 import (
-	"fmt"
-	"strconv"
 	"time"
+
+	"github.com/dapr/components-contrib/metadata"
 )
 
 const (
-	maxRetries             = "maxRetries"
-	maxRetryBackoff        = "maxRetryBackoff"
-	ttlInSeconds           = "ttlInSeconds"
-	queryIndexes           = "queryIndexes"
-	defaultBase            = 10
-	defaultBitSize         = 0
 	defaultMaxRetries      = 3
 	defaultMaxRetryBackoff = time.Second * 2
 )
 
 type Metadata struct {
-	MaxRetries      int
-	MaxRetryBackoff time.Duration
-	TTLInSeconds    *int
-	QueryIndexes    string
+	MaxRetries      int           `json:"maxRetries,string,omitempty"`
+	MaxRetryBackoff time.Duration `json:"maxRetryBackoff,string,omitempty"`
+	TTLInSeconds    *int          `json:"ttlInSeconds,string,omitempty"`
+	QueryIndexes    string        `json:"queryIndexes,omitempty"`
+	RedisVersion    string        `json:"redisVersion,omitempty"`
 }
 
 func ParseRedisMetadata(properties map[string]string) (Metadata, error) {
 	m := Metadata{}
 
 	m.MaxRetries = defaultMaxRetries
-	if val, ok := properties[maxRetries]; ok && val != "" {
-		parsedVal, err := strconv.ParseInt(val, defaultBase, defaultBitSize)
-		if err != nil {
-			return m, fmt.Errorf("redis store error: can't parse maxRetries field: %s", err)
-		}
-		m.MaxRetries = int(parsedVal)
-	}
-
 	m.MaxRetryBackoff = defaultMaxRetryBackoff
-	if val, ok := properties[maxRetryBackoff]; ok && val != "" {
-		parsedVal, err := strconv.ParseInt(val, defaultBase, defaultBitSize)
-		if err != nil {
-			return m, fmt.Errorf("redis store error: can't parse maxRetryBackoff field: %s", err)
-		}
-		m.MaxRetryBackoff = time.Duration(parsedVal)
-	}
-
-	if val, ok := properties[ttlInSeconds]; ok && val != "" {
-		parsedVal, err := strconv.ParseInt(val, defaultBase, defaultBitSize)
-		if err != nil {
-			return m, fmt.Errorf("redis store error: can't parse ttlInSeconds field: %s", err)
-		}
-		intVal := int(parsedVal)
-		m.TTLInSeconds = &intVal
-	} else {
-		m.TTLInSeconds = nil
-	}
+	metadata.DecodeMetadata(properties, &m)
 
-	if val, ok := properties[queryIndexes]; ok && val != "" {
-		m.QueryIndexes = val
-	}
 	return m, nil
 }