[WIP] Trigger.Filter now supports the CNCF Subscription API

config/core/resources/trigger.yaml

@@ -100,6 +100,42 @@ spec:
                     description: 'Attributes filters events by exact match on event context attributes. Each key in the map is compared with the equivalent key in the event context. An event passes the filter if all values are equal to the specified values.  Nested context attributes are not supported as keys. Only string values are supported. '
                     type: object
                     x-kubernetes-preserve-unknown-fields: true
+                  exact:
+                    description: 'Exact evaluates to true if the value of the matching CloudEvents attribute is matches exactly the String value specified (case sensitive). Exact must contain exactly one property, where the key is the name of the CloudEvents attribute to be matched, and its value is the String value to use in the comparison. The attribute name and value specified in the filter express cannot be be empty strings.'
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                  prefix:
+                    description: 'Prefix evaluates to true if the value of the matching CloudEvents attribute starts with the String value specified (case sensitive). Prefix must contain exactly one property, where the key is the name of the CloudEvents attribute to be matched, and its value is the String value to use in the comparison. The attribute name and value specified in the filter express cannot be be empty strings.'
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                  suffix:
+                    description: 'Suffix evaluates to true if the value of the matching CloudEvents attribute ends with the String value specified (case sensitive). | Suffix must contain exactly one property, where the key is the name of the CloudEvents attribute to be matched, and its value is the String value to use in the comparison. he attribute name and value specified in the filter express cannot be be empty strings.'
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                  not:
+                    description: 'Not evaluates to true if the nested expression evaluates to false.'
+                    type: object
+                    # Because kube doesn't allow to use $ref, we can't recursively define this schema.
+                    x-kubernetes-preserve-unknown-fields: true
+                  all:
+                    description: 'All evaluates to true if all the nested expressions evaluate to true. All must contain at least one filter expression.'
+                    type: array
+                    items:
+                      # Because kube doesn't allow to use $ref, we can't recursively reference to the filter schema.
+                      type: object
+                      description: "Sub schema"
+                      x-kubernetes-preserve-unknown-fields: true
+                  any:
+                    description: 'Any evaluates to true if at least one of the nested expressions evaluate to true. Any must contain at least one filter expression.'
+                    type: array
+                    items:
+                      # Because kube doesn't allow to use $ref, we can't recursively reference to the filter schema.
+                      type: object
+                      description: "Sub schema"
+                      x-kubernetes-preserve-unknown-fields: true
+                # This allows extension filter dialects
+                additionalProperties: true
+                x-kubernetes-preserve-unknown-fields: true
               subscriber:
                 description: Subscriber is the addressable that receives events from the Broker that pass the Filter. It is required.
                 type: object

pkg/apis/eventing/v1/trigger_types.go

@@ -79,6 +79,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 filters are specified, then the same semantics of TriggerFilter.All is applied.
 	//
 	// +optional
 	Filter *TriggerFilter `json:"filter,omitempty"`
@@ -92,6 +93,9 @@ type TriggerSpec struct {
 	Delivery *eventingduckv1.DeliverySpec `json:"delivery,omitempty"`
 }
 
+// TriggerFilter allows to define a filter expression.
+// If multiple filters are specified, then the same semantics of TriggerFilter.All is applied.
+// If no filter dialect or empty object is specified, then the filter always accept the events.
 type TriggerFilter struct {
 	// Attributes filters events by exact match on event context attributes.
 	// Each key in the map is compared with the equivalent key in the event
@@ -102,6 +106,55 @@ type TriggerFilter struct {
 	//
 	// +optional
 	Attributes TriggerFilterAttributes `json:"attributes,omitempty"`
+
+	// All evaluates to true if all the nested expressions evaluate to true.
+	//
+	// All must contain at least one filter expression.
+	//
+	// +optional
+	All []TriggerFilter `json:"all,omitempty"`
+
+	// Any evaluates to true if at least one of the nested expressions evaluate to true.
+	//
+	// Any must contain at least one filter expression.
+	//
+	// +optional
+	Any []TriggerFilter `json:"any,omitempty"`
+
+	// Not evaluates to true if the nested expression evaluates to false.
+	//
+	// +optional
+	Not *TriggerFilter `json:"not,omitempty"`
+
+	// Exact evaluates to true if the value of the matching CloudEvents attribute is matches exactly the String value specified (case sensitive).
+	// Exact must contain exactly one property, where the key is the name of the CloudEvents attribute to be matched, and its value is the String value to use in the comparison.
+	//
+	// The attribute name and value specified in the filter express cannot be be empty strings.
+	//
+	// +optional
+	Exact map[string]string `json:"exact,omitempty"`
+
+	// Prefix evaluates to true if the value of the matching CloudEvents attribute starts with the String value specified (case sensitive).
+	// Prefix must contain exactly one property, where the key is the name of the CloudEvents attribute to be matched, and its value is the String value to use in the comparison.
+	//
+	// The attribute name and value specified in the filter express cannot be be empty strings.
+	//
+	// +optional
+	Prefix map[string]string `json:"prefix,omitempty"`
+
+	// Suffix evaluates to true if the value of the matching CloudEvents attribute ends with the String value specified (case sensitive).
+	// Suffix must contain exactly one property, where the key is the name of the CloudEvents attribute to be matched, and its value is the String value to use in the comparison.
+	//
+	// The attribute name and value specified in the filter express cannot be be empty strings.
+	//
+	// +optional
+	Suffix map[string]string `json:"suffix,omitempty"`
+
+	// Extensions includes the list of additional filter dialects supported by specific broker implementations.
+	// Check out the documentation of the broker implementation you're using to know about what additional filters are supported.
+	//
+	// +optional
+	Extensions map[string]*runtime.RawExtension `json:",inline"`
 }
 
 // TriggerFilterAttributes is a map of context attribute names to values for

pkg/apis/eventing/v1/trigger_validation.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"regexp"
+	"strings"
 
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/kmp"
@@ -53,16 +54,8 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(fe)
 	}
 
-	if ts.Filter != nil {
-		for attr := range map[string]string(ts.Filter.Attributes) {
-			if !validAttributeName.MatchString(attr) {
-				fe := &apis.FieldError{
-					Message: fmt.Sprintf("Invalid attribute name: %q", attr),
-					Paths:   []string{"filter.attributes"},
-				}
-				errs = errs.Also(fe)
-			}
-		}
+	for _, err := range validateFilterSpec(ts.Filter, []string{"filter"}) {
+		errs = errs.Also(err)
 	}
 
 	if fe := ts.Subscriber.Validate(ctx); fe != nil {
@@ -78,6 +71,113 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 	return errs
 }
 
+func validateFilterSpec(filter *TriggerFilter, path []string) (errs []*apis.FieldError) {
+	if filter == nil {
+		return nil
+	}
+
+	// Validate Attributes
+	for attr := range map[string]string(filter.Attributes) {
+		if !validAttributeName.MatchString(attr) {
+			errs = append(errs, &apis.FieldError{
+				Message: fmt.Sprintf("Invalid attribute name: %q", attr),
+				Paths:   []string{strings.Join(append(path, "attributes"), ".")},
+			})
+		}
+	}
+
+	// Validate Exact
+	if filter.Exact != nil {
+		if len(filter.Exact) != 1 {
+			errs = append(errs, &apis.FieldError{
+				Message: "Exact can have only one key-value",
+				Paths:   []string{strings.Join(append(path, "exact"), ".")},
+			})
+		}
+		for attr := range filter.Exact {
+			if !validAttributeName.MatchString(attr) {
+				errs = append(errs, &apis.FieldError{
+					Message: fmt.Sprintf("Invalid attribute name: %q", attr),
+					Paths:   []string{strings.Join(append(path, "exact"), ".")},
+				})
+			}
+		}
+	}
+
+	// Validate Prefix
+	if filter.Prefix != nil {
+		if len(filter.Prefix) != 1 {
+			errs = append(errs, &apis.FieldError{
+				Message: "Prefix can have only one key-value",
+				Paths:   []string{strings.Join(append(path, "prefix"), ".")},
+			})
+		}
+		for attr := range filter.Prefix {
+			if !validAttributeName.MatchString(attr) {
+				errs = append(errs, &apis.FieldError{
+					Message: fmt.Sprintf("Invalid attribute name: %q", attr),
+					Paths:   []string{strings.Join(append(path, "prefix"), ".")},
+				})
+			}
+		}
+	}
+
+	// Validate Suffix
+	if filter.Suffix != nil {
+		if len(filter.Suffix) != 1 {
+			errs = append(errs, &apis.FieldError{
+				Message: "Suffix can have only one key-value",
+				Paths:   []string{strings.Join(append(path, "suffix"), ".")},
+			})
+		}
+		for attr := range filter.Suffix {
+			if !validAttributeName.MatchString(attr) {
+				errs = append(errs, &apis.FieldError{
+					Message: fmt.Sprintf("Invalid attribute name: %q", attr),
+					Paths:   []string{strings.Join(append(path, "suffix"), ".")},
+				})
+			}
+		}
+	}
+
+	// Validate All
+	if filter.All != nil {
+		if len(filter.All) < 1 {
+			errs = append(errs, &apis.FieldError{
+				Message: "All must contain at least one nested filter",
+				Paths:   []string{strings.Join(append(path, "all"), ".")},
+			})
+		}
+
+		for i, f := range filter.All {
+			f := f
+			errs = append(errs, validateFilterSpec(&f, append(path, "all", fmt.Sprintf("[%d]", i)))...)
+		}
+	}
+
+	// Validate Any
+	if filter.Any != nil {
+		if len(filter.Any) < 1 {
+			errs = append(errs, &apis.FieldError{
+				Message: "Any must contain at least one nested filter",
+				Paths:   []string{strings.Join(append(path, "any"), ".")},
+			})
+		}
+
+		for i, f := range filter.Any {
+			f := f
+			errs = append(errs, validateFilterSpec(&f, append(path, "any", fmt.Sprintf("[%d]", i)))...)
+		}
+	}
+
+	// Validate Not
+	if filter.Not != nil {
+		errs = append(errs, validateFilterSpec(filter.Not, append(path, "not"))...)
+	}
+
+	return
+}
+
 // CheckImmutableFields checks that any immutable fields were not changed.
 func (t *Trigger) CheckImmutableFields(ctx context.Context, original *Trigger) *apis.FieldError {
 	if original == nil {