List applying policies in InMemoryChannels status (#8011)

* List applying EventPolicies in InMemoryChannels status

* Add & update integration tests

* Reconcile IMC only on relevant EventPolicy changes

* Only list ready EventPolicies in .status.policies

* Make reason for not being ready a bit more speaking (NotReady -> EventPoliciesNotReady)

* Make EventPolicy EventHandler generic

* Move EventPolicyEventHandler to auth package

* Refactor to use EventPolicy helper functions from reconciler/testing package

* Add unit test for GetApplyingResourcesOfEventPolicyForGK

* Fix linter issues

* Match everything in namespace on empty .spec.to

* GetApplyingResourcesOfEventPolicyForGK should return list of unique elements

* Added very basic unit test for EventPolicyEventHandler

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/auth/event_policy.go

@@ -20,6 +20,9 @@ import (
 	"fmt"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/cache"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -86,6 +89,71 @@ func GetEventPoliciesForResource(lister listerseventingv1alpha1.EventPolicyListe
 	return relevantPolicies, nil
 }
 
+// GetApplyingResourcesOfEventPolicyForGK returns all applying resource names of GK of the given event policy.
+// It returns only the names, as the resources are part of the same namespace as the event policy.
+//
+// This function is kind of the "inverse" of GetEventPoliciesForResource.
+func GetApplyingResourcesOfEventPolicyForGK(eventPolicy *v1alpha1.EventPolicy, gk schema.GroupKind, gkIndexer cache.Indexer) ([]string, error) {
+	applyingResources := map[string]struct{}{}
+
+	if eventPolicy.Spec.To == nil {
+		// empty .spec.to matches everything in namespace
+
+		err := cache.ListAllByNamespace(gkIndexer, eventPolicy.Namespace, labels.Everything(), func(i interface{}) {
+			name := i.(metav1.Object).GetName()
+			applyingResources[name] = struct{}{}
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to list all %s %s resources in %s: %w", gk.Group, gk.Kind, eventPolicy.Namespace, err)
+		}
+	} else {
+		for _, to := range eventPolicy.Spec.To {
+			if to.Ref != nil {
+				toGV, err := schema.ParseGroupVersion(to.Ref.APIVersion)
+				if err != nil {
+					return nil, fmt.Errorf("could not parse group version of %q: %w", to.Ref.APIVersion, err)
+				}
+
+				if strings.EqualFold(toGV.Group, gk.Group) &&
+					strings.EqualFold(to.Ref.Kind, gk.Kind) {
+
+					applyingResources[to.Ref.Name] = struct{}{}
+				}
+			}
+
+			if to.Selector != nil {
+				selectorGV, err := schema.ParseGroupVersion(to.Selector.APIVersion)
+				if err != nil {
+					return nil, fmt.Errorf("could not parse group version of %q: %w", to.Selector.APIVersion, err)
+				}
+
+				if strings.EqualFold(selectorGV.Group, gk.Group) &&
+					strings.EqualFold(to.Selector.Kind, gk.Kind) {
+
+					selector, err := metav1.LabelSelectorAsSelector(to.Selector.LabelSelector)
+					if err != nil {
+						return nil, fmt.Errorf("could not parse label selector %v: %w", to.Selector.LabelSelector, err)
+					}
+
+					err = cache.ListAllByNamespace(gkIndexer, eventPolicy.Namespace, selector, func(i interface{}) {
+						name := i.(metav1.Object).GetName()
+						applyingResources[name] = struct{}{}
+					})
+					if err != nil {
+						return nil, fmt.Errorf("could not list resources of GK in %q namespace for selector %v: %w", eventPolicy.Namespace, selector, err)
+					}
+				}
+			}
+		}
+	}
+
+	res := []string{}
+	for name := range applyingResources {
+		res = append(res, name)
+	}
+	return res, nil
+}
+
 // ResolveSubjects returns the OIDC service accounts names for the objects referenced in the EventPolicySpecFrom.
 func ResolveSubjects(resolver *resolver.AuthenticatableResolver, eventPolicy *v1alpha1.EventPolicy) ([]string, error) {
 	allSAs := []string{}
@@ -145,3 +213,77 @@ func SubjectContained(sub string, allowedSubs []string) bool {
 
 	return false
 }
+
+func handleApplyingResourcesOfEventPolicy(eventPolicy *v1alpha1.EventPolicy, gk schema.GroupKind, indexer cache.Indexer, handlerFn func(key types.NamespacedName) error) error {
+	applyingResources, err := GetApplyingResourcesOfEventPolicyForGK(eventPolicy, gk, indexer)
+	if err != nil {
+		return fmt.Errorf("could not get applying resources of eventpolicy: %w", err)
+	}
+
+	for _, resourceName := range applyingResources {
+		err := handlerFn(types.NamespacedName{
+			Namespace: eventPolicy.Namespace,
+			Name:      resourceName,
+		})
+
+		if err != nil {
+			return fmt.Errorf("could not handle resource %q: %w", resourceName, err)
+		}
+	}
+
+	return nil
+}
+
+// EventPolicyEventHandler returns an ResourceEventHandler, which passes the referencing resources of the EventPolicy
+// to the enqueueFn if the EventPolicy was referencing or got updated and now is referencing the resource of the given GVK.
+func EventPolicyEventHandler(indexer cache.Indexer, gk schema.GroupKind, enqueueFn func(key types.NamespacedName)) cache.ResourceEventHandler {
+	return cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			eventPolicy, ok := obj.(*v1alpha1.EventPolicy)
+			if !ok {
+				return
+			}
+
+			handleApplyingResourcesOfEventPolicy(eventPolicy, gk, indexer, func(key types.NamespacedName) error {
+				enqueueFn(key)
+				return nil
+			})
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			// Here we need to check if the old or the new EventPolicy was referencing the given GVK
+			oldEventPolicy, ok := oldObj.(*v1alpha1.EventPolicy)
+			if !ok {
+				return
+			}
+			newEventPolicy, ok := newObj.(*v1alpha1.EventPolicy)
+			if !ok {
+				return
+			}
+
+			// make sure, we handle the keys only once
+			toHandle := map[types.NamespacedName]struct{}{}
+			addToHandleList := func(key types.NamespacedName) error {
+				toHandle[key] = struct{}{}
+				return nil
+			}
+
+			handleApplyingResourcesOfEventPolicy(oldEventPolicy, gk, indexer, addToHandleList)
+			handleApplyingResourcesOfEventPolicy(newEventPolicy, gk, indexer, addToHandleList)
+
+			for k := range toHandle {
+				enqueueFn(k)
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			eventPolicy, ok := obj.(*v1alpha1.EventPolicy)
+			if !ok {
+				return
+			}
+
+			handleApplyingResourcesOfEventPolicy(eventPolicy, gk, indexer, func(key types.NamespacedName) error {
+				enqueueFn(key)
+				return nil
+			})
+		},
+	}
+}