liangqi
/
RedisShake
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

finish test sync and restore

Browse Source
v4
vinllen 6 years ago
parent eec7e3bcee
commit 75e0f1a049
3 changed files with 45 additions and 38 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 20
      src/redis-shake/common/utils.go
  2. 57
      src/redis-shake/sync.go
  3. 6
      src/vendor/vendor.json

20
src/redis-shake/common/utils.go
Unescape View File

@ -30,13 +30,19 @@ import (
)
func OpenRedisConn(target []string, auth_type, passwd string, isCluster bool) redigo.Conn {
return OpenRedisConnWithTimeout(target, auth_type, passwd, 0, 0, isCluster)
}
func OpenRedisConnWithTimeout(target []string, auth_type, passwd string, readTimeout, writeTimeout time.Duration,
isCluster bool) redigo.Conn {
// return redigo.NewConn(OpenNetConn(target, auth_type, passwd), readTimeout, writeTimeout)
if isCluster {
cluster, err := redigoCluster.NewCluster(
&redigoCluster.Options{
StartNodes: target,
ConnTimeout: 1 * time.Second,
ReadTimeout: 500 * time.Millisecond,
WriteTimeout: 500 * time.Millisecond,
ReadTimeout: readTimeout,
WriteTimeout: writeTimeout,
KeepAlive: 16,
AliveTime: 60 * time.Second,
Password: passwd,
@ -47,14 +53,10 @@ func OpenRedisConn(target []string, auth_type, passwd string, isCluster bool) re
}
return NewClusterConn(cluster, 4096)
} else {
return redigo.NewConn(OpenNetConn(target[0], auth_type, passwd), 0, 0)
return redigo.NewConn(OpenNetConn(target[0], auth_type, passwd), readTimeout, writeTimeout)
}
}
func OpenRedisConnWithTimeout(target, auth_type, passwd string, readTimeout, writeTimeout time.Duration) redigo.Conn {
return redigo.NewConn(OpenNetConn(target, auth_type, passwd), readTimeout, writeTimeout)
}
func OpenNetConn(target, auth_type, passwd string) net.Conn {
d := net.Dialer{
KeepAlive: time.Duration(conf.Options.KeepAlive) * time.Second,
@ -64,9 +66,9 @@ func OpenNetConn(target, auth_type, passwd string) net.Conn {
log.PanicErrorf(err, "cannot connect to '%s'", target)
}
log.Infof("try to auth address[%v] with type[%v]", target, auth_type)
// log.Infof("try to auth address[%v] with type[%v]", target, auth_type)
AuthPassword(c, auth_type, passwd)
log.Info("auth OK!")
// log.Info("auth OK!")
return c
}

57
src/redis-shake/sync.go
Unescape View File

@ -71,7 +71,7 @@ func (cmd *CmdSync) Main() {
id int
source string
sourcePassword string
target string
target []string
targetPassword string
}
@ -80,9 +80,14 @@ func (cmd *CmdSync) Main() {
syncChan := make(chan syncNode, total)
cmd.dbSyncers = make([]*dbSyncer, total)
for i, source := range conf.Options.SourceAddressList {
// round-robin pick
pick := utils.PickTargetRoundRobin(len(conf.Options.TargetAddressList))
target := conf.Options.TargetAddressList[pick]
var target []string
if conf.Options.TargetType == conf.RedisTypeCluster {
target = conf.Options.TargetAddressList
} else {
// round-robin pick
pick := utils.PickTargetRoundRobin(len(conf.Options.TargetAddressList))
target = []string{conf.Options.TargetAddressList[pick]}
}
nd := syncNode{
id: i,
@ -130,13 +135,12 @@ func (cmd *CmdSync) Main() {
/*------------------------------------------------------*/
// one sync link corresponding to one dbSyncer
func NewDbSyncer(id int, source, sourcePassword, target, targetPassword string, httpPort int) *dbSyncer {
func NewDbSyncer(id int, source, sourcePassword string, target []string, targetPassword string, httpPort int) *dbSyncer {
ds := &dbSyncer{
id: id,
source: source,
sourcePassword: sourcePassword,
//todo
//target: target,
target: target,
targetPassword: targetPassword,
httpProfilePort: httpPort,
waitFull: make(chan struct{}),
@ -239,18 +243,16 @@ func (ds *dbSyncer) sync() {
go heartbeatCtl.Start()
}
// reader := bufio.NewReaderSize(input, utils.ReaderBufferSize)
reader := bufio.NewReaderSize(input, utils.ReaderBufferSize)
// sync rdb
base.Status = "full"
// todo
// ds.syncRDBFile(reader, ds.target, conf.Options.TargetAuthType, ds.targetPassword, nsize)
ds.syncRDBFile(reader, ds.target, conf.Options.TargetAuthType, ds.targetPassword, nsize)
// sync increment
base.Status = "incr"
close(ds.waitFull)
// todo
//ds.syncCommand(reader, ds.target, conf.Options.TargetAuthType, ds.targetPassword)
ds.syncCommand(reader, ds.target, conf.Options.TargetAuthType, ds.targetPassword)
}
func (ds *dbSyncer) sendSyncCmd(master, auth_type, passwd string) (net.Conn, int64) {
@ -389,7 +391,7 @@ func (ds *dbSyncer) pSyncPipeCopy(c net.Conn, br *bufio.Reader, bw *bufio.Writer
}
}
func (ds *dbSyncer) syncRDBFile(reader *bufio.Reader, target, auth_type, passwd string, nsize int64) {
func (ds *dbSyncer) syncRDBFile(reader *bufio.Reader, target []string, auth_type, passwd string, nsize int64) {
pipe := utils.NewRDBLoader(reader, &ds.rbytes, base.RDBPipeSize)
wait := make(chan struct{})
go func() {
@ -399,8 +401,7 @@ func (ds *dbSyncer) syncRDBFile(reader *bufio.Reader, target, auth_type, passwd
for i := 0; i < conf.Options.Parallel; i++ {
go func() {
defer wg.Done()
// todo
c := utils.OpenRedisConn([]string{target}, auth_type, passwd, false)
c := utils.OpenRedisConn(target, auth_type, passwd, conf.Options.TargetType == conf.RedisTypeCluster)
defer c.Close()
var lastdb uint32 = 0
for e := range pipe {
@ -467,8 +468,11 @@ func (ds *dbSyncer) syncRDBFile(reader *bufio.Reader, target, auth_type, passwd
log.Infof("dbSyncer[%v] sync rdb done", ds.id)
}
func (ds *dbSyncer) syncCommand(reader *bufio.Reader, target, auth_type, passwd string) {
c := utils.OpenRedisConnWithTimeout(target, auth_type, passwd, time.Duration(10)*time.Minute, time.Duration(10)*time.Minute)
func (ds *dbSyncer) syncCommand(reader *bufio.Reader, target []string, auth_type, passwd string) {
readeTimeout := time.Duration(10)*time.Minute
writeTimeout := time.Duration(10)*time.Minute
isCluster := conf.Options.TargetType == conf.RedisTypeCluster
c := utils.OpenRedisConnWithTimeout(target, auth_type, passwd, readeTimeout, writeTimeout, isCluster)
defer c.Close()
ds.sendBuf = make(chan cmdDetail, conf.Options.SenderCount)
@ -481,8 +485,8 @@ func (ds *dbSyncer) syncCommand(reader *bufio.Reader, target, auth_type, passwd
return
}
srcConn := utils.OpenRedisConnWithTimeout(ds.source, conf.Options.SourceAuthType, ds.sourcePassword,
time.Duration(10)*time.Minute, time.Duration(10)*time.Minute)
srcConn := utils.OpenRedisConnWithTimeout([]string{ds.source}, conf.Options.SourceAuthType, ds.sourcePassword,
readeTimeout, writeTimeout, false)
ticker := time.NewTicker(10 * time.Second)
for range ticker.C {
offset, err := utils.GetFakeSlaveOffset(srcConn)
@ -493,11 +497,11 @@ func (ds *dbSyncer) syncCommand(reader *bufio.Reader, target, auth_type, passwd
// Reconnect while network error happen
if err == io.EOF {
srcConn = utils.OpenRedisConnWithTimeout(ds.source, conf.Options.SourceAuthType,
ds.sourcePassword, time.Duration(10)*time.Minute, time.Duration(10)*time.Minute)
srcConn = utils.OpenRedisConnWithTimeout([]string{ds.source}, conf.Options.SourceAuthType,
ds.sourcePassword, readeTimeout, writeTimeout, false)
} else if _, ok := err.(net.Error); ok {
srcConn = utils.OpenRedisConnWithTimeout(ds.source, conf.Options.SourceAuthType,
ds.sourcePassword, time.Duration(10)*time.Minute, time.Duration(10)*time.Minute)
srcConn = utils.OpenRedisConnWithTimeout([]string{ds.source}, conf.Options.SourceAuthType,
ds.sourcePassword, readeTimeout, writeTimeout, false)
}
} else {
// ds.SyncStat.SetOffset(offset)
@ -673,6 +677,7 @@ func (ds *dbSyncer) syncCommand(reader *bufio.Reader, target, auth_type, passwd
noFlushCount += 1
ds.forward.Incr()
ds.wbytes.Add(int64(length))
metric.GetMetric(ds.id).AddPushCmdCount(1)
metric.GetMetric(ds.id).AddNetworkFlow(uint64(length))
sendId.Incr()
@ -700,9 +705,9 @@ func (ds *dbSyncer) syncCommand(reader *bufio.Reader, target, auth_type, passwd
nstat := ds.Stat()
var b bytes.Buffer
fmt.Fprintf(&b, "dbSyncer[%v] sync: ", ds.id)
fmt.Fprintf(&b, " +forward=%-6d", nstat.forward-lstat.forward)
fmt.Fprintf(&b, " +nbypass=%-6d", nstat.nbypass-lstat.nbypass)
fmt.Fprintf(&b, " +nbytes=%d", nstat.wbytes-lstat.wbytes) // todo
fmt.Fprintf(&b, " +forwardCommands=%-6d", nstat.forward-lstat.forward)
fmt.Fprintf(&b, " +filterCommands=%-6d", nstat.nbypass-lstat.nbypass)
fmt.Fprintf(&b, " +writeBytes=%d", nstat.wbytes-lstat.wbytes)
log.Info(b.String())
lstat = nstat
}

6
src/vendor/vendor.json vendored
Unescape View File

@ -81,10 +81,10 @@
"revisionTime": "2019-03-04T09:57:49Z"
},
{
"checksumSHA1": "aDV/4PzdvqpUsJ6xZF90V3heV/I=",
"checksumSHA1": "uma2AeftGc7LC/UB0QY2JSnxl3w=",
"path": "github.com/vinllen/redis-go-cluster",
"revision": "fe39f513526e1ad0fa164a89c094de860937b2ed",
"revisionTime": "2019-05-24T15:20:27Z"
"revision": "b6031aad42dc289f4696f835b2fe6dc0da49c8ee",
"revisionTime": "2019-05-25T06:57:32Z"
},
{
"checksumSHA1": "TM3Neoy1xRAKyZYMGzKc41sDFW4=",

Write Preview
Loading…
Cancel
Save