Fix multi-tenant exemplar matchers

CHANGELOG.md

@@ -13,7 +13,8 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 ### Fixed
 - [#5502](https://github.com/thanos-io/thanos/pull/5502) Receive: Handle exemplar storage errors as conflict error.
-- [#5534](https://github.com/thanos-io/thanos/pull/5534) Query: Set struct return by query api alerts same as prometheus api
+- [#5534](https://github.com/thanos-io/thanos/pull/5534) Query: Set struct return by query api alerts same as prometheus api.
+- [#5554](https://github.com/thanos-io/thanos/pull/5554) Query/Receiver: Fix querying exemplars from multi-tenant receivers.
 
 ### Added
 

pkg/exemplars/proxy.go

@@ -6,6 +6,7 @@ package exemplars
 import (
 	"context"
 	"io"
+	"strings"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -79,11 +80,14 @@ func (s *Proxy) Exemplars(req *exemplarspb.ExemplarsRequest, srv exemplarspb.Exe
 		exemplars []*exemplarspb.ExemplarData
 	)
 
+	queryParts := make([]string, 0)
+	labelMatchers := make([]string, 0)
 	for _, st := range s.exemplars() {
-		query := ""
+		queryParts = queryParts[:0]
+
 	Matchers:
 		for _, matchers := range selectors {
-			metricsSelector := ""
+			matcherSet := make(map[string]struct{})
 			for _, m := range matchers {
 				for _, ls := range st.LabelSets {
 					if lv := ls.Get(m.Name); lv != "" {
@@ -96,27 +100,27 @@ func (s *Proxy) Exemplars(req *exemplarspb.ExemplarsRequest, srv exemplarspb.Exe
 							continue
 						}
 					}
-					if metricsSelector == "" {
-						metricsSelector += m.String()
-					} else {
-						metricsSelector += ", " + m.String()
-					}
+					matcherSet[m.String()] = struct{}{}
 				}
 			}
-			// Construct the query by concatenating metric selectors with '+'.
-			// We cannot preserve the original query info, but the returned
-			// results are the same.
-			if query == "" {
-				query += "{" + metricsSelector + "}"
-			} else {
-				query += " + {" + metricsSelector + "}"
+
+			labelMatchers = labelMatchers[:0]
+			for m := range matcherSet {
+				labelMatchers = append(labelMatchers, m)
 			}
+
+			queryParts = append(queryParts, "{"+strings.Join(labelMatchers, ", ")+"}")
 		}
 
 		// No matchers match this store.
-		if query == "" {
+		if len(queryParts) == 0 {
 			continue
 		}
+
+		// Construct the query by concatenating metric selectors with '+'.
+		// We cannot preserve the original query info, but the returned
+		// results are the same.
+		query := strings.Join(queryParts, "+")
 		r := &exemplarspb.ExemplarsRequest{
 			Start:                   req.Start,
 			End:                     req.End,

pkg/exemplars/proxy_test.go

@@ -5,6 +5,7 @@ package exemplars
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"os"
 	"reflect"
@@ -14,8 +15,11 @@ import (
 	"github.com/go-kit/log"
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql/parser"
 	"go.uber.org/atomic"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/thanos-io/thanos/pkg/exemplars/exemplarspb"
 	"github.com/thanos-io/thanos/pkg/store/labelpb"
@@ -49,9 +53,33 @@ func (t *testExemplarClient) Recv() (*exemplarspb.ExemplarsResponse, error) {
 }
 
 func (t *testExemplarClient) Exemplars(ctx context.Context, in *exemplarspb.ExemplarsRequest, opts ...grpc.CallOption) (exemplarspb.Exemplars_ExemplarsClient, error) {
+	expr, err := parser.ParseExpr(in.Query)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	if err := t.assertUniqueMatchers(expr); err != nil {
+		return nil, err
+	}
+
 	return t, t.exemplarErr
 }
 
+func (t *testExemplarClient) assertUniqueMatchers(expr parser.Expr) error {
+	matchersList := parser.ExtractSelectors(expr)
+	for _, matchers := range matchersList {
+		matcherSet := make(map[string]struct{})
+		for _, matcher := range matchers {
+			if _, ok := matcherSet[matcher.String()]; ok {
+				return status.Error(codes.Internal, fmt.Sprintf("duplicate matcher set found %s", matcher))
+			}
+			matcherSet[matcher.String()] = struct{}{}
+		}
+	}
+
+	return nil
+}
+
 var _ exemplarspb.ExemplarsClient = &testExemplarClient{}
 
 type testExemplarServer struct {
@@ -94,7 +122,7 @@ func TestProxy(t *testing.T) {
 		{
 			name: "proxy success",
 			request: &exemplarspb.ExemplarsRequest{
-				Query:                   "http_request_duration_bucket",
+				Query:                   `http_request_duration_bucket`,
 				PartialResponseStrategy: storepb.PartialResponseStrategy_WARN,
 			},
 			clients: []*exemplarspb.ExemplarStore{
@@ -105,7 +133,38 @@ func TestProxy(t *testing.T) {
 							Exemplars:    []*exemplarspb.Exemplar{{Value: 1}},
 						}),
 					},
-					LabelSets: []labels.Labels{labels.FromMap(map[string]string{"cluster": "A"})},
+					LabelSets: []labels.Labels{
+						labels.FromMap(map[string]string{"cluster": "A"}),
+						labels.FromMap(map[string]string{"cluster": "B"}),
+					},
+				},
+			},
+			server: &testExemplarServer{},
+			wantResponses: []*exemplarspb.ExemplarsResponse{
+				exemplarspb.NewExemplarsResponse(&exemplarspb.ExemplarData{
+					SeriesLabels: labelpb.ZLabelSet{Labels: labelpb.ZLabelsFromPromLabels(labels.FromMap(map[string]string{"__name__": "http_request_duration_bucket"}))},
+					Exemplars:    []*exemplarspb.Exemplar{{Value: 1}},
+				}),
+			},
+		},
+		{
+			name: "proxy success with multiple selectors",
+			request: &exemplarspb.ExemplarsRequest{
+				Query:                   `http_request_duration_bucket{region="us-east1"} / on (region) group_left() http_request_duration_bucket`,
+				PartialResponseStrategy: storepb.PartialResponseStrategy_WARN,
+			},
+			clients: []*exemplarspb.ExemplarStore{
+				{
+					ExemplarsClient: &testExemplarClient{
+						response: exemplarspb.NewExemplarsResponse(&exemplarspb.ExemplarData{
+							SeriesLabels: labelpb.ZLabelSet{Labels: labelpb.ZLabelsFromPromLabels(labels.FromMap(map[string]string{"__name__": "http_request_duration_bucket"}))},
+							Exemplars:    []*exemplarspb.Exemplar{{Value: 1}},
+						}),
+					},
+					LabelSets: []labels.Labels{
+						labels.FromMap(map[string]string{"cluster": "A"}),
+						labels.FromMap(map[string]string{"cluster": "B"}),
+					},
 				},
 			},
 			server: &testExemplarServer{},
@@ -119,7 +178,7 @@ func TestProxy(t *testing.T) {
 		{
 			name: "warning proxy success",
 			request: &exemplarspb.ExemplarsRequest{
-				Query:                   "http_request_duration_bucket",
+				Query:                   `http_request_duration_bucket`,
 				PartialResponseStrategy: storepb.PartialResponseStrategy_WARN,
 			},
 			clients: []*exemplarspb.ExemplarStore{