added script filter (#1894)

* feat: added lua script as event filter

Signed-off-by: Vaibhav Page <vaibhav.page@gmail.com>

* feat: added script filter

Signed-off-by: Vaibhav Page <vaibhav.page@gmail.com>

* chore: ran codegen

Signed-off-by: Vaibhav Page <vaibhav.page@gmail.com>

api/jsonschema/schema.json

@@ -3134,6 +3134,10 @@
           },
           "type": "array"
         },
+        "script": {
+          "description": "Script refers to a Lua script evaluated to determine the validity of an event.",
+          "type": "string"
+        },
         "time": {
           "$ref": "#/definitions/io.argoproj.sensor.v1alpha1.TimeFilter",
           "description": "Time filter on the event with escalation"

examples/sensors/filter-script.yaml

@@ -0,0 +1,55 @@
+# Event Payload
+#
+#  {
+#    "a": "b",
+#    "c": 10,
+#    "d": {
+#      "e": "z"
+#    }
+#  }
+#
+
+apiVersion: argoproj.io/v1alpha1
+kind: Sensor
+metadata:
+  name: with-script-filter
+spec:
+  dependencies:
+    - name: test-dep
+      eventSourceName: webhook
+      eventName: example
+      filters:
+        script: |-
+          if event.a == "b" and event.d.e == "z" then return true else return false end
+  triggers:
+    - template:
+        name: workflow
+        k8s:
+          operation: create
+          source:
+            resource:
+              apiVersion: argoproj.io/v1alpha1
+              kind: Workflow
+              metadata:
+                generateName: workflow-
+              spec:
+                entrypoint: whalesay
+                arguments:
+                  parameters:
+                    - name: message
+                      # value will get overridden by the event payload
+                      value: hello world
+                templates:
+                  - name: whalesay
+                    inputs:
+                      parameters:
+                        - name: message
+                    container:
+                      image: docker/whalesay:latest
+                      command: [cowsay]
+                      args: ["{{inputs.parameters.message}}"]
+          parameters:
+            - src:
+                dependencyName: test-dep
+                dataKey: name
+              dest: spec.arguments.parameters.0.value

pkg/apis/sensor/v1alpha1/types.go

@@ -231,6 +231,8 @@ type EventDependencyFilter struct {
 	// Available values: and (&&), or (||)
 	// Is optional and if left blank treated as and (&&).
 	ExprLogicalOperator LogicalOperator `json:"exprLogicalOperator,omitempty" protobuf:"bytes,6,opt,name=exprLogicalOperator,casttype=ExprLogicalOperator"`
+	// Script refers to a Lua script evaluated to determine the validity of an event.
+	Script string `json:"script,omitempty" protobuf:"bytes,7,opt,name=script"`
 }
 
 type ExprFilter struct {

sensors/dependencies/filter.go

@@ -18,6 +18,7 @@ package dependencies
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"regexp"
@@ -30,8 +31,8 @@ import (
 	"github.com/Masterminds/sprig/v3"
 	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1"
-	"github.com/argoproj/pkg/json"
 	"github.com/tidwall/gjson"
+	lua "github.com/yuin/gopher-lua"
 )
 
 const (
@@ -87,14 +88,19 @@ func filterEvent(filter *v1alpha1.EventDependencyFilter, operator v1alpha1.Logic
 		errMessages = append(errMessages, timeErr.Error())
 	}
 
+	scriptFilter, err := filterScript(filter.Script, event)
+	if err != nil {
+		return false, err
+	}
+
 	if operator == v1alpha1.OrLogicalOperator {
 		if len(errMessages) > 0 {
-			return exprFilter || dataFilter || ctxFilter || timeFilter,
+			return exprFilter || dataFilter || ctxFilter || timeFilter || scriptFilter,
 				errors.New(strings.Join(errMessages, errMsgListSeparator))
 		}
-		return exprFilter || dataFilter || ctxFilter || timeFilter, nil
+		return exprFilter || dataFilter || ctxFilter || timeFilter || scriptFilter, nil
 	}
-	return exprFilter && dataFilter && ctxFilter && timeFilter, nil
+	return exprFilter && dataFilter && ctxFilter && timeFilter && scriptFilter, nil
 }
 
 // filterExpr applies expression based filters against event data
@@ -111,7 +117,7 @@ func filterExpr(filters []v1alpha1.ExprFilter, operator v1alpha1.LogicalOperator
 	if payload == nil {
 		return true, nil
 	}
-	if !json.IsJSON(payload) {
+	if !gjson.Valid(string(payload)) {
 		return false, fmt.Errorf(errMsgTemplate, "expr", "event data not valid JSON")
 	}
 
@@ -195,7 +201,7 @@ func filterData(filters []v1alpha1.DataFilter, operator v1alpha1.LogicalOperator
 	if payload == nil {
 		return true, nil
 	}
-	if !json.IsJSON(payload) {
+	if !gjson.Valid(string(payload)) {
 		return false, fmt.Errorf(errMsgTemplate, "data", "event data not valid JSON")
 	}
 
@@ -477,3 +483,38 @@ func filterTime(timeFilter *v1alpha1.TimeFilter, eventTime time.Time) (bool, err
 		return (eventTime.After(startTime) || eventTime.Equal(startTime)) || eventTime.Before(stopTime), nil
 	}
 }
+
+func filterScript(script string, event *v1alpha1.Event) (bool, error) {
+	if script == "" {
+		return true, nil
+	}
+	if event == nil {
+		return false, fmt.Errorf("nil event")
+	}
+	payload := event.Data
+	if payload == nil {
+		return true, nil
+	}
+	var js *json.RawMessage
+	if err := json.Unmarshal(payload, &js); err != nil {
+		return false, err
+	}
+	var jsData []byte
+	jsData, err := json.Marshal(js)
+	if err != nil {
+		return false, err
+	}
+	l := lua.NewState()
+	defer l.Close()
+	var payloadJson map[string]interface{}
+	if err = json.Unmarshal(jsData, &payloadJson); err != nil {
+		return false, err
+	}
+	lEvent := mapToTable(payloadJson)
+	l.SetGlobal("event", lEvent)
+	if err = l.DoString(script); err != nil {
+		return false, err
+	}
+	lv := l.Get(-1)
+	return lv == lua.LTrue, nil
+}

sensors/dependencies/filter_script_test.go

@@ -0,0 +1,67 @@
+package dependencies
+
+import (
+	"testing"
+
+	"github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestScriptFilter(t *testing.T) {
+	tests := []struct {
+		script   string
+		event    *v1alpha1.Event
+		result   bool
+		hasError bool
+	}{
+		{
+			script: `
+if event.a == "hello" then return true else return false end
+`,
+			event: &v1alpha1.Event{
+				Data: []byte(`{"a":"hello"}`),
+			},
+			result:   true,
+			hasError: false,
+		},
+		{
+			script: `
+if event.a == "hello" and event.b == "world" then return false else return true end
+`,
+			event: &v1alpha1.Event{
+				Data: []byte(`{"a":"hello","b":"world"}`),
+			},
+			result:   false,
+			hasError: false,
+		},
+		{
+			script: `
+if event.a == "hello" return false else return true end
+`,
+			event: &v1alpha1.Event{
+				Data: []byte(`{"a":"hello"}`),
+			},
+			result:   false,
+			hasError: true,
+		},
+		{
+			script: `
+if a.a == "hello" then return true else return false end
+`,
+			event: &v1alpha1.Event{
+				Data: []byte(`{"a":"hello"}`),
+			},
+			result:   false,
+			hasError: true,
+		},
+	}
+	for _, tt := range tests {
+		result, err := filterScript(tt.script, tt.event)
+		if tt.hasError {
+			assert.NotNil(t, err)
+		} else {
+			assert.Nil(t, err)
+			assert.Equal(t, tt.result, result)
+		}
+	}
+}