go beetle: reset cursor when current master changes
MODIFY
go/copy_keys.go
go/copy_keys.go
@@ -29,6 +29,7 @@ type CopierState struct {
redis *redis.Client // current source connection
targetRedis *redis.Client // target connection
keySuffixes []string
+ cursor uint64
}
func (s *CopierState) key(msgId, suffix string) string {
@@ -105,7 +106,6 @@ func (s *CopierState) copyMessageKeys(key string, threshold uint64) (bool, error
func (s *CopierState) copyKeys(db int) {
var copied int
- var cursor uint64
defer func() { logInfo("copied %d keys from db %d", copied, db) }()
ticker := time.NewTicker(100 * time.Millisecond)
s.targetRedis = redis.NewClient(&redis.Options{Addr: s.opts.TargetRedis, DB: db})
@@ -119,16 +119,16 @@ copying:
return
}
if s.getMaster(db) {
- if cursor == 0 {
+ if s.cursor == 0 {
logInfo("starting SCAN on db %d", db)
}
- logDebug("cursor: %d", cursor)
+ logDebug("cursor: %d", s.cursor)
var err error
var keys []string
- keys, cursor, err = s.redis.Scan(cursor, keyPattern, 1000).Result()
+ keys, s.cursor, err = s.redis.Scan(s.cursor, keyPattern, 1000).Result()
if err != nil {
logError("starting over: %v", err)
- cursor = 0
+ s.cursor = 0
copied = 0
continue copying
}
@@ -142,7 +142,7 @@ copying:
ok, err := s.copyMessageKeys(key, threshold)
if err != nil {
logError("starting over: %v", err)
- cursor = 0
+ s.cursor = 0
copied = 0
goto copying
}
@@ -150,7 +150,7 @@ copying:
copied++
}
}
- if cursor == 0 {
+ if s.cursor == 0 {
return
}
}
@@ -163,6 +163,7 @@ func (s *CopierState) getMaster(db int) bool {
if s.currentMaster != server || s.currentDB != db {
s.currentMaster = server
s.currentDB = db
+ s.cursor = 0
if server == "" {
if s.redis != nil {
s.redis.Close()

MODIFY
go/delete_keys.go
go/delete_keys.go
@@ -27,6 +27,7 @@ type DeleterState struct {
currentDB int
redis *redis.Client // current connection
keySuffixes []string
+ cursor uint64
}
func (s *DeleterState) key(msgId, suffix string) string {
@@ -83,7 +84,6 @@ func (s *DeleterState) deleteMessageKeys(key string, threshold uint64) (bool, er
func (s *DeleterState) deleteKeys(db int) {
var deleted int
- var cursor uint64
defer func() { logInfo("deleted %d keys in db %d", deleted, db) }()
ticker := time.NewTicker(100 * time.Millisecond)
keyPattern := "msgid:" + s.opts.QueuePrefix + "*:expires"
@@ -96,16 +96,16 @@ deleting:
return
}
if s.getMaster(db) {
- if cursor == 0 {
+ if s.cursor == 0 {
logInfo("starting SCAN on db %d", db)
}
- logDebug("cursor: %d", cursor)
+ logDebug("cursor: %d", s.cursor)
var err error
var keys []string
- keys, cursor, err = s.redis.Scan(cursor, keyPattern, 1000).Result()
+ keys, s.cursor, err = s.redis.Scan(s.cursor, keyPattern, 1000).Result()
if err != nil {
logError("starting over: %v", err)
- cursor = 0
+ s.cursor = 0
deleted = 0
continue deleting
}
@@ -117,7 +117,7 @@ deleting:
removed, err := s.deleteMessageKeys(key, threshold)
if err != nil {
logError("starting over: %v", err)
- cursor = 0
+ s.cursor = 0
deleted = 0
goto deleting
}
@@ -125,7 +125,7 @@ deleting:
deleted++
}
}
- if cursor == 0 {
+ if s.cursor == 0 {
return
}
}
@@ -138,6 +138,7 @@ func (s *DeleterState) getMaster(db int) bool {
if s.currentMaster != server || s.currentDB != db {
s.currentMaster = server
s.currentDB = db
+ s.cursor = 0
if server == "" {
if s.redis != nil {
s.redis.Close()

MODIFY
go/garbage_collect_keys.go
go/garbage_collect_keys.go
@@ -33,6 +33,7 @@ type GCState struct {
redis *redis.Client // current connection
keySuffixes []string
expiries map[string]map[int]int
+ cursor uint64
}
func (s *GCState) key(msgId, suffix string) string {
@@ -210,7 +211,6 @@ func (s *GCState) gcKey(key string, threshold uint64) (bool, error) {
func (s *GCState) garbageCollectKeys(db int) bool {
var total, expired int
- var cursor uint64
defer func() { logInfo("expired %d keys out of %d in db %d", expired, total, db) }()
ticker := time.NewTicker(1 * time.Second)
threshold := time.Now().Unix() + int64(s.opts.GcThreshold)
@@ -219,13 +219,13 @@ func (s *GCState) garbageCollectKeys(db int) bool {
return false
}
if s.getMaster(db) {
- if cursor == 0 {
+ if s.cursor == 0 {
logInfo("starting SCAN on db %d", db)
}
- logDebug("cursor: %d", cursor)
+ logDebug("s.cursor: %d", s.cursor)
var err error
var keys []string
- keys, cursor, err = s.redis.Scan(cursor, "msgid:*:expires", 10000).Result()
+ keys, s.cursor, err = s.redis.Scan(s.cursor, "msgid:*:expires", 10000).Result()
if err != nil {
logError("starting over: %v", err)
return true
@@ -245,7 +245,7 @@ func (s *GCState) garbageCollectKeys(db int) bool {
expired++
}
}
- if cursor == 0 {
+ if s.cursor == 0 {
return false
}
}
@@ -298,6 +298,7 @@ func (s *GCState) getMaster(db int) bool {
if s.currentMaster != server || s.currentDB != db {
s.currentMaster = server
s.currentDB = db
+ s.cursor = 0
if server == "" {
if s.redis != nil {
s.redis.Close()

go beetle: reset cursor when current master changes