List applying policies in InMemoryChannels status

config/channels/in-memory-channel/roles/controller-clusterrole.yaml

@@ -45,6 +45,14 @@ rules:
       - inmemorychannels
     verbs:
       - patch
+  - apiGroups:
+      - eventing.knative.dev
+    resources:
+      - eventpolicies
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - ""
     resources:

pkg/apis/messaging/v1/in_memory_channel_lifecycle.go

@@ -33,6 +33,7 @@ var imcCondSet = apis.NewLivingConditionSet(
 	InMemoryChannelConditionAddressable,
 	InMemoryChannelConditionChannelServiceReady,
 	InMemoryChannelConditionDeadLetterSinkResolved,
+	InMemoryChannelConditionEventPoliciesReady,
 )
 
 const (
@@ -64,6 +65,10 @@ const (
 	// InMemoryChannelConditionDeadLetterSinkResolved has status True when there is a Dead Letter Sink ref or URI
 	// defined in the Spec.Delivery, is a valid destination and its correctly resolved into a valid URI
 	InMemoryChannelConditionDeadLetterSinkResolved apis.ConditionType = "DeadLetterSinkResolved"
+
+	// InMemoryChannelConditionEventPoliciesReady has status True when all the applying EventPolicies for this
+	// InMemoryChannel are ready.
+	InMemoryChannelConditionEventPoliciesReady apis.ConditionType = "EventPoliciesReady"
 )
 
 // GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
@@ -182,3 +187,19 @@ func (imcs *InMemoryChannelStatus) MarkDeadLetterSinkResolvedFailed(reason, mess
 	imcs.DeliveryStatus = eventingduck.DeliveryStatus{}
 	imcCondSet.Manage(imcs).MarkFalse(InMemoryChannelConditionDeadLetterSinkResolved, reason, messageFormat, messageA...)
 }
+
+func (imcs *InMemoryChannelStatus) MarkEventPoliciesFailed(reason, messageFormat string, messageA ...interface{}) {
+	imcCondSet.Manage(imcs).MarkFalse(InMemoryChannelConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}
+
+func (imcs *InMemoryChannelStatus) MarkEventPoliciesUnknown(reason, messageFormat string, messageA ...interface{}) {
+	imcCondSet.Manage(imcs).MarkUnknown(InMemoryChannelConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}
+
+func (imcs *InMemoryChannelStatus) MarkEventPoliciesTrue() {
+	imcCondSet.Manage(imcs).MarkTrue(InMemoryChannelConditionEventPoliciesReady)
+}
+
+func (imcs *InMemoryChannelStatus) MarkEventPoliciesTrueWithReason(reason, messageFormat string, messageA ...interface{}) {
+	imcCondSet.Manage(imcs).MarkTrueWithReason(InMemoryChannelConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}

pkg/apis/messaging/v1/in_memory_channel_lifecycle_test.go

@@ -137,6 +137,9 @@ func TestInMemoryChannelInitializeConditions(t *testing.T) {
 					}, {
 						Type:   InMemoryChannelConditionEndpointsReady,
 						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   InMemoryChannelConditionEventPoliciesReady,
+						Status: corev1.ConditionUnknown,
 					}, {
 						Type:   InMemoryChannelConditionReady,
 						Status: corev1.ConditionUnknown,
@@ -177,6 +180,9 @@ func TestInMemoryChannelInitializeConditions(t *testing.T) {
 					}, {
 						Type:   InMemoryChannelConditionEndpointsReady,
 						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   InMemoryChannelConditionEventPoliciesReady,
+						Status: corev1.ConditionUnknown,
 					}, {
 						Type:   InMemoryChannelConditionReady,
 						Status: corev1.ConditionUnknown,
@@ -217,6 +223,9 @@ func TestInMemoryChannelInitializeConditions(t *testing.T) {
 					}, {
 						Type:   InMemoryChannelConditionEndpointsReady,
 						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   InMemoryChannelConditionEventPoliciesReady,
+						Status: corev1.ConditionUnknown,
 					}, {
 						Type:   InMemoryChannelConditionReady,
 						Status: corev1.ConditionUnknown,
@@ -244,6 +253,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name                    string
 		markServiceReady        bool
 		markChannelServiceReady bool
+		markEventPolicyReady    bool
 		setAddress              bool
 		markEndpointsReady      bool
 		DLSResolved             *bool
@@ -253,6 +263,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name:                    "all happy",
 		markServiceReady:        true,
 		markChannelServiceReady: true,
+		markEventPolicyReady:    true,
 		markEndpointsReady:      true,
 		dispatcherStatus:        deploymentStatusReady,
 		setAddress:              true,
@@ -262,6 +273,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name:                    "service not ready",
 		markServiceReady:        false,
 		markChannelServiceReady: false,
+		markEventPolicyReady:    true,
 		markEndpointsReady:      true,
 		dispatcherStatus:        deploymentStatusReady,
 		setAddress:              true,
@@ -271,6 +283,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name:                    "endpoints not ready",
 		markServiceReady:        true,
 		markChannelServiceReady: false,
+		markEventPolicyReady:    true,
 		markEndpointsReady:      false,
 		dispatcherStatus:        deploymentStatusReady,
 		setAddress:              true,
@@ -281,6 +294,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		markServiceReady:        true,
 		markEndpointsReady:      true,
 		markChannelServiceReady: false,
+		markEventPolicyReady:    true,
 		dispatcherStatus:        deploymentStatusNotReady,
 		setAddress:              true,
 		wantReady:               false,
@@ -289,6 +303,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name:                    "address not set",
 		markServiceReady:        true,
 		markChannelServiceReady: false,
+		markEventPolicyReady:    true,
 		markEndpointsReady:      true,
 		dispatcherStatus:        deploymentStatusReady,
 		setAddress:              false,
@@ -298,6 +313,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name:                    "channel service not ready",
 		markServiceReady:        true,
 		markChannelServiceReady: false,
+		markEventPolicyReady:    true,
 		markEndpointsReady:      true,
 		dispatcherStatus:        deploymentStatusReady,
 		setAddress:              true,
@@ -307,6 +323,7 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name:                    "dls sad",
 		markServiceReady:        true,
 		markChannelServiceReady: false,
+		markEventPolicyReady:    true,
 		markEndpointsReady:      true,
 		dispatcherStatus:        deploymentStatusReady,
 		setAddress:              true,
@@ -316,6 +333,17 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 		name:                    "dls not configured",
 		markServiceReady:        true,
 		markChannelServiceReady: false,
+		markEventPolicyReady:    true,
+		markEndpointsReady:      true,
+		dispatcherStatus:        deploymentStatusReady,
+		setAddress:              true,
+		wantReady:               false,
+		DLSResolved:             &trueVal,
+	}, {
+		name:                    "EventPolicy not ready",
+		markServiceReady:        true,
+		markChannelServiceReady: true,
+		markEventPolicyReady:    false,
 		markEndpointsReady:      true,
 		dispatcherStatus:        deploymentStatusReady,
 		setAddress:              true,
@@ -336,6 +364,11 @@ func TestInMemoryChannelIsReady(t *testing.T) {
 			} else {
 				cs.MarkChannelServiceFailed("NotReadyChannelService", "testing")
 			}
+			if test.markEventPolicyReady {
+				cs.MarkEventPoliciesTrue()
+			} else {
+				cs.MarkEndpointsFailed("NotReadyEventPolicy", "testing")
+			}
 			if test.setAddress {
 				cs.SetAddress(&duckv1.Addressable{URL: &apis.URL{Scheme: "http", Host: "foo.bar"}})
 			}
@@ -437,6 +470,7 @@ func TestInMemoryChannelStatus_SetAddressable(t *testing.T) {
 func ReadyBrokerStatusWithoutDLS() *InMemoryChannelStatus {
 	imcs := &InMemoryChannelStatus{}
 	imcs.MarkChannelServiceTrue()
+	imcs.MarkEventPoliciesTrue()
 	imcs.MarkDeadLetterSinkNotConfigured()
 	imcs.MarkEndpointsTrue()
 	imcs.SetAddress(&duckv1.Addressable{URL: apis.HTTP("example.com")})

pkg/reconciler/channel/channel_test.go

@@ -135,7 +135,8 @@ func TestReconcile(t *testing.T) {
 				WithInMemoryChannelEndpointsReady(),
 				WithInMemoryChannelChannelServiceReady(),
 				WithInMemoryChannelAddress(backingChannelAddressable),
-				WithInMemoryChannelDLSUnknown()),
+				WithInMemoryChannelDLSUnknown(),
+				WithInMemoryChannelEventPoliciesReady()),
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: NewChannel(channelName, testNS,
@@ -165,7 +166,8 @@ func TestReconcile(t *testing.T) {
 				WithInMemoryChannelChannelServiceReady(),
 				WithInMemoryChannelSubscribers(subscribers()),
 				WithInMemoryChannelAddress(backingChannelAddressable),
-				WithInMemoryChannelDLSUnknown()),
+				WithInMemoryChannelDLSUnknown(),
+				WithInMemoryChannelEventPoliciesReady()),
 		},
 	}, {
 		Name: "Backing channel created",
@@ -259,7 +261,8 @@ func TestReconcile(t *testing.T) {
 				WithInMemoryChannelChannelServiceReady(),
 				WithInMemoryChannelSubscribers(subscribers()),
 				WithInMemoryChannelAddress(backingChannelAddressable),
-				WithInMemoryChannelDLSUnknown()),
+				WithInMemoryChannelDLSUnknown(),
+				WithInMemoryChannelEventPoliciesReady()),
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: NewChannel(channelName, testNS,
@@ -293,7 +296,8 @@ func TestReconcile(t *testing.T) {
 				WithInMemoryChannelAddress(backingChannelAddressable),
 				WithInMemoryChannelSubscribers(subscribers()),
 				WithInMemoryChannelStatusSubscribers(subscriberStatuses()),
-				WithInMemoryChannelDLSUnknown()),
+				WithInMemoryChannelDLSUnknown(),
+				WithInMemoryChannelEventPoliciesReady()),
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: NewChannel(channelName, testNS,

pkg/reconciler/eventpolicy.go

@@ -0,0 +1,113 @@
+/*
+Copyright 2023 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package reconciler
+
+import (
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/cache"
+	"knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+)
+
+// enqueueApplyingResourcesOfEventPolicy checks if the given GVK is referenced in the given EventPolicy.
+// If so, it enqueues it into the enqueueFn().
+func enqueueApplyingResourcesOfEventPolicy(indexer cache.Indexer, gvk schema.GroupVersionKind, policyObj interface{}, enqueueFn func(key types.NamespacedName)) {
+	eventPolicy, ok := policyObj.(*v1alpha1.EventPolicy)
+	if !ok {
+		return
+	}
+
+	for _, to := range eventPolicy.Spec.To {
+		if to.Ref != nil {
+			toGV, err := schema.ParseGroupVersion(to.Ref.APIVersion)
+			if err != nil {
+				continue
+			}
+
+			if strings.EqualFold(toGV.Group, gvk.Group) &&
+				strings.EqualFold(to.Ref.Kind, gvk.Kind) {
+
+				enqueueFn(types.NamespacedName{
+					Namespace: eventPolicy.Namespace,
+					Name:      to.Ref.Name,
+				})
+			}
+		}
+
+		if to.Selector != nil {
+			selectorGV, err := schema.ParseGroupVersion(to.Selector.APIVersion)
+			if err != nil {
+				continue
+			}
+
+			if strings.EqualFold(selectorGV.Group, gvk.Group) &&
+				strings.EqualFold(to.Selector.Kind, gvk.Kind) {
+
+				selector, err := metav1.LabelSelectorAsSelector(to.Selector.LabelSelector)
+				if err != nil {
+					continue
+				}
+
+				resources := []metav1.Object{}
+				err = cache.ListAllByNamespace(indexer, eventPolicy.Namespace, selector, func(i interface{}) {
+					resources = append(resources, i.(metav1.Object))
+				})
+				if err != nil {
+					continue
+				}
+
+				for _, resource := range resources {
+					enqueueFn(types.NamespacedName{
+						Namespace: resource.GetNamespace(),
+						Name:      resource.GetName(),
+					})
+				}
+			}
+		}
+	}
+}
+
+// EventPolicyEventHandler returns an ResourceEventHandler, which enqueues the referencing resources of the EventPolicy
+// if the EventPolicy was referencing or got updated and now is referencing the resource of the given GVK.
+func EventPolicyEventHandler(indexer cache.Indexer, gvk schema.GroupVersionKind, enqueueFn func(key types.NamespacedName)) cache.ResourceEventHandler {
+	return cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			enqueueApplyingResourcesOfEventPolicy(indexer, gvk, obj, enqueueFn)
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			// Here we need to check if the old or the new EventPolicy was referencing the given GVK
+
+			// make sure, we enqueue the keys only once
+			toEnqueue := map[types.NamespacedName]struct{}{}
+			addToEnqueueList := func(key types.NamespacedName) {
+				toEnqueue[key] = struct{}{}
+			}
+			enqueueApplyingResourcesOfEventPolicy(indexer, gvk, oldObj, addToEnqueueList)
+			enqueueApplyingResourcesOfEventPolicy(indexer, gvk, newObj, addToEnqueueList)
+
+			for k := range toEnqueue {
+				enqueueFn(k)
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			enqueueApplyingResourcesOfEventPolicy(indexer, gvk, obj, enqueueFn)
+		},
+	}
+}

pkg/reconciler/inmemorychannel/controller/controller.go

@@ -19,18 +19,20 @@ package controller
 import (
 	"context"
 
-	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/logging"
-
 	"github.com/kelseyhightower/envconfig"
 	"k8s.io/client-go/tools/cache"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	"knative.dev/eventing/pkg/reconciler"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
 
 	"knative.dev/pkg/resolver"
 
 	"knative.dev/eventing/pkg/apis/feature"
+	"knative.dev/eventing/pkg/client/injection/informers/eventing/v1alpha1/eventpolicy"
 	"knative.dev/eventing/pkg/client/injection/informers/messaging/v1/inmemorychannel"
 	inmemorychannelreconciler "knative.dev/eventing/pkg/client/injection/reconciler/messaging/v1/inmemorychannel"
 	"knative.dev/eventing/pkg/eventingtls"
@@ -65,6 +67,7 @@ func NewController(
 	serviceAccountInformer := serviceaccount.Get(ctx)
 	roleBindingInformer := rolebinding.Get(ctx)
 	secretInformer := secretinformer.Get(ctx)
+	eventPolicyInformer := eventpolicy.Get(ctx)
 
 	r := &Reconciler{
 		kubeClientSet:        kubeclient.Get(ctx),
@@ -75,6 +78,7 @@ func NewController(
 		serviceAccountLister: serviceAccountInformer.Lister(),
 		roleBindingLister:    roleBindingInformer.Lister(),
 		secretLister:         secretInformer.Lister(),
+		eventPolicyLister:    eventPolicyInformer.Lister(),
 	}
 
 	env := &envConfig{}
@@ -140,6 +144,12 @@ func NewController(
 		Handler:    controller.HandleAll(globalResync),
 	})
 
+	imcGVK := messagingv1.SchemeGroupVersion.WithKind("InMemoryChannel")
+
+	// Enqueue the InMemoryChannel, if we have an EventPolicy which was referencing
+	// or got updated and now is referencing the InMemoryChannel
+	eventPolicyInformer.Informer().AddEventHandler(reconciler.EventPolicyEventHandler(inmemorychannelInformer.Informer().GetIndexer(), imcGVK, impl.EnqueueKey))
+
 	// Setup the watch on the config map of dispatcher config
 	configStore := config.NewEventDispatcherConfigStore(logging.FromContext(ctx))
 	configStore.WatchConfigs(cmw)