go beetle: added command to copy keys to another redis instance
MODIFY
Makefile
Makefile
@@ -41,7 +41,7 @@ install: $(GO_INSTALL_TARGETS)
uninstall:
cd $(BIN_DIR) && rm -f $(GO_INSTALL_TARGETS) $(SCRIPTS)
-GO_MODULES = $(patsubst %,$(GO_SRC)/%, client.go server.go datatypes.go server_state.go failover_state.go redis.go redis_shim.go redis_server_info.go logging.go version.go garbage_collect_keys.go notification_mailer.go config.go delete_keys.go dump_expiries.go)
+GO_MODULES = $(patsubst %,$(GO_SRC)/%, client.go server.go datatypes.go server_state.go failover_state.go redis.go redis_shim.go redis_server_info.go logging.go version.go garbage_collect_keys.go notification_mailer.go config.go delete_keys.go copy_keys.go dump_expiries.go)
.packr:
go get github.com/gobuffalo/packr/packr

MODIFY
go/beetle.go
go/beetle.go
@@ -37,13 +37,15 @@ var opts struct {
ConsulUrl string `long:"consul" optional:"t" optional-value:"http://127.0.0.1:8500" description:"Specifies consul server url to use for retrieving config values. If given without argument, tries to contact local consul agent."`
GcThreshold int `long:"redis-gc-threshold" description:"Number of seconds to wait until considering an expired redis key eligible for garbage collection. Defaults to 3600 (1 hour)."`
GcDatabases string `long:"redis-gc-databases" description:"Database numbers to collect keys from (e.g. 0,4). Defaults to 4."`
- GcSystem string `long:"redis-gc-system" description:"Redis system from which to collect keys. Defaults to 'system'."`
+ GcSystem string `long:"redis-gc-system" default:"system" description:"Redis system from which to collect keys."`
GcKeyFile string `long:"redis-gc-key-file" description:"File with keys to collect."`
MailTo string `long:"mail-to" description:"Send notifcation mails to this address."`
DialTimeout int `long:"dial-timeout" description:"Number of seconds to wait until a connection attempt to the master times out. Defaults to 5."`
ConfidenceLevel string `long:"redis-failover-confidence-level" description:"A number between 0 and 100, defining the percent of clients which have to agree in an election process. Values are clamped to the interval [0,100]. Defaults to 100."`
- Queue string `long:"queue" description:"Name of the queue for which to delete all redis keys."`
- DeleteBefore time.Duration `long:"delete-before" description:"Delete keys which do expire before the given date/time."`
+ DeleteBefore time.Duration `long:"delete-before" description:"Delete keys which do expire before the given time."`
+ CopyAfter time.Duration `long:"copy-after" description:"Copy keys which do expire after the given time."`
+ TargetRedis string `long:"target-redis" description:"Specifies the target server for the copy_keys command (host:port)."`
+ QueuePrefix string `long:"queue-prefix" description:"Specifies the queue prefix for matching keys to be deleted/copied."`
}
// Verbose stores verbosity or logging purposoes.
@@ -134,18 +136,35 @@ func (x *CmdRunDeleteKeys) Execute(args []string) error {
if opts.GcSystem == "" {
opts.GcSystem = "system"
}
- if opts.Queue == "" {
- return fmt.Errorf("please specify a queue using the --queue option")
- }
return RunDeleteKeys(DeleteKeysOptions{
RedisMasterFile: initialConfig.RedisMasterFile,
Databases: initialConfig.GcDatabases,
System: opts.GcSystem,
- Queue: opts.Queue,
+ QueuePrefix: opts.QueuePrefix,
DeleteBefore: opts.DeleteBefore,
})
}
+// CmdRunCopyKeys is used when the program arguments tell us to copy redis keys.
+type CmdRunCopyKeys struct{}
+
+var cmdRunCopyKeys CmdRunCopyKeys
+
+// Execute copy redis keys.
+func (x *CmdRunCopyKeys) Execute(args []string) error {
+ if opts.GcSystem == "" {
+ opts.GcSystem = "system"
+ }
+ return RunCopyKeys(CopyKeysOptions{
+ RedisMasterFile: initialConfig.RedisMasterFile,
+ Databases: initialConfig.GcDatabases,
+ System: opts.GcSystem,
+ TargetRedis: opts.TargetRedis,
+ QueuePrefix: opts.QueuePrefix,
+ CopyAfter: opts.CopyAfter,
+ })
+}
+
// CmdRunDumpExpiries is used when the program arguments tell us to dump redis epxiries.
type CmdRunDumpExpiries struct{}
@@ -339,7 +358,8 @@ func main() {
parser.AddCommand("configuration_server", "run redis configuration server", "", &cmdRunServer)
parser.AddCommand("dump", "dump configuration after merging all config sources and exit", "", &cmdPrintConfig)
parser.AddCommand("garbage_collect_deduplication_store", "garbage collect keys on redis servers", "", &cmdRunGCKeys)
- parser.AddCommand("delete_queue_keys", "delete all keys for a given queue on redis servers", "", &cmdRunDeleteKeys)
+ parser.AddCommand("delete_queue_keys", "delete all keys for a given queue prefix on redis servers", "", &cmdRunDeleteKeys)
+ parser.AddCommand("copy_queue_keys", "copy all keys for a given queue prefix from current master to a given redis server", "", &cmdRunCopyKeys)
parser.AddCommand("dump_expiries", "print all expiry values from redis master", "", &cmdRunDumpExpiries)
parser.AddCommand("notification_mailer", "listen to system notifications and send them via /usr/sbin/sendmail", "", &cmdRunMailer)
parser.CommandHandler = cmdHandler

ADD
go/copy_keys.go
+package main
+
+import (
+ "fmt"
+ "regexp"
+ "strconv"
+ "strings"
+ "time"
+
+ "gopkg.in/redis.v5"
+)
+
+// CopyKeysOptions are provided by the caller of RunCopyKeys.
+type CopyKeysOptions struct {
+ RedisMasterFile string // Path to the redis master file.
+ Databases string // List of databases to scan.
+ System string // Name of redis system from which to copy keys.
+ TargetRedis string // Redis connection spec of the system to copy keys to.
+ QueuePrefix string // Copy keys for queues starting with the given prefix.
+ CopyAfter time.Duration // Copy keys which expire after the given time.
+}
+
+// CopyerState holds options and copier state, most importantly the current redis
+// connection and database.
+type CopierState struct {
+ opts CopyKeysOptions
+ currentMaster string
+ currentDB int
+ redis *redis.Client // current source connection
+ targetRedis *redis.Client // target connection
+ keySuffixes []string
+}
+
+func (s *CopierState) key(msgId, suffix string) string {
+ return fmt.Sprintf("%s:%s", msgId, suffix)
+}
+
+func (s *CopierState) keys(msgId string) []string {
+ res := make([]string, 0)
+ for _, suffix := range s.keySuffixes {
+ res = append(res, s.key(msgId, suffix))
+ }
+ return res
+}
+
+func (s *CopierState) msgId(key string) string {
+ re := regexp.MustCompile("^(msgid:[^:]+:[-0-9a-f]+):.*$")
+ matches := re.FindStringSubmatch(key)
+ if len(matches) == 0 {
+ logError("msgid could not be extracted from key '%s'", key)
+ return ""
+ }
+ return matches[1]
+}
+
+func (s *CopierState) matches(key string) string {
+ re := regexp.MustCompile("^msgid:(" + s.opts.QueuePrefix + ".*):[-0-9a-f]+:.*$")
+ matches := re.FindStringSubmatch(key)
+ if len(matches) == 0 {
+ return ""
+ }
+ return matches[1]
+}
+
+func (s *CopierState) copyMessageKeys(key string, threshold uint64) (bool, error) {
+ v, err := s.redis.Get(key).Result()
+ if err != nil {
+ if err == redis.Nil {
+ logDebug("key not found: %s", key)
+ return false, nil
+ }
+ return false, err
+ }
+ expires, err := strconv.ParseUint(v, 10, 64)
+ if err != nil {
+ return false, err
+ }
+ if expires < threshold {
+ t := time.Duration(threshold-expires) * time.Second
+ logDebug("key %s expires in %s", key, t)
+ return false, err
+ }
+ t := time.Duration(expires-threshold) * time.Second
+ logDebug("key %s will expire in %s", key, t)
+ msgID := s.msgId(key)
+ if msgID == "" {
+ return false, nil
+ }
+ keys := s.keys(msgID)
+ values, err := s.redis.MGet(keys...).Result()
+ if err != nil {
+ return false, err
+ }
+ pairs := make([]interface{}, 0, 2*len(keys))
+ for i := range keys {
+ if values[i] != nil {
+ pairs = append(pairs, keys[i], values[i])
+ }
+ }
+ logDebug("copying keys: %v", pairs)
+ result, err := s.targetRedis.MSet(pairs...).Result()
+ logDebug("copying keys returned: %v", result)
+ return err == nil, err
+}
+
+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})
+ keyPattern := "msgid:" + s.opts.QueuePrefix + "*:expires"
+ expiry := time.Now().Add(s.opts.CopyAfter)
+ logInfo("copying keys for queue prefix '%s' expiring after %s", s.opts.QueuePrefix, expiry.Format(time.RFC3339))
+ threshold := uint64(expiry.Unix())
+copying:
+ for range ticker.C {
+ if interrupted {
+ return
+ }
+ if s.getMaster(db) {
+ if cursor == 0 {
+ logInfo("starting SCAN on db %d", db)
+ }
+ logDebug("cursor: %d", cursor)
+ var err error
+ var keys []string
+ keys, cursor, err = s.redis.Scan(cursor, keyPattern, 1000).Result()
+ if err != nil {
+ logError("starting over: %v", err)
+ cursor = 0
+ copied = 0
+ continue copying
+ }
+ for _, key := range keys {
+ if interrupted {
+ return
+ }
+ if s.matches(key) == "" {
+ continue
+ }
+ ok, err := s.copyMessageKeys(key, threshold)
+ if err != nil {
+ logError("starting over: %v", err)
+ cursor = 0
+ copied = 0
+ goto copying
+ }
+ if ok {
+ copied++
+ }
+ }
+ if cursor == 0 {
+ return
+ }
+ }
+ }
+}
+
+func (s *CopierState) getMaster(db int) bool {
+ systems := RedisMastersFromMasterFile(s.opts.RedisMasterFile)
+ server := systems[s.opts.System]
+ if s.currentMaster != server || s.currentDB != db {
+ s.currentMaster = server
+ s.currentDB = db
+ if server == "" {
+ if s.redis != nil {
+ s.redis.Close()
+ }
+ s.redis = nil
+ } else {
+ s.redis = redis.NewClient(&redis.Options{Addr: server, DB: db})
+ }
+ }
+ if s.redis == nil {
+ logError("could not determine redis master: %v, db: %d", s.currentMaster, db)
+ }
+ return s.redis != nil
+}
+
+// RunCopyKeys deletes all keys for a given queue on the redis
+// master using the redis SCAN operation. Restarts from the beginning,
+// should the master change while running the scan. Terminates as soon
+// as a full scan has been performed on all databases.
+func RunCopyKeys(opts CopyKeysOptions) error {
+ logDebug("deleting keys with options: %+v", opts)
+ state := &CopierState{opts: opts}
+ state.keySuffixes = []string{"status", "ack_count", "timeout", "delay", "attempts", "exceptions", "mutex", "expires"}
+ for _, s := range strings.Split(opts.Databases, ",") {
+ if interrupted {
+ break
+ }
+ if s != "" {
+ db, err := strconv.Atoi(s)
+ if err != nil {
+ logError("%v", err)
+ continue
+ }
+ state.copyKeys(db)
+ }
+ }
+ if state.redis != nil {
+ return state.redis.Close()
+ }
+ return nil
+}

MODIFY
go/delete_keys.go
go/delete_keys.go
@@ -15,7 +15,7 @@ type DeleteKeysOptions struct {
RedisMasterFile string // Path to the redis master file.
Databases string // List of databases to scan.
System string // Name of redis system for which to delete keys.
- Queue string // Name of the queue for which to delete all keys
+ QueuePrefix string // Delete keys for queues starting with the given prefix.
DeleteBefore time.Duration // Delete keys which expire before the given time.
}
@@ -85,10 +85,10 @@ 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(1 * time.Second)
- keyPattern := "msgid:" + s.opts.Queue + ":*:expires"
+ ticker := time.NewTicker(100 * time.Millisecond)
+ keyPattern := "msgid:" + s.opts.QueuePrefix + "*:expires"
expiry := time.Now().Add(s.opts.DeleteBefore)
- logInfo("deleting keys for queue %s expiring before %s", s.opts.Queue, expiry.Format(time.RFC3339))
+ logInfo("deleting keys with queue prefix '%s' expiring before %s", s.opts.QueuePrefix, expiry.Format(time.RFC3339))
threshold := uint64(expiry.Unix())
deleting:
for range ticker.C {
@@ -102,7 +102,7 @@ deleting:
logDebug("cursor: %d", cursor)
var err error
var keys []string
- keys, cursor, err = s.redis.Scan(cursor, keyPattern, 10000).Result()
+ keys, cursor, err = s.redis.Scan(cursor, keyPattern, 1000).Result()
if err != nil {
logError("starting over: %v", err)
cursor = 0

go beetle: added command to copy keys to another redis instance