JS expressions for trigger filter

config/core/resources/trigger.yaml

@@ -59,6 +59,9 @@ spec:
               filter:
                 type: object
                 properties:
+                  jsExpression:
+                    type: string
+                    description: "Javascript Expression used for filtering events. If not specified, the behaviour is the same as an expression always true"
                   attributes:
                     type: object
                     description: 'Map of CloudEvents attributes used for filtering events. If not specified, will default to all events'
@@ -120,6 +123,9 @@ spec:
                     the Subscriber. If not specified, will default to allowing all
                     events.'
                 properties:
+                  jsExpression:
+                    type: string
+                    description: "Javascript Expression used for filtering events. If not specified, the behaviour is the same as an expression always true"
                   attributes:
                     type: object
                     description: 'Map of CloudEvents attributes used for filtering events. If not specified, will default to all events'

go.mod

@@ -5,7 +5,10 @@ go 1.14
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cloudevents/sdk-go/v2 v2.2.0
+	github.com/dlclark/regexp2 v1.4.0 // indirect
+	github.com/dop251/goja v0.0.0-20201107160812-7545ac6de48a
 	github.com/ghodss/yaml v1.0.0
+	github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect
 	github.com/golang/protobuf v1.4.3
 	github.com/google/go-cmp v0.5.2
 	github.com/google/gofuzz v1.1.0
@@ -49,6 +52,7 @@ require (
 )
 
 replace (
+	github.com/dop251/goja => github.com/slinkydeveloper/goja v0.0.0-20201117110846-2adc5134db9d
 	github.com/prometheus/client_golang => github.com/prometheus/client_golang v0.9.2
 	k8s.io/api => k8s.io/api v0.18.8
 	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.18.8

go.sum

@@ -145,6 +145,8 @@ github.com/dgryski/go-gk v0.0.0-20140819190930-201884a44051/go.mod h1:qm+vckxRlD
 github.com/dgryski/go-gk v0.0.0-20200319235926-a69029f61654 h1:XOPLOMn/zT4jIgxfxSsoXPxkrzz0FaCHwp33x5POJ+Q=
 github.com/dgryski/go-gk v0.0.0-20200319235926-a69029f61654/go.mod h1:qm+vckxRlDt0aOla0RYJJVeqHZlWfOm2UIxHaqPB46E=
 github.com/dgryski/go-lttb v0.0.0-20180810165845-318fcdf10a77/go.mod h1:Va5MyIzkU0rAM92tn3hb3Anb7oz7KcnixF49+2wOMe4=
+github.com/dlclark/regexp2 v1.4.0 h1:F1rxgk7p4uKjwIQxBs9oAXe5CqrXlCduYEJvrF4u93E=
+github.com/dlclark/regexp2 v1.4.0/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
 github.com/docker/docker v0.7.3-0.20190327010347-be7ac8be2ae0/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-units v0.3.3/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
@@ -239,6 +241,8 @@ github.com/go-openapi/swag v0.19.7/go.mod h1:ao+8BpOPyKdpQz3AOJfbeEVpLmWAvlT1IfT
 github.com/go-openapi/validate v0.18.0/go.mod h1:Uh4HdOzKt19xGIGm1qHf/ofbX1YQ4Y+MYsct2VUrAJ4=
 github.com/go-openapi/validate v0.19.2/go.mod h1:1tRCw7m3jtI8eNWEEliiAqUIcBztB2KDnRCRMUi7GTA=
 github.com/go-openapi/validate v0.19.5/go.mod h1:8DJv2CVJQ6kGNpFW6eV9N3JviE1C85nY1c2z52x1Gk4=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible h1:W1iEw64niKVGogNgBN3ePyLFfuisuzeidWPMPWmECqU=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -585,6 +589,8 @@ github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPx
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0 h1:UBcNElsrwanuuMsnGSlYmtmgbb23qDR5dG+6X6Oo89I=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
+github.com/slinkydeveloper/goja v0.0.0-20201117110846-2adc5134db9d h1:ErtEjtFpSIplqEyFGNmchWoGdfLR1LUB1FZxMvn+OJI=
+github.com/slinkydeveloper/goja v0.0.0-20201117110846-2adc5134db9d/go.mod h1:EIJIr8dMHKqDqRw0vt3ZYiamgzvzXQG5Dg2YQv+/fGo=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=

pkg/apis/eventing/v1/trigger_types.go

@@ -77,6 +77,7 @@ type TriggerSpec struct {
 
 	// Filter is the filter to apply against all events from the Broker. Only events that pass this
 	// filter will be sent to the Subscriber. If not specified, will default to allowing all events.
+	// If multiple filter types are specified, the event will pass only if all filters pass
 	//
 	// +optional
 	Filter *TriggerFilter `json:"filter,omitempty"`
@@ -96,6 +97,12 @@ type TriggerFilter struct {
 	//
 	// +optional
 	Attributes TriggerFilterAttributes `json:"attributes,omitempty"`
+
+	// Javascript Expression to evaluate.
+	// Event fields are available using the dot operator, eg:
+	//   event.id === "aaa" || event.type === "my.type"
+	// +optional
+	JsExpression string `json:"jsExpression,omitempty"`
 }
 
 // TriggerFilterAttributes is a map of context attribute names to values for

pkg/apis/eventing/v1/trigger_validation.go

@@ -26,6 +26,8 @@ import (
 	"knative.dev/pkg/kmp"
 
 	corev1 "k8s.io/api/core/v1"
+
+	"knative.dev/eventing/pkg/eventfilter/jsengine"
 )
 
 var (
@@ -63,6 +65,16 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 				errs = errs.Also(fe)
 			}
 		}
+		if ts.Filter.JsExpression != "" {
+			_, err := jsengine.ParseFilterExpr(ts.Filter.JsExpression)
+			if err != nil {
+				fe := &apis.FieldError{
+					Message: fmt.Sprintf("Invalid filter expression: %q", err),
+					Paths:   []string{"filter.expression"},
+				}
+				errs = errs.Also(fe)
+			}
+		}
 	}
 
 	if fe := ts.Subscriber.Validate(ctx); fe != nil {

pkg/apis/eventing/v1beta1/trigger_conversion.go

@@ -20,8 +20,9 @@ import (
 	"context"
 	"fmt"
 
-	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/pkg/apis"
+
+	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 )
 
 // ConvertTo implements apis.Convertible
@@ -38,6 +39,7 @@ func (source *Trigger) ConvertTo(_ context.Context, to apis.Convertible) error {
 			for k, v := range source.Spec.Filter.Attributes {
 				sink.Spec.Filter.Attributes[k] = v
 			}
+			sink.Spec.Filter.JsExpression = source.Spec.Filter.JsExpression
 		}
 		sink.Status.Status = source.Status.Status
 		sink.Status.SubscriberURI = source.Status.SubscriberURI
@@ -60,7 +62,8 @@ func (sink *Trigger) ConvertFrom(_ context.Context, from apis.Convertible) error
 				attributes[k] = v
 			}
 			sink.Spec.Filter = &TriggerFilter{
-				Attributes: attributes,
+				Attributes:   attributes,
+				JsExpression: source.Spec.Filter.JsExpression,
 			}
 		}
 		sink.Status.Status = source.Status.Status

pkg/apis/eventing/v1beta1/trigger_conversion_test.go

@@ -23,9 +23,10 @@ import (
 	"github.com/google/go-cmp/cmp"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+
+	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 )
 
 func TestTriggerConversionBadType(t *testing.T) {
@@ -126,6 +127,30 @@ func TestTriggerConversionRoundTripV1beta1(t *testing.T) {
 				},
 			},
 		},
+	}, {name: "filter rules with expressions",
+		in: &Trigger{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:       "trigger-name",
+				Namespace:  "trigger-ns",
+				Generation: 17,
+			},
+			Spec: TriggerSpec{
+				Broker: "default",
+				Filter: &TriggerFilter{
+					Attributes:   TriggerFilterAttributes{"source": "mysource", "type": "mytype", "customkey": "customvalue"},
+					JsExpression: "event.id == 1234",
+				},
+			},
+			Status: TriggerStatus{
+				Status: duckv1.Status{
+					ObservedGeneration: 1,
+					Conditions: duckv1.Conditions{{
+						Type:   "Ready",
+						Status: "True",
+					}},
+				},
+			},
+		},
 	}, {name: "full configuration",
 		in: &Trigger{
 			ObjectMeta: metav1.ObjectMeta{
@@ -255,6 +280,30 @@ func TestTriggerConversionRoundTripV1(t *testing.T) {
 				},
 			},
 		},
+	}, {name: "filter rules, many",
+		in: &v1.Trigger{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:       "trigger-name",
+				Namespace:  "trigger-ns",
+				Generation: 17,
+			},
+			Spec: v1.TriggerSpec{
+				Broker: "default",
+				Filter: &v1.TriggerFilter{
+					Attributes:   v1.TriggerFilterAttributes{"source": "mysource", "type": "mytype", "customkey": "customvalue"},
+					JsExpression: "event.id == 1234",
+				},
+			},
+			Status: v1.TriggerStatus{
+				Status: duckv1.Status{
+					ObservedGeneration: 1,
+					Conditions: duckv1.Conditions{{
+						Type:   "Ready",
+						Status: "True",
+					}},
+				},
+			},
+		},
 	}, {name: "full configuration",
 		in: &v1.Trigger{
 			ObjectMeta: metav1.ObjectMeta{

pkg/apis/eventing/v1beta1/trigger_types.go

@@ -80,6 +80,7 @@ type TriggerSpec struct {
 
 	// Filter is the filter to apply against all events from the Broker. Only events that pass this
 	// filter will be sent to the Subscriber. If not specified, will default to allowing all events.
+	// If multiple filter types are specified, the event will pass only if all filters pass
 	//
 	// +optional
 	Filter *TriggerFilter `json:"filter,omitempty"`
@@ -99,6 +100,12 @@ type TriggerFilter struct {
 	//
 	// +optional
 	Attributes TriggerFilterAttributes `json:"attributes,omitempty"`
+
+	// Javascript Expression to evaluate.
+	// Event fields are available using the dot operator, eg:
+	//   event.id === "aaa" || event.type === "my.type"
+	// +optional
+	JsExpression string `json:"jsExpression,omitempty"`
 }
 
 // TriggerFilterAttributes is a map of context attribute names to values for

pkg/apis/eventing/v1beta1/trigger_validation.go

@@ -26,6 +26,8 @@ import (
 	"knative.dev/pkg/kmp"
 
 	corev1 "k8s.io/api/core/v1"
+
+	"knative.dev/eventing/pkg/eventfilter/jsengine"
 )
 
 var (
@@ -59,6 +61,16 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 				errs = errs.Also(fe)
 			}
 		}
+		if ts.Filter.JsExpression != "" {
+			_, err := jsengine.ParseFilterExpr(ts.Filter.JsExpression)
+			if err != nil {
+				fe := &apis.FieldError{
+					Message: fmt.Sprintf("Invalid filter expression: %q", err),
+					Paths:   []string{"filter.expression"},
+				}
+				errs = errs.Also(fe)
+			}
+		}
 	}
 
 	if fe := ts.Subscriber.Validate(ctx); fe != nil {

pkg/eventfilter/benchmarks/jsengine_benchmark_test.go

@@ -0,0 +1,71 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package benchmarks
+
+import (
+	"fmt"
+	"testing"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+
+	"knative.dev/eventing/pkg/eventfilter"
+	"knative.dev/eventing/pkg/eventfilter/jsengine"
+)
+
+func BenchmarkJsEngineFilter(b *testing.B) {
+	event := cetest.FullEvent()
+
+	RunFilterBenchmarks(b,
+		func(i interface{}) eventfilter.Filter {
+			f, _ := jsengine.NewJsFilter(i.(string))
+			return f
+		},
+		FilterBenchmark{
+			name:  "Pass with exact match of id",
+			arg:   fmt.Sprintf(`event.id === "%s"`, event.ID()),
+			event: event,
+		},
+		FilterBenchmark{
+			name: "Pass with exact match of all context attributes (except time)",
+			arg: fmt.Sprintf(
+				`event.id === "%s" && event.source === "%s" && event.type === "%s" && event.dataschema === "%s" && event.datacontenttype === "%s" && event.subject === "%s"`,
+				event.ID(),
+				event.Source(),
+				event.Type(),
+				event.DataSchema(),
+				event.DataContentType(),
+				event.Subject(),
+			),
+			event: event,
+		},
+		FilterBenchmark{
+			name:  "No pass with exact match of id and source",
+			arg:   `event.id === "qwertyuiopasdfghjklzxcvbnm" && event.source === "qwertyuiopasdfghjklzxcvbnm"`,
+			event: event,
+		},
+		FilterBenchmark{
+			name:  "No pass with if then",
+			arg:   fmt.Sprintf(`(event.id === "%s") ? event.type === "---%s" : true`, event.ID(), event.Type()),
+			event: event,
+		},
+		FilterBenchmark{
+			name:  "No pass with nested logic",
+			arg:   fmt.Sprintf(`(event.type === "---%s") || (event.type === "%s" ? event.id !== "%s" : event.id === "%s")`, event.Type(), event.Type(), event.ID(), event.ID()),
+			event: event,
+		},
+	)
+}

pkg/eventfilter/jsengine/filter.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package jsengine
+
+import (
+	"context"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/dop251/goja"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing/pkg/eventfilter"
+)
+
+type jsFilter goja.Runtime
+
+func (j *jsFilter) Filter(ctx context.Context, event cloudevents.Event) eventfilter.FilterResult {
+	pass, err := runFilter(event, (*goja.Runtime)(j))
+	if err != nil {
+		logging.FromContext(ctx).Warn("Error while trying to run the js expression filter: ", err)
+		return eventfilter.FailFilter
+	}
+	if pass {
+		return eventfilter.PassFilter
+	} else {
+		return eventfilter.FailFilter
+	}
+}
+
+func NewJsFilter(src string) (eventfilter.Filter, error) {
+	p, err := ParseFilterExpr(src)
+	if err != nil {
+		return nil, err
+	}
+	engine := goja.New()
+	if _, err := runProgramWithSafeTimeout(timeout, engine, p); err != nil {
+		return nil, err
+	}
+	return (*jsFilter)(engine), nil
+}