Attributes filters in Triggers

pkg/apis/eventing/v1alpha1/trigger_defaults.go

@@ -33,16 +33,4 @@ func (ts *TriggerSpec) SetDefaults(ctx context.Context) {
 	if ts.Filter == nil {
 		ts.Filter = &TriggerFilter{}
 	}
-
-	// Note that this logic will need to change once there are other filtering options, as it should
-	// only apply if no other filter is applied.
-	if ts.Filter.SourceAndType == nil {
-		ts.Filter.SourceAndType = &TriggerFilterSourceAndType{}
-	}
-	if ts.Filter.SourceAndType.Type == "" {
-		ts.Filter.SourceAndType.Type = TriggerAnyFilter
-	}
-	if ts.Filter.SourceAndType.Source == "" {
-		ts.Filter.SourceAndType.Source = TriggerAnyFilter
-	}
 }

pkg/apis/eventing/v1alpha1/trigger_defaults_test.go

@@ -24,18 +24,19 @@ import (
 )
 
 var (
-	defaultBroker = "default"
-	otherBroker   = "other_broker"
-
-	otherTriggerFilter = &TriggerFilter{
-		SourceAndType: &TriggerFilterSourceAndType{
+	defaultBroker        = "default"
+	otherBroker          = "other_broker"
+	defaultTriggerFilter = &TriggerFilter{}
+	otherTriggerFilter   = &TriggerFilter{
+		DeprecatedSourceAndType: &TriggerFilterSourceAndType{
 			Type:   "other_type",
 			Source: "other_source"},
 	}
+
 	defaultTrigger = Trigger{
 		Spec: TriggerSpec{
 			Broker: defaultBroker,
-			Filter: defaultTriggerFilter(),
+			Filter: defaultTriggerFilter,
 		},
 	}
 )
@@ -51,7 +52,7 @@ func TestTriggerDefaults(t *testing.T) {
 		},
 		"nil filter": {
 			initial:  Trigger{Spec: TriggerSpec{Broker: otherBroker}},
-			expected: Trigger{Spec: TriggerSpec{Broker: otherBroker, Filter: defaultTriggerFilter()}},
+			expected: Trigger{Spec: TriggerSpec{Broker: otherBroker, Filter: defaultTriggerFilter}},
 		},
 		"nil broker and nil filter": {
 			initial:  Trigger{},
@@ -67,12 +68,3 @@ func TestTriggerDefaults(t *testing.T) {
 		})
 	}
 }
-
-func defaultTriggerFilter() *TriggerFilter {
-	// Can't just be a package level var because it gets mutated.
-	return &TriggerFilter{
-		SourceAndType: &TriggerFilterSourceAndType{
-			Type:   TriggerAnyFilter,
-			Source: TriggerAnyFilter},
-	}
-}

pkg/apis/eventing/v1alpha1/trigger_lifecycle_test.go

@@ -300,14 +300,14 @@ func TestTriggerAnnotateUserInfo(t *testing.T) {
 		}, {
 			name:       "update trigger which has no annotations without diff",
 			user:       u1,
-			this:       &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter()}},
-			prev:       &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter()}},
+			this:       &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter}},
+			prev:       &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter}},
 			wantedAnns: map[string]string{},
 		}, {
 			name: "update trigger which has annotations without diff",
 			user: u2,
-			this: withUserAnns(u1, u1, &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter()}}),
-			prev: withUserAnns(u1, u1, &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter()}}),
+			this: withUserAnns(u1, u1, &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter}}),
+			prev: withUserAnns(u1, u1, &Trigger{Spec: TriggerSpec{Broker: defaultBroker, Filter: defaultTriggerFilter}}),
 			wantedAnns: map[string]string{
 				eventing.CreatorAnnotation: u1,
 				eventing.UpdaterAnnotation: u1,

pkg/apis/eventing/v1alpha1/trigger_types.go

@@ -74,17 +74,37 @@ type TriggerSpec struct {
 }
 
 type TriggerFilter struct {
-	SourceAndType *TriggerFilterSourceAndType `json:"sourceAndType,omitempty"`
+	// DeprecatedSourceAndType filters events based on exact matches on the
+	// CloudEvents type and source attributes. This field has been replaced by the
+	// Attributes field.
+	//
+	// +optional
+	DeprecatedSourceAndType *TriggerFilterSourceAndType `json:"sourceAndType,omitempty"`
+
+	// 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.
+	//
+	// +optional
+	Attributes *TriggerFilterAttributes `json:"attributes,omitempty"`
 }
 
 // TriggerFilterSourceAndType filters events based on exact matches on the cloud event's type and
 // source attributes. Only exact matches will pass the filter. Either or both type and source can
-// use the value 'Any' to indicate all strings match.
+// use the value '' to indicate all strings match.
 type TriggerFilterSourceAndType struct {
 	Type   string `json:"type,omitempty"`
 	Source string `json:"source,omitempty"`
 }
 
+// TriggerFilterAttributes is a map of context attribute names to values for
+// filtering by equality. Only exact matches will pass the filter. You can use the value ''
+// to indicate all strings match.
+type TriggerFilterAttributes map[string]string
+
 // TriggerStatus represents the current state of a Trigger.
 type TriggerStatus struct {
 	// inherits duck/v1beta1 Status, which currently provides:

pkg/apis/eventing/v1alpha1/trigger_validation.go

@@ -18,15 +18,24 @@ package v1alpha1
 
 import (
 	"context"
+	"fmt"
+	"regexp"
 
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/kmp"
 )
 
+var (
+	// Only allow lowercase alphanumeric, starting with letters.
+	validAttributeName = regexp.MustCompile(`^[a-z][a-z0-9]*$`)
+)
+
+// Validate the Trigger.
 func (t *Trigger) Validate(ctx context.Context) *apis.FieldError {
 	return t.Spec.Validate(ctx).ViaField("spec")
 }
 
+// Validate the TriggerSpec.
 func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 	var errs *apis.FieldError
 	if ts.Broker == "" {
@@ -39,9 +48,39 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(fe)
 	}
 
-	if ts.Filter != nil && ts.Filter.SourceAndType == nil {
-		fe := apis.ErrMissingField("filter.sourceAndType")
-		errs = errs.Also(fe)
+	if ts.Filter != nil {
+		filtersSpecified := make([]string, 0)
+
+		if ts.Filter.DeprecatedSourceAndType != nil {
+			filtersSpecified = append(filtersSpecified, "filter.sourceAndType")
+		}
+
+		if ts.Filter.Attributes != nil {
+			filtersSpecified = append(filtersSpecified, "filter.attributes")
+			if len(*ts.Filter.Attributes) == 0 {
+				fe := &apis.FieldError{
+					Message: "At least one filtered attribute must be specified",
+					Paths:   []string{"filter.attributes"},
+				}
+				errs = errs.Also(fe)
+			} else {
+				attrs := map[string]string(*ts.Filter.Attributes)
+				for attr := range attrs {
+					if !validAttributeName.MatchString(attr) {
+						fe := &apis.FieldError{
+							Message: fmt.Sprintf("Invalid attribute name: %s", attr),
+							Paths:   []string{"filter.attributes"},
+						}
+						errs = errs.Also(fe)
+					}
+				}
+			}
+		}
+
+		if len(filtersSpecified) > 1 {
+			fe := apis.ErrMultipleOneOf(filtersSpecified...)
+			errs = errs.Also(fe)
+		}
 	}
 
 	if isSubscriberSpecNilOrEmpty(ts.Subscriber) {
@@ -54,6 +93,7 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 	return errs
 }
 
+// CheckImmutableFields checks that any immutable fields were not changed.
 func (t *Trigger) CheckImmutableFields(ctx context.Context, og apis.Immutable) *apis.FieldError {
 	if og == nil {
 		return nil

pkg/apis/eventing/v1alpha1/trigger_validation_test.go

@@ -27,10 +27,18 @@ import (
 )
 
 var (
-	validTriggerFilter = &TriggerFilter{
-		SourceAndType: &TriggerFilterSourceAndType{
+	validEmptyFilter         = &TriggerFilter{}
+	validSourceAndTypeFilter = &TriggerFilter{
+		DeprecatedSourceAndType: &TriggerFilterSourceAndType{
 			Type:   "other_type",
-			Source: "other_source"},
+			Source: "other_source",
+		},
+	}
+	validAttributesFilter = &TriggerFilter{
+		Attributes: &TriggerFilterAttributes{
+			"type":   "other_type",
+			"source": "other_source",
+		},
 	}
 	validSubscriber = &SubscriberSpec{
 		Ref: &corev1.ObjectReference{
@@ -80,7 +88,7 @@ func TestTriggerSpecValidation(t *testing.T) {
 		name: "missing broker",
 		ts: &TriggerSpec{
 			Broker:     "",
-			Filter:     validTriggerFilter,
+			Filter:     validSourceAndTypeFilter,
 			Subscriber: validSubscriber,
 		},
 		want: func() *apis.FieldError {
@@ -98,21 +106,72 @@ func TestTriggerSpecValidation(t *testing.T) {
 			return fe
 		}(),
 	}, {
-		name: "missing filter.sourceAndType",
+		name: "missing attributes keys",
 		ts: &TriggerSpec{
-			Broker:     "test_broker",
-			Filter:     &TriggerFilter{},
+			Broker: "test_broker",
+			Filter: &TriggerFilter{
+				Attributes: &TriggerFilterAttributes{},
+			},
+			Subscriber: validSubscriber,
+		},
+		want: &apis.FieldError{
+			Message: "At least one filtered attribute must be specified",
+			Paths:   []string{"filter.attributes"},
+		},
+	}, {
+		name: "invalid attribute name, start with number",
+		ts: &TriggerSpec{
+			Broker: "test_broker",
+			Filter: &TriggerFilter{
+				Attributes: &TriggerFilterAttributes{
+					"0invalid": "my-value",
+				},
+			},
+			Subscriber: validSubscriber,
+		},
+		want: &apis.FieldError{
+			Message: "Invalid attribute name: 0invalid",
+			Paths:   []string{"filter.attributes"},
+		},
+	}, {
+		name: "invalid attribute name, capital letters",
+		ts: &TriggerSpec{
+			Broker: "test_broker",
+			Filter: &TriggerFilter{
+				Attributes: &TriggerFilterAttributes{
+					"invALID": "my-value",
+				},
+			},
+			Subscriber: validSubscriber,
+		},
+		want: &apis.FieldError{
+			Message: "Invalid attribute name: invALID",
+			Paths:   []string{"filter.attributes"},
+		},
+	}, {
+		name: "multiple oneof sourceAndType and attributes",
+		ts: &TriggerSpec{
+			Broker: "test_broker",
+			Filter: &TriggerFilter{
+				DeprecatedSourceAndType: &TriggerFilterSourceAndType{
+					Type:   "other_type",
+					Source: "other_source",
+				},
+				Attributes: &TriggerFilterAttributes{
+					"type": "other_type",
+				},
+			},
 			Subscriber: validSubscriber,
 		},
 		want: func() *apis.FieldError {
-			fe := apis.ErrMissingField("filter.sourceAndType")
+			fe := apis.ErrMultipleOneOf("filter.sourceAndType", "filter.attributes")
 			return fe
 		}(),
 	}, {
 		name: "missing subscriber",
 		ts: &TriggerSpec{
 			Broker: "test_broker",
-			Filter: validTriggerFilter,
+			Filter: validSourceAndTypeFilter,
 		},
 		want: func() *apis.FieldError {
 			fe := apis.ErrMissingField("subscriber")
@@ -122,15 +181,49 @@ func TestTriggerSpecValidation(t *testing.T) {
 		name: "missing subscriber.ref.name",
 		ts: &TriggerSpec{
 			Broker:     "test_broker",
-			Filter:     validTriggerFilter,
+			Filter:     validSourceAndTypeFilter,
 			Subscriber: invalidSubscriber,
 		},
 		want: func() *apis.FieldError {
 			fe := apis.ErrMissingField("subscriber.ref.name")
 			return fe
 		}(),
-	},
-	}
+	}, {
+		name: "missing broker",
+		ts: &TriggerSpec{
+			Broker:     "",
+			Filter:     validSourceAndTypeFilter,
+			Subscriber: validSubscriber,
+		},
+		want: func() *apis.FieldError {
+			fe := apis.ErrMissingField("broker")
+			return fe
+		}(),
+	}, {
+		name: "valid empty filter",
+		ts: &TriggerSpec{
+			Broker:     "test_broker",
+			Filter:     validEmptyFilter,
+			Subscriber: validSubscriber,
+		},
+		want: &apis.FieldError{},
+	}, {
+		name: "valid SourceAndType filter",
+		ts: &TriggerSpec{
+			Broker:     "test_broker",
+			Filter:     validSourceAndTypeFilter,
+			Subscriber: validSubscriber,
+		},
+		want: &apis.FieldError{},
+	}, {
+		name: "valid Attributes filter",
+		ts: &TriggerSpec{
+			Broker:     "test_broker",
+			Filter:     validAttributesFilter,
+			Subscriber: validSubscriber,
+		},
+		want: &apis.FieldError{},
+	}}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
@@ -191,7 +284,7 @@ func TestTriggerImmutableFields(t *testing.T) {
 		original: &Trigger{
 			Spec: TriggerSpec{
 				Broker: "broker",
-				Filter: validTriggerFilter,
+				Filter: validSourceAndTypeFilter,
 			},
 		},
 		want: nil,