support tcp keepalive

conf/redis-shake.conf

@@ -126,6 +126,12 @@ sender.count = 5000
 # 用于metric统计时延的队列
 sender.delay_channel_size = 65535
 
+# enable keep_alive option in TCP when connecting redis.
+# the unit is second.
+# 0 means disable.
+# TCP keep-alive保活参数，单位秒，0表示不启用。
+keep_alive = 0
+
 # ----------------splitter----------------
 # below variables are useless for current opensource version so don't set.
 

src/redis-shake/common/utils.go

@@ -34,7 +34,10 @@ func OpenRedisConnWithTimeout(target, auth_type, passwd string, readTimeout, wri
 }
 
 func OpenNetConn(target, auth_type, passwd string) net.Conn {
-	c, err := net.Dial("tcp", target)
+	d := net.Dialer{
+		KeepAlive: time.Duration(conf.Options.KeepAlive) * time.Second,
+	}
+	c, err := d.Dial("tcp", target)
 	if err != nil {
 		log.PanicErrorf(err, "cannot connect to '%s'", target)
 	}

src/redis-shake/configure/configure.go

@@ -39,6 +39,7 @@ type Configuration struct {
 	SenderSize                uint64   `config:"sender.size"`
 	SenderCount               uint     `config:"sender.count"`
 	SenderDelayChannelSize    uint     `config:"sender.delay_channel_size"`
+	KeepAlive                 uint     `config:"keep_alive"`
 
 	// inner variables
 	ReplaceHashTag bool   `config:"replace_hash_tag"`

src/redis-shake/sync.go

@@ -361,7 +361,7 @@ func (cmd *CmdSync) SyncCommand(reader *bufio.Reader, target, auth_type, passwd
 			offset, err := utils.GetFakeSlaveOffset(srcConn)
 			if err != nil {
 				// log.PurePrintf("%s\n", NewLogItem("GetFakeSlaveOffsetFail", "WARN", NewErrorLogDetail("", err.Error())))
-				log.Warnf("Event:GetFakeSlaveOffsetFail\tId:%s\tError:%s", conf.Options.Id, err.Error())
+				log.Warnf("Event:GetFakeSlaveOffsetFail\tId:%s\tWarn:%s", conf.Options.Id, err.Error())
 
 				// Reconnect while network error happen
 				if err == io.EOF {