feat: update go-redis to v8, support ctx in redis methods (#1507)

* feat: update go-redis to v8, support ctx in redis methods

* fix compile errors

* chore: remove unused const

* chore: add tracing log on redis

core/limit/periodlimit_test.go

@@ -23,10 +23,9 @@ func TestPeriodLimit_RedisUnavailable(t *testing.T) {
 
 	const (
 		seconds = 1
-		total   = 100
 		quota   = 5
 	)
-	l := NewPeriodLimit(seconds, quota, redis.NewRedis(s.Addr(), redis.NodeType), "periodlimit")
+	l := NewPeriodLimit(seconds, quota, redis.New(s.Addr()), "periodlimit")
 	s.Close()
 	val, err := l.Take("first")
 	assert.NotNil(t, err)

core/stores/redis/hook.go

@@ -0,0 +1,83 @@
+package redis
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	red "github.com/go-redis/redis/v8"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/mapping"
+	"github.com/zeromicro/go-zero/core/timex"
+)
+
+var (
+	startTimeKey = contextKey("startTime")
+	durationHook = hook{}
+)
+
+type (
+	contextKey string
+	hook       struct{}
+)
+
+func (h hook) BeforeProcess(ctx context.Context, _ red.Cmder) (context.Context, error) {
+	return context.WithValue(ctx, startTimeKey, timex.Now()), nil
+}
+
+func (h hook) AfterProcess(ctx context.Context, cmd red.Cmder) error {
+	val := ctx.Value(startTimeKey)
+	if val == nil {
+		return nil
+	}
+
+	start, ok := val.(time.Duration)
+	if !ok {
+		return nil
+	}
+
+	duration := timex.Since(start)
+	if duration > slowThreshold.Load() {
+		logDuration(ctx, cmd, duration)
+	}
+
+	return nil
+}
+
+func (h hook) BeforeProcessPipeline(ctx context.Context, _ []red.Cmder) (context.Context, error) {
+	return context.WithValue(ctx, startTimeKey, timex.Now()), nil
+}
+
+func (h hook) AfterProcessPipeline(ctx context.Context, cmds []red.Cmder) error {
+	if len(cmds) == 0 {
+		return nil
+	}
+
+	val := ctx.Value(startTimeKey)
+	if val == nil {
+		return nil
+	}
+
+	start, ok := val.(time.Duration)
+	if !ok {
+		return nil
+	}
+
+	duration := timex.Since(start)
+	if duration > slowThreshold.Load()*time.Duration(len(cmds)) {
+		logDuration(ctx, cmds[0], duration)
+	}
+
+	return nil
+}
+
+func logDuration(ctx context.Context, cmd red.Cmder, duration time.Duration) {
+	var buf strings.Builder
+	for i, arg := range cmd.Args() {
+		if i > 0 {
+			buf.WriteByte(' ')
+		}
+		buf.WriteString(mapping.Repr(arg))
+	}
+	logx.WithContext(ctx).WithDuration(duration).Slowf("[REDIS] slowcall on executing: %s", buf.String())
+}

core/stores/redis/hook_test.go

@@ -0,0 +1,138 @@
+package redis
+
+import (
+	"context"
+	"log"
+	"strings"
+	"testing"
+	"time"
+
+	red "github.com/go-redis/redis/v8"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHookProcessCase1(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	ctx, err := durationHook.BeforeProcess(context.Background(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Nil(t, durationHook.AfterProcess(ctx, red.NewCmd(context.Background())))
+	assert.False(t, strings.Contains(buf.String(), "slow"))
+}
+
+func TestHookProcessCase2(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	ctx, err := durationHook.BeforeProcess(context.Background(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(slowThreshold.Load() + time.Millisecond)
+
+	assert.Nil(t, durationHook.AfterProcess(ctx, red.NewCmd(context.Background(), "foo", "bar")))
+	assert.True(t, strings.Contains(buf.String(), "slow"))
+}
+
+func TestHookProcessCase3(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	assert.Nil(t, durationHook.AfterProcess(context.Background(), red.NewCmd(context.Background())))
+	assert.True(t, buf.Len() == 0)
+}
+
+func TestHookProcessCase4(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	ctx := context.WithValue(context.Background(), startTimeKey, "foo")
+	assert.Nil(t, durationHook.AfterProcess(ctx, red.NewCmd(context.Background())))
+	assert.True(t, buf.Len() == 0)
+}
+
+func TestHookProcessPipelineCase1(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	ctx, err := durationHook.BeforeProcessPipeline(context.Background(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Nil(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{
+		red.NewCmd(context.Background()),
+	}))
+	assert.False(t, strings.Contains(buf.String(), "slow"))
+}
+
+func TestHookProcessPipelineCase2(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	ctx, err := durationHook.BeforeProcessPipeline(context.Background(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(slowThreshold.Load() + time.Millisecond)
+
+	assert.Nil(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{
+		red.NewCmd(context.Background(), "foo", "bar"),
+	}))
+	assert.True(t, strings.Contains(buf.String(), "slow"))
+}
+
+func TestHookProcessPipelineCase3(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	assert.Nil(t, durationHook.AfterProcessPipeline(context.Background(), []red.Cmder{
+		red.NewCmd(context.Background()),
+	}))
+	assert.True(t, buf.Len() == 0)
+}
+
+func TestHookProcessPipelineCase4(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	ctx := context.WithValue(context.Background(), startTimeKey, "foo")
+	assert.Nil(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{
+		red.NewCmd(context.Background()),
+	}))
+	assert.True(t, buf.Len() == 0)
+}
+
+func TestHookProcessPipelineCase5(t *testing.T) {
+	writer := log.Writer()
+	var buf strings.Builder
+	log.SetOutput(&buf)
+	defer log.SetOutput(writer)
+
+	ctx := context.WithValue(context.Background(), startTimeKey, "foo")
+	assert.Nil(t, durationHook.AfterProcessPipeline(ctx, nil))
+	assert.True(t, buf.Len() == 0)
+}

core/stores/redis/process.go

@@ -1,32 +0,0 @@
-package redis
-
-import (
-	"strings"
-
-	red "github.com/go-redis/redis"
-	"github.com/zeromicro/go-zero/core/logx"
-	"github.com/zeromicro/go-zero/core/mapping"
-	"github.com/zeromicro/go-zero/core/timex"
-)
-
-func checkDuration(proc func(red.Cmder) error) func(red.Cmder) error {
-	return func(cmd red.Cmder) error {
-		start := timex.Now()
-
-		defer func() {
-			duration := timex.Since(start)
-			if duration > slowThreshold.Load() {
-				var buf strings.Builder
-				for i, arg := range cmd.Args() {
-					if i > 0 {
-						buf.WriteByte(' ')
-					}
-					buf.WriteString(mapping.Repr(arg))
-				}
-				logx.WithDuration(duration).Slowf("[REDIS] slowcall on executing: %s", buf.String())
-			}
-		}()
-
-		return proc(cmd)
-	}
-}

core/stores/redis/redis_test.go

@@ -1,6 +1,7 @@
 package redis
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"io"
@@ -9,7 +10,7 @@ import (
 	"time"
 
 	"github.com/alicebob/miniredis/v2"
-	red "github.com/go-redis/redis"
+	red "github.com/go-redis/redis/v8"
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stringx"
@@ -964,13 +965,14 @@ func TestRedis_SortedSet(t *testing.T) {
 		assert.NotNil(t, err)
 		client.Zadd("second", 2, "aa")
 		client.Zadd("third", 3, "bbb")
-		val, err = client.Zunionstore("union", ZStore{
+		val, err = client.Zunionstore("union", &ZStore{
+			Keys:      []string{"second", "third"},
 			Weights:   []float64{1, 2},
 			Aggregate: "SUM",
-		}, "second", "third")
+		})
 		assert.Nil(t, err)
 		assert.Equal(t, int64(2), val)
-		_, err = New(client.Addr, badType()).Zunionstore("union", ZStore{})
+		_, err = New(client.Addr, badType()).Zunionstore("union", &ZStore{})
 		assert.NotNil(t, err)
 		vals, err = client.Zrange("union", 0, 10000)
 		assert.Nil(t, err)
@@ -988,9 +990,9 @@ func TestRedis_Pipelined(t *testing.T) {
 		}))
 		err := client.Pipelined(
 			func(pipe Pipeliner) error {
-				pipe.Incr("pipelined_counter")
+				pipe.Incr(context.Background(), "pipelined_counter")
-				pipe.Expire("pipelined_counter", time.Hour)
+				pipe.Expire(context.Background(), "pipelined_counter", time.Hour)
-				pipe.ZAdd("zadd", Z{Score: 12, Member: "zadd"})
+				pipe.ZAdd(context.Background(), "zadd", &Z{Score: 12, Member: "zadd"})
 				return nil
 			},
 		)
@@ -1187,6 +1189,6 @@ type mockedNode struct {
 	RedisNode
 }
 
-func (n mockedNode) BLPop(timeout time.Duration, keys ...string) *red.StringSliceCmd {
+func (n mockedNode) BLPop(ctx context.Context, timeout time.Duration, keys ...string) *red.StringSliceCmd {
-	return red.NewStringSliceCmd("foo", "bar")
+	return red.NewStringSliceCmd(context.Background(), "foo", "bar")
 }

core/stores/redis/redisblockingnode.go

@@ -3,7 +3,7 @@ package redis
 import (
 	"fmt"
 
-	red "github.com/go-redis/redis"
+	red "github.com/go-redis/redis/v8"
 	"github.com/zeromicro/go-zero/core/logx"
 )
 

core/stores/redis/redisclientmanager.go

@@ -4,7 +4,7 @@ import (
 	"crypto/tls"
 	"io"
 
-	red "github.com/go-redis/redis"
+	red "github.com/go-redis/redis/v8"
 	"github.com/zeromicro/go-zero/core/syncx"
 )
 
@@ -32,7 +32,8 @@ func getClient(r *Redis) (*red.Client, error) {
 			MinIdleConns: idleConns,
 			TLSConfig:    tlsConfig,
 		})
-		store.WrapProcess(checkDuration)
+		store.AddHook(durationHook)
+
 		return store, nil
 	})
 	if err != nil {

core/stores/redis/redisclustermanager.go

@@ -4,7 +4,7 @@ import (
 	"crypto/tls"
 	"io"
 
-	red "github.com/go-redis/redis"
+	red "github.com/go-redis/redis/v8"
 	"github.com/zeromicro/go-zero/core/syncx"
 )
 
@@ -25,7 +25,7 @@ func getCluster(r *Redis) (*red.ClusterClient, error) {
 			MinIdleConns: idleConns,
 			TLSConfig:    tlsConfig,
 		})
-		store.WrapProcess(checkDuration)
+		store.AddHook(durationHook)
 
 		return store, nil
 	})

core/stores/redis/redislock.go

@@ -5,7 +5,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	red "github.com/go-redis/redis"
+	red "github.com/go-redis/redis/v8"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stringx"
 )

core/stores/redis/redistest/redistest.go

@@ -17,10 +17,12 @@ func CreateRedis() (r *redis.Redis, clean func(), err error) {
 
 	return redis.New(mr.Addr()), func() {
 		ch := make(chan lang.PlaceholderType)
+
 		go func() {
 			mr.Close()
 			close(ch)
 		}()
+
 		select {
 		case <-ch:
 		case <-time.After(time.Second):

go.mod

@@ -7,7 +7,6 @@ require (
 	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/alicebob/miniredis/v2 v2.17.0
 	github.com/globalsign/mgo v0.0.0-20181015135952-eeefdecb41b8
-	github.com/go-redis/redis v6.15.9+incompatible
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/golang-jwt/jwt/v4 v4.2.0
 	github.com/golang/mock v1.6.0
@@ -42,6 +41,7 @@ require (
 require (
 	github.com/fatih/color v1.10.0 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-redis/redis/v8 v8.11.4
 	github.com/mattn/go-runewidth v0.0.13 // indirect
 	github.com/openzipkin/zipkin-go v0.4.0 // indirect
 	golang.org/x/net v0.0.0-20220114011407-0dd24b26b47d // indirect

go.sum

@@ -62,8 +62,9 @@ github.com/bkaradzic/go-lz4 v1.0.0/go.mod h1:0YdlkowM3VswSROI7qDxhRvJ3sLhlFrRRwj
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
-github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
@@ -87,6 +88,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 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/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
@@ -142,8 +145,8 @@ github.com/go-openapi/jsonreference v0.19.3/go.mod h1:rjx6GuL8TTa9VaixXglHmQmIL9
 github.com/go-openapi/spec v0.19.3/go.mod h1:FpwSN1ksY1eteniUU7X0N/BgJ7a4WvBFVA8Lj9mJglo=
 github.com/go-openapi/swag v0.19.2/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
-github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=
+github.com/go-redis/redis/v8 v8.11.4 h1:kHoYkfZP6+pe04aFTnhDH6GDROa5yJdHJVNxV3F46Tg=
-github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
+github.com/go-redis/redis/v8 v8.11.4/go.mod h1:2Z2wHZXdQpCDXEGzqMockDpNyYvi2l4Pxt6RJr792+w=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=
 github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=

tools/goctl/go.mod

@@ -13,5 +13,5 @@ require (
 	github.com/urfave/cli v1.22.5
 	github.com/zeromicro/antlr v0.0.1
 	github.com/zeromicro/ddl-parser v0.0.0-20210712021150-63520aca7348
-	github.com/zeromicro/go-zero v1.3.0-beta
+	github.com/zeromicro/go-zero v1.3.0
 )

tools/goctl/go.sum

@@ -360,8 +360,8 @@ github.com/zeromicro/antlr v0.0.1 h1:CQpIn/dc0pUjgGQ81y98s/NGOm2Hfru2NNio2I9mQgk
 github.com/zeromicro/antlr v0.0.1/go.mod h1:nfpjEwFR6Q4xGDJMcZnCL9tEfQRgszMwu3rDz2Z+p5M=
 github.com/zeromicro/ddl-parser v0.0.0-20210712021150-63520aca7348 h1:OhxL9tn28gDeJVzreIUiE5oVxZCjL3tBJ0XBNw8p5R8=
 github.com/zeromicro/ddl-parser v0.0.0-20210712021150-63520aca7348/go.mod h1:ISU/8NuPyEpl9pa17Py9TBPetMjtsiHrb9f5XGiYbo8=
-github.com/zeromicro/go-zero v1.3.0-beta h1:jNwtjGzAV6PSklpXbLCmQph7BzfH7abcUIOG9odWMsM=
+github.com/zeromicro/go-zero v1.3.0 h1:Eyn36yBtR043sm4YKmxR6eS3UA/GtZDktQ+UqIJ3Lm0=
-github.com/zeromicro/go-zero v1.3.0-beta/go.mod h1:Hy4o1VFAt32lXaQMbaBhoFeZjA/rJqJ4PTGNdGsURcc=
+github.com/zeromicro/go-zero v1.3.0/go.mod h1:Hy4o1VFAt32lXaQMbaBhoFeZjA/rJqJ4PTGNdGsURcc=
 go.etcd.io/etcd/api/v3 v3.5.1/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=
 go.etcd.io/etcd/client/pkg/v3 v3.5.1/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3YSwc9/Ac1g=
 go.etcd.io/etcd/client/v3 v3.5.1/go.mod h1:OnjH4M8OnAotwaB2l9bVgZzRFKru7/ZMoS46OtKyd3Q=