Query Label Values from block store

Signed-off-by: Goutham Veeramachaneni <[email protected]>

Author: gouthamve

CHANGELOG.md

@@ -13,7 +13,7 @@
 * [ENHANCEMENT] Added `cortex_alertmanager_config_hash` metric to expose hash of Alertmanager Config loaded per user. #3388
 * [ENHANCEMENT] Query-Frontend / Query-Scheduler: New component called "Query-Scheduler" has been introduced. Query-Scheduler is simply a queue of requests, moved outside of Query-Frontend. This allows Query-Frontend to be scaled separately from number of queues. To make Query-Frontend and Querier use Query-Scheduler, they need to be started with `-frontend.scheduler-address` and `-querier.scheduler-address` options respectively. #3374 #3471
 * [ENHANCEMENT] Query-frontend / Querier / Ruler: added `-querier.max-query-lookback` to limit how long back data (series and metadata) can be queried. This setting can be overridden on a per-tenant basis and is enforced in the query-frontend, querier and ruler. #3452 #3458
-* [ENHANCEMENT] Querier: added `-querier.query-store-for-labels-enabled` to query store for series API. Only works with blocks storage engine. #3461
+* [ENHANCEMENT] Querier: added `-querier.query-store-for-labels-enabled` to query store for label names, label values and series APIs. Only works with blocks storage engine. #3461 #3520
 * [ENHANCEMENT] Ingester: exposed `-blocks-storage.tsdb.wal-segment-size-bytes` config option to customise the TSDB WAL segment max size. #3476
 * [ENHANCEMENT] Compactor: concurrently run blocks cleaner for multiple tenants. Concurrency can be configured via `-compactor.cleanup-concurrency`. #3483
 * [ENHANCEMENT] Compactor: shuffle tenants before running compaction. #3483

docs/api/_index.md

@@ -294,7 +294,7 @@ GET,POST <prometheus-http-prefix>/api/v1/labels
 GET,POST <legacy-http-prefix>/api/v1/labels
 ```
 
-Get label names of ingested series. Differently than Prometheus and due to scalability and performances reasons, Cortex currently ignores the `start` and `end` request parameters and always fetches the label names from in-memory data stored in the ingesters.
+Get label names of ingested series. Differently than Prometheus and due to scalability and performances reasons, Cortex currently ignores the `start` and `end` request parameters and always fetches the label names from in-memory data stored in the ingesters. There is experimental support to query the long-term store with the *blocks* storage engine when `-querier.query-store-for-labels-enabled` is set.
 
 _For more information, please check out the Prometheus [get label names](https://prometheus.io/docs/prometheus/latest/querying/api/#getting-label-names) documentation._
 
@@ -309,7 +309,7 @@ GET <prometheus-http-prefix>/api/v1/label/{name}/values
 GET <legacy-http-prefix>/api/v1/label/{name}/values
 ```
 
-Get label values for a given label name. Differently than Prometheus and due to scalability and performances reasons, Cortex currently ignores the `start` and `end` request parameters and always fetches the label values from in-memory data stored in the ingesters.
+Get label values for a given label name. Differently than Prometheus and due to scalability and performances reasons, Cortex currently ignores the `start` and `end` request parameters and always fetches the label values from in-memory data stored in the ingesters. There is experimental support to query the long-term store with the *blocks* storage engine when `-querier.query-store-for-labels-enabled` is set.
 
 _For more information, please check out the Prometheus [get label values](https://prometheus.io/docs/prometheus/latest/querying/api/#querying-label-values) documentation._
 

integration/querier_test.go

@@ -536,7 +536,7 @@ func testMetadataQueriesWithBlocksStorage(
 			labelValuesTests: []labelValuesTest{
 				{
 					label: labels.MetricName,
-					resp:  []string{lastSeriesInIngesterBlocksName, firstSeriesInIngesterHeadName},
+					resp:  []string{lastSeriesInStorageName, lastSeriesInIngesterBlocksName, firstSeriesInIngesterHeadName},
 				},
 			},
 			labelNames: []string{labels.MetricName, lastSeriesInStorageName, lastSeriesInIngesterBlocksName, firstSeriesInIngesterHeadName},
@@ -563,7 +563,7 @@ func testMetadataQueriesWithBlocksStorage(
 			labelValuesTests: []labelValuesTest{
 				{
 					label: labels.MetricName,
-					resp:  []string{firstSeriesInIngesterHeadName},
+					resp:  []string{lastSeriesInStorageName, firstSeriesInIngesterHeadName},
 				},
 			},
 			labelNames: []string{labels.MetricName, lastSeriesInStorageName, firstSeriesInIngesterHeadName},
@@ -590,7 +590,7 @@ func testMetadataQueriesWithBlocksStorage(
 				for _, val := range lvt.resp {
 					exp = append(exp, model.LabelValue(val))
 				}
-				require.ElementsMatch(t, exp, labelsRes)
+				require.Equal(t, exp, labelsRes)
 			}
 
 			labelNames, err := c.LabelNames(tc.from, tc.to)

pkg/querier/blocks_store_queryable.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -293,11 +294,6 @@ func (q *blocksStoreQuerier) Select(_ bool, sp *storage.SelectHints, matchers ..
 	return q.selectSorted(sp, matchers...)
 }
 
-func (q *blocksStoreQuerier) LabelValues(name string) ([]string, storage.Warnings, error) {
-	// Cortex doesn't use this. It will ask ingesters for metadata.
-	return nil, nil, errors.New("not implemented")
-}
-
 func (q *blocksStoreQuerier) LabelNames() ([]string, storage.Warnings, error) {
 	spanLog, spanCtx := spanlogger.New(q.ctx, "blocksStoreQuerier.LabelNames")
 	defer spanLog.Span.Finish()
@@ -333,6 +329,41 @@ func (q *blocksStoreQuerier) LabelNames() ([]string, storage.Warnings, error) {
 	return strutil.MergeSlices(resNameSets...), resWarnings, nil
 }
 
+func (q *blocksStoreQuerier) LabelValues(name string) ([]string, storage.Warnings, error) {
+	spanLog, spanCtx := spanlogger.New(q.ctx, "blocksStoreQuerier.LabelValues")
+	defer spanLog.Span.Finish()
+
+	minT, maxT := q.minT, q.maxT
+
+	var (
+		resValueSets = [][]string{}
+		resWarnings  = storage.Warnings(nil)
+
+		resultMtx sync.Mutex
+	)
+
+	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64) ([]ulid.ULID, error) {
+		valueSets, warnings, queriedBlocks, err := q.fetchLabelValuesFromStore(spanCtx, name, clients, minT, maxT)
+		if err != nil {
+			return nil, err
+		}
+
+		resultMtx.Lock()
+		resValueSets = append(resValueSets, valueSets...)
+		resWarnings = append(resWarnings, warnings...)
+		resultMtx.Unlock()
+
+		return queriedBlocks, nil
+	}
+
+	err := q.queryWithConsistencyCheck(spanCtx, spanLog, minT, maxT, queryFunc)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return strutil.MergeSlices(resValueSets...), resWarnings, nil
+}
+
 func (q *blocksStoreQuerier) Close() error {
 	return nil
 }
@@ -690,6 +721,85 @@ func (q *blocksStoreQuerier) fetchLabelNamesFromStore(
 	return nameSets, warnings, queriedBlocks, nil
 }
 
+func (q *blocksStoreQuerier) fetchLabelValuesFromStore(
+	ctx context.Context,
+	name string,
+	clients map[BlocksStoreClient][]ulid.ULID,
+	minT int64,
+	maxT int64,
+) ([][]string, storage.Warnings, []ulid.ULID, error) {
+	var (
+		reqCtx        = grpc_metadata.AppendToOutgoingContext(ctx, cortex_tsdb.TenantIDExternalLabel, q.userID)
+		g, gCtx       = errgroup.WithContext(reqCtx)
+		mtx           = sync.Mutex{}
+		nameSets      = [][]string{}
+		warnings      = storage.Warnings(nil)
+		queriedBlocks = []ulid.ULID(nil)
+		spanLog       = spanlogger.FromContext(ctx)
+	)
+
+	// Concurrently fetch series from all clients.
+	for c, blockIDs := range clients {
+		// Change variables scope since it will be used in a goroutine.
+		c := c
+		blockIDs := blockIDs
+
+		g.Go(func() error {
+			req, err := createLabelValuesRequest(minT, maxT, name, blockIDs)
+			if err != nil {
+				return errors.Wrapf(err, "failed to create label names request")
+			}
+
+			valuesResp, err := c.LabelValues(gCtx, req)
+			if err != nil {
+				return errors.Wrapf(err, "failed to fetch series from %s", c)
+			}
+
+			myQueriedBlocks := []ulid.ULID(nil)
+			if valuesResp.Hints != nil {
+				hints := hintspb.LabelValuesResponseHints{}
+				if err := types.UnmarshalAny(valuesResp.Hints, &hints); err != nil {
+					return errors.Wrapf(err, "failed to unmarshal label names hints from %s", c)
+				}
+
+				ids, err := convertBlockHintsToULIDs(hints.QueriedBlocks)
+				if err != nil {
+					return errors.Wrapf(err, "failed to parse queried block IDs from received hints")
+				}
+
+				myQueriedBlocks = ids
+			}
+
+			level.Debug(spanLog).Log("msg", "received label names from store-gateway",
+				"instance", c,
+				"num labels", len(valuesResp.Values),
+				"requested blocks", strings.Join(convertULIDsToString(blockIDs), " "),
+				"queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " "))
+
+			// Values returned need not be sorted, but we need them to be sorted so we can merge.
+			sort.Strings(valuesResp.Values)
+
+			// Store the result.
+			mtx.Lock()
+			nameSets = append(nameSets, valuesResp.Values)
+			for _, w := range valuesResp.Warnings {
+				warnings = append(warnings, errors.New(w))
+			}
+			queriedBlocks = append(queriedBlocks, myQueriedBlocks...)
+			mtx.Unlock()
+
+			return nil
+		})
+	}
+
+	// Wait until all client requests complete.
+	if err := g.Wait(); err != nil {
+		return nil, nil, nil, err
+	}
+
+	return nameSets, warnings, queriedBlocks, nil
+}
+
 func createSeriesRequest(minT, maxT int64, matchers []storepb.LabelMatcher, skipChunks bool, blockIDs []ulid.ULID) (*storepb.SeriesRequest, error) {
 	// Selectively query only specific blocks.
 	hints := &hintspb.SeriesRequestHints{

pkg/querier/blocks_store_queryable_test.go

@@ -796,9 +796,9 @@ func (m *storeGatewayClientMock) LabelNames(context.Context, *storepb.LabelNames
 	return nil, nil
 }
 
-// func (m *storeGatewayClientMock) LabelValues(context.Context, *storepb.LabelValuesRequest, ...grpc.CallOption) (*storepb.LabelValuesResponse, error) {
-// 	return nil, nil
-// }
+func (m *storeGatewayClientMock) LabelValues(context.Context, *storepb.LabelValuesRequest, ...grpc.CallOption) (*storepb.LabelValuesResponse, error) {
+	return nil, nil
+}
 
 func (m *storeGatewayClientMock) RemoteAddress() string {
 	return m.remoteAddr

pkg/querier/querier.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -376,7 +377,51 @@ func (q querier) Select(_ bool, sp *storage.SelectHints, matchers ...*labels.Mat
 
 // LabelsValue implements storage.Querier.
 func (q querier) LabelValues(name string) ([]string, storage.Warnings, error) {
-	return q.metadataQuerier.LabelValues(name)
+	if !q.queryStoreForLabels {
+		return q.metadataQuerier.LabelValues(name)
+	}
+
+	if len(q.queriers) == 1 {
+		return q.queriers[0].LabelValues(name)
+	}
+
+	// Using an errgroup here instead of channels, etc because this
+	// is a better model imo and we should move to this everywhere.
+	var (
+		g, _     = errgroup.WithContext(q.ctx)
+		sets     = [][]string{}
+		warnings = storage.Warnings(nil)
+
+		resMtx sync.Mutex
+	)
+
+	for _, querier := range q.queriers {
+		// Need to reassign as the original variable will change and can't be relied on in a goroutine.
+		querier := querier
+		g.Go(func() error {
+			myValues, myWarnings, err := querier.LabelValues(name)
+			if err != nil {
+				return err
+			}
+
+			// We need values to be sorted we can merge them.
+			sort.Strings(myValues)
+
+			resMtx.Lock()
+			sets = append(sets, myValues)
+			warnings = append(warnings, myWarnings...)
+			resMtx.Unlock()
+
+			return nil
+		})
+	}
+
+	err := g.Wait()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return strutil.MergeSlices(sets...), warnings, nil
 }
 
 func (q querier) LabelNames() ([]string, storage.Warnings, error) {

pkg/querier/store_gateway_client_test.go

@@ -79,6 +79,6 @@ func (m *mockStoreGatewayServer) LabelNames(context.Context, *storepb.LabelNames
 	return nil, nil
 }
 
-// func (m *mockStoreGatewayServer) LabelValues(context.Context, *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) {
-// 	return nil, nil
-// }
+func (m *mockStoreGatewayServer) LabelValues(context.Context, *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) {
+	return nil, nil
+}

pkg/storegateway/bucket_stores.go

@@ -266,22 +266,22 @@ func (u *BucketStores) LabelNames(ctx context.Context, req *storepb.LabelNamesRe
 }
 
 // LabelValues implements the Storegateway proto service.
-// func (u *BucketStores) LabelValues(ctx context.Context, req *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) {
-// 	spanLog, spanCtx := spanlogger.New(ctx, "BucketStores.LabelNames")
-// 	defer spanLog.Span.Finish()
-
-// 	userID := getUserIDFromGRPCContext(spanCtx)
-// 	if userID == "" {
-// 		return nil, fmt.Errorf("no userID")
-// 	}
-
-// 	store := u.getStore(userID)
-// 	if store == nil {
-// 		return nil, nil
-// 	}
-
-// 	return store.LabelValues(ctx, req)
-// }
+func (u *BucketStores) LabelValues(ctx context.Context, req *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) {
+	spanLog, spanCtx := spanlogger.New(ctx, "BucketStores.LabelNames")
+	defer spanLog.Span.Finish()
+
+	userID := getUserIDFromGRPCContext(spanCtx)
+	if userID == "" {
+		return nil, fmt.Errorf("no userID")
+	}
+
+	store := u.getStore(userID)
+	if store == nil {
+		return nil, nil
+	}
+
+	return store.LabelValues(ctx, req)
+}
 
 // scanUsers in the bucket and return the list of found users. If an error occurs while
 // iterating the bucket, it may return both an error and a subset of the users in the bucket.

pkg/storegateway/gateway.go

@@ -325,9 +325,9 @@ func (g *StoreGateway) LabelNames(ctx context.Context, req *storepb.LabelNamesRe
 }
 
 // LabelValues implements the Storegateway proto service.
-// func (g *StoreGateway) LabelValues(ctx context.Context, req *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) {
-// 	return g.stores.LabelValues(ctx, req)
-// }
+func (g *StoreGateway) LabelValues(ctx context.Context, req *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) {
+	return g.stores.LabelValues(ctx, req)
+}
 
 func (g *StoreGateway) OnRingInstanceRegister(_ *ring.BasicLifecycler, ringDesc ring.Desc, instanceExists bool, instanceID string, instanceDesc ring.IngesterDesc) (ring.IngesterState, ring.Tokens) {
 	// When we initialize the store-gateway instance in the ring we want to start from