Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Filter external labels from matchers on LabelValues/LabelNames #5596

Merged
merged 4 commits into from
Aug 17, 2022
Merged

Filter external labels from matchers on LabelValues/LabelNames #5596

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
02dfb61
Filter external labels from matchers on LabelValues/LabelNames
oronsh Aug 15, 2022
893ba45
Fix linting
oronsh Aug 15, 2022
f497e09
Fix linter
oronsh Aug 15, 2022
d561e4c
Fix comments to be full Sentences
oronsh Aug 17, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 42 additions & 14 deletions pkg/store/bucket.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1254,6 +1254,11 @@ func (s *BucketStore) LabelNames(ctx context.Context, req *storepb.LabelNamesReq
if len(reqBlockMatchers) > 0 && !b.matchRelabelLabels(reqBlockMatchers) {
continue
}
// Filter external labels from matchers.
reqSeriesMatchersNoExtLabels, ok := b.FilterExtLabelsMatchers(reqSeriesMatchers)
if !ok {
continue
}

resHints.AddQueriedBlock(b.meta.ULID)

Expand All @@ -1270,7 +1275,7 @@ func (s *BucketStore) LabelNames(ctx context.Context, req *storepb.LabelNamesReq
defer runutil.CloseWithLogOnErr(s.logger, indexr, "label names")

var result []string
if len(reqSeriesMatchers) == 0 {
if len(reqSeriesMatchersNoExtLabels) == 0 {
// Do it via index reader to have pending reader registered correctly.
// LabelNames are already sorted.
res, err := indexr.block.indexHeaderReader.LabelNames()
Expand All @@ -1288,7 +1293,7 @@ func (s *BucketStore) LabelNames(ctx context.Context, req *storepb.LabelNamesReq

result = strutil.MergeSlices(res, extRes)
} else {
seriesSet, _, err := blockSeries(newCtx, b.extLset, indexr, nil, reqSeriesMatchers, nil, seriesLimiter, true, req.Start, req.End, nil, nil)
seriesSet, _, err := blockSeries(newCtx, b.extLset, indexr, nil, reqSeriesMatchersNoExtLabels, nil, seriesLimiter, true, req.Start, req.End, nil, nil)
if err != nil {
return errors.Wrapf(err, "fetch series for block %s", b.meta.ULID)
}
Expand Down Expand Up @@ -1341,6 +1346,24 @@ func (s *BucketStore) LabelNames(ctx context.Context, req *storepb.LabelNamesReq
}, nil
}

func (b *bucketBlock) FilterExtLabelsMatchers(matchers []*labels.Matcher) ([]*labels.Matcher, bool) {
// We filter external labels from matchers so we won't try to match series on them.
var result []*labels.Matcher
for _, m := range matchers {
// Get value of external label from block.
v := b.extLset.Get(m.Name)
// If value is empty string the matcher is a valid one since it's not part of external labels.
if v == "" {
result = append(result, m)
} else if v != "" && v != m.Value {
// If matcher is external label but value is different we don't want to look in block anyway.
return []*labels.Matcher{}, false
}
}

return result, true
}

// LabelValues implements the storepb.StoreServer interface.
func (s *BucketStore) LabelValues(ctx context.Context, req *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) {
reqSeriesMatchers, err := storepb.MatchersToPromMatchers(req.Matchers...)
Expand All @@ -1366,16 +1389,6 @@ func (s *BucketStore) LabelValues(ctx context.Context, req *storepb.LabelValuesR
}
}

// If we have series matchers, add <labelName> != "" matcher, to only select series that have given label name.
if len(reqSeriesMatchers) > 0 {
m, err := labels.NewMatcher(labels.MatchNotEqual, req.Label, "")
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}

reqSeriesMatchers = append(reqSeriesMatchers, m)
}

s.mtx.RLock()

var mtx sync.Mutex
Expand All @@ -1391,6 +1404,21 @@ func (s *BucketStore) LabelValues(ctx context.Context, req *storepb.LabelValuesR
if len(reqBlockMatchers) > 0 && !b.matchRelabelLabels(reqBlockMatchers) {
continue
}
// Filter external labels from matchers.
reqSeriesMatchersNoExtLabels, ok := b.FilterExtLabelsMatchers(reqSeriesMatchers)
if !ok {
continue
}

// If we have series matchers, add <labelName> != "" matcher, to only select series that have given label name.
if len(reqSeriesMatchersNoExtLabels) > 0 {
m, err := labels.NewMatcher(labels.MatchNotEqual, req.Label, "")
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}

reqSeriesMatchersNoExtLabels = append(reqSeriesMatchersNoExtLabels, m)
}

resHints.AddQueriedBlock(b.meta.ULID)

Expand All @@ -1406,7 +1434,7 @@ func (s *BucketStore) LabelValues(ctx context.Context, req *storepb.LabelValuesR
defer runutil.CloseWithLogOnErr(s.logger, indexr, "label values")

var result []string
if len(reqSeriesMatchers) == 0 {
if len(reqSeriesMatchersNoExtLabels) == 0 {
// Do it via index reader to have pending reader registered correctly.
res, err := indexr.block.indexHeaderReader.LabelValues(req.Label)
if err != nil {
Expand All @@ -1419,7 +1447,7 @@ func (s *BucketStore) LabelValues(ctx context.Context, req *storepb.LabelValuesR
}
result = res
} else {
seriesSet, _, err := blockSeries(newCtx, b.extLset, indexr, nil, reqSeriesMatchers, nil, seriesLimiter, true, req.Start, req.End, nil, nil)
seriesSet, _, err := blockSeries(newCtx, b.extLset, indexr, nil, reqSeriesMatchersNoExtLabels, nil, seriesLimiter, true, req.Start, req.End, nil, nil)
if err != nil {
return errors.Wrapf(err, "fetch series for block %s", b.meta.ULID)
}
Expand Down
46 changes: 46 additions & 0 deletions pkg/store/bucket_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,52 @@ func TestBucketBlock_Property(t *testing.T) {
properties.TestingRun(t)
}

func TestBucketFilterExtLabelsMatchers(t *testing.T) {
defer testutil.TolerantVerifyLeak(t)

dir := t.TempDir()
bkt, err := filesystem.NewBucket(dir)
testutil.Ok(t, err)
defer func() { testutil.Ok(t, bkt.Close()) }()

blockID := ulid.MustNew(1, nil)
meta := &metadata.Meta{
BlockMeta: tsdb.BlockMeta{ULID: blockID},
Thanos: metadata.Thanos{
Labels: map[string]string{
"a": "b",
"c": "d",
},
},
}
b, _ := newBucketBlock(context.Background(), log.NewNopLogger(), newBucketStoreMetrics(nil), meta, bkt, path.Join(dir, blockID.String()), nil, nil, nil, nil)
ms := []*labels.Matcher{
{Type: labels.MatchNotEqual, Name: "a", Value: "b"},
}
res, _ := b.FilterExtLabelsMatchers(ms)
testutil.Equals(t, len(res), 0)

ms = []*labels.Matcher{
{Type: labels.MatchNotEqual, Name: "a", Value: "a"},
}
_, ok := b.FilterExtLabelsMatchers(ms)
testutil.Equals(t, ok, false)

ms = []*labels.Matcher{
{Type: labels.MatchNotEqual, Name: "a", Value: "a"},
{Type: labels.MatchNotEqual, Name: "c", Value: "d"},
}
res, _ = b.FilterExtLabelsMatchers(ms)
testutil.Equals(t, len(res), 0)

ms = []*labels.Matcher{
{Type: labels.MatchNotEqual, Name: "a2", Value: "a"},
}
res, _ = b.FilterExtLabelsMatchers(ms)
testutil.Equals(t, len(res), 1)
testutil.Equals(t, res, ms)
}

func TestBucketBlock_matchLabels(t *testing.T) {
defer testutil.TolerantVerifyLeak(t)

Expand Down