WIP

Signed-off-by: Alan Protasio <[email protected]>

Author: alanprot

Makefile

@@ -95,6 +95,7 @@ pkg/ring/ring.pb.go: pkg/ring/ring.proto
 pkg/frontend/v1/frontendv1pb/frontend.pb.go: pkg/frontend/v1/frontendv1pb/frontend.proto
 pkg/frontend/v2/frontendv2pb/frontend.pb.go: pkg/frontend/v2/frontendv2pb/frontend.proto
 pkg/querier/tripperware/queryrange/queryrange.pb.go: pkg/querier/tripperware/queryrange/queryrange.proto
+pkg/querier/tripperware/instantquery/instantquery.pb.go: pkg/querier/tripperware/instantquery/instantquery.proto
 pkg/querier/tripperware/query.pb.go: pkg/querier/tripperware/query.proto
 pkg/querier/stats/stats.pb.go: pkg/querier/stats/stats.proto
 pkg/distributor/ha_tracker.pb.go: pkg/distributor/ha_tracker.proto
@@ -202,6 +203,7 @@ lint:
 		./pkg/frontend/... \
 		./pkg/querier/tenantfederation/... \
 		./pkg/querier/tripperware/... \
+		./pkg/querier/tripperware/instantquery/... \
 		./pkg/querier/tripperware/queryrange/...
 
 	# Ensure packages that no longer use a global logger don't reintroduce it

docs/configuration/config-file-reference.md

@@ -126,6 +126,11 @@ api:
 # Cortex query-frontend.
 [query_range: <query_range_config>]
 
+query:
+  # Number of shards to use when distributing shardable PromQL queries.
+  # CLI flag: -querier.vertical-shard-size
+  [vertical_shard_size: <int> | default = 0]
+
 # The blocks_storage_config configures the blocks storage.
 [blocks_storage: <blocks_storage_config>]
 

pkg/cortex/cortex.go

@@ -40,6 +40,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/querier"
 	"github.com/cortexproject/cortex/pkg/querier/tenantfederation"
 	"github.com/cortexproject/cortex/pkg/querier/tripperware"
+	"github.com/cortexproject/cortex/pkg/querier/tripperware/instantquery"
 	"github.com/cortexproject/cortex/pkg/querier/tripperware/queryrange"
 	querier_worker "github.com/cortexproject/cortex/pkg/querier/worker"
 	"github.com/cortexproject/cortex/pkg/ring"
@@ -107,6 +108,7 @@ type Config struct {
 	Worker           querier_worker.Config           `yaml:"frontend_worker"`
 	Frontend         frontend.CombinedFrontendConfig `yaml:"frontend"`
 	QueryRange       queryrange.Config               `yaml:"query_range"`
+	Query            instantquery.Config             `yaml:"query"`
 	BlocksStorage    tsdb.BlocksStorageConfig        `yaml:"blocks_storage"`
 	Compactor        compactor.Config                `yaml:"compactor"`
 	StoreGateway     storegateway.Config             `yaml:"store_gateway"`
@@ -153,6 +155,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	c.Worker.RegisterFlags(f)
 	c.Frontend.RegisterFlags(f)
 	c.QueryRange.RegisterFlags(f)
+	c.Query.RegisterFlags(f)
 	c.BlocksStorage.RegisterFlags(f)
 	c.Compactor.RegisterFlags(f)
 	c.StoreGateway.RegisterFlags(f)

pkg/cortex/modules.go

@@ -34,6 +34,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/querier"
 	"github.com/cortexproject/cortex/pkg/querier/tenantfederation"
 	"github.com/cortexproject/cortex/pkg/querier/tripperware"
+	"github.com/cortexproject/cortex/pkg/querier/tripperware/instantquery"
 	"github.com/cortexproject/cortex/pkg/querier/tripperware/queryrange"
 	querier_worker "github.com/cortexproject/cortex/pkg/querier/worker"
 	"github.com/cortexproject/cortex/pkg/ring"
@@ -434,15 +435,20 @@ func (t *Cortex) initDeleteRequestsStore() (serv services.Service, err error) {
 // initQueryFrontendTripperware instantiates the tripperware used by the query frontend
 // to optimize Prometheus query requests.
 func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err error) {
+	t.Cfg.QueryRange.VerticalShardSize = t.Cfg.Query.VerticalShardSize
 	queryRangeMiddlewares, cache, err := queryrange.Middlewares(
 		t.Cfg.QueryRange,
 		util_log.Logger,
 		t.Overrides,
-		queryrange.PrometheusCodec,
 		queryrange.PrometheusResponseExtractor{},
 		prometheus.DefaultRegisterer,
 		t.TombstonesLoader,
 	)
+	if err != nil {
+		return nil, err
+	}
+
+	instantQueryMiddlewares, err := instantquery.Middlewares(t.Cfg.Query, util_log.Logger, t.Overrides)
 
 	if err != nil {
 		return nil, err
@@ -452,7 +458,9 @@ func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err erro
 		prometheus.DefaultRegisterer,
 		t.Cfg.QueryRange.ForwardHeaders,
 		queryRangeMiddlewares,
+		instantQueryMiddlewares,
 		queryrange.PrometheusCodec,
+		instantquery.InstantQueryCodec,
 	)
 
 	return services.NewIdleService(nil, func(_ error) error {

pkg/ingester/ingester.go

@@ -29,6 +29,7 @@ import (
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/shipper"
+	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"github.com/weaveworks/common/httpgrpc"
 	"go.uber.org/atomic"
 	"golang.org/x/sync/errgroup"
@@ -37,6 +38,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/chunk/encoding"
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/ingester/client"
+	"github.com/cortexproject/cortex/pkg/querysharding"
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
@@ -1173,6 +1175,13 @@ func (i *Ingester) Query(ctx context.Context, req *client.QueryRequest) (*client
 		return nil, err
 	}
 
+	matchers, sm, err := querysharding.ExtractShardingMatchers(matchers)
+	if err != nil {
+		return nil, err
+	}
+
+	defer sm.Close()
+
 	i.metrics.queries.Inc()
 
 	db := i.getTSDB(userID)
@@ -1197,6 +1206,9 @@ func (i *Ingester) Query(ctx context.Context, req *client.QueryRequest) (*client
 	result := &client.QueryResponse{}
 	for ss.Next() {
 		series := ss.At()
+		if sm.IsSharded() && !sm.MatchesLabels(series.Labels()) {
+			continue
+		}
 
 		ts := cortexpb.TimeSeries{
 			Labels: cortexpb.FromLabelsToLabelAdapters(series.Labels()),
@@ -1600,6 +1612,13 @@ func (i *Ingester) QueryStream(req *client.QueryRequest, stream client.Ingester_
 		return err
 	}
 
+	matchers, shardMatcher, err := querysharding.ExtractShardingMatchers(matchers)
+	if err != nil {
+		return err
+	}
+
+	defer shardMatcher.Close()
+
 	i.metrics.queries.Inc()
 
 	db := i.getTSDB(userID)
@@ -1609,7 +1628,7 @@ func (i *Ingester) QueryStream(req *client.QueryRequest, stream client.Ingester_
 
 	numSamples := 0
 	numSeries := 0
-	numSeries, numSamples, err = i.queryStreamChunks(ctx, db, int64(from), int64(through), matchers, stream)
+	numSeries, numSamples, err = i.queryStreamChunks(ctx, db, int64(from), int64(through), matchers, shardMatcher, stream)
 
 	if err != nil {
 		return err
@@ -1622,7 +1641,7 @@ func (i *Ingester) QueryStream(req *client.QueryRequest, stream client.Ingester_
 }
 
 // queryStreamChunks streams metrics from a TSDB. This implements the client.IngesterServer interface
-func (i *Ingester) queryStreamChunks(ctx context.Context, db *userTSDB, from, through int64, matchers []*labels.Matcher, stream client.Ingester_QueryStreamServer) (numSeries, numSamples int, _ error) {
+func (i *Ingester) queryStreamChunks(ctx context.Context, db *userTSDB, from, through int64, matchers []*labels.Matcher, sm *storepb.ShardMatcher, stream client.Ingester_QueryStreamServer) (numSeries, numSamples int, _ error) {
 	q, err := db.ChunkQuerier(ctx, from, through)
 	if err != nil {
 		return 0, 0, err
@@ -1640,6 +1659,10 @@ func (i *Ingester) queryStreamChunks(ctx context.Context, db *userTSDB, from, th
 	for ss.Next() {
 		series := ss.At()
 
+		if sm.IsSharded() && !sm.MatchesLabels(series.Labels()) {
+			continue
+		}
+
 		// convert labels to LabelAdapter
 		ts := client.TimeSeriesChunk{
 			Labels: cortexpb.FromLabelsToLabelAdapters(series.Labels()),

pkg/querier/blocks_store_queryable.go

@@ -33,6 +33,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/querier/series"
 	"github.com/cortexproject/cortex/pkg/querier/stats"
+	"github.com/cortexproject/cortex/pkg/querysharding"
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/ring/kv"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
@@ -409,9 +410,8 @@ func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*
 	}
 
 	var (
-		convertedMatchers = convertMatchersToLabelMatcher(matchers)
-		resSeriesSets     = []storage.SeriesSet(nil)
-		resWarnings       = storage.Warnings(nil)
+		resSeriesSets = []storage.SeriesSet(nil)
+		resWarnings   = storage.Warnings(nil)
 
 		maxChunksLimit  = q.limits.MaxChunksPerQueryFromStore(q.userID)
 		leftChunksLimit = maxChunksLimit
@@ -420,7 +420,7 @@ func (q *blocksStoreQuerier) selectSorted(sp *storage.SelectHints, matchers ...*
 	)
 
 	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64) ([]ulid.ULID, error) {
-		seriesSets, queriedBlocks, warnings, numChunks, err := q.fetchSeriesFromStores(spanCtx, sp, clients, minT, maxT, matchers, convertedMatchers, maxChunksLimit, leftChunksLimit)
+		seriesSets, queriedBlocks, warnings, numChunks, err := q.fetchSeriesFromStores(spanCtx, sp, clients, minT, maxT, matchers, maxChunksLimit, leftChunksLimit)
 		if err != nil {
 
 			return nil, err
@@ -562,7 +562,6 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 	minT int64,
 	maxT int64,
 	matchers []*labels.Matcher,
-	convertedMatchers []storepb.LabelMatcher,
 	maxChunksLimit int,
 	leftChunksLimit int,
 ) ([]storage.SeriesSet, []ulid.ULID, storage.Warnings, int, error) {
@@ -578,6 +577,12 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 		queryLimiter  = limiter.QueryLimiterFromContextWithFallback(ctx)
 		reqStats      = stats.FromContext(ctx)
 	)
+	matchers, shardingInfo, err := querysharding.ExtractShardingInfo(matchers)
+	convertedMatchers := convertMatchersToLabelMatcher(matchers)
+
+	if err != nil {
+		return nil, nil, nil, 0, err
+	}
 
 	// Concurrently fetch series from all clients.
 	for c, blockIDs := range clients {
@@ -595,7 +600,7 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 			// retrieved.
 			skipChunks := sp != nil && sp.Func == "series"
 
-			req, err := createSeriesRequest(minT, maxT, convertedMatchers, skipChunks, blockIDs)
+			req, err := createSeriesRequest(minT, maxT, convertedMatchers, shardingInfo, skipChunks, blockIDs)
 			if err != nil {
 				return errors.Wrapf(err, "failed to create series request")
 			}
@@ -875,7 +880,7 @@ func (q *blocksStoreQuerier) fetchLabelValuesFromStore(
 	return valueSets, warnings, queriedBlocks, nil
 }
 
-func createSeriesRequest(minT, maxT int64, matchers []storepb.LabelMatcher, skipChunks bool, blockIDs []ulid.ULID) (*storepb.SeriesRequest, error) {
+func createSeriesRequest(minT, maxT int64, matchers []storepb.LabelMatcher, shardingInfo *storepb.ShardInfo, skipChunks bool, blockIDs []ulid.ULID) (*storepb.SeriesRequest, error) {
 	// Selectively query only specific blocks.
 	hints := &hintspb.SeriesRequestHints{
 		BlockMatchers: []storepb.LabelMatcher{
@@ -899,6 +904,7 @@ func createSeriesRequest(minT, maxT int64, matchers []storepb.LabelMatcher, skip
 		PartialResponseStrategy: storepb.PartialResponseStrategy_ABORT,
 		Hints:                   anyHints,
 		SkipChunks:              skipChunks,
+		ShardInfo:               shardingInfo,
 	}, nil
 }
 

pkg/querier/tripperware/instantquery/InstantQueryMiddlewares.go

@@ -0,0 +1,32 @@
+package instantquery
+
+import (
+	"flag"
+
+	"github.com/go-kit/log"
+
+	"github.com/cortexproject/cortex/pkg/querier/tripperware"
+)
+
+// Config for query_range middleware chain.
+type Config struct {
+	VerticalShardSize int `yaml:"vertical_shard_size"`
+}
+
+// RegisterFlags adds the flags required to config this to the given FlagSet.
+func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
+	f.IntVar(&cfg.VerticalShardSize, "querier.vertical-shard-size", 0, "Number of shards to use when distributing shardable PromQL queries.")
+}
+
+func Middlewares(
+	cfg Config,
+	log log.Logger,
+	limits tripperware.Limits,
+) ([]tripperware.Middleware, error) {
+	var m []tripperware.Middleware
+
+	if cfg.VerticalShardSize > 1 {
+		m = append(m, tripperware.ShardByMiddleware(log, limits, InstantQueryCodec, cfg.VerticalShardSize))
+	}
+	return m, nil
+}