(ctx context.Context)
| 951 | } |
| 952 | |
| 953 | func (r *Ruler) listRulesShuffleSharding(ctx context.Context) (map[string]rulespb.RuleGroupList, map[string]rulespb.RuleGroupList, error) { |
| 954 | users, err := r.store.ListAllUsers(ctx) |
| 955 | if err != nil { |
| 956 | return nil, nil, errors.Wrap(err, "unable to list users of ruler") |
| 957 | } |
| 958 | |
| 959 | // Only users in userRings will be used in the to load the rules. |
| 960 | userRings := map[string]ring.ReadRing{} |
| 961 | for _, u := range users { |
| 962 | if shardSize := r.limits.RulerTenantShardSize(u); shardSize > 0 { |
| 963 | subRing := r.ring.ShuffleShard(u, r.getShardSizeForUser(u)) |
| 964 | |
| 965 | // Include the user only if it belongs to this ruler shard. |
| 966 | if subRing.HasInstance(r.lifecycler.GetInstanceID()) { |
| 967 | userRings[u] = subRing |
| 968 | } |
| 969 | } else { |
| 970 | // A shard size of 0 means shuffle sharding is disabled for this specific user. |
| 971 | // In that case we use the full ring so that rule groups will be sharded across all rulers. |
| 972 | userRings[u] = r.ring |
| 973 | } |
| 974 | } |
| 975 | |
| 976 | if len(userRings) == 0 { |
| 977 | r.ruleGroupMetrics.UpdateRuleGroupsInStore(make(map[string]int)) |
| 978 | return nil, nil, nil |
| 979 | } |
| 980 | |
| 981 | userCh := make(chan string, len(userRings)) |
| 982 | for u := range userRings { |
| 983 | userCh <- u |
| 984 | } |
| 985 | close(userCh) |
| 986 | |
| 987 | mu := sync.Mutex{} |
| 988 | owned := map[string]rulespb.RuleGroupList{} |
| 989 | backedUp := map[string]rulespb.RuleGroupList{} |
| 990 | gLock := sync.Mutex{} |
| 991 | ruleGroupCounts := make(map[string]int, len(userRings)) |
| 992 | |
| 993 | concurrency := min(len(userRings), loadRulesConcurrency) |
| 994 | |
| 995 | g, gctx := errgroup.WithContext(ctx) |
| 996 | for range concurrency { |
| 997 | g.Go(func() error { |
| 998 | for userID := range userCh { |
| 999 | groups, err := r.store.ListRuleGroupsForUserAndNamespace(gctx, userID, "") |
| 1000 | if err != nil { |
| 1001 | return errors.Wrapf(err, "failed to fetch rule groups for user %s", userID) |
| 1002 | } |
| 1003 | gLock.Lock() |
| 1004 | ruleGroupCounts[userID] = len(groups) |
| 1005 | gLock.Unlock() |
| 1006 | |
| 1007 | filterOwned := r.filterRuleGroups(userID, groups, userRings[userID]) |
| 1008 | var filterBackup []*rulespb.RuleGroupDesc |
| 1009 | if r.cfg.RulesBackupEnabled() { |
| 1010 | filterBackup = r.filterBackupRuleGroups(userID, groups, filterOwned, userRings[userID]) |
no test coverage detected