NewCleanupCommand creates a new command to clean up unused redis keys.
(maker Maker, baseLogger log.Logger)
| 20 | |
| 21 | // NewCleanupCommand creates a new command to clean up unused redis keys. |
| 22 | func NewCleanupCommand(maker Maker, baseLogger log.Logger) *cobra.Command { |
| 23 | const cursorFileName = "redis-scan-cursor" |
| 24 | var ( |
| 25 | logger = logging.WithLevel(baseLogger) |
| 26 | cursorPath string |
| 27 | batchSize int64 |
| 28 | prefix string |
| 29 | instance string |
| 30 | ) |
| 31 | |
| 32 | type stats struct { |
| 33 | scanned uint64 |
| 34 | removed uint64 |
| 35 | } |
| 36 | |
| 37 | removeKeys := func(ctx context.Context, cursor *uint64, stats *stats, threshold time.Duration) error { |
| 38 | var ( |
| 39 | keys []string |
| 40 | err error |
| 41 | ) |
| 42 | |
| 43 | logger.Info(fmt.Sprintf("scanning redis keys from cursor %d", *cursor)) |
| 44 | if err := os.WriteFile(cursorPath, []byte(fmt.Sprintf("%d", *cursor)), os.ModePerm); err != nil { |
| 45 | logger.Err("cannot store cursor to cursor location", err, nil) |
| 46 | } |
| 47 | |
| 48 | redisClient, err := maker.Make(instance) |
| 49 | if err != nil { |
| 50 | return fmt.Errorf("cannot find redis instance under the name of %s: %w", instance, err) |
| 51 | } |
| 52 | keys, *cursor, err = redisClient.Scan(ctx, *cursor, prefix+"*", batchSize).Result() |
| 53 | if err != nil { |
| 54 | return err |
| 55 | } |
| 56 | stats.scanned += uint64(len(keys)) |
| 57 | |
| 58 | var wg sync.WaitGroup |
| 59 | for _, key := range keys { |
| 60 | wg.Add(1) |
| 61 | go func(key string) { |
| 62 | idleTime, _ := redisClient.ObjectIdleTime(ctx, key).Result() |
| 63 | if idleTime > threshold { |
| 64 | logger.Info(fmt.Sprintf("removing %s from redis as it is %s old", key, idleTime)) |
| 65 | redisClient.Del(ctx, key) |
| 66 | atomic.AddUint64(&stats.removed, 1) |
| 67 | } |
| 68 | wg.Done() |
| 69 | }(key) |
| 70 | } |
| 71 | wg.Wait() |
| 72 | return nil |
| 73 | } |
| 74 | |
| 75 | initCursor := func() (uint64, error) { |
| 76 | cursorPath = filepath.Join(os.TempDir(), fmt.Sprintf("%s.%s.txt", cursorFileName, instance)) |
| 77 | if _, err := os.Stat(cursorPath); os.IsNotExist(err) { |
| 78 | return 0, nil |
| 79 | } |