go beetle: added command to delete keys for a given queue
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)
+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)
.packr:
go get github.com/gobuffalo/packr/packr

MODIFY
go/beetle.go
go/beetle.go
@@ -9,6 +9,7 @@ import (
"strconv"
"strings"
"syscall"
+ "time"
// "github.com/davecgh/go-spew/spew"
"github.com/jessevdk/go-flags"
@@ -41,6 +42,8 @@ var opts struct {
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."`
}
// Verbose stores verbosity or logging purposoes.
@@ -121,6 +124,28 @@ func (x *CmdRunGCKeys) Execute(args []string) error {
})
}
+// CmdRunDeleteKeys is used when the program arguments tell us to garbage collect redis keys.
+type CmdRunDeleteKeys struct{}
+
+var cmdRunDeleteKeys CmdRunDeleteKeys
+
+// Execute delete redis keys.
+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,
+ DeleteBefore: opts.DeleteBefore,
+ })
+}
+
func init() {
ReportVersionIfRequestedAndExit()
opts.Id = getFQDN()
@@ -296,7 +321,8 @@ func main() {
parser.AddCommand("configuration_client", "run redis configuration client", "", &cmdRunClient)
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 server", "", &cmdRunGCKeys)
+ 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("notification_mailer", "listen to system notifications and send them via /usr/sbin/sendmail", "", &cmdRunMailer)
parser.CommandHandler = cmdHandler

ADD
go/delete_keys.go
+package main
+
+import (
+ "fmt"
+ "regexp"
+ "strconv"
+ "strings"
+ "time"
+
+ "gopkg.in/redis.v5"
+)
+
+// DeleteKeysOptions are provided by the caller of RunDeleteKeys.
+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
+ DeleteBefore time.Duration // Delete keys which expire before the given time.
+}
+
+// DeleterState holds options and collector state, most importantly the current redis
+// connection and database.
+type DeleterState struct {
+ opts DeleteKeysOptions
+ currentMaster string
+ currentDB int
+ redis *redis.Client // current connection
+ keySuffixes []string
+}
+
+func (s *DeleterState) key(msgId, suffix string) string {
+ return fmt.Sprintf("%s:%s", msgId, suffix)
+}
+
+func (s *DeleterState) keys(msgId string) []string {
+ res := make([]string, 0)
+ for _, suffix := range s.keySuffixes {
+ res = append(res, s.key(msgId, suffix))
+ }
+ return res
+}
+
+func (s *DeleterState) 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 *DeleterState) deleteMessageKeys(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(expires-threshold) * time.Second
+ logDebug("key %s expires in %s", key, t)
+ return false, err
+ }
+ t := time.Duration(threshold-expires) * time.Second
+ logDebug("key %s has expired %s ago", key, t)
+ msgID := s.msgId(key)
+ if msgID == "" {
+ return false, nil
+ }
+ keys := s.keys(msgID)
+ // logDebug("deleting keys: %s", strings.Join(keys, ", "))
+ _, err = s.redis.Del(keys...).Result()
+ return err == nil, err
+}
+
+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"
+ expiry := time.Now().Add(s.opts.DeleteBefore)
+ logInfo("deleting keys for queue %s expiring before %s", s.opts.Queue, expiry.Format(time.RFC3339))
+ threshold := uint64(expiry.Unix())
+deleting:
+ 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, 10000).Result()
+ if err != nil {
+ logError("starting over: %v", err)
+ cursor = 0
+ deleted = 0
+ continue deleting
+ }
+ logDebug("retrieved %d keys from db %d", len(keys), db)
+ for _, key := range keys {
+ if interrupted {
+ return
+ }
+ removed, err := s.deleteMessageKeys(key, threshold)
+ if err != nil {
+ logError("starting over: %v", err)
+ cursor = 0
+ deleted = 0
+ goto deleting
+ }
+ if removed {
+ deleted++
+ }
+ }
+ if cursor == 0 {
+ return
+ }
+ }
+ }
+}
+
+func (s *DeleterState) 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
+}
+
+// RunDeleteKeys 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 RunDeleteKeys(opts DeleteKeysOptions) error {
+ logDebug("deleting keys with options: %+v", opts)
+ state := &DeleterState{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.deleteKeys(db)
+ }
+ }
+ if state.redis != nil {
+ return state.redis.Close()
+ }
+ return nil
+}

MODIFY
go/garbage_collect_keys.go
go/garbage_collect_keys.go
@@ -91,7 +91,7 @@ func (s *GCState) garbageCollectKeys(db int) {
collecting:
for range ticker.C {
if interrupted {
- break
+ return
}
if s.getMaster(db) {
if cursor == 0 {
@@ -112,6 +112,9 @@ collecting:
total += len(keys)
threshold := time.Now().Unix() + int64(s.opts.GcThreshold)
for _, key := range keys {
+ if interrupted {
+ return
+ }
collected, err := s.gcKey(key, uint64(threshold))
if err != nil {
logError("starting over: %v", err)

MODIFY
go/go.sum
go/go.sum
@@ -221,9 +221,11 @@ github.com/karrick/godirwalk v1.7.8/go.mod h1:2c9FRhkDxdIbgkOnCEvnSWs71Bhugbl46s
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
github.com/konsorten/go-windows-terminal-sequences v0.0.0-20180402223658-b729f2633dfe/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/pty v1.1.3/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
@@ -394,6 +396,7 @@ gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4
gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc/go.mod h1:m7x9LTH6d71AHyAX77c9yqWCCa3UKHcVEj9y7hAtKDk=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=

go beetle: added command to delete keys for a given queue