Kiu/VictoriaMetrics
Kiu/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-01-20 07:19:17 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

lib/storage: optimize convert multiple values regexp filter to composite tag filter (#1610)

Browse Source 
* lib/storage: optimize convert multiple values regexp filter to composite tag filter

* Apply suggestions from code review

Co-authored-by: Aliaksandr Valialkin <valyala@gmail.com>
This commit is contained in:
faceair 2021-09-14 17:47:07 +08:00 committed by GitHub
parent bc2d05be8e
commit 00cbb099b6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 351 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
lib/storage/tag_filters.go
View File

@ -18,34 +18,58 @@ import (
//
// This converts `foo{bar="baz",x=~"a.+"}` to `{foo=bar="baz",foo=x=~"a.+"} filter.
func convertToCompositeTagFilterss(tfss []*TagFilters) []*TagFilters {
tfssNew := make([]*TagFilters, len(tfss))
for i, tfs := range tfss {
tfssNew[i] = convertToCompositeTagFilters(tfs)
tfssNew := make([]*TagFilters, 0, len(tfss))
for _, tfs := range tfss {
tfssNew = append(tfss, convertToCompositeTagFilters(tfs)...)
}
return tfssNew
}
func convertToCompositeTagFilters(tfs *TagFilters) *TagFilters {
// Search for metric name filter, which must be used for creating composite filters.
var name []byte
func convertToCompositeTagFilters(tfs *TagFilters) []*TagFilters {
tfssCompiled := make([]*TagFilters, 0)
// Search for filters on metric name, which will be used for creating composite filters.
var names [][]byte
hasPositiveFilter := false
for _, tf := range tfs.tfs {
if len(tf.key) == 0 && !tf.isNegative && !tf.isRegexp {
name = tf.value
names = [][]byte{tf.value}
} else if len(tf.key) == 0 && !tf.isNegative && tf.isRegexp && len(tf.orSuffixes) > 0 {
// Split the filter {__name__=~"name1|...|nameN", other_filters}
// into `name1{other_filters}`, ..., `nameN{other_filters}`
// and generate composite filters for each of them
names = names[:0]
for _, orSuffix := range tf.orSuffixes {
names = append(names, []byte(orSuffix))
}
} else if !tf.isNegative && !tf.isEmptyMatch {
hasPositiveFilter = true
}
}
if len(name) == 0 {
if len(names) == 0 {
tfssCompiled = append(tfssCompiled, tfs)
atomic.AddUint64(&compositeFilterMissingConversions, 1)
return tfs
return tfssCompiled
}
tfsNew := make([]tagFilter, 0, len(tfs.tfs))
var compositeKey []byte
compositeFilters := 0
for _, name := range names {
tfsNew := make([]tagFilter, 0, len(tfs.tfs))
for _, tf := range tfs.tfs {
if len(tf.key) == 0 {
if !hasPositiveFilter || tf.isNegative || tf.isRegexp || string(tf.value) != string(name) {
sameOrSuffixes := true
if len(names) != len(tf.orSuffixes) {
sameOrSuffixes = false
} else {
for i, orSuffix := range tf.orSuffixes {
if string(names[i]) != orSuffix {
sameOrSuffixes = false
break
}
}
}
if !hasPositiveFilter || tf.isNegative || tf.isRegexp && !sameOrSuffixes || !tf.isRegexp && string(tf.value) != string(name) {
tfsNew = append(tfsNew, tf)
}
continue
@ -54,6 +78,7 @@ func convertToCompositeTagFilters(tfs *TagFilters) *TagFilters {
tfsNew = append(tfsNew, tf)
continue
}
compositeKey = marshalCompositeTagKey(compositeKey[:0], name, tf.key)
var tfNew tagFilter
if err := tfNew.Init(tfs.commonPrefix, compositeKey, tf.value, tf.isNegative, tf.isRegexp); err != nil {
@ -62,14 +87,17 @@ func convertToCompositeTagFilters(tfs *TagFilters) *TagFilters {
tfsNew = append(tfsNew, tfNew)
compositeFilters++
}
if compositeFilters == 0 {
atomic.AddUint64(&compositeFilterMissingConversions, 1)
return tfs
}
tfsCompiled := NewTagFilters()
tfsCompiled.tfs = tfsNew
tfssCompiled = append(tfssCompiled, tfsCompiled)
}
if compositeFilters == 0 {
tfssCompiled = append(tfssCompiled[:0], tfs)
atomic.AddUint64(&compositeFilterMissingConversions, 1)
return tfssCompiled
}
atomic.AddUint64(&compositeFilterSuccessConversions, 1)
return tfsCompiled
return tfssCompiled
}
var (

203
lib/storage/tag_filters_test.go
View File

@ -7,7 +7,7 @@ import (
)
func TestConvertToCompositeTagFilters(t *testing.T) {
f := func(tfs, resultExpected []TagFilter) {
f := func(tfs []TagFilter, resultExpected [][]TagFilter) {
t.Helper()
tfsCompiled := NewTagFilters()
for _, tf := range tfs {
@ -15,23 +15,27 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
t.Fatalf("cannot add tf=%s: %s", tf.String(), err)
}
}
resultCompiled := convertToCompositeTagFilters(tfsCompiled)
result := make([]TagFilter, len(resultCompiled.tfs))
resultCompileds := convertToCompositeTagFilters(tfsCompiled)
result := make([][]TagFilter, len(resultCompileds))
for i, resultCompiled := range resultCompileds {
tfs := make([]TagFilter, len(resultCompiled.tfs))
for i, tf := range resultCompiled.tfs {
result[i] = TagFilter{
tfs[i] = TagFilter{
Key: tf.key,
Value: tf.value,
IsNegative: tf.isNegative,
IsRegexp: tf.isRegexp,
}
}
result[i] = tfs
}
if !reflect.DeepEqual(result, resultExpected) {
t.Fatalf("unexpected result;\ngot\n%+v\nwant\n%+v", result, resultExpected)
}
}
// Empty filters
f(nil, []TagFilter{})
f(nil, [][]TagFilter{{}})
// A single non-name filter
f([]TagFilter{
@ -41,13 +45,15 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("foo"),
Value: []byte("bar"),
IsNegative: false,
IsRegexp: false,
},
},
})
// Multiple non-name filters
@ -64,7 +70,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: true,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("foo"),
Value: []byte("bar"),
@ -77,6 +84,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: true,
IsRegexp: false,
},
},
})
// A single name filter
@ -87,13 +95,15 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar"),
IsNegative: false,
IsRegexp: false,
},
},
})
// Two name filters
@ -110,7 +120,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar"),
@ -123,6 +134,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
},
})
// A name filter with non-name filter.
@ -139,13 +151,15 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
IsNegative: false,
IsRegexp: false,
},
},
})
// A name filter with a single negative filter
@ -162,7 +176,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: true,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar"),
@ -175,6 +190,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: true,
IsRegexp: false,
},
},
})
// A name filter with a negative and a positive filter
@ -197,7 +213,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: true,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
@ -210,6 +227,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: true,
},
},
})
// Two name filters with non-name filter.
@ -232,7 +250,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar"),
@ -245,6 +264,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
},
})
// A name filter with regexp non-name filter, which can be converted to non-regexp.
@ -261,13 +281,15 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: true,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
IsNegative: false,
IsRegexp: false,
},
},
})
// A name filter with regexp non-name filter.
@ -284,13 +306,15 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: true,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc.+"),
IsNegative: false,
IsRegexp: true,
},
},
})
// A name filter with graphite filter.
@ -307,7 +331,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar"),
@ -320,6 +345,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
},
})
// A name filter with non-name filter and a graphite filter.
@ -342,7 +368,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
@ -355,6 +382,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
},
})
// Regexp name filter, which can be converted to non-regexp, with non-name filter.
@ -371,13 +399,144 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
IsNegative: false,
IsRegexp: false,
},
},
})
// Multiple values regexp filter, which can be converted to non-regexp, with non-name filter.
f([]TagFilter{
{
Key: nil,
Value: []byte("bar|foo"),
IsNegative: false,
IsRegexp: true,
},
{
Key: []byte("foo"),
Value: []byte("abc"),
IsNegative: false,
IsRegexp: false,
},
}, [][]TagFilter{
{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
IsNegative: false,
IsRegexp: false,
},
},
{
{
Key: []byte("\xfe\x03foofoo"),
Value: []byte("abc"),
IsNegative: false,
IsRegexp: false,
},
},
})
// Two multiple values regexp filter, which can be converted to non-regexp, with non-name filter.
f([]TagFilter{
{
Key: nil,
Value: []byte("bar|foo"),
IsNegative: false,
IsRegexp: true,
},
{
Key: nil,
Value: []byte("abc|def"),
IsNegative: false,
IsRegexp: true,
},
{
Key: []byte("face"),
Value: []byte("air"),
IsNegative: false,
IsRegexp: false,
},
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar|foo"),
IsNegative: false,
IsRegexp: true,
},
{
Key: []byte("\xfe\x03abcface"),
Value: []byte("air"),
IsNegative: false,
IsRegexp: false,
},
},
{
{
Key: nil,
Value: []byte("bar|foo"),
IsNegative: false,
IsRegexp: true,
},
{
Key: []byte("\xfe\x03defface"),
Value: []byte("air"),
IsNegative: false,
IsRegexp: false,
},
},
})
// Multiple values regexp filter with a single negative filter
f([]TagFilter{
{
Key: nil,
Value: []byte("bar|foo"),
IsNegative: false,
IsRegexp: true,
},
{
Key: []byte("foo"),
Value: []byte("abc"),
IsNegative: true,
IsRegexp: false,
},
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar|foo"),
IsNegative: false,
IsRegexp: true,
},
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
IsNegative: true,
IsRegexp: false,
},
},
{
{
Key: nil,
Value: []byte("bar|foo"),
IsNegative: false,
IsRegexp: true,
},
{
Key: []byte("\xfe\x03foofoo"),
Value: []byte("abc"),
IsNegative: true,
IsRegexp: false,
},
},
})
// Regexp name filter with non-name filter.
@ -394,7 +553,8 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: true,
IsRegexp: false,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar.+"),
@ -407,6 +567,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: true,
IsRegexp: false,
},
},
})
// Regexp non-name filter, which matches anything.
@ -423,13 +584,15 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
IsNegative: false,
IsRegexp: true,
},
}, []TagFilter{
}, [][]TagFilter{
{
{
Key: nil,
Value: []byte("bar"),
IsNegative: false,
IsRegexp: false,
},
},
})
}