Fix test + changelog

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

Author: alanprot

CHANGELOG.md

@@ -56,6 +56,7 @@
 * [FEATURE] Added `-api.http-request-headers-to-log` allowing for the addition of HTTP Headers to logs #4803
 * [FEATURE] Distributor: Added a new limit `-validation.max-labels-size-bytes` allowing to limit the combined size of labels for each timeseries. #4848
 * [FEATURE] Storage/Bucket: Added `-*.s3.bucket-lookup-type` allowing to configure the s3 bucket lookup type. #4794
+* [FEATURE] QueryFrontend: Implement experimental vertical sharding at query frontend for range/instance queries. #4863
 * [BUGFIX] Memberlist: Add join with no retrying when starting service. #4804
 * [BUGFIX] Ruler: Fix /ruler/rule_groups returns YAML with extra fields. #4767
 * [BUGFIX] Respecting `-tracing.otel.sample-ratio` configuration when enabling OpenTelemetry tracing with X-ray. #4862

pkg/querier/tripperware/test_shard_by_query_utils.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"sort"
 	"testing"
 	"time"
 
@@ -92,6 +93,11 @@ http_requests_total`,
 			expression:     "sum by (pod) (http_requests_total)",
 			shardingLabels: []string{"pod"},
 		},
+		{
+			name:           "aggregation with comparison",
+			expression:     "avg by (Roles,type) (rss_service_message_handling) > 0.5",
+			shardingLabels: []string{"Roles", "type"},
+		},
 		{
 			name:           "multiple aggregations with grouping",
 			expression:     "max by (pod) (sum by (pod, cluster) (http_requests_total))",
@@ -185,23 +191,25 @@ http_requests_total`,
 	}
 	tests := []testCase{
 		{
-			name:        "should shard range query when query is shardable",
-			path:        `/api/v1/query_range?end=1&start=0&step=120&query=sum(metric) by (pod,cluster_name)`,
-			isShardable: true,
-			codec:       shardedPrometheusCodec,
-			shardSize:   2,
+			name:           "should shard range query when query is shardable",
+			path:           `/api/v1/query_range?end=1&start=0&step=120&query=sum(metric) by (pod,cluster_name)`,
+			isShardable:    true,
+			codec:          shardedPrometheusCodec,
+			shardingLabels: []string{"pod", "cluster_name"},
+			shardSize:      2,
 			responses: []string{
 				`{"status":"success","data":{"resultType":"matrix","result":[{"metric":{"__name__":"metric","__job__":"a"},"values":[[1,"1"],[2,"2"],[3,"3"]]}],"stats":{"samples":{"totalQueryableSamples":6,"totalQueryableSamplesPerStep":[[1,1],[2,2],[3,3]]}}}}`,
 				`{"status":"success","data":{"resultType":"matrix","result":[{"metric":{"__name__":"metric","__job__":"b"},"values":[[1,"1"],[2,"2"],[3,"3"]]}],"stats":{"samples":{"totalQueryableSamples":6,"totalQueryableSamplesPerStep":[[1,1],[2,2],[3,3]]}}}}`,
 			},
 			response: `{"status":"success","data":{"resultType":"matrix","result":[{"metric":{"__job__":"a","__name__":"metric"},"values":[[1,"1"],[2,"2"],[3,"3"]]},{"metric":{"__job__":"b","__name__":"metric"},"values":[[1,"1"],[2,"2"],[3,"3"]]}],"stats":{"samples":{"totalQueryableSamples":12,"totalQueryableSamplesPerStep":[[1,2],[2,4],[3,6]]}}}}`,
 		},
 		{
-			name:        "should shard instant query when query is shardable",
-			path:        `/api/v1/query?time=120&query=sum(metric) by (pod,cluster_name)`,
-			codec:       instantQueryCodec,
-			shardSize:   2,
-			isShardable: true,
+			name:           "should shard instant query when query is shardable",
+			path:           `/api/v1/query?time=120&query=sum(metric) by (pod,cluster_name)`,
+			codec:          instantQueryCodec,
+			shardSize:      2,
+			shardingLabels: []string{"pod", "cluster_name"},
+			isShardable:    true,
 			responses: []string{
 				`{"status":"success","data":{"resultType":"vector","result":[{"metric":{"__name__":"up","job":"foo"},"value":[1,"1"]}],"stats":{"samples":{"totalQueryableSamples":10,"totalQueryableSamplesPerStep":[[1,10]]}}}}`,
 				`{"status":"success","data":{"resultType":"vector","result":[{"metric":{"__name__":"up","job":"bar"},"value":[2,"2"]}],"stats":{"samples":{"totalQueryableSamples":10,"totalQueryableSamplesPerStep":[[1,10]]}}}}`,
@@ -258,11 +266,12 @@ http_requests_total`,
 			response: `{"status":"success","data":{"resultType":"vector","result":[{"metric":{"__name__":"up","job":"bar"},"value":[2,"2"]},{"metric":{"__name__":"up","job":"foo"},"value":[1,"1"]}],"stats":{"samples":{"totalQueryableSamples":20,"totalQueryableSamplesPerStep":[[1,20]]}}}}`,
 		})
 		tests = append(tests, testCase{
-			name:        fmt.Sprintf("shardable query_range: %s", query.name),
-			path:        fmt.Sprintf(`/api/v1/query_range?start=1&end=2&step=1&query=%s`, url.QueryEscape(query.expression)),
-			codec:       shardedPrometheusCodec,
-			isShardable: true,
-			shardSize:   2,
+			name:           fmt.Sprintf("shardable query_range: %s", query.name),
+			path:           fmt.Sprintf(`/api/v1/query_range?start=1&end=2&step=1&query=%s`, url.QueryEscape(query.expression)),
+			codec:          shardedPrometheusCodec,
+			isShardable:    true,
+			shardSize:      2,
+			shardingLabels: query.shardingLabels,
 			responses: []string{
 				`{"status":"success","data":{"resultType":"matrix","result":[{"metric":{"__name__":"metric","__job__":"a"},"values":[[1,"1"],[2,"2"],[3,"3"]]}],"stats":{"samples":{"totalQueryableSamples":6,"totalQueryableSamplesPerStep":[[1,1],[2,2],[3,3]]}}}}`,
 				`{"status":"success","data":{"resultType":"matrix","result":[{"metric":{"__name__":"metric","__job__":"b"},"values":[[1,"1"],[2,"2"],[3,"3"]]}],"stats":{"samples":{"totalQueryableSamples":6,"totalQueryableSamplesPerStep":[[1,1],[2,2],[3,3]]}}}}`,
@@ -273,28 +282,33 @@ http_requests_total`,
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			sort.Strings(tt.shardingLabels)
 			s := httptest.NewServer(
 				http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 					q := r.FormValue("query")
 					expr, _ := parser.ParseExpr(q)
-					shardInfo := storepb.ShardInfo{}
+					shardIndex := int64(0)
 
 					parser.Inspect(expr, func(n parser.Node, _ []parser.Node) error {
 						if selector, ok := n.(*parser.VectorSelector); ok {
 							for _, matcher := range selector.LabelMatchers {
 								if matcher.Name == querysharding.CortexShardByLabel {
 
 									decoded, _ := base64.StdEncoding.DecodeString(matcher.Value)
+									shardInfo := storepb.ShardInfo{}
 									err := shardInfo.Unmarshal(decoded)
 									require.NoError(t, err)
+									sort.Strings(shardInfo.Labels)
+									require.Equal(t, tt.shardingLabels, shardInfo.Labels)
+									require.Equal(t, tt.isShardable, shardInfo.TotalShards > 0)
+									shardIndex = shardInfo.ShardIndex
 								}
 							}
 						}
 						return nil
 					})
 
-					require.Equal(t, tt.isShardable, shardInfo.TotalShards > 0)
-					_, _ = w.Write([]byte(tt.responses[shardInfo.ShardIndex]))
+					_, _ = w.Write([]byte(tt.responses[shardIndex]))
 				}),
 			)
 			defer s.Close()