Add SubscriptionsAPI filters to APIServerSource

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: "disabled"

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

@@ -2101,7 +2101,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
@@ -5327,6 +5327,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>
@@ -5934,6 +5953,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

@@ -34,6 +34,8 @@ import (
 
 	"knative.dev/eventing/pkg/adapter/v2"
 	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+	brokerfilter "knative.dev/eventing/pkg/broker/filter"
+	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
 )
 
 type envConfig struct {
@@ -71,6 +73,7 @@ func (a *apiServerAdapter) start(ctx context.Context, stopCh <-chan struct{}) er
 		logger:              a.logger,
 		ref:                 a.config.EventMode == v1.ReferenceMode,
 		apiServerSourceName: a.name,
+		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(a.logger.Desugar(), a.config.Filters)...),
 	}
 	if a.config.ResourceOwner != nil {
 		a.logger.Infow("will be filtered",

pkg/adapter/apiserver/adapter_test.go

@@ -34,6 +34,9 @@ import (
 	dynamicfake "k8s.io/client-go/dynamic/fake"
 	kubetesting "k8s.io/client-go/testing"
 	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	brokerfilter "knative.dev/eventing/pkg/broker/filter"
+	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
 	rectesting "knative.dev/eventing/pkg/reconciler/testing"
 	"knative.dev/pkg/logging"
 	pkgtesting "knative.dev/pkg/reconciler/testing"
@@ -289,21 +292,27 @@ func validateNotSent(t *testing.T, ce *adaptertest.TestCloudEventsClient, want s
 
 func makeResourceAndTestingClient() (*resourceDelegate, *adaptertest.TestCloudEventsClient) {
 	ce := adaptertest.NewTestClient()
+	logger := zap.NewExample().Sugar()
+
 	return &resourceDelegate{
 		ce:                  ce,
 		source:              "unit-test",
 		apiServerSourceName: apiServerSourceNameTest,
-		logger:              zap.NewExample().Sugar(),
+		logger:              logger,
+		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), []eventingv1.SubscriptionsAPIFilter{})...),
 	}, ce
 }
 
 func makeRefAndTestingClient() (*resourceDelegate, *adaptertest.TestCloudEventsClient) {
 	ce := adaptertest.NewTestClient()
+	logger := zap.NewExample().Sugar()
+
 	return &resourceDelegate{
 		ce:                  ce,
 		source:              "unit-test",
 		apiServerSourceName: apiServerSourceNameTest,
 		logger:              zap.NewExample().Sugar(),
 		ref:                 true,
+		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), []eventingv1.SubscriptionsAPIFilter{})...),
 	}, ce
 }

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

@@ -24,45 +24,52 @@ import (
 	"go.uber.org/zap"
 	"k8s.io/client-go/tools/cache"
 	"knative.dev/eventing/pkg/adapter/apiserver/events"
+	"knative.dev/eventing/pkg/eventfilter"
 )
 
 type resourceDelegate struct {
 	ce                  cloudevents.Client
 	source              string
 	ref                 bool
 	apiServerSourceName string
+	filter              eventfilter.Filter
 
 	logger *zap.SugaredLogger
 }
 
 var _ cache.Store = (*resourceDelegate)(nil)
 
 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
 	}
+
+	filterResult := a.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/adapter/apiserver/delegate_test.go

@@ -18,7 +18,12 @@ package apiserver
 import (
 	"testing"
 
+	"go.uber.org/zap"
+	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/sources"
+	brokerfilter "knative.dev/eventing/pkg/broker/filter"
+	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
 )
 
 func TestResourceAddEvent(t *testing.T) {
@@ -68,3 +73,40 @@ func TestResourceStub(t *testing.T) {
 	d.Replace(nil, "")
 	d.Resync()
 }
+
+func TestFilterFails(t *testing.T) {
+	ce := adaptertest.NewTestClient()
+	filters := []eventingv1.SubscriptionsAPIFilter{{
+		Exact: map[string]string{
+			"type": "dev.knative.apiserver.resource.add",
+		},
+	}}
+	logger := zap.NewExample().Sugar()
+	delegate := &resourceDelegate{
+		ce:                  ce,
+		source:              "unit-test",
+		apiServerSourceName: apiServerSourceNameTest,
+		logger:              logger,
+		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), filters)...),
+	}
+
+	delegate.Update(simplePod("unit", "test"))
+	validateNotSent(t, ce, sources.ApiServerSourceUpdateEventType)
+}
+
+func TestEmptyFiltersList(t *testing.T) {
+	ce := adaptertest.NewTestClient()
+	filters := []eventingv1.SubscriptionsAPIFilter{}
+
+	logger := zap.NewExample().Sugar()
+	delegate := &resourceDelegate{
+		ce:                  ce,
+		source:              "unit-test",
+		apiServerSourceName: apiServerSourceNameTest,
+		logger:              logger,
+		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), filters)...),
+	}
+
+	delegate.Update(simplePod("unit", "test"))
+	validateSent(t, ce, sources.ApiServerSourceUpdateEventType)
+}

pkg/apis/feature/features.go

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

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
 }