Replace keep_entry API with query API

Author: jotak

docs/api.md

@@ -162,7 +162,7 @@ Following is the supported API format for filter transformations:
                     remove_entry_if_equal: removes the entry if the field value equals specified value
                     remove_entry_if_not_equal: removes the entry if the field value does not equal specified value
                     remove_entry_all_satisfied: removes the entry if all of the defined rules are satisfied
-                    keep_entry_all_satisfied: keeps the entry if the set of rules are all satisfied
+                    keep_entry_query: keeps the entry if it matches the query
                     add_field: adds (input) field to the entry; overrides previous value if present (key=input, value=value)
                     add_field_if_doesnt_exist: adds a field to the entry if the field does not exist
                     add_field_if: add output field set to assignee if input field satisfies criteria from parameters field
@@ -188,18 +188,7 @@ Following is the supported API format for filter transformations:
                              input: entry input field
                              value: specified value of input field:
                              castInt: set true to cast the value field as an int (numeric values are float64 otherwise)
-                 keepEntryAllSatisfied: configuration for keep_entry rule
-                         type: (enum) one of the following:
-                            keep_entry_if_exists: keeps the entry if the field exists
-                            keep_entry_if_doesnt_exist: keeps the entry if the field does not exist
-                            keep_entry_if_equal: keeps the entry if the field value equals specified value
-                            keep_entry_if_not_equal: keeps the entry if the field value does not equal specified value
-                            keep_entry_if_regex_match: keeps the entry if the field value matches the specified regex
-                            keep_entry_if_not_regex_match: keeps the entry if the field value does not match the specified regex
-                         keepEntry: configuration for keep_entry_* rules
-                             input: entry input field
-                             value: specified value of input field:
-                             castInt: set true to cast the value field as an int (numeric values are float64 otherwise)
+                 keepEntryQuery: configuration for keep_entry rule
                  keepEntrySampling: sampling value for keep_entry type: 1 flow on <sampling> is kept
                  addField: configuration for add_field rule
                      input: entry input field

pkg/api/transform_filter.go

@@ -37,7 +37,7 @@ const (
 	RemoveEntryIfEqual       TransformFilterEnum = "remove_entry_if_equal"        // removes the entry if the field value equals specified value
 	RemoveEntryIfNotEqual    TransformFilterEnum = "remove_entry_if_not_equal"    // removes the entry if the field value does not equal specified value
 	RemoveEntryAllSatisfied  TransformFilterEnum = "remove_entry_all_satisfied"   // removes the entry if all of the defined rules are satisfied
-	KeepEntryAllSatisfied    TransformFilterEnum = "keep_entry_all_satisfied"     // keeps the entry if the set of rules are all satisfied
+	KeepEntryQuery           TransformFilterEnum = "keep_entry_query"             // keeps the entry if it matches the query
 	AddField                 TransformFilterEnum = "add_field"                    // adds (input) field to the entry; overrides previous value if present (key=input, value=value)
 	AddFieldIfDoesntExist    TransformFilterEnum = "add_field_if_doesnt_exist"    // adds a field to the entry if the field does not exist
 	AddFieldIf               TransformFilterEnum = "add_field_if"                 // add output field set to assignee if input field satisfies criteria from parameters field
@@ -56,23 +56,12 @@ const (
 	RemoveEntryIfNotEqualD    TransformFilterRemoveEntryEnum = "remove_entry_if_not_equal"    // removes the entry if the field value does not equal specified value
 )
 
-type TransformFilterKeepEntryEnum string
-
-const (
-	KeepEntryIfExists        TransformFilterKeepEntryEnum = "keep_entry_if_exists"          // keeps the entry if the field exists
-	KeepEntryIfDoesntExist   TransformFilterKeepEntryEnum = "keep_entry_if_doesnt_exist"    // keeps the entry if the field does not exist
-	KeepEntryIfEqual         TransformFilterKeepEntryEnum = "keep_entry_if_equal"           // keeps the entry if the field value equals specified value
-	KeepEntryIfNotEqual      TransformFilterKeepEntryEnum = "keep_entry_if_not_equal"       // keeps the entry if the field value does not equal specified value
-	KeepEntryIfRegexMatch    TransformFilterKeepEntryEnum = "keep_entry_if_regex_match"     // keeps the entry if the field value matches the specified regex
-	KeepEntryIfNotRegexMatch TransformFilterKeepEntryEnum = "keep_entry_if_not_regex_match" // keeps the entry if the field value does not match the specified regex
-)
-
 type TransformFilterRule struct {
 	Type                    TransformFilterEnum              `yaml:"type,omitempty" json:"type,omitempty" doc:"(enum) one of the following:"`
 	RemoveField             *TransformFilterGenericRule      `yaml:"removeField,omitempty" json:"removeField,omitempty" doc:"configuration for remove_field rule"`
 	RemoveEntry             *TransformFilterGenericRule      `yaml:"removeEntry,omitempty" json:"removeEntry,omitempty" doc:"configuration for remove_entry_* rules"`
 	RemoveEntryAllSatisfied []*RemoveEntryRule               `yaml:"removeEntryAllSatisfied,omitempty" json:"removeEntryAllSatisfied,omitempty" doc:"configuration for remove_entry_all_satisfied rule"`
-	KeepEntryAllSatisfied   []*KeepEntryRule                 `yaml:"keepEntryAllSatisfied,omitempty" json:"keepEntryAllSatisfied,omitempty" doc:"configuration for keep_entry rule"`
+	KeepEntryQuery          string                           `yaml:"keepEntryQuery,omitempty" json:"keepEntryQuery,omitempty" doc:"configuration for keep_entry rule"`
 	KeepEntrySampling       uint16                           `yaml:"keepEntrySampling,omitempty" json:"keepEntrySampling,omitempty" doc:"sampling value for keep_entry type: 1 flow on <sampling> is kept"`
 	AddField                *TransformFilterGenericRule      `yaml:"addField,omitempty" json:"addField,omitempty" doc:"configuration for add_field rule"`
 	AddFieldIfDoesntExist   *TransformFilterGenericRule      `yaml:"addFieldIfDoesntExist,omitempty" json:"addFieldIfDoesntExist,omitempty" doc:"configuration for add_field_if_doesnt_exist rule"`
@@ -93,9 +82,6 @@ func (r *TransformFilterRule) preprocess() {
 	for i := range r.RemoveEntryAllSatisfied {
 		r.RemoveEntryAllSatisfied[i].RemoveEntry.preprocess()
 	}
-	for i := range r.KeepEntryAllSatisfied {
-		r.KeepEntryAllSatisfied[i].KeepEntry.preprocess()
-	}
 	for i := range r.ConditionalSampling {
 		r.ConditionalSampling[i].preprocess()
 	}
@@ -127,11 +113,6 @@ type RemoveEntryRule struct {
 	RemoveEntry *TransformFilterGenericRule    `yaml:"removeEntry,omitempty" json:"removeEntry,omitempty" doc:"configuration for remove_entry_* rules"`
 }
 
-type KeepEntryRule struct {
-	Type      TransformFilterKeepEntryEnum `yaml:"type,omitempty" json:"type,omitempty" doc:"(enum) one of the following:"`
-	KeepEntry *TransformFilterGenericRule  `yaml:"keepEntry,omitempty" json:"keepEntry,omitempty" doc:"configuration for keep_entry_* rules"`
-}
-
 type SamplingCondition struct {
 	Value uint16             `yaml:"value,omitempty" json:"value,omitempty" doc:"sampling value: 1 flow on <sampling> is kept"`
 	Rules []*RemoveEntryRule `yaml:"rules,omitempty" json:"rules,omitempty" doc:"rules to be satisfied for this sampling configuration"`

pkg/pipeline/transform/transform_filter.go

@@ -27,6 +27,7 @@ import (
 	"github.com/Knetic/govaluate"
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
+	"github.com/netobserv/flowlogs-pipeline/pkg/dsl"
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils/filters"
 	"github.com/sirupsen/logrus"
@@ -39,12 +40,12 @@ var (
 
 type Filter struct {
 	Rules     []api.TransformFilterRule
-	KeepRules []predicatesRule
+	KeepRules []predicateRule
 }
 
-type predicatesRule struct {
-	predicates []filters.Predicate
-	sampling   uint16
+type predicateRule struct {
+	predicate filters.Predicate
+	sampling  uint16
 }
 
 // Transform transforms a flow; if false is returned as a second argument, the entry is dropped
@@ -55,7 +56,7 @@ func (f *Filter) Transform(entry config.GenericMap) (config.GenericMap, bool) {
 	if len(f.KeepRules) > 0 {
 		keep := false
 		for _, r := range f.KeepRules {
-			if applyPredicates(outputEntry, r) {
+			if applyPredicate(outputEntry, r) {
 				keep = true
 				break
 			}
@@ -162,9 +163,9 @@ func applyRule(entry config.GenericMap, labels map[string]string, rule *api.Tran
 		return !isRemoveEntrySatisfied(entry, rule.RemoveEntryAllSatisfied)
 	case api.ConditionalSampling:
 		return sample(entry, rule.ConditionalSampling)
-	case api.KeepEntryAllSatisfied:
+	case api.KeepEntryQuery:
 		// This should be processed only in "applyPredicates". Failure to do so is a bug.
-		tlog.Panicf("unexpected KeepEntryAllSatisfied: %v", rule)
+		tlog.Panicf("unexpected KeepEntryQuery: %v", rule)
 	default:
 		tlog.Panicf("unknown type %s for transform.Filter rule: %v", rule.Type, rule)
 	}
@@ -181,16 +182,11 @@ func isRemoveEntrySatisfied(entry config.GenericMap, rules []*api.RemoveEntryRul
 	return true
 }
 
-func applyPredicates(entry config.GenericMap, rule predicatesRule) bool {
+func applyPredicate(entry config.GenericMap, rule predicateRule) bool {
 	if !rollSampling(rule.sampling) {
 		return false
 	}
-	for _, p := range rule.predicates {
-		if !p(entry) {
-			return false
-		}
-	}
-	return true
+	return rule.predicate(entry)
 }
 
 func sample(entry config.GenericMap, rules []*api.SamplingCondition) bool {
@@ -209,22 +205,21 @@ func rollSampling(value uint16) bool {
 // NewTransformFilter create a new filter transform
 func NewTransformFilter(params config.StageParam) (Transformer, error) {
 	tlog.Debugf("entering NewTransformFilter")
-	keepRules := []predicatesRule{}
+	keepRules := []predicateRule{}
 	rules := []api.TransformFilterRule{}
 	if params.Transform != nil && params.Transform.Filter != nil {
 		params.Transform.Filter.Preprocess()
 		for i := range params.Transform.Filter.Rules {
 			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)
+			if baseRules.Type == api.KeepEntryQuery {
+				predicate, err := dsl.Parse(baseRules.KeepEntryQuery)
+				if err != nil {
+					return nil, err
 				}
-				keepRules = append(keepRules, pr)
+				keepRules = append(keepRules, predicateRule{
+					sampling:  baseRules.KeepEntrySampling,
+					predicate: predicate,
+				})
 			} else {
 				rules = append(rules, *baseRules)
 			}

pkg/pipeline/transform/transform_filter_test.go

@@ -676,34 +676,8 @@ func Test_Transform_KeepEntry(t *testing.T) {
 	newFilter := api.TransformFilter{
 		Rules: []api.TransformFilterRule{
 			{
-				Type: api.KeepEntryAllSatisfied,
-				KeepEntryAllSatisfied: []*api.KeepEntryRule{
-					{
-						Type: api.KeepEntryIfEqual,
-						KeepEntry: &api.TransformFilterGenericRule{
-							Input: "namespace",
-							Value: "A",
-						},
-					},
-					{
-						Type: api.KeepEntryIfExists,
-						KeepEntry: &api.TransformFilterGenericRule{
-							Input: "workload",
-						},
-					},
-				},
-			},
-			{
-				Type: api.KeepEntryAllSatisfied,
-				KeepEntryAllSatisfied: []*api.KeepEntryRule{
-					{
-						Type: api.KeepEntryIfRegexMatch,
-						KeepEntry: &api.TransformFilterGenericRule{
-							Input: "service",
-							Value: "abc.+",
-						},
-					},
-				},
+				Type:           api.KeepEntryQuery,
+				KeepEntryQuery: `(namespace="A" and with(workload)) or service=~"abc.+"`,
 			},
 		},
 	}
@@ -742,29 +716,13 @@ func Test_Transform_KeepEntrySampling(t *testing.T) {
 	newFilter := api.TransformFilter{
 		Rules: []api.TransformFilterRule{
 			{
-				Type: api.KeepEntryAllSatisfied,
-				KeepEntryAllSatisfied: []*api.KeepEntryRule{
-					{
-						Type: api.KeepEntryIfEqual,
-						KeepEntry: &api.TransformFilterGenericRule{
-							Input: "namespace",
-							Value: "A",
-						},
-					},
-				},
+				Type:              api.KeepEntryQuery,
+				KeepEntryQuery:    `namespace="A"`,
 				KeepEntrySampling: 10,
 			},
 			{
-				Type: api.KeepEntryAllSatisfied,
-				KeepEntryAllSatisfied: []*api.KeepEntryRule{
-					{
-						Type: api.KeepEntryIfEqual,
-						KeepEntry: &api.TransformFilterGenericRule{
-							Input: "namespace",
-							Value: "B",
-						},
-					},
-				},
+				Type:           api.KeepEntryQuery,
+				KeepEntryQuery: `namespace="B"`,
 			},
 		},
 	}

pkg/utils/filters/filters.go

@@ -1,11 +1,9 @@
 package filters
 
 import (
-	"fmt"
 	"regexp"
 	"strings"
 
-	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
 )
@@ -113,41 +111,3 @@ func injectVars(flow config.GenericMap, filterValue string, varLookups [][]strin
 	}
 	return injected
 }
-
-func FromKeepEntry(from *api.KeepEntryRule) (Predicate, error) {
-	switch from.Type {
-	case api.KeepEntryIfExists:
-		return Presence(from.KeepEntry.Input), nil
-	case api.KeepEntryIfDoesntExist:
-		return Absence(from.KeepEntry.Input), nil
-	case api.KeepEntryIfEqual:
-		return Equal(from.KeepEntry.Input, from.KeepEntry.Value, true), nil
-	case api.KeepEntryIfNotEqual:
-		return NotEqual(from.KeepEntry.Input, from.KeepEntry.Value, true), nil
-	case api.KeepEntryIfRegexMatch:
-		if r, err := compileRegex(from.KeepEntry); err != nil {
-			return nil, err
-		} else {
-			return Regex(from.KeepEntry.Input, r), nil
-		}
-	case api.KeepEntryIfNotRegexMatch:
-		if r, err := compileRegex(from.KeepEntry); err != nil {
-			return nil, err
-		} else {
-			return NotRegex(from.KeepEntry.Input, r), nil
-		}
-	}
-	return nil, fmt.Errorf("keep entry rule type not recognized: %s", from.Type)
-}
-
-func compileRegex(from *api.TransformFilterGenericRule) (*regexp.Regexp, error) {
-	s, ok := from.Value.(string)
-	if !ok {
-		return nil, fmt.Errorf("invalid regex keep rule: rule value must be a string [%v]", from)
-	}
-	r, err := regexp.Compile(s)
-	if err != nil {
-		return nil, fmt.Errorf("invalid regex keep rule: cannot compile regex [%w]", err)
-	}
-	return r, nil
-}