added method to collect keys given a file with msg ids
MODIFY
go/src/github.com/xing/beetle/beetle.go
go/src/github.com/xing/beetle/beetle.go
@@ -36,6 +36,7 @@ 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."`
+ GcKeyFile string `long:"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."`
@@ -111,6 +112,7 @@ func (x *CmdRunGCKeys) Execute(args []string) error {
RedisMasterFile: initialConfig.RedisMasterFile,
GcThreshold: initialConfig.GcThreshold,
GcDatabases: initialConfig.GcDatabases,
+ GcKeyFile: opts.GcKeyFile,
})
}

MODIFY
go/src/github.com/xing/beetle/garbage_collect_keys.go
go/src/github.com/xing/beetle/garbage_collect_keys.go
package main
import (
+ "bufio"
"fmt"
+ "os"
"regexp"
"strconv"
"strings"
@@ -15,6 +17,7 @@ type GCOptions struct {
RedisMasterFile string // Path to the redis master file.
GcThreshold int // Number of seconds after which a key should be considered collectible.
GcDatabases string // List of databases to scan.
+ GcKeyFile string // Name of file containing keys to collect.
}
// GCState holds options and collector state, most importantly the current redis
@@ -51,6 +54,10 @@ func (s *GCState) msgId(key string) string {
func (s *GCState) gcKey(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)
@@ -74,7 +81,7 @@ func (s *GCState) gcKey(key string, threshold uint64) (bool, error) {
func (s *GCState) garbageCollectKeys(db int) {
var total, expired int
var cursor uint64
- defer logInfo("expired %d keys out of %d in db %d", expired, total, db)
+ defer func() { logInfo("expired %d keys out of %d in db %d", expired, total, db) }()
ticker := time.NewTicker(1 * time.Second)
collecting:
for _ = range ticker.C {
@@ -109,7 +116,7 @@ collecting:
goto collecting
}
if collected {
- expired += 1
+ expired++
}
}
if cursor == 0 {
@@ -119,6 +126,45 @@ collecting:
}
}
+func (s *GCState) garbageCollectKeysFromFile(db int, filePath string) {
+ var total, expired int
+ defer func() { logInfo("expired %d keys out of %d in db %d", expired, total, db) }()
+
+ file, err := os.Open(filePath)
+ if err != nil {
+ logError("%v", err)
+ return
+ }
+ defer file.Close()
+
+ s.getMaster(db)
+
+ threshold := time.Now().Unix() + int64(s.opts.GcThreshold)
+ re := regexp.MustCompile("^(msgid:[^:]+:[-0-9a-f]+):expires$")
+ scanner := bufio.NewScanner(file)
+ for scanner.Scan() {
+ if interrupted {
+ break
+ }
+ line := scanner.Text()
+ if !re.MatchString(line) {
+ continue
+ }
+ total++
+ collected, err := s.gcKey(line, uint64(threshold))
+ if err != nil {
+ logError("could not collect %s: %v", line, err)
+ continue
+ }
+ if collected {
+ expired++
+ }
+ }
+ if err := scanner.Err(); err != nil {
+ logError("%v", err)
+ }
+}
+
func (s *GCState) getMaster(db int) bool {
server := ReadRedisMasterFile(s.opts.RedisMasterFile)
if s.currentMaster != server || s.currentDB != db {
@@ -157,7 +203,11 @@ func RunGarbageCollectKeys(opts GCOptions) error {
logError("%v", err)
continue
}
+ if opts.GcKeyFile == "" {
state.garbageCollectKeys(db)
+ } else {
+ state.garbageCollectKeysFromFile(db, opts.GcKeyFile)
+ }
}
}
if state.redis != nil {

added method to collect keys given a file with msg ids