go beetle: added command to dump expiries
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 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)
+GO_MODULES = $(patsubst %,$(GO_SRC)/%, client.go server.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)
.packr:
go get github.com/gobuffalo/packr/packr

MODIFY
go/beetle.go
go/beetle.go
@@ -124,7 +124,7 @@ func (x *CmdRunGCKeys) Execute(args []string) error {
})
}
-// CmdRunDeleteKeys is used when the program arguments tell us to garbage collect redis keys.
+// CmdRunDeleteKeys is used when the program arguments tell us to delete redis keys.
type CmdRunDeleteKeys struct{}
var cmdRunDeleteKeys CmdRunDeleteKeys
@@ -146,6 +146,23 @@ func (x *CmdRunDeleteKeys) Execute(args []string) error {
})
}
+// CmdRunDumpExpiries is used when the program arguments tell us to dump redis epxiries.
+type CmdRunDumpExpiries struct{}
+
+var cmdRunDumpExpiries CmdRunDumpExpiries
+
+// Execute delete redis keys.
+func (x *CmdRunDumpExpiries) Execute(args []string) error {
+ if opts.GcSystem == "" {
+ opts.GcSystem = "system"
+ }
+ return RunDumpExpiries(DumpExpiriesOptions{
+ RedisMasterFile: initialConfig.RedisMasterFile,
+ Databases: initialConfig.GcDatabases,
+ System: opts.GcSystem,
+ })
+}
+
func init() {
ReportVersionIfRequestedAndExit()
opts.Id = getFQDN()
@@ -323,6 +340,7 @@ func main() {
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("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/dump_expiries.go
+package main
+
+import (
+ "fmt"
+ "regexp"
+ "strconv"
+ "strings"
+ "time"
+
+ "gopkg.in/redis.v5"
+)
+
+// DumpExpiriesOptions are provided by the caller of RunDumpExpiries.
+type DumpExpiriesOptions 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.
+}
+
+// DumperState holds options and collector state, most importantly the current redis
+// connection and database.
+type DumperState struct {
+ opts DumpExpiriesOptions
+ currentMaster string
+ currentDB int
+ redis *redis.Client // current connection
+ keySuffixes []string
+}
+
+func (s *DumperState) key(msgId, suffix string) string {
+ return fmt.Sprintf("%s:%s", msgId, suffix)
+}
+
+func (s *DumperState) keys(msgId string) []string {
+ res := make([]string, 0)
+ for _, suffix := range s.keySuffixes {
+ res = append(res, s.key(msgId, suffix))
+ }
+ return res
+}
+
+func (s *DumperState) 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 *DumperState) printExpiry(key string) (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
+ }
+ fmt.Printf("%s:%s\n", key, v)
+ return true, nil
+}
+
+func (s *DumperState) dumpKeys(db int) {
+ var dumped int
+ var cursor uint64
+ defer func() { logInfo("dumped %d keys in db %d", dumped, db) }()
+ ticker := time.NewTicker(1 * time.Second)
+dumping:
+ 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, "msgid:*:expires", 10000).Result()
+ if err != nil {
+ logError("starting over: %v", err)
+ cursor = 0
+ dumped = 0
+ continue dumping
+ }
+ logDebug("retrieved %d keys from db %d", len(keys), db)
+ for _, key := range keys {
+ if interrupted {
+ return
+ }
+ printed, err := s.printExpiry(key)
+ if err != nil {
+ logError("starting over: %v", err)
+ cursor = 0
+ dumped = 0
+ goto dumping
+ }
+ if printed {
+ dumped++
+ }
+ }
+ if cursor == 0 {
+ return
+ }
+ }
+ }
+}
+
+func (s *DumperState) 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
+}
+
+// RunDumpExpiries 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 RunDumpExpiries(opts DumpExpiriesOptions) error {
+ logDebug("dumping keys with options: %+v", opts)
+ state := &DumperState{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.dumpKeys(db)
+ }
+ }
+ if state.redis != nil {
+ return state.redis.Close()
+ }
+ return nil
+}

go beetle: added command to dump expiries