feat!: migrate to go-redis as redis client

configs/service.yaml

@@ -5,7 +5,9 @@ open_saves_project: ""
 log_level: "info"
 
 redis_address: "localhost:6379"
-redis_pool_maxIdle: 500
-redis_pool_maxActive: 10000
-redis_pool_idleTimeout: 0
-redis_pool_wait: false
+redis_max_idle: 500
+redis_pool_size: 10000
+redis_idle_timeout: 0
+redis_max_retries: 3
+redis_min_retry_backoff: "8ms"
+redis_max_retry_backoff: "512ms"

go.mod

@@ -26,9 +26,11 @@ require (
 	github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 // indirect
 	github.com/cncf/xds/go v0.0.0-20211130200136-a8f946100490 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/envoyproxy/go-control-plane v0.10.1 // indirect
 	github.com/envoyproxy/protoc-gen-validate v0.6.2 // indirect
 	github.com/fsnotify/fsnotify v1.5.1 // indirect
+	github.com/go-redis/redis/v8 v8.11.5 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/go-cmp v0.5.6 // indirect
@@ -47,7 +49,7 @@ require (
 	go.opencensus.io v0.23.0 // indirect
 	golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
-	golang.org/x/sys v0.0.0-20211210111614-af8b64212486 // indirect
+	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/appengine v1.6.7 // indirect

go.sum

@@ -146,6 +146,8 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/denisenkom/go-mssqldb v0.9.0/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
 github.com/devigned/tab v0.1.1/go.mod h1:XG9mPq0dFghrYvoBF3xdRrJzSTX1b7IQrvaL9mzjeJY=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/dimchansky/utfbom v1.1.0/go.mod h1:rO41eb7gLfo8SF1jd9F8HplJm1Fewwi4mQvIirEdv+8=
 github.com/dimchansky/utfbom v1.1.1/go.mod h1:SxdoEBH5qIqFocHMyGOXVAybYJdr71b1Q/j0mACtrfE=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -184,6 +186,8 @@ github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvSc
 github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
 github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
 github.com/go-playground/validator/v10 v10.2.0/go.mod h1:uOYAAleCW8F/7oMFd6aG0GOhaH6EGOAJShg8Id5JGkI=
+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-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gobwas/httphead v0.0.0-20180130184737-2c6c146eadee/go.mod h1:L0fX3K22YWvt/FAX9NnzrNzcI4wNYi9Yku4O0LKYflo=
@@ -672,8 +676,9 @@ golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211124211545-fe61309f8881/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20211210111614-af8b64212486 h1:5hpz5aRr+W1erYCL5JRhSUBJRph7l9XkNveoExlrKYk=
 golang.org/x/sys v0.0.0-20211210111614-af8b64212486/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

internal/app/server/open_saves_test.go

@@ -16,12 +16,13 @@ package server
 
 import (
 	"context"
-	"github.com/googleforgames/open-saves/internal/pkg/config"
 	"io"
 	"net"
 	"testing"
 	"time"
 
+	"github.com/googleforgames/open-saves/internal/pkg/config"
+
 	"cloud.google.com/go/datastore"
 	"github.com/google/uuid"
 	pb "github.com/googleforgames/open-saves/api"
@@ -62,13 +63,9 @@ func getOpenSavesServer(ctx context.Context, t *testing.T, cloud string) (*openS
 			Project: testProject,
 		},
 		RedisConfig: config.RedisConfig{
-			Address: testCacheAddr,
-			Pool: config.RedisPool{
-				MaxIdle:     500,
-				MaxActive:   10000,
-				IdleTimeout: 0,
-				Wait:        false,
-			},
+			Address:     testCacheAddr,
+			PoolSize:    10000,
+			IdleTimeout: 0,
 		},
 	}
 	impl, err := newOpenSavesServer(ctx, cfg)

internal/pkg/cache/redis/redis.go

@@ -16,170 +16,66 @@ package redis
 
 import (
 	"context"
-	"errors"
-	"syscall"
 
-	"github.com/gomodule/redigo/redis"
+	"github.com/go-redis/redis/v8"
 	"github.com/googleforgames/open-saves/internal/pkg/config"
 )
 
-const BrokenPipeRetries = 3
-
 // Redis is an implementation of the cache.Cache interface.
 type Redis struct {
-	redisPool *redis.Pool
+	c *redis.Client
 }
 
 // NewRedis creates a new Redis instance.
-func NewRedis(address string, opts ...redis.DialOption) *Redis {
+func NewRedis(address string) *Redis {
 	cfg := &config.RedisConfig{
 		Address: address,
-		Pool: config.RedisPool{
-			MaxIdle:     500,
-			MaxActive:   10000,
-			IdleTimeout: 0,
-			Wait:        false,
-		},
 	}
 
-	return NewRedisWithConfig(cfg, opts...)
+	return NewRedisWithConfig(cfg)
 }
 
 // NewRedis creates a new Redis instance.
-func NewRedisWithConfig(cfg *config.RedisConfig, opts ...redis.DialOption) *Redis {
-	rp := &redis.Pool{
-		Dial: func() (redis.Conn, error) {
-			return redis.Dial("tcp", cfg.Address, opts...)
-		},
-		MaxIdle:     cfg.Pool.MaxIdle,
-		MaxActive:   cfg.Pool.MaxActive,
-		IdleTimeout: cfg.Pool.IdleTimeout,
-		Wait:        cfg.Pool.Wait,
+func NewRedisWithConfig(cfg *config.RedisConfig) *Redis {
+	o := &redis.Options{
+		Addr:         cfg.Address,
+		MinIdleConns: cfg.MinIdleConns,
+		PoolSize:     cfg.PoolSize,
+		IdleTimeout:  cfg.IdleTimeout,
 	}
+
+	c := redis.NewClient(o)
+
 	return &Redis{
-		redisPool: rp,
+		c: c,
 	}
 }
 
 // Set adds a key-value pair to the redis instance.
 func (r *Redis) Set(ctx context.Context, key string, value []byte) error {
-	return retryOnBrokenPipe(BrokenPipeRetries, func() error {
-		conn, err := r.redisPool.GetContext(ctx)
-		if err != nil {
-			return err
-		}
-		defer conn.Close()
-
-		_, err = conn.Do("SET", key, value)
-		if err != nil {
-			return err
-		}
-		return nil
-	})
+	return r.c.Set(ctx, key, string(value), 0).Err()
 }
 
 // Get retrieves the value for a given key.
 func (r *Redis) Get(ctx context.Context, key string) ([]byte, error) {
-	return retryOnBrokenPipeBytes(BrokenPipeRetries, func() ([]byte, error) {
-		conn, err := r.redisPool.GetContext(ctx)
-		if err != nil {
-			return nil, err
-		}
-		defer conn.Close()
-
-		val, err := redis.Bytes(conn.Do("GET", key))
-		if err != nil {
-			return nil, err
-		}
-		return val, nil
-	})
+	val, err := r.c.Get(ctx, key).Result()
+	if err != nil {
+		return nil, err
+	}
+	return []byte(val), nil
 }
 
 // Delete deletes the key from the redis instance.
 func (r *Redis) Delete(ctx context.Context, key string) error {
-	return retryOnBrokenPipe(BrokenPipeRetries, func() error {
-		conn, err := r.redisPool.GetContext(ctx)
-		if err != nil {
-			return err
-		}
-		defer conn.Close()
-
-		_, err = conn.Do("DEL", key)
-		if err != nil {
-			return err
-		}
-		return nil
-	})
+	return r.c.Del(ctx, key).Err()
 }
 
 // FlushAll removes all key-value pairs from the redis instance.
 func (r *Redis) FlushAll(ctx context.Context) error {
-	return retryOnBrokenPipe(BrokenPipeRetries, func() error {
-		conn, err := r.redisPool.GetContext(ctx)
-		if err != nil {
-			return err
-		}
-		defer conn.Close()
-
-		_, err = conn.Do("FLUSHALL")
-		if err != nil {
-			return err
-		}
-		return nil
-	})
+	return r.c.FlushAll(ctx).Err()
 }
 
 // ListKeys lists all the keys in the redis instance.
 func (r *Redis) ListKeys(ctx context.Context) ([]string, error) {
-	return retryOnBrokenPipeStrings(BrokenPipeRetries, func() ([]string, error) {
-		conn, err := r.redisPool.GetContext(ctx)
-		if err != nil {
-			return nil, err
-		}
-		defer conn.Close()
-
-		keys, err := redis.Strings(conn.Do("KEYS", "*"))
-		if err != nil {
-			return nil, err
-		}
-		return keys, nil
-	})
-}
-
-func retryOnBrokenPipe(attempts int, fn func() error) error {
-	if err := fn(); err != nil {
-		if errors.Is(err, syscall.EPIPE) {
-			if attempts--; attempts > 0 {
-				return retryOnBrokenPipe(attempts, fn)
-			}
-		}
-		return err
-	}
-	return nil
-}
-
-func retryOnBrokenPipeBytes(attempts int, fn func() ([]byte, error)) ([]byte, error) {
-	b, err := fn()
-	if err != nil {
-		if errors.Is(err, syscall.EPIPE) {
-			if attempts--; attempts > 0 {
-				return retryOnBrokenPipeBytes(attempts, fn)
-			}
-		}
-		return nil, err
-	}
-	return b, nil
-}
-
-func retryOnBrokenPipeStrings(attempts int, fn func() ([]string, error)) ([]string, error) {
-	s, err := fn()
-	if err != nil {
-		if errors.Is(err, syscall.EPIPE) {
-			if attempts--; attempts > 0 {
-				return retryOnBrokenPipeStrings(attempts, fn)
-			}
-		}
-		return nil, err
-	}
-	return s, nil
+	return r.c.Keys(ctx, "*").Result()
 }

internal/pkg/cache/redis/redis_test.go

@@ -43,11 +43,11 @@ func TestRedis_All(t *testing.T) {
 
 	val, err := r.Get(ctx, "hello")
 	assert.NoError(t, err)
-	assert.Equal(t, val, by)
+	assert.Equal(t, by, val)
 
 	keys, err = r.ListKeys(ctx)
 	assert.NoError(t, err)
-	assert.Equal(t, keys, []string{"hello"})
+	assert.Equal(t, []string{"hello"}, keys)
 
 	assert.NoError(t, r.Delete(ctx, "hello"))
 

internal/pkg/config/loader.go

@@ -2,12 +2,13 @@ package config
 
 import (
 	"fmt"
-	log "github.com/sirupsen/logrus"
-	"github.com/spf13/pflag"
-	"github.com/spf13/viper"
 	"os"
 	"strconv"
 	"time"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
 )
 
 func Load(path string) (*ServiceConfig, error) {
@@ -94,15 +95,14 @@ func Load(path string) (*ServiceConfig, error) {
 	}
 
 	// Redis configuration
-	redisPool := RedisPool{
-		MaxIdle:     viper.GetInt(RedisPoolMaxIdle),
-		MaxActive:   viper.GetInt(RedisPoolMaxActive),
-		IdleTimeout: time.Duration(viper.GetUint(RedisPoolIdleTimeout)) * time.Second,
-		Wait:        viper.GetBool(RedisPoolWait),
-	}
 	redisConfig := RedisConfig{
-		Address: viper.GetString(RedisAddress),
-		Pool:    redisPool,
+		Address:         viper.GetString(RedisAddress),
+		MaxRetries:      viper.GetInt(RedisMaxRetries),
+		MinRetyBackoff:  viper.GetDuration(RedisMinRetryBackoff),
+		MaxRetryBackoff: viper.GetDuration(RedisMaxRetryBackoff),
+		MinIdleConns:    viper.GetInt(RedisMinIdleConns),
+		PoolSize:        viper.GetInt(RedisPoolSize),
+		IdleTimeout:     time.Duration(viper.GetUint(RedisIdleTimeout)) * time.Second,
 	}
 
 	return &ServiceConfig{

internal/pkg/config/model.go

@@ -10,10 +10,12 @@ const (
 	LogLevel         = "log_level"
 
 	RedisAddress         = "redis_address"
-	RedisPoolMaxIdle     = "redis_pool_maxIdle"
-	RedisPoolMaxActive   = "redis_pool_maxActive"
-	RedisPoolIdleTimeout = "redis_pool_idleTimeout"
-	RedisPoolWait        = "redis_pool_wait"
+	RedisMinIdleConns    = "redis_min_idle_conns"
+	RedisPoolSize        = "redis_pool_size"
+	RedisIdleTimeout     = "redis_idle_timeout"
+	RedisMaxRetries      = "redis_max_retries"
+	RedisMinRetryBackoff = "redis_min_retry_backoff"
+	RedisMaxRetryBackoff = "redis_max_retry_backoff"
 )
 
 type ServiceConfig struct {
@@ -30,15 +32,15 @@ type ServerConfig struct {
 	Project string
 }
 
+// RedisConfig as defined in https://pkg.go.dev/github.com/go-redis/redis/v8#Options
 type RedisConfig struct {
 	Address string
-	Pool    RedisPool
-}
 
-// RedisPool as defined in https://pkg.go.dev/github.com/gomodule/redigo/redis#Pool
-type RedisPool struct {
-	MaxIdle     int
-	MaxActive   int
-	IdleTimeout time.Duration
-	Wait        bool
+	MaxRetries      int
+	MinRetyBackoff  time.Duration
+	MaxRetryBackoff time.Duration
+
+	MinIdleConns int
+	PoolSize     int
+	IdleTimeout  time.Duration
 }