Share filter predicate code (prom+transform)

Move prom-encode predicates filtering code to its own package and share
it with "keep_entry" transforms

Author: jotak

pkg/api/transform_filter.go

@@ -17,23 +17,14 @@
 
 package api
 
-import (
-	"errors"
-	"regexp"
-)
-
 type TransformFilter struct {
 	Rules []TransformFilterRule `yaml:"rules,omitempty" json:"rules,omitempty" doc:"list of filter rules, each includes:"`
 }
 
-func (tf *TransformFilter) Preprocess() error {
-	var errs []error
+func (tf *TransformFilter) Preprocess() {
 	for i := range tf.Rules {
-		if err := tf.Rules[i].preprocess(); err != nil {
-			errs = append(errs, err)
-		}
+		tf.Rules[i].preprocess()
 	}
-	return errors.Join(errs...)
 }
 
 type TransformFilterEnum string
@@ -92,35 +83,22 @@ type TransformFilterRule struct {
 	ConditionalSampling     []*SamplingCondition             `yaml:"conditionalSampling,omitempty" json:"conditionalSampling,omitempty" doc:"sampling configuration rules"`
 }
 
-func (r *TransformFilterRule) preprocess() error {
-	var errs []error
+func (r *TransformFilterRule) preprocess() {
 	if r.RemoveField != nil {
-		if err := r.RemoveField.preprocess(false); err != nil {
-			errs = append(errs, err)
-		}
+		r.RemoveField.preprocess()
 	}
 	if r.RemoveEntry != nil {
-		if err := r.RemoveEntry.preprocess(false); err != nil {
-			errs = append(errs, err)
-		}
+		r.RemoveEntry.preprocess()
 	}
 	for i := range r.RemoveEntryAllSatisfied {
-		if err := r.RemoveEntryAllSatisfied[i].RemoveEntry.preprocess(false); err != nil {
-			errs = append(errs, err)
-		}
+		r.RemoveEntryAllSatisfied[i].RemoveEntry.preprocess()
 	}
 	for i := range r.KeepEntryAllSatisfied {
-		err := r.KeepEntryAllSatisfied[i].KeepEntry.preprocess(r.KeepEntryAllSatisfied[i].Type == KeepEntryIfRegexMatch || r.KeepEntryAllSatisfied[i].Type == KeepEntryIfNotRegexMatch)
-		if err != nil {
-			errs = append(errs, err)
-		}
+		r.KeepEntryAllSatisfied[i].KeepEntry.preprocess()
 	}
 	for i := range r.ConditionalSampling {
-		if err := r.ConditionalSampling[i].preprocess(); err != nil {
-			errs = append(errs, err)
-		}
+		r.ConditionalSampling[i].preprocess()
 	}
-	return errors.Join(errs...)
 }
 
 type TransformFilterGenericRule struct {
@@ -129,25 +107,12 @@ type TransformFilterGenericRule struct {
 	CastInt bool        `yaml:"castInt,omitempty" json:"castInt,omitempty" doc:"set true to cast the value field as an int (numeric values are float64 otherwise)"`
 }
 
-func (r *TransformFilterGenericRule) preprocess(isRegex bool) error {
-	if isRegex {
-		if s, ok := r.Value.(string); ok {
-			v, err := regexp.Compile(s)
-			if err != nil {
-				r.Value = nil
-				return err
-			}
-			r.Value = v
-		} else {
-			r.Value = nil
-			return errors.New("regex filter expects string value")
-		}
-	} else if r.CastInt {
+func (r *TransformFilterGenericRule) preprocess() {
+	if r.CastInt {
 		if f, ok := r.Value.(float64); ok {
 			r.Value = int(f)
 		}
 	}
-	return nil
 }
 
 type TransformFilterRuleWithAssignee struct {
@@ -172,12 +137,8 @@ type SamplingCondition struct {
 	Rules []*RemoveEntryRule `yaml:"rules,omitempty" json:"rules,omitempty" doc:"rules to be satisfied for this sampling configuration"`
 }
 
-func (s *SamplingCondition) preprocess() error {
-	var errs []error
+func (s *SamplingCondition) preprocess() {
 	for i := range s.Rules {
-		if err := s.Rules[i].RemoveEntry.preprocess(false); err != nil {
-			errs = append(errs, err)
-		}
+		s.Rules[i].RemoveEntry.preprocess()
 	}
-	return errors.Join(errs...)
 }

pkg/pipeline/encode/metrics/preprocess.go

@@ -8,12 +8,9 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
+	"github.com/netobserv/flowlogs-pipeline/pkg/utils/filters"
 )
 
-type Predicate func(config.GenericMap) bool
-
-var variableExtractor = regexp.MustCompile(`\$\(([^\)]+)\)`)
-
 type Preprocessed struct {
 	*api.MetricsItem
 	filters         []preprocessedFilter
@@ -99,17 +96,19 @@ func NotRegex(filter api.MetricsFilter) Predicate {
 func filterToPredicate(filter api.MetricsFilter) Predicate {
 	switch filter.Type {
 	case api.MetricFilterEqual:
-		return Equal(filter)
+		return filters.Equal(filter.Key, filter.Value, true)
 	case api.MetricFilterNotEqual:
-		return NotEqual(filter)
+		return filters.NotEqual(filter.Key, filter.Value, true)
 	case api.MetricFilterPresence:
-		return Presence(filter)
+		return filters.Presence(filter.Key)
 	case api.MetricFilterAbsence:
-		return Absence(filter)
+		return filters.Absence(filter.Key)
 	case api.MetricFilterRegex:
-		return Regex(filter)
+		r, _ := regexp.Compile(filter.Value)
+		return filters.Regex(filter.Key, r)
 	case api.MetricFilterNotRegex:
-		return NotRegex(filter)
+		r, _ := regexp.Compile(filter.Value)
+		return filters.NotRegex(filter.Key, r)
 	}
 	// Default = Exact
 	return Equal(filter)

pkg/pipeline/transform/transform_filter.go

@@ -28,6 +28,7 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
+	"github.com/netobserv/flowlogs-pipeline/pkg/utils/filters"
 	"github.com/sirupsen/logrus"
 )
 
@@ -38,7 +39,12 @@ var (
 
 type Filter struct {
 	Rules     []api.TransformFilterRule
-	KeepRules []api.TransformFilterRule
+	KeepRules []predicatesRule
+}
+
+type predicatesRule struct {
+	predicates []filters.Predicate
+	sampling   uint16
 }
 
 // Transform transforms a flow; if false is returned as a second argument, the entry is dropped
@@ -48,9 +54,8 @@ func (f *Filter) Transform(entry config.GenericMap) (config.GenericMap, bool) {
 	labels := make(map[string]string)
 	if len(f.KeepRules) > 0 {
 		keep := false
-		for i := range f.KeepRules {
-			tlog.Tracef("keep rule = %v", f.KeepRules[i])
-			if applyRule(outputEntry, labels, &f.KeepRules[i]) {
+		for _, r := range f.KeepRules {
+			if applyPredicates(outputEntry, r) {
 				keep = true
 				break
 			}
@@ -158,7 +163,8 @@ func applyRule(entry config.GenericMap, labels map[string]string, rule *api.Tran
 	case api.ConditionalSampling:
 		return sample(entry, rule.ConditionalSampling)
 	case api.KeepEntryAllSatisfied:
-		return rollSampling(rule.KeepEntrySampling) && isKeepEntrySatisfied(entry, rule.KeepEntryAllSatisfied)
+		// This should be processed only in "applyPredicates". Failure to do so is a bug.
+		tlog.Panicf("unexpected KeepEntryAllSatisfied: %v", rule)
 	default:
 		tlog.Panicf("unknown type %s for transform.Filter rule: %v", rule.Type, rule)
 	}
@@ -175,58 +181,18 @@ func isRemoveEntrySatisfied(entry config.GenericMap, rules []*api.RemoveEntryRul
 	return true
 }
 
-func isKeepEntrySatisfied(entry config.GenericMap, rules []*api.KeepEntryRule) bool {
-	for _, r := range rules {
-		val, ok := entry[r.KeepEntry.Input]
-		switch r.Type {
-		case api.KeepEntryIfExists:
-			if !ok {
-				return false
-			}
-		case api.KeepEntryIfDoesntExist:
-			if ok {
-				return false
-			}
-		case api.KeepEntryIfEqual:
-			if !ok || val != r.KeepEntry.Value {
-				return false
-			}
-		case api.KeepEntryIfNotEqual:
-			if ok && val == r.KeepEntry.Value {
-				return false
-			}
-		case api.KeepEntryIfRegexMatch:
-			if ok {
-				match, ok := checkRegex(r.KeepEntry.Value, val)
-				if !ok || !match {
-					return false
-				}
-			} else {
-				return false
-			}
-		case api.KeepEntryIfNotRegexMatch:
-			if ok {
-				match, ok := checkRegex(r.KeepEntry.Value, val)
-				if !ok || match {
-					return false
-				}
-			} else {
-				return false
-			}
+func applyPredicates(entry config.GenericMap, rule predicatesRule) bool {
+	if !rollSampling(rule.sampling) {
+		return false
+	}
+	for _, p := range rule.predicates {
+		if !p(entry) {
+			return false
 		}
 	}
 	return true
 }
 
-// Returns (valid, match)
-func checkRegex(maybeReg any, value any) (bool, bool) {
-	reg, ok := maybeReg.(*regexp.Regexp)
-	if !ok {
-		return false, false
-	}
-	return true, reg.MatchString(utils.ConvertToString(value))
-}
-
 func sample(entry config.GenericMap, rules []*api.SamplingCondition) bool {
 	for _, r := range rules {
 		if isRemoveEntrySatisfied(entry, r.Rules) {
@@ -243,17 +209,24 @@ func rollSampling(value uint16) bool {
 // NewTransformFilter create a new filter transform
 func NewTransformFilter(params config.StageParam) (Transformer, error) {
 	tlog.Debugf("entering NewTransformFilter")
-	keepRules := []api.TransformFilterRule{}
+	keepRules := []predicatesRule{}
 	rules := []api.TransformFilterRule{}
 	if params.Transform != nil && params.Transform.Filter != nil {
-		if err := params.Transform.Filter.Preprocess(); err != nil {
-			return nil, err
-		}
+		params.Transform.Filter.Preprocess()
 		for i := range params.Transform.Filter.Rules {
-			if params.Transform.Filter.Rules[i].Type == api.KeepEntryAllSatisfied {
-				keepRules = append(keepRules, params.Transform.Filter.Rules[i])
+			baseRules := &params.Transform.Filter.Rules[i]
+			if baseRules.Type == api.KeepEntryAllSatisfied {
+				pr := predicatesRule{sampling: baseRules.KeepEntrySampling}
+				for _, keepRule := range baseRules.KeepEntryAllSatisfied {
+					pred, err := filters.FromKeepEntry(keepRule)
+					if err != nil {
+						return nil, err
+					}
+					pr.predicates = append(pr.predicates, pred)
+				}
+				keepRules = append(keepRules, pr)
 			} else {
-				rules = append(rules, params.Transform.Filter.Rules[i])
+				rules = append(rules, *baseRules)
 			}
 		}
 	}

pkg/utils/convert.go

@@ -246,9 +246,9 @@ func ConvertToBool(unk interface{}) (bool, error) {
 func ConvertToString(unk interface{}) string {
 	switch i := unk.(type) {
 	case float64:
-		return strconv.FormatFloat(i, 'E', -1, 64)
+		return strconv.FormatFloat(i, 'f', -1, 64)
 	case float32:
-		return strconv.FormatFloat(float64(i), 'E', -1, 32)
+		return strconv.FormatFloat(float64(i), 'f', -1, 32)
 	case int64:
 		return strconv.FormatInt(i, 10)
 	case int32: