Add SubscriptionsAPI filters to APIServerSource

This MR introduces the `filters` key in the APIServerSource Spec.
This new field allows users to filter which messages are sent from the
APIServerSource to the specified sink. The filter language
is the new SubscriptionsAPI, that allows for powerful filtering.

Signed-off-by: Hector Martinez <[email protected]>

config/core/configmaps/features.yaml

@@ -60,3 +60,6 @@ data:
   # For more details: https://github.com/knative/eventing/issues/7739
   cross-namespace-event-links: "disabled"
 
+  # ALPHA feature: The new-apiserversource-filters flag allows you to use the new `filters` field
+  # in APIServerSource objects with its rich filtering capabilities.
+  new-apiserversource-filters: "enabled"

config/core/resources/apiserversource.yaml

@@ -67,6 +67,7 @@ spec:
         properties:
           spec:
             type: object
+            x-kubernetes-preserve-unknown-fields: true
             required:
               - resources
             properties:

docs/eventing-api.md

@@ -2089,7 +2089,7 @@ resolved delivery options.</p>
 <h3 id="eventing.knative.dev/v1.SubscriptionsAPIFilter">SubscriptionsAPIFilter
 </h3>
 <p>
-(<em>Appears on:</em><a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">SubscriptionsAPIFilter</a>, <a href="#eventing.knative.dev/v1.TriggerSpec">TriggerSpec</a>)
+(<em>Appears on:</em><a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">SubscriptionsAPIFilter</a>, <a href="#eventing.knative.dev/v1.TriggerSpec">TriggerSpec</a>, <a href="#sources.knative.dev/v1.ApiServerSourceSpec">ApiServerSourceSpec</a>)
 </p>
 <p>
 <p>SubscriptionsAPIFilter allows defining a filter expression using CloudEvents
@@ -5313,6 +5313,25 @@ Kubernetes meta/v1.LabelSelector
 should be watched by the source.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>filters</code><br/>
+<em>
+<a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">
+[]SubscriptionsAPIFilter
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Filters is an experimental field that conforms to the CNCF CloudEvents Subscriptions
+API. It&rsquo;s an array of filter expressions that evaluate to true or false.
+If any filter expression in the array evaluates to false, the event MUST
+NOT be sent to the Sink. If all the filter expressions in the array
+evaluate to true, the event MUST be attempted to be delivered. Absence of
+a filter or empty array implies a value of true.</p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -5920,6 +5939,25 @@ Kubernetes meta/v1.LabelSelector
 should be watched by the source.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>filters</code><br/>
+<em>
+<a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">
+[]SubscriptionsAPIFilter
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Filters is an experimental field that conforms to the CNCF CloudEvents Subscriptions
+API. It&rsquo;s an array of filter expressions that evaluate to true or false.
+If any filter expression in the array evaluates to false, the event MUST
+NOT be sent to the Sink. If all the filter expressions in the array
+evaluate to true, the event MUST be attempted to be delivered. Absence of
+a filter or empty array implies a value of true.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="sources.knative.dev/v1.ApiServerSourceStatus">ApiServerSourceStatus

pkg/adapter/apiserver/adapter.go

@@ -71,6 +71,7 @@ func (a *apiServerAdapter) start(ctx context.Context, stopCh <-chan struct{}) er
 		logger:              a.logger,
 		ref:                 a.config.EventMode == v1.ReferenceMode,
 		apiServerSourceName: a.name,
+		filters:             a.config.Filters,
 	}
 	if a.config.ResourceOwner != nil {
 		a.logger.Infow("will be filtered",

pkg/adapter/apiserver/config.go

@@ -17,6 +17,8 @@ package apiserver
 
 import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	v1 "knative.dev/eventing/pkg/apis/sources/v1"
 )
 
@@ -56,4 +58,14 @@ type Config struct {
 	// Defaults to `Reference`
 	// +optional
 	EventMode string `json:"mode,omitempty"`
+
+	// Filters is an experimental field that conforms to the CNCF CloudEvents Subscriptions
+	// API. It's an array of filter expressions that evaluate to true or false.
+	// If any filter expression in the array evaluates to false, the event MUST
+	// NOT be sent to the Sink. If all the filter expressions in the array
+	// evaluate to true, the event MUST be attempted to be delivered. Absence of
+	// a filter or empty array implies a value of true.
+	//
+	// +optional
+	Filters []eventingv1.SubscriptionsAPIFilter `json:"filters,omitempty"`
 }

pkg/adapter/apiserver/delegate.go

@@ -18,51 +18,69 @@ package apiserver
 
 import (
 	"context"
+	"sync"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/google/uuid"
 	"go.uber.org/zap"
 	"k8s.io/client-go/tools/cache"
 	"knative.dev/eventing/pkg/adapter/apiserver/events"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	brokerfilter "knative.dev/eventing/pkg/broker/filter"
+	"knative.dev/eventing/pkg/eventfilter"
+	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
 )
 
 type resourceDelegate struct {
 	ce                  cloudevents.Client
 	source              string
 	ref                 bool
 	apiServerSourceName string
+	filters             []eventingv1.SubscriptionsAPIFilter
 
 	logger *zap.SugaredLogger
 }
 
 var _ cache.Store = (*resourceDelegate)(nil)
+var filterSync sync.Once
+var filter eventfilter.Filter
 
 func (a *resourceDelegate) Add(obj interface{}) error {
-	ctx, event, err := events.MakeAddEvent(a.source, a.apiServerSourceName, obj, a.ref)
-	if err != nil {
-		a.logger.Infow("event creation failed", zap.Error(err))
-		return err
-	}
-	a.sendCloudEvent(ctx, event)
-	return nil
+	return a.handleKubernetesObject(events.MakeAddEvent, obj)
 }
 
 func (a *resourceDelegate) Update(obj interface{}) error {
-	ctx, event, err := events.MakeUpdateEvent(a.source, a.apiServerSourceName, obj, a.ref)
-	if err != nil {
-		a.logger.Info("event creation failed", zap.Error(err))
-		return err
-	}
-	a.sendCloudEvent(ctx, event)
-	return nil
+	return a.handleKubernetesObject(events.MakeUpdateEvent, obj)
 }
 
 func (a *resourceDelegate) Delete(obj interface{}) error {
-	ctx, event, err := events.MakeDeleteEvent(a.source, a.apiServerSourceName, obj, a.ref)
+	return a.handleKubernetesObject(events.MakeDeleteEvent, obj)
+
+}
+
+// makeEventFunc represents the signature of the functions `events.Make*Event` so they can
+// be passed as a parameter
+type makeEventFunc func(string, string, interface{}, bool) (context.Context, cloudevents.Event, error)
+
+func (a *resourceDelegate) handleKubernetesObject(makeEvent makeEventFunc, obj interface{}) error {
+	ctx, event, err := makeEvent(a.source, a.apiServerSourceName, obj, a.ref)
+
 	if err != nil {
-		a.logger.Info("event creation failed", zap.Error(err))
+		a.logger.Infow("event creation failed", zap.Error(err))
 		return err
 	}
+
+	filterSync.Do(func() {
+		a.logger.Debug("initializing filters once")
+		filter = subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(a.logger.Desugar(), a.filters)...)
+	})
+
+	filterResult := filter.Filter(ctx, event)
+	if filterResult == eventfilter.FailFilter {
+		a.logger.Debugf("event type %s filtered out", event.Type())
+		return nil
+	}
+
 	a.sendCloudEvent(ctx, event)
 	return nil
 }

pkg/apis/feature/features.go

@@ -61,6 +61,7 @@ func newDefaults() Flags {
 		TransportEncryption: Disabled,
 		OIDCAuthentication:  Disabled,
 		EvenTypeAutoCreate:  Disabled,
+		NewAPIServerFilters: Enabled,
 	}
 }
 

pkg/apis/feature/flag_names.go

@@ -27,4 +27,5 @@ const (
 	OIDCAuthentication       = "authentication-oidc"
 	NodeSelectorLabel        = "apiserversources-nodeselector-"
 	CrossNamespaceEventLinks = "cross-namespace-event-links"
+	NewAPIServerFilters      = "new-apiserversource-filters"
 )

pkg/apis/sources/v1/apiserver_types.go

@@ -19,6 +19,7 @@ package v1
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
@@ -85,6 +86,16 @@ type ApiServerSourceSpec struct {
 	// should be watched by the source.
 	// +optional
 	NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty"`
+
+	// Filters is an experimental field that conforms to the CNCF CloudEvents Subscriptions
+	// API. It's an array of filter expressions that evaluate to true or false.
+	// If any filter expression in the array evaluates to false, the event MUST
+	// NOT be sent to the Sink. If all the filter expressions in the array
+	// evaluate to true, the event MUST be attempted to be delivered. Absence of
+	// a filter or empty array implies a value of true.
+	//
+	// +optional
+	Filters []eventingv1.SubscriptionsAPIFilter `json:"filters,omitempty"`
 }
 
 // ApiServerSourceStatus defines the observed state of ApiServerSource

pkg/apis/sources/v1/apiserver_validation.go

@@ -22,6 +22,8 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 )
 
@@ -73,5 +75,22 @@ func (cs *ApiServerSourceSpec) Validate(ctx context.Context) *apis.FieldError {
 		}
 	}
 	errs = errs.Also(cs.SourceSpec.Validate(ctx))
+	errs = errs.Also(validateSubscriptionAPIFiltersList(ctx, cs.Filters).ViaField("filters"))
+	return errs
+}
+
+func validateSubscriptionAPIFiltersList(ctx context.Context, filters []eventingv1.SubscriptionsAPIFilter) (errs *apis.FieldError) {
+	if !feature.FromContext(ctx).IsEnabled(feature.NewAPIServerFilters) {
+		if len(filters) != 0 {
+			return errs.Also(apis.ErrGeneric("Filters is not empty but the NewAPIServerFilters feature is disabled."))
+		}
+
+		return nil
+	}
+
+	for i, f := range filters {
+		f := f
+		errs = errs.Also(eventingv1.ValidateSubscriptionAPIFilter(ctx, &f)).ViaIndex(i)
+	}
 	return errs
 }

pkg/apis/sources/v1/apiserver_validation_test.go

@@ -23,9 +23,11 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"github.com/google/go-cmp/cmp"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 )
 
@@ -266,3 +268,72 @@ func TestAPIServerValidationCallsSpecValidation(t *testing.T) {
 	err := source.Validate(context.TODO())
 	assert.EqualError(t, err, "missing field(s): spec.resources", "Spec is not validated!")
 }
+
+func TestAPIServerFiltersValidation(t *testing.T) {
+	tests := []struct {
+		name         string
+		featureState feature.Flag
+		want         error
+		filters      []eventingv1.SubscriptionsAPIFilter
+	}{{
+		name:         "an error is raised if the feature is disabled but filters are specified",
+		featureState: feature.Disabled,
+		filters: []eventingv1.SubscriptionsAPIFilter{{
+			Prefix: map[string]string{
+				"invALID": "abc",
+			},
+		}},
+		want: apis.ErrGeneric("Filters is not empty but the NewAPIServerFilters feature is disabled."),
+	}, {
+		name:         "filters are validated when the feature is enabled",
+		featureState: feature.Enabled,
+		filters: []eventingv1.SubscriptionsAPIFilter{{
+			Prefix: map[string]string{
+				"invALID": "abc",
+			},
+		}},
+		want: apis.ErrInvalidKeyName("invALID", apis.CurrentField,
+			"Attribute name must start with a letter and can only contain "+
+				"lowercase alphanumeric").ViaFieldKey("prefix", "invALID").ViaFieldIndex("filters", 0),
+	}, {
+		name:         "validation works for valid filters",
+		featureState: feature.Enabled,
+		filters: []eventingv1.SubscriptionsAPIFilter{{
+			Exact: map[string]string{"myattr": "myval"},
+		}},
+		want: nil,
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			featureContext := feature.ToContext(context.TODO(), feature.Flags{
+				feature.NewAPIServerFilters: test.featureState,
+			})
+			apiserversource := &ApiServerSourceSpec{
+				Filters:   test.filters,
+				EventMode: "Resource",
+				Resources: []APIVersionKindSelector{{
+					APIVersion: "v1",
+					Kind:       "Foo",
+				}},
+				SourceSpec: duckv1.SourceSpec{
+					Sink: duckv1.Destination{
+						Ref: &duckv1.KReference{
+							APIVersion: "v1",
+							Kind:       "broker",
+							Name:       "default",
+						},
+					},
+				},
+			}
+			got := apiserversource.Validate(featureContext)
+			if test.want != nil {
+				if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
+					t.Errorf("APIServerSourceSpec.Validate (-want, +got) = %v", diff)
+				}
+			} else if got != nil {
+				t.Errorf("APIServerSourceSpec.Validate wanted nil, got = %v", got.Error())
+			}
+		})
+	}
+}