Add EventPolicy reconciliation for Parallel

pkg/reconciler/parallel/parallel.go

@@ -22,6 +22,7 @@
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -36,6 +37,7 @@
 	pkgreconciler "knative.dev/pkg/reconciler"
 
 	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
 	"knative.dev/eventing/pkg/apis/feature"
 	v1 "knative.dev/eventing/pkg/apis/flows/v1"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
@@ -144,6 +146,11 @@
 		return fmt.Errorf("error removing unwanted Subscriptions: %w", err)
 	}
 
+	// Reconcile EventPolicies for the parallel.
+	if err := r.reconcileEventPolicies(ctx, p, ingressChannel, channels, filterSubs, featureFlags); err != nil {
+		return fmt.Errorf("failed to reconcile EventPolicies for Parallel: %w", err)
+	}
+
 	err := auth.UpdateStatusWithEventPolicies(featureFlags, &p.Status.AppliedEventPoliciesStatus, &p.Status, r.eventPolicyLister, v1.SchemeGroupVersion.WithKind("Parallel"), p.ObjectMeta)
 	if err != nil {
 		return fmt.Errorf("could not update parallel status with EventPolicies: %v", err)
@@ -349,3 +356,139 @@
 
 	return nil
 }
+
+func (r *Reconciler) reconcileEventPolicies(ctx context.Context, p *v1.Parallel, ingressChannel *duckv1.Channelable,
+	channels []*duckv1.Channelable, filterSubs []*messagingv1.Subscription, featureFlags feature.Flags) error {
+
+	if !featureFlags.IsOIDCAuthentication() {
+		return r.cleanupAllEventPolicies(ctx, p)
+	}
+	// list all the existing event policies for the parallel.
+	existingPolicies, err := r.listEventPoliciesForParallel(p)
+	if err != nil {
+		return fmt.Errorf("failed to list existing event policies for parallel: %w", err)
+	}
+	// make a map of existing event policies for easy and efficient lookup.
+	existingPolicyMap := make(map[string]*eventingv1alpha1.EventPolicy)
+	for _, policy := range existingPolicies {
+		existingPolicyMap[policy.Name] = policy
+	}
+
+	// prepare the list of event policies to create, update and delete.
+	var policiesToCreate, policiesToUpdate, policiesToDelete []*eventingv1alpha1.EventPolicy
+	policiesToDelete = make([]*eventingv1alpha1.EventPolicy, 0, len(existingPolicyMap))
+
+	for i, channel := range channels {
+		filterSub := filterSubs[i]
+		expectedPolicy := resources.MakeEventPolicyForParallelChannel(p, channel, filterSub)
+		if existingPolicy, ok := existingPolicyMap[expectedPolicy.Name]; ok {
+			if !equality.Semantic.DeepDerivative(existingPolicy.Spec, expectedPolicy.Spec) {
+				policiesToUpdate = append(policiesToUpdate, expectedPolicy)
+			}
+			delete(existingPolicyMap, expectedPolicy.Name)
+		} else {
+			policiesToCreate = append(policiesToCreate, expectedPolicy)
+		}
+	}
+
+	// prepare the event policies for the ingress channel.
+	ingressChannelEventPolicies, err := r.prepareIngressChannelEventpolicies(p, ingressChannel)
+	if err != nil {
+		return fmt.Errorf("failed to prepare event policies for ingress channel: %w", err)
+	}
+
+	for _, policy := range ingressChannelEventPolicies {
+		if existingPolicy, ok := existingPolicyMap[policy.Name]; ok {
+			if !equality.Semantic.DeepDerivative(existingPolicy.Spec, policy.Spec) {
+				policiesToUpdate = append(policiesToUpdate, policy)
+			}
+			delete(existingPolicyMap, policy.Name)
+		} else {
+			policiesToCreate = append(policiesToCreate, policy)
+		}
+	}
+
+	// delete the remaining event policies in the map.
+	for _, policy := range existingPolicyMap {
+		policiesToDelete = append(policiesToDelete, policy)
+	}
+
+	// now that we have the list of event policies to create, update and delete, we can perform the operations.
+	if err := r.createEventPolicies(ctx, policiesToCreate); err != nil {
+		return fmt.Errorf("failed to create event policies: %w", err)
+	}
+	if err := r.updateEventPolicies(ctx, policiesToUpdate); err != nil {
+		return fmt.Errorf("failed to update event policies: %w", err)
+	}
+	if err := r.deleteEventPolicies(ctx, policiesToDelete); err != nil {
+		return fmt.Errorf("failed to delete event policies: %w", err)
+	}
+
+	return nil
+}
+
+func (r *Reconciler) createEventPolicies(ctx context.Context, policies []*eventingv1alpha1.EventPolicy) error {
+	for _, policy := range policies {
+		_, err := r.eventingClientSet.EventingV1alpha1().EventPolicies(policy.Namespace).Create(ctx, policy, metav1.CreateOptions{})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *Reconciler) updateEventPolicies(ctx context.Context, policies []*eventingv1alpha1.EventPolicy) error {
+	for _, policy := range policies {
+		_, err := r.eventingClientSet.EventingV1alpha1().EventPolicies(policy.Namespace).Update(ctx, policy, metav1.UpdateOptions{})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *Reconciler) deleteEventPolicies(ctx context.Context, policies []*eventingv1alpha1.EventPolicy) error {
+	for _, policy := range policies {
+		err := r.eventingClientSet.EventingV1alpha1().EventPolicies(policy.Namespace).Delete(ctx, policy.Name, metav1.DeleteOptions{})
+		if err != nil && !apierrs.IsNotFound(err) {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *Reconciler) prepareIngressChannelEventpolicies(p *v1.Parallel, ingressChannel *duckv1.Channelable) ([]*eventingv1alpha1.EventPolicy, error) {
+	applyingEventPoliciesForParallel, err := auth.GetEventPoliciesForResource(r.eventPolicyLister, v1.SchemeGroupVersion.WithKind("Parallel"), p.ObjectMeta)
+	if err != nil {
+		return nil, fmt.Errorf("could not get EventPolicies for Parallel %s/%s: %w", p.Namespace, p.Name, err)
+	}
+
+	if len(applyingEventPoliciesForParallel) == 0 {
+		return nil, nil
+	}
+
+	var ingressChannelEventPolicies []*eventingv1alpha1.EventPolicy
+	for _, eventPolicy := range applyingEventPoliciesForParallel {
+		ingressChannelEventPolicies = append(ingressChannelEventPolicies, resources.MakeEventPolicyForParallelIngressChannel(p, ingressChannel, eventPolicy))
+	}
+
+	return ingressChannelEventPolicies, nil
+}
+
+func (r *Reconciler) cleanupAllEventPolicies(ctx context.Context, p *v1.Parallel) error {
+	// list all the event policies for the parallel.
+	eventPolicies, err := r.listEventPoliciesForParallel(p)
+	if err != nil {
+		return err
+	}
+	return r.deleteEventPolicies(ctx, eventPolicies)
+}
+
+// listEventPoliciesForParallel lists all EventPolicies (e.g. the policies for the input channel and the intermediate channels)
+// created during reconcileKind that are associated with the given Parallel.
+func (r *Reconciler) listEventPoliciesForParallel(p *v1.Parallel) ([]*eventingv1alpha1.EventPolicy, error) {
+	labelSelector := labels.SelectorFromSet(map[string]string{
+		resources.ParallelChannelEventPolicyLabelPrefix + "parallel-name": p.Name,
+	})
+	return r.eventPolicyLister.EventPolicies(p.Namespace).List(labelSelector)
+}

pkg/reconciler/parallel/parallel_test.go

@@ -35,6 +35,8 @@
 	clientgotesting "k8s.io/client-go/testing"
 
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/eventing/pkg/client/injection/ducks/duck/v1/channelable"
 	"knative.dev/eventing/pkg/duck"
 	"knative.dev/pkg/apis"
@@ -65,14 +67,20 @@
 	subscriberGVK = metav1.GroupVersionKind{
 		Group:   "messaging.knative.dev",
 		Version: "v1",
-		Kind:    "Subscriber",
+		Kind:    "Subscription",
 	}
 
 	parallelGVK = metav1.GroupVersionKind{
 		Group:   "flows.knative.dev",
 		Version: "v1",
 		Kind:    "Parallel",
 	}
+
+	channelV1GVK = metav1.GroupVersionKind{
+		Group:   "messaging.knative.dev",
+		Version: "v1",
+		Kind:    "InMemoryChannel",
+	}
 )
 
 func TestAllBranches(t *testing.T) {
@@ -708,6 +716,97 @@
 						SubscriptionStatus:       createParallelSubscriptionStatus(parallelName, 0, corev1.ConditionFalse),
 					}})),
 			}},
+		}, {
+			Name: "AuthZ Enablled with single branch, with filter, no EventPolicies",
+			Key:  pKey,
+			Objects: []runtime.Object{
+				NewFlowsParallel(parallelName, testNS,
+					WithInitFlowsParallelConditions,
+					WithFlowsParallelChannelTemplateSpec(imc),
+					WithFlowsParallelBranches([]v1.ParallelBranch{
+						{Filter: createFilter(0), Subscriber: createSubscriber(0)},
+					}))},
+			WantErr: false,
+			WantCreates: []runtime.Object{
+				createChannel(parallelName),
+				createBranchChannel(parallelName, 0),
+				resources.NewFilterSubscription(0, NewFlowsParallel(parallelName, testNS, WithFlowsParallelChannelTemplateSpec(imc), WithFlowsParallelBranches([]v1.ParallelBranch{
+					{Filter: createFilter(0), Subscriber: createSubscriber(0)},
+				}))),
+				resources.NewSubscription(0, NewFlowsParallel(parallelName, testNS, WithFlowsParallelChannelTemplateSpec(imc), WithFlowsParallelBranches([]v1.ParallelBranch{
+					{Filter: createFilter(0), Subscriber: createSubscriber(0)},
+				}))),
+				makeEventPolicy(parallelName, resources.ParallelBranchChannelName(parallelName, 0), 0),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: NewFlowsParallel(parallelName, testNS,
+					WithInitFlowsParallelConditions,
+					WithFlowsParallelChannelTemplateSpec(imc),
+					WithFlowsParallelBranches([]v1.ParallelBranch{{Filter: createFilter(0), Subscriber: createSubscriber(0)}}),
+					WithFlowsParallelChannelsNotReady("ChannelsNotReady", "Channels are not ready yet, or there are none"),
+					WithFlowsParallelAddressableNotReady("emptyAddress", "addressable is nil"),
+					WithFlowsParallelSubscriptionsNotReady("SubscriptionsNotReady", "Subscriptions are not ready yet, or there are none"),
+					WithFlowsParallelIngressChannelStatus(createParallelChannelStatus(parallelName, corev1.ConditionFalse)),
+					WithFlowsParallelEventPoliciesReadyBecauseNoPolicyAndOIDCEnabled(),
+					WithFlowsParallelBranchStatuses([]v1.ParallelBranchStatus{{
+						FilterSubscriptionStatus: createParallelFilterSubscriptionStatus(parallelName, 0, corev1.ConditionFalse),
+						FilterChannelStatus:      createParallelBranchChannelStatus(parallelName, 0, corev1.ConditionFalse),
+						SubscriptionStatus:       createParallelSubscriptionStatus(parallelName, 0, corev1.ConditionFalse),
+					}})),
+			}},
+			Ctx: feature.ToContext(context.Background(), feature.Flags{
+				feature.OIDCAuthentication:       feature.Enabled,
+				feature.AuthorizationDefaultMode: feature.AuthorizationAllowSameNamespace,
+			}),
+		}, {
+			Name: "AuthZ Enablled with single branch, with filter, with Parallel EventPolicy",
+			Key:  pKey,
+			Objects: []runtime.Object{
+				NewFlowsParallel(parallelName, testNS,
+					WithInitFlowsParallelConditions,
+					WithFlowsParallelChannelTemplateSpec(imc),
+					WithFlowsParallelBranches([]v1.ParallelBranch{
+						{Filter: createFilter(0), Subscriber: createSubscriber(0)},
+					})),
+				NewEventPolicy(readyEventPolicyName, testNS,
+					WithReadyEventPolicyCondition,
+					WithEventPolicyToRef(parallelGVK, parallelName),
+				),
+			},
+			WantErr: false,
+			WantCreates: []runtime.Object{
+				createChannel(parallelName),
+				createBranchChannel(parallelName, 0),
+				resources.NewFilterSubscription(0, NewFlowsParallel(parallelName, testNS, WithFlowsParallelChannelTemplateSpec(imc), WithFlowsParallelBranches([]v1.ParallelBranch{
+					{Filter: createFilter(0), Subscriber: createSubscriber(0)},
+				}))),
+				resources.NewSubscription(0, NewFlowsParallel(parallelName, testNS, WithFlowsParallelChannelTemplateSpec(imc), WithFlowsParallelBranches([]v1.ParallelBranch{
+					{Filter: createFilter(0), Subscriber: createSubscriber(0)},
+				}))),
+				makeEventPolicy(parallelName, resources.ParallelBranchChannelName(parallelName, 0), 0),
+				makeIngressChannelEventPolicy(parallelName, resources.ParallelChannelName(parallelName)),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: NewFlowsParallel(parallelName, testNS,
+					WithInitFlowsParallelConditions,
+					WithFlowsParallelChannelTemplateSpec(imc),
+					WithFlowsParallelBranches([]v1.ParallelBranch{{Filter: createFilter(0), Subscriber: createSubscriber(0)}}),
+					WithFlowsParallelChannelsNotReady("ChannelsNotReady", "Channels are not ready yet, or there are none"),
+					WithFlowsParallelAddressableNotReady("emptyAddress", "addressable is nil"),
+					WithFlowsParallelSubscriptionsNotReady("SubscriptionsNotReady", "Subscriptions are not ready yet, or there are none"),
+					WithFlowsParallelIngressChannelStatus(createParallelChannelStatus(parallelName, corev1.ConditionFalse)),
+					WithFlowsParallelEventPoliciesReady(),
+					WithFlowsParallelEventPoliciesListed(readyEventPolicyName),
+					WithFlowsParallelBranchStatuses([]v1.ParallelBranchStatus{{
+						FilterSubscriptionStatus: createParallelFilterSubscriptionStatus(parallelName, 0, corev1.ConditionFalse),
+						FilterChannelStatus:      createParallelBranchChannelStatus(parallelName, 0, corev1.ConditionFalse),
+						SubscriptionStatus:       createParallelSubscriptionStatus(parallelName, 0, corev1.ConditionFalse),
+					}})),
+			}},
+			Ctx: feature.ToContext(context.Background(), feature.Flags{
+				feature.OIDCAuthentication:       feature.Enabled,
+				feature.AuthorizationDefaultMode: feature.AuthorizationAllowSameNamespace,
+			}),
 		},
 	}
 
@@ -890,3 +989,47 @@
 		},
 	}
 }
+
+func makeEventPolicy(parallelName, channelName string, branch int) *eventingv1alpha1.EventPolicy {
+	return NewEventPolicy(resources.ParallelEventPolicyName(parallelName, channelName), testNS,
+		WithEventPolicyToRef(channelV1GVK, channelName),
+		// from a subscription
+		WithEventPolicyFrom(subscriberGVK, resources.ParallelFilterSubscriptionName(parallelName, branch), testNS),
+		WithEventPolicyOwnerReferences([]metav1.OwnerReference{
+			{
+				APIVersion: "flows.knative.dev/v1",
+				Kind:       "Parallel",
+				Name:       parallelName,
+			},
+		}...),
+		WithEventPolicyLabels(resources.LabelsForParallelChannelsEventPolicy(parallelName)),
+	)
+}
+
+func makeParallelEventPolicy(parallelName string) *eventingv1alpha1.EventPolicy {
+	return NewEventPolicy(resources.ParallelEventPolicyName(parallelName, ""), testNS,
+		// from a subscription
+		WithEventPolicyOwnerReferences([]metav1.OwnerReference{
+			{
+				APIVersion: "flows.knative.dev/v1",
+				Kind:       "Parallel",
+				Name:       parallelName,
+			},
+		}...),
+	)
+}
+
+func makeIngressChannelEventPolicy(parallelName, channelName string) *eventingv1alpha1.EventPolicy {
+	return NewEventPolicy(resources.ParallelEventPolicyName(parallelName, channelName), testNS,
+		WithEventPolicyToRef(channelV1GVK, channelName),
+		// from a subscription
+		WithEventPolicyOwnerReferences([]metav1.OwnerReference{
+			{
+				APIVersion: "flows.knative.dev/v1",
+				Kind:       "Parallel",
+				Name:       parallelName,
+			},
+		}...),
+		WithEventPolicyLabels(resources.LabelsForParallelChannelsEventPolicy(parallelName)),
+	)
+}