Merge pull request #115 from alibaba/bugfix-1.6.11

Bugfix 1.6.11

ChangeLog

@@ -1,4 +1,8 @@
-2019-06-25 Alibaba Cloud.
+2019-07-04 Alibaba Cloud.
+	* VERSION: 1.6.11
+	* BUGFIX: adapt "redis-go-cluster" driver to fix bug of big key syncing
+	block in `sync` mode. See #114
+2019-07-03 Alibaba Cloud.
 	* VERSION: 1.6.10
 	* IMPROVE: support print Lua in `decode` mode.
 	* BUGFIX: merge metric panic PR#111

conf/redis-shake.conf

@@ -6,7 +6,7 @@ id = redis-shake
 
 # log file，日志文件，不配置将打印到stdout (e.g. /var/log/redis-shake.log )
 log.file =
-# log level: "none", "error", "warn", "info", "all". default is "info".
+# log level: "none", "error", "warn", "info", "debug", "all". default is "info". "debug" == "all"
 log.level = info
 # pid path，进程文件存储地址（e.g. /var/run/)，不配置将默认输出到执行下面,
 # 注意这个是目录，真正的pid是`{pid_path}/{id}.pid`

src/redis-shake/common/cluster.go

@@ -1,8 +1,10 @@
 package utils
 
 import (
+
 	redigoCluster "github.com/vinllen/redis-go-cluster"
 	redigo "github.com/garyburd/redigo/redis"
+	"pkg/libs/log"
 )
 
 /* implement redigo.Conn(https://github.com/garyburd/redigo)
@@ -52,12 +54,37 @@ func (cc *ClusterConn) Send(commandName string, args ...interface{}) error {
 // send batcher and put the return into recvChan
 func (cc *ClusterConn) Flush() error {
 	ret, err := cc.client.RunBatch(cc.batcher)
+	defer func() {
 		cc.batcher = nil // reset batcher
+	}()
+
+	if err != nil {
 		cc.recvChan <- reply{
-		answer: ret,
+			answer: nil,
 			err:    err,
 		}
 
+		return err
+	}
+
+	// for redis-go-cluster driver, "Receive" function returns all the replies once flushed.
+	// However, this action is different with redigo driver that "Receive" only returns 1
+	// reply each time.
+
+	retLength := len(ret)
+	availableSize := cap(cc.recvChan) - len(cc.recvChan)
+	if availableSize < retLength {
+		log.Warnf("available channel size[%v] less than current returned batch size[%v]", availableSize, retLength)
+	}
+	log.Debugf("cluster flush batch with size[%v], return replies size[%v]", cc.batcher.GetBatchSize(), retLength)
+
+	for _, ele := range ret {
+		cc.recvChan <- reply{
+			answer: ele,
+			err:    err,
+		}
+	}
+
 	return err
 }
 

src/redis-shake/common/common.go

@@ -26,6 +26,7 @@ const (
 	LogLevelError = "error"
 	LogLevelWarn  = "warn"
 	LogLevelInfo  = "info"
+	LogLevelDebug = "debug"
 	LogLevelAll   = "all"
 
 	TencentCluster = "tencent_cluster"

src/redis-shake/common/utils.go

@@ -332,6 +332,9 @@ func set(c redigo.Conn, key []byte, value []byte) {
 }
 
 func flushAndCheckReply(c redigo.Conn, count int) {
+	// for redis-go-cluster driver, "Receive" function returns all the replies once flushed.
+	// However, this action is different with redigo driver that "Receive" only returns 1
+	// reply each time.
 	c.Flush()
 	for j := 0; j < count; j++ {
 		_, err := c.Receive()

src/redis-shake/main/main.go

@@ -266,6 +266,8 @@ func sanitizeOptions(tp string) error {
 		fallthrough
 	case utils.LogLevelInfo:
 		logDeepLevel = log.LEVEL_INFO
+	case utils.LogLevelDebug:
+		fallthrough
 	case utils.LogLevelAll:
 		logDeepLevel = log.LEVEL_DEBUG
 	default:

src/vendor/vendor.json

@@ -159,10 +159,10 @@
 			"revisionTime": "2019-03-04T09:57:49Z"
 		},
 		{
-			"checksumSHA1": "UXns0bW61NZgqtFBLj6jPgAwF+U=",
+			"checksumSHA1": "IjUkzECf826IH8K1bexM/uockIk=",
 			"path": "github.com/vinllen/redis-go-cluster",
-			"revision": "6ae0947e93ad83020f121a9b179d95e117fc0a5c",
-			"revisionTime": "2019-07-03T02:46:24Z"
+			"revision": "ffc34c0f905b65e6b71e07fd68a1d50b435c60f5",
+			"revisionTime": "2019-07-04T07:53:38Z"
 		},
 		{
 			"checksumSHA1": "TM3Neoy1xRAKyZYMGzKc41sDFW4=",