Merge pull request #20 from alibaba/bugfix-1.2.2

Bugfix 1.2.2

ChangeLog

@@ -1,3 +1,7 @@
+2019-04-03 Alibaba Cloud.
+	* version: 1.2.2
+	* BUGFIX: support 5.0 rdb RDB_OPCODE_MODULE_AUX, RDB_OPCODE_IDLE and
+	RDB_OPCODE_FREQ type.
 2019-03-27 Alibaba Cloud.
 	* version: 1.2.1
 	* IMPROVE: support syncing lua script in RDB syncing.

README.md

@@ -53,11 +53,11 @@ Add tag when releasing: "release-v{version}-{date}". for example: "release-v1.0.
 ---
 Run `./bin/redis-shake.darwin64` or `redis-shake.linux64` which is built in OSX and Linux respectively.<br>
 Or you can build redis-shake yourself according to the following steps:
-*  git clone https://github.com/aliyun/redis-shake.git
+*  git clone https://github.com/alibaba/RedisShake.git
-*  cd redis-shake/src/vendor
+*  cd RedisShake/src/vendor
 *  GOPATH=\`pwd\`/../..; govendor sync     #please note: must install govendor first and then pull all dependencies: `go get -u github.com/kardianos/govendor`
 *  cd ../../ && ./build.sh
-*  ./bin/collector -type=$(type_must_be_sync_dump_restore_or_decode) -conf=conf/redis-shake.conf #please note: user must modify collector.conf first to match needs.
+*  ./bin/redis-shake -type=$(type_must_be_sync_dump_restore_or_decode) -conf=conf/redis-shake.conf #please note: user must modify collector.conf first to match needs.
 
 # Shake series tool
 ---

conf/redis-shake.conf

@@ -2,8 +2,10 @@
 
 # id
 id = redis-shake
-# log file，日志文件，不配置将打印到stdout
+# log file，日志文件，不配置将打印到stdout (e.g. /var/log/redis-shake.log )
 log_file =
+# pid path，进程文件存储地址，不配置将输出到项目目录下 （e.g. /var/run/ )
+pid_path =
 
 # pprof port
 system_profile = 9310

src/pkg/rdb/loader.go

@@ -157,8 +157,28 @@ func (l *Loader) NextBinEntry() (*BinEntry, error) {
 			l.db = dbnum
 		case rdbFlagEOF:
 			return nil, nil
-		case rdbFlagOnlyValue:
+		case rdbFlagModuleAux:
-			fallthrough
+			// currently, ignore this filed
+			_, err := l.ReadLength() // module-id
+			if err != nil {
+				return nil, err
+			}
+			// skip module
+			if err = rdbLoadCheckModuleValue(l); err != nil {
+				return nil, err
+			}
+		case rdbFlagIdle:
+			// ignore idle because target redis doesn't support this for given key
+			_, err := l.ReadLength()
+			if err != nil {
+				return nil, err
+			}
+		case rdbFlagFreq:
+			// ignore freq because target redis doesn't support this for given key
+			_, err := l.readUint8()
+			if err != nil {
+				return nil, err
+			}
 		default:
 			var key []byte
 			if l.remainMember == 0 {

src/pkg/rdb/reader.go

@@ -35,13 +35,23 @@ const (
 	RdbTypeQuicklist       = 14
 	RDBTypeStreamListPacks = 15 // stream
 
-	rdbFlagOnlyValue = 0xf9
+	rdbFlagModuleAux = 0xf7
+	rdbFlagIdle      = 0xf8
+	rdbFlagFreq      = 0xf9
 	RdbFlagAUX       = 0xfa
 	rdbFlagResizeDB  = 0xfb
 	rdbFlagExpiryMS  = 0xfc
 	rdbFlagExpiry    = 0xfd
 	rdbFlagSelectDB  = 0xfe
 	rdbFlagEOF       = 0xff
+
+	// Module serialized values sub opcodes
+	rdbModuleOpcodeEof    = 0
+	rdbModuleOpcodeSint   = 1
+	rdbModuleOpcodeUint   = 2
+	rdbModuleOpcodeFloat  = 3
+	rdbModuleOpcodeDouble = 4
+	rdbModuleOpcodeString = 5
 )
 
 const (

src/redis-shake/common/mix.go

@@ -22,8 +22,13 @@ func WritePid(id string) (err error) {
 	return nil
 }
 
-func WritePidById(id string) error {
+func WritePidById(id string, path string) error {
-	dir, _ := os.Getwd()
+	var dir string
+	if path == "" {
+		dir, _ = os.Getwd()
+	} else {
+		dir = path
+	}
 	pidfile := filepath.Join(dir, id) + ".pid"
 	if err := WritePid(pidfile); err != nil {
 		return err

src/redis-shake/configure/configure.go

@@ -40,6 +40,7 @@ type Configuration struct {
 	SenderCount               uint     `config:"sender.count"`
 	SenderDelayChannelSize    uint     `config:"sender.delay_channel_size"`
 	KeepAlive                 uint     `config:"keep_alive"`
+	PidPath                   string   `config:"pid_path"`
 
 	// inner variables
 	ReplaceHashTag bool   `config:"replace_hash_tag"`

src/redis-shake/main/main.go

@@ -5,40 +5,45 @@
 package main
 
 import (
+	"encoding/json"
 	"flag"
 	"fmt"
+	"math"
+	_ "net/http/pprof"
 	"os"
 	"os/signal"
-	"syscall"
+	"reflect"
-	"runtime/debug"
-	"time"
 	"runtime"
-	"math"
+	"runtime/debug"
-	_ "net/http/pprof"
-	"strings"
 	"strconv"
-	"encoding/json"
+	"strings"
-	"reflect"
+	"syscall"
+	"time"
 
+	"pkg/libs/log"
+	"redis-shake"
+	"redis-shake/base"
 	"redis-shake/common"
 	"redis-shake/configure"
 	"redis-shake/metric"
-	"redis-shake"
-	"redis-shake/base"
 	"redis-shake/restful"
-	"pkg/libs/log"
 
 	"github.com/gugemichael/nimo4go"
 	logRotate "gopkg.in/natefinch/lumberjack.v2"
 )
 
-type Exit struct {Code int}
+type Exit struct{ Code int }
 
-const(
+const (
 	TypeDecode  = "decode"
 	TypeRestore = "restore"
 	TypeDump    = "dump"
 	TypeSync    = "sync"
+
+	defaultHttpPort    = 20881
+	defaultSystemPort  = 20882
+	defaultSenderSize  = 65535
+	defaultSenderCount = 1024
 )
 
 func main() {
@@ -83,7 +88,7 @@ func main() {
 	utils.Welcome()
 	utils.StartTime = fmt.Sprintf("%v", time.Now().Format(utils.GolangSecurityTime))
 
-	if err = utils.WritePidById(conf.Options.Id); err != nil {
+	if err = utils.WritePidById(conf.Options.Id, conf.Options.PidPath); err != nil {
 		crash(fmt.Sprintf("write pid failed. %v", err), -5)
 	}
 
@@ -172,7 +177,9 @@ func sanitizeOptions(tp string) error {
 		runtime.GOMAXPROCS(conf.Options.NCpu)
 	}
 
-	if conf.Options.Parallel == 0 || conf.Options.Parallel > 1024 {
+	if conf.Options.Parallel == 0 { // not set
+		conf.Options.Parallel = 1
+	} else if conf.Options.Parallel > 1024 {
 		return fmt.Errorf("parallel[%v] should in (0, 1024]", conf.Options.Parallel)
 	} else {
 		conf.Options.Parallel = int(math.Max(float64(conf.Options.Parallel), float64(conf.Options.NCpu)))
@@ -204,7 +211,7 @@ func sanitizeOptions(tp string) error {
 	}
 
 	if conf.Options.LogFile != "" {
-		conf.Options.LogFile = fmt.Sprintf("%s.log", conf.Options.Id)
+		//conf.Options.LogFile = fmt.Sprintf("%s.log", conf.Options.Id)
 
 		utils.LogRotater = &logRotate.Logger{
 			Filename:   conf.Options.LogFile,
@@ -215,9 +222,9 @@ func sanitizeOptions(tp string) error {
 		log.StdLog = log.New(utils.LogRotater, "")
 	}
 
-	// heartbeat
+	// heartbeat, 86400 = 1 day
-	if conf.Options.HeartbeatInterval <= 0 || conf.Options.HeartbeatInterval > 86400 {
+	if conf.Options.HeartbeatInterval > 86400 {
-		return fmt.Errorf("HeartbeatInterval[%v] should in (0, 86400]", conf.Options.HeartbeatInterval)
+		return fmt.Errorf("HeartbeatInterval[%v] should in [0, 86400]", conf.Options.HeartbeatInterval)
 	}
 	if conf.Options.HeartbeatNetworkInterface == "" {
 		conf.Options.HeartbeatIp = "127.0.0.1"
@@ -240,7 +247,7 @@ func sanitizeOptions(tp string) error {
 			if n, err := strconv.ParseInt(conf.Options.FakeTime[1:], 10, 64); err != nil {
 				return fmt.Errorf("parse fake_time failed[%v]", err)
 			} else {
-				conf.Options.ShiftTime = time.Duration(n * int64(time.Millisecond) - time.Now().UnixNano())
+				conf.Options.ShiftTime = time.Duration(n*int64(time.Millisecond) - time.Now().UnixNano())
 			}
 		default:
 			if t, err := time.Parse("2006-01-02 15:04:05", conf.Options.FakeTime); err != nil {
@@ -273,25 +280,38 @@ func sanitizeOptions(tp string) error {
 		// pass, >= 0 means enable
 	}
 
-	if conf.Options.HttpProfile <= 0 || conf.Options.HttpProfile > 65535 {
+	if conf.Options.HttpProfile < 0 || conf.Options.HttpProfile > 65535 {
-		return fmt.Errorf("HttpProfile[%v] should in (0, 65535]", conf.Options.HttpProfile)
+		return fmt.Errorf("HttpProfile[%v] should in [0, 65535]", conf.Options.HttpProfile)
+	} else if conf.Options.HttpProfile  == 0 {
+		// set to default when not set
+		conf.Options.HttpProfile = defaultHttpPort
 	}
-	if conf.Options.SystemProfile <= 0 || conf.Options.SystemProfile > 65535 {
+
-		return fmt.Errorf("SystemProfile[%v] should in (0, 65535]", conf.Options.SystemProfile)
+	if conf.Options.SystemProfile < 0 || conf.Options.SystemProfile > 65535 {
+		return fmt.Errorf("SystemProfile[%v] should in [0, 65535]", conf.Options.SystemProfile)
+	} else if conf.Options.SystemProfile  == 0 {
+		// set to default when not set
+		conf.Options.SystemProfile = defaultSystemPort
 	}
 
-	if conf.Options.SenderSize <= 0 || conf.Options.SenderSize >= 1073741824 {
+	if conf.Options.SenderSize < 0 || conf.Options.SenderSize >= 1073741824 {
-		return fmt.Errorf("SenderSize[%v] should in (0, 1073741824]", conf.Options.SenderSize)
+		return fmt.Errorf("SenderSize[%v] should in [0, 1073741824]", conf.Options.SenderSize)
+	} else if conf.Options.SenderSize  == 0 {
+		// set to default when not set
+		conf.Options.SenderSize = defaultSenderSize
 	}
 
-	if conf.Options.SenderCount <= 0 || conf.Options.SenderCount >= 100000 {
+	if conf.Options.SenderCount < 0 || conf.Options.SenderCount >= 100000 {
-		return fmt.Errorf("SenderCount[%v] should in (0, 100000]", conf.Options.SenderCount)
+		return fmt.Errorf("SenderCount[%v] should in [0, 100000]", conf.Options.SenderCount)
+	} else if conf.Options.SenderCount  == 0 {
+		// set to default when not set
+		conf.Options.SenderCount = defaultSenderCount
 	}
 
 	if tp == TypeRestore || tp == TypeSync {
 		// get target redis version and set TargetReplace.
 		if conf.Options.TargetRedisVersion, err = utils.GetRedisVersion(conf.Options.TargetAddress,
-				conf.Options.TargetAuthType, conf.Options.TargetPasswordRaw); err != nil {
+			conf.Options.TargetAuthType, conf.Options.TargetPasswordRaw); err != nil {
 			return fmt.Errorf("get target redis version failed[%v]", err)
 		} else {
 			if strings.HasPrefix(conf.Options.TargetRedisVersion, "4.") ||