Fixes the Series function to hanlde properly sharding. (#4563)

Before they would try to match the injected label but this label is only used by the store and not returned in the chunks causing all chunks to be filtered out. Fixes #4547 Signed-off-by: Cyril Tovena <[email protected]>
grafana · Oct 27, 2021 · 165ff31 · 165ff31
1 parent e49e318
commit 165ff31
Show file tree

Hide file tree

Showing 3 changed files with 127 additions and 1 deletion.
diff --git a/pkg/querier/queryrange/codec.go b/pkg/querier/queryrange/codec.go
@@ -238,6 +238,7 @@ func (Codec) DecodeRequest(_ context.Context, r *http.Request) (queryrange.Reque
 			StartTs: req.Start.UTC(),
 			EndTs:   req.End.UTC(),
 			Path:    r.URL.Path,
+			Shards:  req.Shards,
 		}, nil
 	case LabelNamesOp:
 		req, err := loghttp.ParseLabelQuery(r)

diff --git a/pkg/storage/store.go b/pkg/storage/store.go
@@ -303,6 +303,9 @@ func (s *store) GetSeries(ctx context.Context, req logql.SelectLogParams) ([]log
 	outer:
 		for _, chk := range group {
 			for _, matcher := range matchers {
+				if matcher.Name == astmapper.ShardLabel || matcher.Name == labels.MetricName {
+					continue
+				}
 				if !matcher.Matches(chk.Chunk.Metric.Get(matcher.Name)) {
 					continue outer
 				}

diff --git a/pkg/storage/store_test.go b/pkg/storage/store_test.go
@@ -752,7 +752,7 @@ func Test_store_decodeReq_Matchers(t *testing.T) {
 			},
 		},
 		{
-			"unsharded",
+			"sharded",
 			newQuery(
 				"{foo=~\"ba.*\"}", from, from.Add(6*time.Millisecond),
 				[]astmapper.ShardAnnotation{
@@ -1169,3 +1169,125 @@ func Test_OverlappingChunks(t *testing.T) {
 	require.Equal(t, "1", it.Entry().Line)
 	require.False(t, it.Next())
 }
+
+func Test_GetSeries(t *testing.T) {
+	var (
+		store = &store{
+			Store: newMockChunkStore([]*logproto.Stream{
+				{
+					Labels: `{foo="bar",buzz="boo"}`,
+					Entries: []logproto.Entry{
+						{Timestamp: time.Unix(0, 1), Line: "1"},
+					},
+				},
+				{
+					Labels: `{foo="buzz"}`,
+					Entries: []logproto.Entry{
+						{Timestamp: time.Unix(0, 1), Line: "1"},
+					},
+				},
+				{
+					Labels: `{bar="foo"}`,
+					Entries: []logproto.Entry{
+						{Timestamp: time.Unix(0, 1), Line: "1"},
+					},
+				},
+			}),
+			cfg: Config{
+				MaxChunkBatchSize: 10,
+			},
+			chunkMetrics: NilMetrics,
+		}
+		ctx            = user.InjectOrgID(context.Background(), "test-user")
+		expectedSeries = []logproto.SeriesIdentifier{
+			{
+				Labels: map[string]string{"bar": "foo"},
+			},
+			{
+				Labels: map[string]string{"foo": "bar", "buzz": "boo"},
+			},
+			{
+				Labels: map[string]string{"foo": "buzz"},
+			},
+		}
+	)
+
+	for _, tt := range []struct {
+		name           string
+		req            logql.SelectLogParams
+		expectedSeries []logproto.SeriesIdentifier
+	}{
+		{
+			"all series",
+			logql.SelectLogParams{
+				QueryRequest: &logproto.QueryRequest{
+					Selector: ``,
+					Start:    time.Unix(0, 0),
+					End:      time.Unix(0, 10),
+				},
+			},
+			expectedSeries,
+		},
+		{
+			"all series with sharding",
+			logql.SelectLogParams{
+				QueryRequest: &logproto.QueryRequest{
+					Selector: ``,
+					Start:    time.Unix(0, 0),
+					End:      time.Unix(0, 10),
+					Shards:   []string{astmapper.ShardAnnotation{Shard: 1, Of: 16}.String()},
+				},
+			},
+			expectedSeries,
+		},
+		{
+			"selected series",
+			logql.SelectLogParams{
+				QueryRequest: &logproto.QueryRequest{
+					Selector: `{buzz=~".oo"}`,
+					Start:    time.Unix(0, 0),
+					End:      time.Unix(0, 10),
+				},
+			},
+			[]logproto.SeriesIdentifier{
+				{
+					Labels: map[string]string{"foo": "bar", "buzz": "boo"},
+				},
+			},
+		},
+		{
+			"selected series with sharding",
+			logql.SelectLogParams{
+				QueryRequest: &logproto.QueryRequest{
+					Selector: `{buzz=~".oo"}`,
+					Start:    time.Unix(0, 0),
+					End:      time.Unix(0, 10),
+					Shards:   []string{astmapper.ShardAnnotation{Shard: 1, Of: 16}.String()},
+				},
+			},
+			[]logproto.SeriesIdentifier{
+				{
+					Labels: map[string]string{"foo": "bar", "buzz": "boo"},
+				},
+			},
+		},
+		{
+			"no match",
+			logql.SelectLogParams{
+				QueryRequest: &logproto.QueryRequest{
+					Selector: `{buzz=~"foo"}`,
+					Start:    time.Unix(0, 0),
+					End:      time.Unix(0, 10),
+				},
+			},
+			[]logproto.SeriesIdentifier{},
+		},
+	} {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			series, err := store.GetSeries(ctx, tt.req)
+			require.NoError(t, err)
+			require.Equal(t, tt.expectedSeries, series)
+		})
+	}
+}