Compactor.ownUser now determines whether the user is owned by a compactor via ring, instead of returning true if shuffle-sharding is enabled

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

Author: roystchiang

pkg/compactor/compactor.go

@@ -818,12 +818,23 @@ func (c *Compactor) ownUser(userID string) (bool, error) {
 		return false, nil
 	}
 
-	// Always owned if sharding is disabled or if using shuffle-sharding as shard ownership
-	// is determined by the shuffle sharding grouper.
-	if !c.compactorCfg.ShardingEnabled || c.compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
+	// Always owned if sharding is disabled
+	if !c.compactorCfg.ShardingEnabled {
 		return true, nil
 	}
 
+	// If using shuffle-sharding, ownership is determined by the ring
+	if c.compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
+		subRing := c.ring.ShuffleShard(userID, c.limits.CompactorTenantShardSize(userID))
+
+		rs, err := subRing.GetAllHealthy(RingOp)
+		if err != nil {
+			return false, err
+		}
+
+		return rs.Includes(c.ringLifecycler.Addr), nil
+	}
+
 	// Hash the user ID.
 	hasher := fnv.New32a()
 	_, _ = hasher.Write([]byte(userID))