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/inmemorychannel/controller/controller.go

@@ -18,19 +18,26 @@ package controller
 
 import (
 	"context"
-
-	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/logging"
+	"strings"
 
 	"github.com/kelseyhightower/envconfig"
+	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"
+	"knative.dev/eventing/pkg/apis/messaging"
+	v1 "knative.dev/eventing/pkg/client/listers/messaging/v1"
+	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 +72,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 +83,7 @@ func NewController(
 		serviceAccountLister: serviceAccountInformer.Lister(),
 		roleBindingLister:    roleBindingInformer.Lister(),
 		secretLister:         secretInformer.Lister(),
+		eventPolicyLister:    eventPolicyInformer.Lister(),
 	}
 
 	env := &envConfig{}
@@ -140,10 +149,88 @@ func NewController(
 		Handler:    controller.HandleAll(globalResync),
 	})
 
+	// Enqueue the InMemoryChannel, if we have an EventPolicy which was referencing
+	// or got updated and now is referencing the InMemoryChannel
+	eventPolicyInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			enqueueApplyingChannelOfEventPolicy(inmemorychannelInformer.Lister(), obj, impl.EnqueueKey)
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			// Here we need to check if either the old or the new EventPolicy was referencing the InMemoryChannel
+
+			alreadyEnqueued := enqueueApplyingChannelOfEventPolicy(inmemorychannelInformer.Lister(), oldObj, impl.EnqueueKey)
+			if !alreadyEnqueued {
+				enqueueApplyingChannelOfEventPolicy(inmemorychannelInformer.Lister(), newObj, impl.EnqueueKey)
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			enqueueApplyingChannelOfEventPolicy(inmemorychannelInformer.Lister(), obj, impl.EnqueueKey)
+		},
+	})
+
 	// Setup the watch on the config map of dispatcher config
 	configStore := config.NewEventDispatcherConfigStore(logging.FromContext(ctx))
 	configStore.WatchConfigs(cmw)
 	r.eventDispatcherConfigStore = configStore
 
 	return impl
 }
+
+// enqueueApplyingChannelOfEventPolicy checks if an InMemoryChannel is referenced in the given EventPolicy.
+// If so, it enqueues the channel into the enqueueFn() and returns true.
+func enqueueApplyingChannelOfEventPolicy(imcLister v1.InMemoryChannelLister, obj interface{}, enqueueFn func(key types.NamespacedName)) bool {
+	eventPolicy, ok := obj.(*v1alpha1.EventPolicy)
+	if !ok {
+		return false
+	}
+
+	for _, to := range eventPolicy.Spec.To {
+		if to.Ref != nil {
+			toGV, err := schema.ParseGroupVersion(to.Ref.APIVersion)
+			if err != nil {
+				return false
+			}
+
+			if strings.EqualFold(toGV.Group, messaging.GroupName) &&
+				strings.EqualFold(to.Ref.Kind, "InMemoryChannel") {
+
+				enqueueFn(types.NamespacedName{
+					Namespace: eventPolicy.Namespace,
+					Name:      to.Ref.Name,
+				})
+				return true
+			}
+		}
+
+		if to.Selector != nil {
+			selectorGV, err := schema.ParseGroupVersion(to.Selector.APIVersion)
+			if err != nil {
+				return false
+			}
+
+			if strings.EqualFold(selectorGV.Group, messaging.GroupName) &&
+				strings.EqualFold(to.Selector.Kind, "InMemoryChannel") {
+
+				selector, err := metav1.LabelSelectorAsSelector(to.Selector.LabelSelector)
+				if err != nil {
+					return false
+				}
+
+				imcs, err := imcLister.InMemoryChannels(eventPolicy.Namespace).List(selector)
+				if err != nil {
+					return false
+				}
+
+				for _, imc := range imcs {
+					enqueueFn(types.NamespacedName{
+						Namespace: eventPolicy.Namespace,
+						Name:      imc.Name,
+					})
+				}
+				return true
+			}
+		}
+	}
+
+	return false
+}

pkg/reconciler/inmemorychannel/controller/controller_test.go

@@ -31,6 +31,7 @@ import (
 	. "knative.dev/pkg/reconciler/testing"
 
 	// Fake injection informers
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1alpha1/eventpolicy/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/inmemorychannel/fake"
 	"knative.dev/eventing/pkg/reconciler/inmemorychannel/controller/config"