GetIngestersForQuery returns a replication set including all ingesters that should be queried to fetch series matching input label matchers.
(ctx context.Context, matchers ...*labels.Matcher)
| 83 | // GetIngestersForQuery returns a replication set including all ingesters that should be queried |
| 84 | // to fetch series matching input label matchers. |
| 85 | func (d *Distributor) GetIngestersForQuery(ctx context.Context, matchers ...*labels.Matcher) (ring.ReplicationSet, error) { |
| 86 | userID, err := users.TenantID(ctx) |
| 87 | if err != nil { |
| 88 | return ring.ReplicationSet{}, err |
| 89 | } |
| 90 | |
| 91 | // If shuffle sharding is enabled we should only query ingesters which are |
| 92 | // part of the tenant's subring. |
| 93 | if d.cfg.ShardingStrategy == util.ShardingStrategyShuffle { |
| 94 | shardSize := d.limits.IngestionTenantShardSize(userID) |
| 95 | lookbackPeriod := d.cfg.ShuffleShardingLookbackPeriod |
| 96 | |
| 97 | if shardSize > 0 && lookbackPeriod > 0 { |
| 98 | return d.ingestersRing.ShuffleShardWithLookback(userID, shardSize, lookbackPeriod, time.Now()).GetReplicationSetForOperation(ring.Read) |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | // If "shard by all labels" is disabled, we can get ingesters by metricName if exists. |
| 103 | if !d.cfg.ShardByAllLabels && len(matchers) > 0 { |
| 104 | metricNameMatcher, _, ok := extract.MetricNameMatcherFromMatchers(matchers) |
| 105 | |
| 106 | if ok && metricNameMatcher.Type == labels.MatchEqual { |
| 107 | return d.ingestersRing.Get(shardByMetricName(userID, metricNameMatcher.Value), ring.Read, nil, nil, nil) |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | return d.ingestersRing.GetReplicationSetForOperation(ring.Read) |
| 112 | } |
| 113 | |
| 114 | // GetIngestersForMetadata returns a replication set including all ingesters that should be queried |
| 115 | // to fetch metadata (eg. label names/values or series). |
no test coverage detected