lib/storage: properly handle queries containing a filter on metric name plus any number of negative filters and zero non-negative filters

Example: `node_cpu_seconds_total{mode!="idle"}`
This commit is contained in:
Aliaksandr Valialkin 2021-02-18 18:32:33 +02:00
parent 902a4f6486
commit 418de71509
3 changed files with 84 additions and 12 deletions

View file

@ -2,6 +2,8 @@
# tip # tip
* BUGFIX: properly handle queries containing a filter on metric name plus any number of negative filters and zero non-negative filters. For example, `node_cpu_seconds_total{mode!="idle"}`. The bug was introduced in [v1.54.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.54.0).
# [v1.54.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.54.0) # [v1.54.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.54.0)

View file

@ -36,13 +36,16 @@ func convertToCompositeTagFilters(tfs *TagFilters) *TagFilters {
} }
} }
if len(name) == 0 { if len(name) == 0 {
// There is no metric name filter, so composite filters cannot be created. // Composite filters cannot be created in the following cases:
// - if there is no filter on metric name
// - if there is no at least a single positive filter.
atomic.AddUint64(&compositeFilterMissingConversions, 1) atomic.AddUint64(&compositeFilterMissingConversions, 1)
return tfs return tfs
} }
tfsNew := make([]tagFilter, 0, len(tfs.tfs)) tfsNew := make([]tagFilter, 0, len(tfs.tfs))
var compositeKey []byte var compositeKey []byte
compositeFilters := 0 compositeFilters := 0
hasPositiveFilter := false
for _, tf := range tfs.tfs { for _, tf := range tfs.tfs {
if len(tf.key) == 0 { if len(tf.key) == 0 {
if tf.isNegative || tf.isRegexp || string(tf.value) != string(name) { if tf.isNegative || tf.isRegexp || string(tf.value) != string(name) {
@ -59,10 +62,13 @@ func convertToCompositeTagFilters(tfs *TagFilters) *TagFilters {
if err := tfNew.Init(tfs.commonPrefix, compositeKey, tf.value, tf.isNegative, tf.isRegexp); err != nil { if err := tfNew.Init(tfs.commonPrefix, compositeKey, tf.value, tf.isNegative, tf.isRegexp); err != nil {
logger.Panicf("BUG: unexpected error when creating composite tag filter for name=%q and key=%q: %s", name, tf.key, err) logger.Panicf("BUG: unexpected error when creating composite tag filter for name=%q and key=%q: %s", name, tf.key, err)
} }
if !tfNew.isNegative {
hasPositiveFilter = true
}
tfsNew = append(tfsNew, tfNew) tfsNew = append(tfsNew, tfNew)
compositeFilters++ compositeFilters++
} }
if compositeFilters == 0 { if compositeFilters == 0 || !hasPositiveFilter {
atomic.AddUint64(&compositeFilterMissingConversions, 1) atomic.AddUint64(&compositeFilterMissingConversions, 1)
return tfs return tfs
} }

View file

@ -156,6 +156,70 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
}, },
}) })
// A name filter with a single negative filter
f([]TagFilter{
{
Key: nil,
Value: []byte("bar"),
IsNegative: false,
IsRegexp: false,
},
{
Key: []byte("foo"),
Value: []byte("abc"),
IsNegative: true,
IsRegexp: false,
},
}, []TagFilter{
{
Key: nil,
Value: []byte("bar"),
IsNegative: false,
IsRegexp: false,
},
{
Key: []byte("foo"),
Value: []byte("abc"),
IsNegative: true,
IsRegexp: false,
},
})
// A name filter with a negative and a positive filter
f([]TagFilter{
{
Key: nil,
Value: []byte("bar"),
IsNegative: false,
IsRegexp: false,
},
{
Key: []byte("foo"),
Value: []byte("abc"),
IsNegative: true,
IsRegexp: false,
},
{
Key: []byte("a"),
Value: []byte("b.+"),
IsNegative: false,
IsRegexp: true,
},
}, []TagFilter{
{
Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"),
IsNegative: true,
IsRegexp: false,
},
{
Key: []byte("\xfe\x03bara"),
Value: []byte("b.+"),
IsNegative: false,
IsRegexp: true,
},
})
// Two name filters with non-name filter. // Two name filters with non-name filter.
f([]TagFilter{ f([]TagFilter{
{ {
@ -191,7 +255,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
}, },
}) })
// A name filter with negative regexp non-name filter, which can be converted to non-regexp. // A name filter with regexp non-name filter, which can be converted to non-regexp.
f([]TagFilter{ f([]TagFilter{
{ {
Key: nil, Key: nil,
@ -202,19 +266,19 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
{ {
Key: []byte("foo"), Key: []byte("foo"),
Value: []byte("abc"), Value: []byte("abc"),
IsNegative: true, IsNegative: false,
IsRegexp: true, IsRegexp: true,
}, },
}, []TagFilter{ }, []TagFilter{
{ {
Key: []byte("\xfe\x03barfoo"), Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"), Value: []byte("abc"),
IsNegative: true, IsNegative: false,
IsRegexp: false, IsRegexp: false,
}, },
}) })
// A name filter with negative regexp non-name filter. // A name filter with regexp non-name filter.
f([]TagFilter{ f([]TagFilter{
{ {
Key: nil, Key: nil,
@ -225,14 +289,14 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
{ {
Key: []byte("foo"), Key: []byte("foo"),
Value: []byte("abc.+"), Value: []byte("abc.+"),
IsNegative: true, IsNegative: false,
IsRegexp: true, IsRegexp: true,
}, },
}, []TagFilter{ }, []TagFilter{
{ {
Key: []byte("\xfe\x03barfoo"), Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc.+"), Value: []byte("abc.+"),
IsNegative: true, IsNegative: false,
IsRegexp: true, IsRegexp: true,
}, },
}) })
@ -277,7 +341,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
{ {
Key: []byte("foo"), Key: []byte("foo"),
Value: []byte("abc"), Value: []byte("abc"),
IsNegative: true, IsNegative: false,
IsRegexp: true, IsRegexp: true,
}, },
{ {
@ -290,7 +354,7 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
{ {
Key: []byte("\xfe\x03barfoo"), Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"), Value: []byte("abc"),
IsNegative: true, IsNegative: false,
IsRegexp: false, IsRegexp: false,
}, },
{ {
@ -312,14 +376,14 @@ func TestConvertToCompositeTagFilters(t *testing.T) {
{ {
Key: []byte("foo"), Key: []byte("foo"),
Value: []byte("abc"), Value: []byte("abc"),
IsNegative: true, IsNegative: false,
IsRegexp: false, IsRegexp: false,
}, },
}, []TagFilter{ }, []TagFilter{
{ {
Key: []byte("\xfe\x03barfoo"), Key: []byte("\xfe\x03barfoo"),
Value: []byte("abc"), Value: []byte("abc"),
IsNegative: true, IsNegative: false,
IsRegexp: false, IsRegexp: false,
}, },
}) })