cherry-pick v1.4.4

ChangeLog

@@ -1,6 +1,16 @@
+2019-06-07 Alibaba Cloud.
+	* VERSION: 1.6.6
+	* cherry-pick merge v1.4.4
+	*
 2019-06-06 Alibaba Cloud.
 	* VERSION: 1.6.5
 	* IMPROVE: run rump in parallel to support several db nodes behind proxy.
+	* BUGFIX: run rump panic when the source is proxy with more than 1 db.
+2019-06-05 Alibaba Cloud.
+	* VERSION: 1.4.4
+	* BUGFIX: modify the ttl from millisecond to second in restore when
+	overpass big key threshold.
+	* IMPROVE: set some default values in configuration.
 2019-05-30 Alibaba Cloud.
 	* VERSION: 1.6.4
 	* BUGFIX: fix bug of `GetDetailedInfo` panic.

src/redis-shake/common/common.go

@@ -36,6 +36,14 @@ var (
 	TargetRoundRobin int
 )
 
+const (
+	KB = 1024
+	MB = 1024 * KB
+	GB = 1024 * MB
+	TB = 1024 * GB
+	PB = 1024 * TB
+)
+
 // read until hit the end of RESP: "\r\n"
 func ReadRESPEnd(c net.Conn) (string, error) {
 	var ret string

src/redis-shake/common/utils.go

@@ -723,7 +723,7 @@ func RestoreRdbEntry(c redigo.Conn, e *rdb.BinEntry) {
 		}
 		restoreQuicklistEntry(c, e)
 		if e.ExpireAt != 0 {
-			r, err := redigo.Int64(c.Do("expire", e.Key, ttlms))
+			r, err := redigo.Int64(c.Do("pexpire", e.Key, ttlms))
 			if err != nil && r != 1 {
 				log.Panicf("expire ", string(e.Key), err)
 			}
@@ -755,7 +755,7 @@ func RestoreRdbEntry(c redigo.Conn, e *rdb.BinEntry) {
 		}
 		restoreBigRdbEntry(c, e)
 		if e.ExpireAt != 0 {
-			r, err := redigo.Int64(c.Do("expire", e.Key, ttlms))
+			r, err := redigo.Int64(c.Do("pexpire", e.Key, ttlms))
 			if err != nil && r != 1 {
 				log.Panicf("expire ", string(e.Key), err)
 			}

src/redis-shake/configure/configure.go

@@ -23,7 +23,7 @@ type Configuration struct {
 	TargetPasswordRaw         string   `config:"target.password_raw"`
 	TargetPasswordEncoding    string   `config:"target.password_encoding"`
 	TargetVersion             uint     `config:"target.version"`
-	TargetDB                  int      `config:"target.db"`
+	TargetDBString            string   `config:"target.db"`
 	TargetAuthType            string   `config:"target.auth_type"`
 	TargetType                string   `config:"target.type"`
 	TargetTLSEnable           bool     `config:"target.tls_enable"`
@@ -67,6 +67,7 @@ type Configuration struct {
 	ShiftTime          time.Duration // shift
 	TargetRedisVersion string        // to_redis_version
 	TargetReplace      bool          // to_replace
+	TargetDB           int           // int type
 }
 
 var Options Configuration

src/redis-shake/main/main.go

@@ -181,8 +181,11 @@ func sanitizeOptions(tp string) error {
 		conf.Options.Parallel = int(math.Max(float64(conf.Options.Parallel), float64(conf.Options.NCpu)))
 	}
 
-	if conf.Options.BigKeyThreshold > 524288000 {
+	// 500 M
-		return fmt.Errorf("BigKeyThreshold[%v] should <= 524288000", conf.Options.BigKeyThreshold)
+	if conf.Options.BigKeyThreshold > 500 * utils.MB {
+		return fmt.Errorf("BigKeyThreshold[%v] should <= 500 MB", conf.Options.BigKeyThreshold)
+	} else if conf.Options.BigKeyThreshold == 0 {
+		conf.Options.BigKeyThreshold = 50 * utils.MB
 	}
 
 	// source password
@@ -266,7 +269,10 @@ func sanitizeOptions(tp string) error {
 	// heartbeat, 86400 = 1 day
 	if conf.Options.HeartbeatInterval > 86400 {
 		return fmt.Errorf("HeartbeatInterval[%v] should in [0, 86400]", conf.Options.HeartbeatInterval)
+	} else if conf.Options.HeartbeatInterval == 0 {
+		conf.Options.HeartbeatInterval = 10
 	}
+
 	if conf.Options.HeartbeatNetworkInterface == "" {
 		conf.Options.HeartbeatIp = "127.0.0.1"
 	} else {
@@ -325,8 +331,14 @@ func sanitizeOptions(tp string) error {
 		}
 	}
 
-	if conf.Options.TargetDB >= 0 {
+	if conf.Options.TargetDBString == "" {
-		// pass, >= 0 means enable
+		conf.Options.TargetDB = -1
+	} else if v, err := strconv.Atoi(conf.Options.TargetDBString); err != nil {
+		return fmt.Errorf("parse target.db[%v] failed[%v]", conf.Options.TargetDBString, err)
+	} else if v < 0 {
+		conf.Options.TargetDB = -1
+	} else {
+		conf.Options.TargetDB = v
 	}
 
 	if conf.Options.HttpProfile < 0 || conf.Options.HttpProfile > 65535 {
@@ -357,7 +369,12 @@ func sanitizeOptions(tp string) error {
 		conf.Options.SenderCount = defaultSenderCount
 	}
 
-	if tp == conf.TypeRestore || tp == conf.TypeSync {
+
+	if conf.Options.SenderDelayChannelSize == 0 {
+		conf.Options.SenderDelayChannelSize = 32
+	}
+
+	if tp == TypeRestore || tp == TypeSync {
 		// get target redis version and set TargetReplace.
 		for _, address := range conf.Options.TargetAddressList {
 			// single connection even if the target is cluster