Function forPartitionRingReplicaSets

modules/querier/partition_ring.go:13–33

forPartitionRingReplicaSets runs f, in parallel, for all live-store instances in the input replicationSets. Return an error if any f fails for any of the input replicationSets.

(ctx context.Context, q *Querier, replicationSets []ring.ReplicationSet, f func(context.Context, TClient) (R, error))

// forPartitionRingReplicaSets runs f, in parallel, for all live-store instances in the input replicationSets.
// Return an error if any f fails for any of the input replicationSets.
func forPartitionRingReplicaSets[R any, TClient any](ctx context.Context, q *Querier, replicationSets []ring.ReplicationSet, f func(context.Context, TClient) (R, error)) ([]R, error) {
	wrappedF := func(ctx context.Context, instance *ring.InstanceDesc) (R, error) {
		client, err := q.liveStorePool.GetClientForInstance(*instance)
		if err != nil {
			var empty R
			return empty, err
		}

		return f(ctx, client.(TClient))
	}

	cleanup := func(_ R) {
		// Nothing to do.
	}

	quorumConfig := q.queryQuorumConfigForReplicationSets(ctx, replicationSets)

	return concurrency.ForEachJobMergeResults(ctx, replicationSets, 0, func(ctx context.Context, set ring.ReplicationSet) ([]R, error) {
		return ring.DoUntilQuorum(ctx, set, quorumConfig, wrappedF, cleanup)
	})
}

// queryQuorumConfigForReplicationSets returns the config to use with "do until quorum" functions when running queries.
func (q *Querier) queryQuorumConfigForReplicationSets(_ context.Context, _ []ring.ReplicationSet) ring.DoUntilQuorumConfig {