Remote execution of query plans in queriers (#12302)

#### What this PR does

This PR adds support for remote execution of query plans in queriers.

Query-frontends can now send a Protobuf message (instead of a smuggled
HTTP request) to queriers via query-schedulers, and queriers will
respond with a stream of results. This has been implemented in a way
where we can add other kinds of query requests and responses in the
future, with the goal of eventually moving away from smuggling HTTP
requests in Protobuf messages.

In the case of queries, query-frontends can now send a query plan and
request evaluation of a particular node in the plan, and queriers will
respond with results for that node.

I've also added a HTTP endpoint to queriers for debugging purposes
(`/api/v1/evaluate`), and a CLI tool to exercise this
(`tools/evaluate-query-plan`). I plan to remove this in a subsequent PR,
once I've added the functionality to query-frontends to send these
requests to queriers.

Not included in this PR:
* batching of responses to query-frontends (each series' data is sent as
its own Protobuf message for now)
* actually sending requests from query-frontends
* metrics and tracing for requests to queriers

There are a number of other items I'll handle in follow up PRs (eg.
propagating strong consistency information and dealing with query stats)
- my goal with this PR is to get the general structure in place and
start getting feedback on the approach before I invest too much more
time in this.

I've chosen not to add a changelog entry given this code is not actively
used at present, but open to other opinions on this.

#### Which issue(s) this PR fixes or relates to

(none)

#### Checklist

- [x] Tests updated.
- [n/a] Documentation added.
- [n/a] `CHANGELOG.md` updated - the order of entries should be
`[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`. If changelog entry
is not needed, please add the `changelog-not-needed` label to the PR.
- [n/a]
[`about-versioning.md`](https://github.com/grafana/mimir/blob/main/docs/sources/mimir/configure/about-versioning.md)
updated with experimental features.

Author: charleskorn

pkg/api/api.go

@@ -476,6 +476,10 @@ func (a *API) RegisterQueryAPI(handler http.Handler, buildInfoHandler http.Handl
 	a.RegisterRoute(path.Join(a.cfg.PrometheusHTTPPrefix, "/api/v1/format_query"), handler, true, true, "GET", "POST")
 }
 
+func (a *API) RegisterEvaluationAPI(handler http.Handler) {
+	a.RegisterRoute("/api/v1/evaluate", handler, true, true, "POST")
+}
+
 func (a *API) RegisterQueryAnalysisAPI(handler http.Handler) {
 	a.RegisterRoute("/api/v1/analyze", handler, true, true, "POST")
 }

pkg/api/handlers.go

@@ -214,6 +214,7 @@ func (cfg *Config) statusFlagsHandler() http.HandlerFunc {
 func NewQuerierHandler(
 	cfg Config,
 	querierCfg querier.Config,
+	dispatcher *querier.Dispatcher,
 	queryable storage.SampleAndChunkQueryable,
 	exemplarQueryable storage.ExemplarQueryable,
 	metadataSupplier querier.MetadataSupplier,
@@ -311,9 +312,8 @@ func NewQuerierHandler(
 	router.Use(querierapi.ConsistencyMiddleware().Wrap)
 
 	// Define the prefixes for all routes
-	prefix := path.Join(cfg.ServerPrefix, cfg.PrometheusHTTPPrefix)
-
-	promRouter := route.New().WithPrefix(path.Join(prefix, "/api/v1"))
+	promPrefix := path.Join(cfg.ServerPrefix, cfg.PrometheusHTTPPrefix, "/api/v1")
+	promRouter := route.New().WithPrefix(path.Join(promPrefix))
 	api.Register(promRouter)
 
 	// Track the requests count in the anonymous usage stats.
@@ -329,19 +329,23 @@ func NewQuerierHandler(
 
 	// TODO(gotjosh): This custom handler is temporary until we're able to vendor the changes in:
 	// https://github.com/prometheus/prometheus/pull/7125/files
-	router.Path(path.Join(prefix, "/api/v1/read")).Methods("POST").Handler(remoteReadStats.Wrap(querier.RemoteReadHandler(queryable, logger, querierCfg)))
-	router.Path(path.Join(prefix, "/api/v1/query")).Methods("GET", "POST").Handler(instantQueryStats.Wrap(promRouter))
-	router.Path(path.Join(prefix, "/api/v1/query_range")).Methods("GET", "POST").Handler(rangeQueryStats.Wrap(promRouter))
-	router.Path(path.Join(prefix, "/api/v1/query_exemplars")).Methods("GET", "POST").Handler(exemplarsQueryStats.Wrap(promRouter))
-	router.Path(path.Join(prefix, "/api/v1/labels")).Methods("GET", "POST").Handler(labelsQueryStats.Wrap(promRouter))
-	router.Path(path.Join(prefix, "/api/v1/label/{name}/values")).Methods("GET").Handler(labelsQueryStats.Wrap(promRouter))
-	router.Path(path.Join(prefix, "/api/v1/series")).Methods("GET", "POST", "DELETE").Handler(seriesQueryStats.Wrap(promRouter))
-	router.Path(path.Join(prefix, "/api/v1/metadata")).Methods("GET").Handler(metadataQueryStats.Wrap(querier.NewMetadataHandler(metadataSupplier)))
-	router.Path(path.Join(prefix, "/api/v1/cardinality/label_names")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.LabelNamesCardinalityHandler(distributor, limits)))
-	router.Path(path.Join(prefix, "/api/v1/cardinality/label_values")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.LabelValuesCardinalityHandler(distributor, limits)))
-	router.Path(path.Join(prefix, "/api/v1/cardinality/active_series")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.ActiveSeriesCardinalityHandler(distributor, limits)))
-	router.Path(path.Join(prefix, "/api/v1/cardinality/active_native_histogram_metrics")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.ActiveNativeHistogramMetricsHandler(distributor, limits)))
-	router.Path(path.Join(prefix, "/api/v1/format_query")).Methods("GET", "POST").Handler(formattingQueryStats.Wrap(promRouter))
+	router.Path(path.Join(promPrefix, "/read")).Methods("POST").Handler(remoteReadStats.Wrap(querier.RemoteReadHandler(queryable, logger, querierCfg)))
+	router.Path(path.Join(promPrefix, "/query")).Methods("GET", "POST").Handler(instantQueryStats.Wrap(promRouter))
+	router.Path(path.Join(promPrefix, "/query_range")).Methods("GET", "POST").Handler(rangeQueryStats.Wrap(promRouter))
+	router.Path(path.Join(promPrefix, "/query_exemplars")).Methods("GET", "POST").Handler(exemplarsQueryStats.Wrap(promRouter))
+	router.Path(path.Join(promPrefix, "/labels")).Methods("GET", "POST").Handler(labelsQueryStats.Wrap(promRouter))
+	router.Path(path.Join(promPrefix, "/label/{name}/values")).Methods("GET").Handler(labelsQueryStats.Wrap(promRouter))
+	router.Path(path.Join(promPrefix, "/series")).Methods("GET", "POST", "DELETE").Handler(seriesQueryStats.Wrap(promRouter))
+	router.Path(path.Join(promPrefix, "/metadata")).Methods("GET").Handler(metadataQueryStats.Wrap(querier.NewMetadataHandler(metadataSupplier)))
+	router.Path(path.Join(promPrefix, "/cardinality/label_names")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.LabelNamesCardinalityHandler(distributor, limits)))
+	router.Path(path.Join(promPrefix, "/cardinality/label_values")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.LabelValuesCardinalityHandler(distributor, limits)))
+	router.Path(path.Join(promPrefix, "/cardinality/active_series")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.ActiveSeriesCardinalityHandler(distributor, limits)))
+	router.Path(path.Join(promPrefix, "/cardinality/active_native_histogram_metrics")).Methods("GET", "POST").Handler(cardinalityQueryStats.Wrap(querier.ActiveNativeHistogramMetricsHandler(distributor, limits)))
+	router.Path(path.Join(promPrefix, "/format_query")).Methods("GET", "POST").Handler(formattingQueryStats.Wrap(promRouter))
+
+	if dispatcher != nil {
+		router.Path("/api/v1/evaluate").Methods("POST").Handler(dispatcher)
+	}
 
 	// Track execution time.
 	return stats.NewWallTimeMiddleware().Wrap(router)

pkg/frontend/v2/frontend_scheduler_adapter.go

@@ -38,7 +38,7 @@ func (a *frontendToSchedulerAdapter) frontendToSchedulerEnqueueRequest(
 		Type:                      schedulerpb.ENQUEUE,
 		QueryID:                   req.queryID,
 		UserID:                    req.userID,
-		HttpRequest:               req.request,
+		Payload:                   &schedulerpb.FrontendToScheduler_HttpRequest{HttpRequest: req.request},
 		FrontendAddress:           frontendAddr,
 		StatsEnabled:              req.statsEnabled,
 		AdditionalQueueDimensions: addlQueueDims,