(ctx context.Context, userID string, rulesRequest RulesRequest)
| 1434 | } |
| 1435 | |
| 1436 | func (r *Ruler) getShardedRules(ctx context.Context, userID string, rulesRequest RulesRequest) (*RulesResponse, error) { |
| 1437 | ring := ring.ReadRing(r.ring) |
| 1438 | |
| 1439 | if shardSize := r.limits.RulerTenantShardSize(userID); shardSize > 0 && r.cfg.ShardingStrategy == util.ShardingStrategyShuffle { |
| 1440 | ring = r.ring.ShuffleShard(userID, r.getShardSizeForUser(userID)) |
| 1441 | } |
| 1442 | |
| 1443 | rulers, failedZones, err := GetReplicationSetForListRule(ring, &r.cfg.Ring) |
| 1444 | if err != nil { |
| 1445 | return nil, err |
| 1446 | } |
| 1447 | |
| 1448 | ctx, err = user.InjectIntoGRPCRequest(ctx) |
| 1449 | if err != nil { |
| 1450 | return nil, fmt.Errorf("unable to inject user ID into grpc request, %v", err) |
| 1451 | } |
| 1452 | |
| 1453 | var ( |
| 1454 | mtx sync.Mutex |
| 1455 | merged []*RulesResponse |
| 1456 | errCount int |
| 1457 | ) |
| 1458 | |
| 1459 | zoneByAddress := make(map[string]string) |
| 1460 | if r.cfg.RulesBackupEnabled() { |
| 1461 | for _, ruler := range rulers.Instances { |
| 1462 | zoneByAddress[ruler.Addr] = ruler.Zone |
| 1463 | } |
| 1464 | } |
| 1465 | // Concurrently fetch rules from all rulers. |
| 1466 | jobs := concurrency.CreateJobsFromStrings(rulers.GetAddresses()) |
| 1467 | err = concurrency.ForEach(ctx, jobs, len(jobs), func(ctx context.Context, job any) error { |
| 1468 | addr := job.(string) |
| 1469 | |
| 1470 | rulerClient, err := r.clientsPool.GetClientFor(addr) |
| 1471 | if err != nil { |
| 1472 | return errors.Wrapf(err, "unable to get client for ruler %s", addr) |
| 1473 | } |
| 1474 | |
| 1475 | ctx, cancel := context.WithTimeout(ctx, r.cfg.ClientTLSConfig.RemoteTimeout) |
| 1476 | defer cancel() |
| 1477 | newGrps, err := rulerClient.Rules(ctx, &RulesRequest{ |
| 1478 | RuleNames: rulesRequest.GetRuleNames(), |
| 1479 | RuleGroupNames: rulesRequest.GetRuleGroupNames(), |
| 1480 | Files: rulesRequest.GetFiles(), |
| 1481 | Type: rulesRequest.GetType(), |
| 1482 | State: rulesRequest.GetState(), |
| 1483 | Health: rulesRequest.GetHealth(), |
| 1484 | Matchers: rulesRequest.GetMatchers(), |
| 1485 | ExcludeAlerts: rulesRequest.GetExcludeAlerts(), |
| 1486 | MaxRuleGroups: rulesRequest.GetMaxRuleGroups(), |
| 1487 | NextToken: rulesRequest.GetNextToken(), |
| 1488 | }) |
| 1489 | |
| 1490 | if err != nil { |
| 1491 | level.Error(r.logger).Log("msg", "unable to retrieve rules from ruler", "addr", addr, "err", err) |
| 1492 | r.rulerGetRulesFailures.WithLabelValues(addr).Inc() |
| 1493 | // If rules backup is enabled and there are enough rulers replicating the rules, we should |
no test coverage detected