complete isUserOwned function in am/ruler

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

pkg/alertmanager/multitenant.go

@@ -801,10 +801,6 @@ func (am *MultitenantAlertmanager) loadAlertmanagerConfigs(ctx context.Context)
 
 	// Filter out users not owned by this shard.
 	for _, userID := range allUserIDs {
-		if !am.allowedTenants.IsAllowed(userID) {
-			level.Debug(am.logger).Log("msg", "ignoring alertmanager for user, not allowed", "user", userID)
-			continue
-		}
 		if am.isUserOwned(userID) {
 			ownedUserIDs = append(ownedUserIDs, userID)
 		}
@@ -823,6 +819,10 @@ func (am *MultitenantAlertmanager) loadAlertmanagerConfigs(ctx context.Context)
 }
 
 func (am *MultitenantAlertmanager) isUserOwned(userID string) bool {
+	if !am.allowedTenants.IsAllowed(userID) {
+		return false
+	}
+
 	// If sharding is disabled, any alertmanager instance owns all users.
 	if !am.cfg.ShardingEnabled {
 		return true

pkg/ruler/ruler.go

@@ -747,7 +747,11 @@ func (r *Ruler) userIndexUpdateLoop(ctx context.Context) {
 			level.Error(r.logger).Log("msg", "context timeout, exit user index update loop", "err", ctx.Err())
 			return
 		case <-ticker.C:
-			owned := r.isUserOwned(userID)
+			owned, err := r.isUserOwned(userID)
+			if err != nil {
+				level.Error(r.logger).Log("msg", "failed to check if compactor owns updating user index", "err", err)
+				continue
+			}
 			if !owned {
 				continue
 			}
@@ -760,20 +764,38 @@ func (r *Ruler) userIndexUpdateLoop(ctx context.Context) {
 	}
 }
 
-func (r *Ruler) isUserOwned(userID string) bool {
+func (r *Ruler) isUserOwned(userID string) (bool, error) {
+	if !r.allowedTenants.IsAllowed(userID) {
+		return false, nil
+	}
+
 	// If sharding is disabled, any ruler instance owns all users.
 	if !r.cfg.EnableSharding {
-		return true
+		return true, nil
+	}
+
+	if r.cfg.ShardingStrategy == util.ShardingStrategyShuffle {
+		shardSize := r.getShardSizeForUser(userID)
+		subRing := r.ring.ShuffleShard(userID, shardSize)
+
+		rs, err := subRing.GetAllHealthy(RingOp)
+		if err != nil {
+			r.ringCheckErrors.Inc()
+			level.Error(r.logger).Log("msg", "failed to get rulers from ring", "user", userID, "err", err)
+			return false, err
+		}
+
+		return rs.Includes(r.lifecycler.GetInstanceAddr()), nil
 	}
 
 	rulers, err := r.ring.Get(users.ShardByUser(userID), RingOp, nil, nil, nil)
 	if err != nil {
 		r.ringCheckErrors.Inc()
 		level.Error(r.logger).Log("msg", "failed to get rulers from ring", "user", userID, "err", err)
-		return false
+		return false, err
 	}
 
-	return rulers.Includes(r.lifecycler.GetInstanceAddr())
+	return rulers.Includes(r.lifecycler.GetInstanceAddr()), nil
 }
 
 func (r *Ruler) syncRules(ctx context.Context, reason string) error {