fix bug where multiple compactors are trying to cleanup the same tenant at once, which results in dangling bucket index

Signed-off-by: Roy Chiang <[email protected]>

Author: roystchiang

pkg/compactor/compactor.go

@@ -457,7 +457,7 @@ func (c *Compactor) starting(ctx context.Context) error {
 	c.bucketClient = bucketindex.BucketWithGlobalMarkers(c.bucketClient)
 
 	// Create the users scanner.
-	c.usersScanner = cortex_tsdb.NewUsersScanner(c.bucketClient, c.ownUser, c.parentLogger)
+	c.usersScanner = cortex_tsdb.NewUsersScanner(c.bucketClient, c.cleanUser, c.parentLogger)
 
 	// Create the blocks cleaner (service).
 	c.blocksCleaner = NewBlocksCleaner(BlocksCleanerConfig{
@@ -814,6 +814,14 @@ func (c *Compactor) discoverUsers(ctx context.Context) ([]string, error) {
 }
 
 func (c *Compactor) ownUser(userID string) (bool, error) {
+	return c.ownUserHelper(userID, false)
+}
+
+func (c *Compactor) cleanUser(userID string) (bool, error) {
+	return c.ownUserHelper(userID, true)
+}
+
+func (c *Compactor) ownUserHelper(userID string, isCleanUp bool) (bool, error) {
 	if !c.allowedTenants.IsAllowed(userID) {
 		return false, nil
 	}
@@ -823,8 +831,8 @@ func (c *Compactor) ownUser(userID string) (bool, error) {
 		return true, nil
 	}
 
-	// If using shuffle-sharding, ownership is determined by the ring
-	if c.compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
+	// If we aren't cleaning up user blocks, and we are using shuffle-sharding, ownership is determined by the ring
+	if !isCleanUp && c.compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
 		subRing := c.ring.ShuffleShard(userID, c.limits.CompactorTenantShardSize(userID))
 
 		rs, err := subRing.GetAllHealthy(RingOp)