NETOBSERV-1425: enhance metrics filters

docs/api.md

@@ -20,19 +20,23 @@ Following is the supported API format for prometheus encode:
                  filter: an optional criterion to filter entries by. Deprecated: use filters instead.
                      key: the key to match and filter by
                      value: the value to match and filter by
-                     type: (enum) the type of filter match: exact (default), presence, absence or regex
-                         exact: match exactly the provided fitler value
+                     type: (enum) the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex
+                         equal: match exactly the provided filter value
+                         not_equal: the value must be different from the provided filter
                          presence: filter key must be present (filter value is ignored)
                          absence: filter key must be absent (filter value is ignored)
-                         regex: match filter value as a regular expression
+                         match_regex: match filter value as a regular expression
+                         not_match_regex: the filter value must not match the provided regular expression
                  filters: a list of criteria to filter entries by
                          key: the key to match and filter by
                          value: the value to match and filter by
-                         type: (enum) the type of filter match: exact (default), presence, absence or regex
-                             exact: match exactly the provided fitler value
+                         type: (enum) the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex
+                             equal: match exactly the provided filter value
+                             not_equal: the value must be different from the provided filter
                              presence: filter key must be present (filter value is ignored)
                              absence: filter key must be absent (filter value is ignored)
-                             regex: match filter value as a regular expression
+                             match_regex: match filter value as a regular expression
+                             not_match_regex: the filter value must not match the provided regular expression
                  valueKey: entry key from which to resolve metric value
                  labels: labels to be associated with the metric
                  buckets: histogram buckets
@@ -353,19 +357,23 @@ Following is the supported API format for writing metrics to an OpenTelemetry co
                  filter: an optional criterion to filter entries by. Deprecated: use filters instead.
                      key: the key to match and filter by
                      value: the value to match and filter by
-                     type: (enum) the type of filter match: exact (default), presence, absence or regex
-                         exact: match exactly the provided fitler value
+                     type: (enum) the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex
+                         equal: match exactly the provided filter value
+                         not_equal: the value must be different from the provided filter
                          presence: filter key must be present (filter value is ignored)
                          absence: filter key must be absent (filter value is ignored)
-                         regex: match filter value as a regular expression
+                         match_regex: match filter value as a regular expression
+                         not_match_regex: the filter value must not match the provided regular expression
                  filters: a list of criteria to filter entries by
                          key: the key to match and filter by
                          value: the value to match and filter by
-                         type: (enum) the type of filter match: exact (default), presence, absence or regex
-                             exact: match exactly the provided fitler value
+                         type: (enum) the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex
+                             equal: match exactly the provided filter value
+                             not_equal: the value must be different from the provided filter
                              presence: filter key must be present (filter value is ignored)
                              absence: filter key must be absent (filter value is ignored)
-                             regex: match filter value as a regular expression
+                             match_regex: match filter value as a regular expression
+                             not_match_regex: the filter value must not match the provided regular expression
                  valueKey: entry key from which to resolve metric value
                  labels: labels to be associated with the metric
                  buckets: histogram buckets

pkg/api/api.go

@@ -50,10 +50,12 @@ const (
 	AddKubernetesRuleType        = "add_kubernetes"
 	AddKubernetesInfraRuleType   = "add_kubernetes_infra"
 	ReinterpretDirectionRuleType = "reinterpret_direction"
-	PromFilterExact              = "exact"
+	PromFilterEqual              = "equal"
+	PromFilterNotEqual           = "not_equal"
 	PromFilterPresence           = "presence"
 	PromFilterAbsence            = "absence"
-	PromFilterRegex              = "regex"
+	PromFilterRegex              = "match_regex"
+	PromFilterNotRegex           = "not_match_regex"
 
 	TagYaml = "yaml"
 	TagDoc  = "doc"

pkg/api/encode_prom.go

@@ -70,14 +70,16 @@ type MetricsItems []MetricsItem
 type MetricsFilter struct {
 	Key   string `yaml:"key" json:"key" doc:"the key to match and filter by"`
 	Value string `yaml:"value" json:"value" doc:"the value to match and filter by"`
-	Type  string `yaml:"type" json:"type" enum:"MetricEncodeFilterTypeEnum" doc:"the type of filter match: exact (default), presence, absence or regex"`
+	Type  string `yaml:"type" json:"type" enum:"MetricEncodeFilterTypeEnum" doc:"the type of filter match: equal (default), not_equal, presence, absence, match_regex or not_match_regex"`
 }
 
 type MetricEncodeFilterTypeEnum struct {
-	Exact    string `yaml:"exact" json:"exact" doc:"match exactly the provided fitler value"`
-	Presence string `yaml:"presence" json:"presence" doc:"filter key must be present (filter value is ignored)"`
-	Absence  string `yaml:"absence" json:"absence" doc:"filter key must be absent (filter value is ignored)"`
-	Regex    string `yaml:"regex" json:"regex" doc:"match filter value as a regular expression"`
+	Equal         string `yaml:"equal" json:"equal" doc:"match exactly the provided filter value"`
+	NotEqual      string `yaml:"not_equal" json:"not_equal" doc:"the value must be different from the provided filter"`
+	Presence      string `yaml:"presence" json:"presence" doc:"filter key must be present (filter value is ignored)"`
+	Absence       string `yaml:"absence" json:"absence" doc:"filter key must be absent (filter value is ignored)"`
+	MatchRegex    string `yaml:"match_regex" json:"match_regex" doc:"match filter value as a regular expression"`
+	NotMatchRegex string `yaml:"not_match_regex" json:"not_match_regex" doc:"the filter value must not match the provided regular expression"`
 }
 
 func MetricEncodeFilterTypeName(t string) string {

pkg/pipeline/encode/encode_prom_metric.go

@@ -3,13 +3,16 @@ package encode
 import (
 	"fmt"
 	"regexp"
+	"strings"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 )
 
 type Predicate func(flow config.GenericMap) bool
 
+var variableExtractor, _ = regexp.Compile(`\$\(([^\)]+)\)`)
+
 type MetricInfo struct {
 	api.MetricsItem
 	FilterPredicates []Predicate
@@ -29,20 +32,30 @@ func Absence(filter api.MetricsFilter) Predicate {
 	}
 }
 
-func Exact(filter api.MetricsFilter) Predicate {
+func Equal(filter api.MetricsFilter) Predicate {
+	varLookups := extractVarLookups(filter.Value)
 	return func(flow config.GenericMap) bool {
 		if val, found := flow[filter.Key]; found {
 			sVal, ok := val.(string)
 			if !ok {
 				sVal = fmt.Sprint(val)
 			}
-			return sVal == filter.Value
+			value := filter.Value
+			if len(varLookups) > 0 {
+				value = injectVars(flow, value, varLookups)
+			}
+			return sVal == value
 		}
 		return false
 	}
 }
 
-func regex(filter api.MetricsFilter) Predicate {
+func NotEqual(filter api.MetricsFilter) Predicate {
+	pred := Equal(filter)
+	return func(flow config.GenericMap) bool { return !pred(flow) }
+}
+
+func Regex(filter api.MetricsFilter) Predicate {
 	r, _ := regexp.Compile(filter.Value)
 	return func(flow config.GenericMap) bool {
 		if val, found := flow[filter.Key]; found {
@@ -56,19 +69,53 @@ func regex(filter api.MetricsFilter) Predicate {
 	}
 }
 
+func NotRegex(filter api.MetricsFilter) Predicate {
+	pred := Regex(filter)
+	return func(flow config.GenericMap) bool { return !pred(flow) }
+}
+
 func filterToPredicate(filter api.MetricsFilter) Predicate {
 	switch filter.Type {
-	case api.PromFilterExact:
-		return Exact(filter)
+	case api.PromFilterEqual:
+		return Equal(filter)
+	case api.PromFilterNotEqual:
+		return NotEqual(filter)
 	case api.PromFilterPresence:
 		return Presence(filter)
 	case api.PromFilterAbsence:
 		return Absence(filter)
 	case api.PromFilterRegex:
-		return regex(filter)
+		return Regex(filter)
+	case api.PromFilterNotRegex:
+		return NotRegex(filter)
 	}
 	// Default = Exact
-	return Exact(filter)
+	return Equal(filter)
+}
+
+func extractVarLookups(value string) [][]string {
+	// Extract list of variables to lookup
+	// E.g: filter "$(SrcAddr):$(SrcPort)" would return [SrcAddr,SrcPort]
+	if len(value) > 0 {
+		return variableExtractor.FindAllStringSubmatch(value, -1)
+	}
+	return nil
+}
+
+func injectVars(flow config.GenericMap, filterValue string, varLookups [][]string) string {
+	injected := filterValue
+	for _, matchGroup := range varLookups {
+		var value string
+		if rawVal, found := flow[matchGroup[1]]; found {
+			if sVal, ok := rawVal.(string); ok {
+				value = sVal
+			} else {
+				value = fmt.Sprint(rawVal)
+			}
+		}
+		injected = strings.ReplaceAll(injected, matchGroup[0], value)
+	}
+	return injected
 }
 
 func CreateMetricInfo(def api.MetricsItem) *MetricInfo {

pkg/pipeline/encode/encode_prom_test.go

@@ -361,7 +361,7 @@ func Test_FilterDirection(t *testing.T) {
 				Name:     "ingress_or_inner_packets_total",
 				Type:     "counter",
 				ValueKey: "packets",
-				Filters:  []api.MetricsFilter{{Key: "dir", Value: "0|2", Type: "regex"}},
+				Filters:  []api.MetricsFilter{{Key: "dir", Value: "0|2", Type: "match_regex"}},
 			},
 		},
 	}
@@ -380,6 +380,63 @@ func Test_FilterDirection(t *testing.T) {
 	require.Contains(t, exposed, `test_ingress_or_inner_packets_total 1010`)
 }
 
+func Test_FilterSameOrDifferentNamespace(t *testing.T) {
+	metrics := []config.GenericMap{
+		{
+			"src-ns":  "a",
+			"dst-ns":  "b",
+			"packets": 10,
+		},
+		{
+			"src-ns":  "b",
+			"dst-ns":  "a",
+			"packets": 200,
+		},
+		{
+			"src-ns":  "a",
+			"dst-ns":  "a",
+			"packets": 3000,
+		},
+		{
+			"src-ns":  "b",
+			"dst-ns":  "b",
+			"packets": 40000,
+		},
+	}
+	params := api.PromEncode{
+		Prefix: "test_",
+		ExpiryTime: api.Duration{
+			Duration: time.Duration(60 * time.Second),
+		},
+		Metrics: []api.MetricsItem{
+			{
+				Name:     "packets_same_namespace_total",
+				Type:     "counter",
+				ValueKey: "packets",
+				Filters:  []api.MetricsFilter{{Key: "src-ns", Value: "$(dst-ns)"}},
+			},
+			{
+				Name:     "packets_different_namespace_total",
+				Type:     "counter",
+				ValueKey: "packets",
+				Filters:  []api.MetricsFilter{{Key: "src-ns", Value: "$(dst-ns)", Type: "not_equal"}},
+			},
+		},
+	}
+
+	encodeProm, err := initProm(&params)
+	require.NoError(t, err)
+	for _, metric := range metrics {
+		encodeProm.Encode(metric)
+	}
+	time.Sleep(100 * time.Millisecond)
+
+	exposed := test.ReadExposedMetrics(t)
+
+	require.Contains(t, exposed, `test_packets_same_namespace_total 43000`)
+	require.Contains(t, exposed, `test_packets_different_namespace_total 210`)
+}
+
 func Test_ValueScale(t *testing.T) {
 	metrics := []config.GenericMap{{"rtt": 15_000_000} /*15ms*/, {"rtt": 110_000_000} /*110ms*/}
 	params := api.PromEncode{
@@ -650,3 +707,12 @@ func Test_MultipleProm(t *testing.T) {
 
 	// TODO: Add test for different addresses, but need to deal with StartPromServer (ListenAndServe)
 }
+
+func Test_Filters_extractVarLookups(t *testing.T) {
+	variables := extractVarLookups("$(abc)--$(def)")
+
+	require.Equal(t, [][]string{{"$(abc)", "abc"}, {"$(def)", "def"}}, variables)
+
+	variables = extractVarLookups("")
+	require.Empty(t, variables)
+}