Add support for new thanos promql engine

CHANGELOG.md

@@ -19,6 +19,7 @@
 * [FEATURE] Query Frontend/Scheduler: Add a new counter metric `cortex_request_queue_requests_total` for total requests going to queue. #5030
 * [FEATURE] Build ARM docker images. #5041
 * [FEATURE] Query-frontend/Querier: Create spans to measure time to merge promql responses. #5041
+* [FEATURE] Querier/Ruler: Support the new thanos promql engine. This is an experimental feature and might change in the future. #5093
 * [FEATURE] Added zstd as an option for grpc compression #5092
 * [BUGFIX] Updated `golang.org/x/net` dependency to fix CVE-2022-27664. #5008
 * [BUGFIX] Fix panic when otel and xray tracing is enabled. #5044

docs/blocks-storage/querier.md

@@ -218,6 +218,12 @@ querier:
   # sharding on read path is disabled).
   # CLI flag: -querier.shuffle-sharding-ingesters-lookback-period
   [shuffle_sharding_ingesters_lookback_period: <duration> | default = 0s]
+
+  # Experimental. Use Thanos promql engine
+  # https://github.com/thanos-community/promql-engine rather than the Prometheus
+  # promql engine.
+  # CLI flag: -querier.thanos-engine
+  [thanos_engine: <boolean> | default = false]
 ```
 
 ### `blocks_storage_config`

docs/configuration/config-file-reference.md

@@ -906,6 +906,12 @@ store_gateway_client:
 # is disabled).
 # CLI flag: -querier.shuffle-sharding-ingesters-lookback-period
 [shuffle_sharding_ingesters_lookback_period: <duration> | default = 0s]
+
+# Experimental. Use Thanos promql engine
+# https://github.com/thanos-community/promql-engine rather than the Prometheus
+# promql engine.
+# CLI flag: -querier.thanos-engine
+[thanos_engine: <boolean> | default = false]
 ```
 
 ### `query_frontend_config`

go.mod

@@ -49,6 +49,7 @@ require (
 	github.com/sony/gobreaker v0.5.0
 	github.com/spf13/afero v1.9.3
 	github.com/stretchr/testify v1.8.1
+	github.com/thanos-community/promql-engine v0.0.0-20230111121531-c293f65f5389
 	github.com/thanos-io/objstore v0.0.0-20221205132204-5aafc0079f06
 	github.com/thanos-io/thanos v0.29.1-0.20230103123855-3327c510076a
 	github.com/uber/jaeger-client-go v2.30.0+incompatible
@@ -209,6 +210,7 @@ require (
 	golang.org/x/text v0.5.0 // indirect
 	golang.org/x/tools v0.4.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
+	gonum.org/v1/gonum v0.12.0 // indirect
 	google.golang.org/api v0.102.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20221027153422-115e99e71e1c // indirect

go.sum

@@ -1490,6 +1490,8 @@ github.com/tchap/go-patricia v2.2.6+incompatible/go.mod h1:bmLyhP68RS6kStMGxByiQ
 github.com/tencentyun/cos-go-sdk-v5 v0.7.34 h1:xm+Pg+6m486y4eugRI7/E4WasbVmpY1hp9QBSRErgp8=
 github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e h1:f1Zsv7OAU9iQhZwigp50Yl38W10g/vd5NC8Rdk1Jzng=
 github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e/go.mod h1:jXcofnrSln/cLI6/dhlBxPQZEEQHVPCcFaH75M+nSzM=
+github.com/thanos-community/promql-engine v0.0.0-20230111121531-c293f65f5389 h1:ZER0Tf+2PnNEd4bcOJFDQ48dOraRyvu5tfYWySMS2S4=
+github.com/thanos-community/promql-engine v0.0.0-20230111121531-c293f65f5389/go.mod h1:GJkKOtKfXos1xbTmHBnX3YTCov8enxdAS1woR6/h4CI=
 github.com/thanos-io/objstore v0.0.0-20221205132204-5aafc0079f06 h1:xUnLk2CwIoJyv6OB4MWC3aOH9TnneSgM5kgTsOmXIuI=
 github.com/thanos-io/objstore v0.0.0-20221205132204-5aafc0079f06/go.mod h1:gdo4vwwonBnheHB/TCwAOUtKJKrLhLtbBVTQR9rN/v0=
 github.com/thanos-io/thanos v0.29.1-0.20230103123855-3327c510076a h1:oN3VupYNkavPRvdXwq71p54SAFSbOGvL0qL7CeKFrJ0=
@@ -2150,6 +2152,8 @@ golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2/go.mod h1:K8+ghG5WaK9qNq
 gonum.org/v1/gonum v0.0.0-20180816165407-929014505bf4/go.mod h1:Y+Yx5eoAFn32cQvJDxZx5Dpnq+c3wtXuadVZAcxbbBo=
 gonum.org/v1/gonum v0.8.2/go.mod h1:oe/vMfY3deqTw+1EZJhuvEW2iwGF1bW9wwu7XCu0+v0=
 gonum.org/v1/gonum v0.9.3/go.mod h1:TZumC3NeyVQskjXqmyWt4S3bINhy7B4eYwW69EbyX+0=
+gonum.org/v1/gonum v0.12.0 h1:xKuo6hzt+gMav00meVPUlXwSdoEJP46BR+wdxQEFK2o=
+gonum.org/v1/gonum v0.12.0/go.mod h1:73TDxJfAAHeA8Mk9mf8NlIppyhQNo5GLTcYeqgo2lvY=
 gonum.org/v1/netlib v0.0.0-20190313105609-8cb42192e0e0/go.mod h1:wa6Ws7BG/ESfp6dHfk7C6KdzKA7wR7u/rKwOGE66zvw=
 gonum.org/v1/plot v0.0.0-20190515093506-e2840ee46a6b/go.mod h1:Wt8AAjI+ypCyYX3nZBvf6cAIx93T+c/OS2HFAYskSZc=
 gonum.org/v1/plot v0.9.0/go.mod h1:3Pcqqmp6RHvJI72kgb8fThyUnav364FOsdDo2aGW5lY=

integration/querier_remote_read_test.go

@@ -6,8 +6,10 @@ package integration
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"testing"
 	"time"
 
@@ -25,99 +27,105 @@ import (
 )
 
 func TestQuerierRemoteRead(t *testing.T) {
-	s, err := e2e.NewScenario(networkName)
-	require.NoError(t, err)
-	defer s.Close()
-
-	flags := mergeFlags(BlocksStorageFlags(), map[string]string{})
-
-	// Start dependencies.
-	minio := e2edb.NewMinio(9000, bucketName)
-	consul := e2edb.NewConsul()
-	require.NoError(t, s.StartAndWaitReady(consul, minio))
-
-	// Start Cortex components for the write path.
-	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
-	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
-	require.NoError(t, s.StartAndWaitReady(distributor, ingester))
-
-	// Wait until the distributor has updated the ring.
-	require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
-
-	// Push a series for each user to Cortex.
-	now := time.Now()
-
-	c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", "user-1")
-	require.NoError(t, err)
-
-	series, expectedVectors := generateSeries("series_1", now)
-	res, err := c.Push(series)
-	require.NoError(t, err)
-	require.Equal(t, 200, res.StatusCode)
-
-	storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
-	require.NoError(t, s.StartAndWaitReady(storeGateway))
-	querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), BlocksStorageFlags(), "")
-	require.NoError(t, s.StartAndWaitReady(querier))
-
-	// Wait until the querier has updated the ring.
-	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(2*512), "cortex_ring_tokens_total"))
-
-	matcher, err := labels.NewMatcher(labels.MatchEqual, "__name__", "series_1")
-	require.NoError(t, err)
-
-	startMs := now.Add(-1*time.Minute).Unix() * 1000
-	endMs := now.Add(time.Minute).Unix() * 1000
-
-	q, err := remote.ToQuery(startMs, endMs, []*labels.Matcher{matcher}, &storage.SelectHints{
-		Step:  1,
-		Start: startMs,
-		End:   endMs,
-	})
-	require.NoError(t, err)
-
-	req := &prompb.ReadRequest{
-		Queries:               []*prompb.Query{q},
-		AcceptedResponseTypes: []prompb.ReadRequest_ResponseType{prompb.ReadRequest_STREAMED_XOR_CHUNKS},
+	for _, thanosEngine := range []bool{false, true} {
+		t.Run(fmt.Sprintf("thanosEngine=%t", thanosEngine), func(t *testing.T) {
+			s, err := e2e.NewScenario(networkName)
+			require.NoError(t, err)
+			defer s.Close()
+
+			flags := mergeFlags(BlocksStorageFlags(), map[string]string{
+				"-querier.thanos-engine": strconv.FormatBool(thanosEngine),
+			})
+
+			// Start dependencies.
+			minio := e2edb.NewMinio(9000, bucketName)
+			consul := e2edb.NewConsul()
+			require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+			// Start Cortex components for the write path.
+			distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+			ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+			require.NoError(t, s.StartAndWaitReady(distributor, ingester))
+
+			// Wait until the distributor has updated the ring.
+			require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+			// Push a series for each user to Cortex.
+			now := time.Now()
+
+			c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", "user-1")
+			require.NoError(t, err)
+
+			series, expectedVectors := generateSeries("series_1", now)
+			res, err := c.Push(series)
+			require.NoError(t, err)
+			require.Equal(t, 200, res.StatusCode)
+
+			storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+			require.NoError(t, s.StartAndWaitReady(storeGateway))
+			querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+			require.NoError(t, s.StartAndWaitReady(querier))
+
+			// Wait until the querier has updated the ring.
+			require.NoError(t, querier.WaitSumMetrics(e2e.Equals(2*512), "cortex_ring_tokens_total"))
+
+			matcher, err := labels.NewMatcher(labels.MatchEqual, "__name__", "series_1")
+			require.NoError(t, err)
+
+			startMs := now.Add(-1*time.Minute).Unix() * 1000
+			endMs := now.Add(time.Minute).Unix() * 1000
+
+			q, err := remote.ToQuery(startMs, endMs, []*labels.Matcher{matcher}, &storage.SelectHints{
+				Step:  1,
+				Start: startMs,
+				End:   endMs,
+			})
+			require.NoError(t, err)
+
+			req := &prompb.ReadRequest{
+				Queries:               []*prompb.Query{q},
+				AcceptedResponseTypes: []prompb.ReadRequest_ResponseType{prompb.ReadRequest_STREAMED_XOR_CHUNKS},
+			}
+
+			data, err := proto.Marshal(req)
+			require.NoError(t, err)
+			compressed := snappy.Encode(nil, data)
+
+			// Call the remote read API endpoint with a timeout.
+			httpReqCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+
+			httpReq, err := http.NewRequestWithContext(httpReqCtx, "POST", "http://"+querier.HTTPEndpoint()+"/prometheus/api/v1/read", bytes.NewReader(compressed))
+			require.NoError(t, err)
+			httpReq.Header.Set("X-Scope-OrgID", "user-1")
+			httpReq.Header.Add("Content-Encoding", "snappy")
+			httpReq.Header.Add("Accept-Encoding", "snappy")
+			httpReq.Header.Set("Content-Type", "application/x-protobuf")
+			httpReq.Header.Set("User-Agent", "Prometheus/1.8.2")
+			httpReq.Header.Set("X-Prometheus-Remote-Read-Version", "0.1.0")
+
+			httpResp, err := http.DefaultClient.Do(httpReq)
+			require.NoError(t, err)
+			require.Equal(t, http.StatusOK, httpResp.StatusCode)
+
+			compressed, err = io.ReadAll(httpResp.Body)
+			require.NoError(t, err)
+
+			uncompressed, err := snappy.Decode(nil, compressed)
+			require.NoError(t, err)
+
+			var resp prompb.ReadResponse
+			err = proto.Unmarshal(uncompressed, &resp)
+			require.NoError(t, err)
+
+			// Validate the returned remote read data matches what was written
+			require.Len(t, resp.Results, 1)
+			require.Len(t, resp.Results[0].Timeseries, 1)
+			require.Len(t, resp.Results[0].Timeseries[0].Labels, 1)
+			require.Equal(t, "series_1", resp.Results[0].Timeseries[0].Labels[0].GetValue())
+			require.Len(t, resp.Results[0].Timeseries[0].Samples, 1)
+			require.Equal(t, int64(expectedVectors[0].Timestamp), resp.Results[0].Timeseries[0].Samples[0].Timestamp)
+			require.Equal(t, float64(expectedVectors[0].Value), resp.Results[0].Timeseries[0].Samples[0].Value)
+		})
 	}
-
-	data, err := proto.Marshal(req)
-	require.NoError(t, err)
-	compressed := snappy.Encode(nil, data)
-
-	// Call the remote read API endpoint with a timeout.
-	httpReqCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	httpReq, err := http.NewRequestWithContext(httpReqCtx, "POST", "http://"+querier.HTTPEndpoint()+"/prometheus/api/v1/read", bytes.NewReader(compressed))
-	require.NoError(t, err)
-	httpReq.Header.Set("X-Scope-OrgID", "user-1")
-	httpReq.Header.Add("Content-Encoding", "snappy")
-	httpReq.Header.Add("Accept-Encoding", "snappy")
-	httpReq.Header.Set("Content-Type", "application/x-protobuf")
-	httpReq.Header.Set("User-Agent", "Prometheus/1.8.2")
-	httpReq.Header.Set("X-Prometheus-Remote-Read-Version", "0.1.0")
-
-	httpResp, err := http.DefaultClient.Do(httpReq)
-	require.NoError(t, err)
-	require.Equal(t, http.StatusOK, httpResp.StatusCode)
-
-	compressed, err = io.ReadAll(httpResp.Body)
-	require.NoError(t, err)
-
-	uncompressed, err := snappy.Decode(nil, compressed)
-	require.NoError(t, err)
-
-	var resp prompb.ReadResponse
-	err = proto.Unmarshal(uncompressed, &resp)
-	require.NoError(t, err)
-
-	// Validate the returned remote read data matches what was written
-	require.Len(t, resp.Results, 1)
-	require.Len(t, resp.Results[0].Timeseries, 1)
-	require.Len(t, resp.Results[0].Timeseries[0].Labels, 1)
-	require.Equal(t, "series_1", resp.Results[0].Timeseries[0].Labels[0].GetValue())
-	require.Len(t, resp.Results[0].Timeseries[0].Samples, 1)
-	require.Equal(t, int64(expectedVectors[0].Timestamp), resp.Results[0].Timeseries[0].Samples[0].Timestamp)
-	require.Equal(t, float64(expectedVectors[0].Value), resp.Results[0].Timeseries[0].Samples[0].Value)
 }