compact: add some concurrency around deletion

Deleting one by one is REALLY slow if you have thousands of blocks to
delete. Use some concurrency by default.

Signed-off-by: Giedrius Statkevičius <[email protected]>

Author: GiedriusS

pkg/compact/blocks_cleaner.go

@@ -5,6 +5,7 @@ package compact
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/go-kit/log"
@@ -15,6 +16,8 @@ import (
 	"github.com/thanos-io/objstore"
 
 	"github.com/thanos-io/thanos/pkg/block"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
+	"github.com/thanos-io/thanos/pkg/errutil"
 )
 
 // BlocksCleaner is a struct that deletes blocks from bucket which are marked for deletion.
@@ -42,23 +45,55 @@ func NewBlocksCleaner(logger log.Logger, bkt objstore.Bucket, ignoreDeletionMark
 // DeleteMarkedBlocks uses ignoreDeletionMarkFilter to gather the blocks that are marked for deletion and deletes those
 // if older than given deleteDelay.
 func (s *BlocksCleaner) DeleteMarkedBlocks(ctx context.Context) (map[ulid.ULID]struct{}, error) {
+	const conc = 32
+
 	level.Info(s.logger).Log("msg", "started cleaning of blocks marked for deletion")
 
-	deletedBlocks := make(map[ulid.ULID]struct{}, 0)
+	var (
+		merr             errutil.SyncMultiError
+		deletedBlocksMtx sync.Mutex
+		deletedBlocks    = make(map[ulid.ULID]struct{}, 0)
+		deletionMarkMap  = s.ignoreDeletionMarkFilter.DeletionMarkBlocks()
+		wg               sync.WaitGroup
+		dm               = make(chan *metadata.DeletionMark, conc)
+	)
 
-	deletionMarkMap := s.ignoreDeletionMarkFilter.DeletionMarkBlocks()
-	for _, deletionMark := range deletionMarkMap {
-		if time.Since(time.Unix(deletionMark.DeletionTime, 0)).Seconds() > s.deleteDelay.Seconds() {
-			if err := block.Delete(ctx, s.logger, s.bkt, deletionMark.ID); err != nil {
-				s.blockCleanupFailures.Inc()
-				return deletedBlocks, errors.Wrap(err, "delete block")
+	wg.Add(conc)
+	for range conc {
+		go func() {
+			defer wg.Done()
+
+			for deletionMark := range dm {
+				if ctx.Err() != nil {
+					return
+				}
+				if time.Since(time.Unix(deletionMark.DeletionTime, 0)).Seconds() > s.deleteDelay.Seconds() {
+					if err := block.Delete(ctx, s.logger, s.bkt, deletionMark.ID); err != nil {
+						s.blockCleanupFailures.Inc()
+						merr.Add(errors.Wrap(err, "delete block"))
+						continue
+					}
+					s.blocksCleaned.Inc()
+					level.Info(s.logger).Log("msg", "deleted block marked for deletion", "block", deletionMark.ID)
+
+					deletedBlocksMtx.Lock()
+					deletedBlocks[deletionMark.ID] = struct{}{}
+					deletedBlocksMtx.Unlock()
+				}
 			}
-			s.blocksCleaned.Inc()
-			level.Info(s.logger).Log("msg", "deleted block marked for deletion", "block", deletionMark.ID)
-			deletedBlocks[deletionMark.ID] = struct{}{}
-		}
+		}()
+	}
+
+	for _, deletionMark := range deletionMarkMap {
+		dm <- deletionMark
+	}
+	close(dm)
+	wg.Wait()
+
+	if ctx.Err() != nil {
+		return deletedBlocks, ctx.Err()
 	}
 
 	level.Info(s.logger).Log("msg", "cleaning of blocks marked for deletion done")
-	return deletedBlocks, nil
+	return deletedBlocks, merr.Err()
 }