Metrics filters: allow OR on same key

Fix issue where several metrics filters targetting the same key would
not be ORed, which should be an expected behaviour.

This is necessary for filters such as "SubnetLabel is absent OR starts
with EXT-", which is a pattern that should become pretty common.

Author: jotak

pkg/pipeline/encode/metrics/filtering.go

@@ -4,15 +4,32 @@ import "github.com/netobserv/flowlogs-pipeline/pkg/config"
 
 func (p *Preprocessed) ApplyFilters(flow config.GenericMap, flatParts []config.GenericMap) (bool, []config.GenericMap) {
 	filteredParts := flatParts
-	for _, filter := range p.filters {
-		if filter.useFlat {
-			filteredParts = filter.filterFlatParts(filteredParts)
-			if len(filteredParts) == 0 {
-				return false, nil
+	// For a given key, all related filters are OR'ed
+	for _, filtersPerKey := range p.filters {
+		allFailed := true
+		for _, filter := range filtersPerKey {
+			passed, nfp := applySingleFilter(flow, &filter, filteredParts)
+			if passed {
+				allFailed = false
+				filteredParts = nfp
+				break
 			}
-		} else if !filter.predicate(flow) {
+		}
+		if allFailed {
+			return false, nil
+		}
+	}
+	return true, filteredParts
+}
+
+func applySingleFilter(flow config.GenericMap, filter *preprocessedFilter, filteredParts []config.GenericMap) (bool, []config.GenericMap) {
+	if filter.useFlat {
+		filteredParts = filter.filterFlatParts(filteredParts)
+		if len(filteredParts) == 0 {
 			return false, nil
 		}
+	} else if !filter.predicate(flow) {
+		return false, nil
 	}
 	return true, filteredParts
 }

pkg/pipeline/encode/metrics/preprocess.go

@@ -10,7 +10,7 @@ import (
 
 type Preprocessed struct {
 	*api.MetricsItem
-	filters         []preprocessedFilter
+	filters         map[string][]preprocessedFilter
 	MappedLabels    []MappedLabel
 	FlattenedLabels []MappedLabel
 }
@@ -60,6 +60,7 @@ func filterToPredicate(filter api.MetricsFilter) filters.Predicate {
 func Preprocess(def *api.MetricsItem) *Preprocessed {
 	mi := Preprocessed{
 		MetricsItem: def,
+		filters:     make(map[string][]preprocessedFilter),
 	}
 	for _, l := range def.Labels {
 		ml := MappedLabel{Source: l, Target: l}
@@ -73,7 +74,7 @@ func Preprocess(def *api.MetricsItem) *Preprocessed {
 		}
 	}
 	for _, f := range def.Filters {
-		mi.filters = append(mi.filters, preprocessedFilter{
+		mi.filters[f.Key] = append(mi.filters[f.Key], preprocessedFilter{
 			predicate: filterToPredicate(f),
 			useFlat:   mi.isFlattened(f.Key),
 		})

pkg/pipeline/encode/metrics/preprocess_test.go

@@ -42,3 +42,24 @@ func Test_Flatten(t *testing.T) {
 		},
 	}, fl)
 }
+
+func Test_ORedFilters(t *testing.T) {
+	// Several filters on the same key are ORed
+	pp := Preprocess(&api.MetricsItem{Filters: []api.MetricsFilter{
+		{
+			Key:  "label",
+			Type: api.MetricFilterAbsence,
+		},
+		{
+			Key:   "label",
+			Type:  api.MetricFilterRegex,
+			Value: "^EXT-.*",
+		},
+	}})
+	keep, _ := pp.ApplyFilters(config.GenericMap{"namespace": "A", "bytes": 7, "label": "Something"}, nil)
+	assert.False(t, keep)
+	keep, _ = pp.ApplyFilters(config.GenericMap{"namespace": "A", "bytes": 7, "label": "EXT-Something"}, nil)
+	assert.True(t, keep)
+	keep, _ = pp.ApplyFilters(config.GenericMap{"namespace": "A", "bytes": 7}, nil)
+	assert.True(t, keep)
+}