Add EventPolicy reconciliation for Sequence

pkg/reconciler/sequence/resources/eventpolicy.go

@@ -0,0 +1,118 @@
+/*
+Copyright 2024 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 resources
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	flowsv1 "knative.dev/eventing/pkg/apis/flows/v1"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	"knative.dev/pkg/kmeta"
+)
+
+const (
+	SequenceChannelEventPolicyLabelPrefix = "flows.knative.dev/"
+	sequenceKind                          = "Sequence"
+)
+
+func MakeEventPolicyForSequenceChannel(s *flowsv1.Sequence, channel *eventingduckv1.Channelable, subscription *messagingv1.Subscription) *eventingv1alpha1.EventPolicy {
+	return &eventingv1alpha1.EventPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: channel.Namespace,
+			Name:      SequenceEventPolicyName(s.Name, channel.Name),
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: flowsv1.SchemeGroupVersion.String(),
+					Kind:       sequenceKind,
+					Name:       s.Name,
+				},
+			},
+			Labels: LabelsForSequenceChannelsEventPolicy(s.Name),
+		},
+		Spec: eventingv1alpha1.EventPolicySpec{
+			To: []eventingv1alpha1.EventPolicySpecTo{
+				{
+					Ref: &eventingv1alpha1.EventPolicyToReference{
+						APIVersion: channel.APIVersion,
+						Kind:       channel.Kind,
+						Name:       channel.Name,
+					},
+				},
+			},
+			From: []eventingv1alpha1.EventPolicySpecFrom{
+				{
+					Ref: &eventingv1alpha1.EventPolicyFromReference{
+						APIVersion: subscription.APIVersion,
+						Kind:       subscription.Kind,
+						Name:       subscription.Name,
+						Namespace:  subscription.Namespace,
+					},
+				},
+			},
+		},
+	}
+}
+
+func LabelsForSequenceChannelsEventPolicy(sequenceName string) map[string]string {
+	return map[string]string{
+		SequenceChannelEventPolicyLabelPrefix + "sequence-group":   flowsv1.SchemeGroupVersion.Group,
+		SequenceChannelEventPolicyLabelPrefix + "sequence-version": flowsv1.SchemeGroupVersion.Version,
+		SequenceChannelEventPolicyLabelPrefix + "sequence-kind":    sequenceKind,
+		SequenceChannelEventPolicyLabelPrefix + "sequence-name":    sequenceName,
+	}
+}
+
+func SequenceEventPolicyName(sequenceName, channelName string) string {
+	// if channel name is empty, it means the event policy is for the output channel
+	if channelName == "" {
+		return kmeta.ChildName(sequenceName, "-ep") // no need to add the channel name
+	} else {
+		return kmeta.ChildName(sequenceName, "-ep-"+channelName)
+	}
+
+}
+
+// MakeEventPolicyForSequenceInputChannel creates an EventPolicy for the input channel of a Sequence
+func MakeEventPolicyForSequenceInputChannel(s *flowsv1.Sequence, inputChannel *eventingduckv1.Channelable, sequencePolicy *eventingv1alpha1.EventPolicy) *eventingv1alpha1.EventPolicy {
+	return &eventingv1alpha1.EventPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: inputChannel.Namespace,
+			Name:      SequenceEventPolicyName(s.Name, inputChannel.Name),
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: flowsv1.SchemeGroupVersion.String(),
+					Kind:       sequenceKind,
+					Name:       s.Name,
+				},
+			},
+			Labels: LabelsForSequenceChannelsEventPolicy(s.Name),
+		},
+		Spec: eventingv1alpha1.EventPolicySpec{
+			To: []eventingv1alpha1.EventPolicySpecTo{
+				{
+					Ref: &eventingv1alpha1.EventPolicyToReference{
+						APIVersion: inputChannel.APIVersion,
+						Kind:       inputChannel.Kind,
+						Name:       inputChannel.Name,
+					},
+				},
+			},
+			From: sequencePolicy.Spec.From,
+		},
+	}
+}

pkg/reconciler/sequence/sequence.go

@@ -22,12 +22,14 @@
 
 	"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"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
+	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
 	"knative.dev/pkg/kmeta"
 
 	duckapis "knative.dev/pkg/apis/duck"
@@ -129,6 +131,11 @@
 		return err
 	}
 
+	// Handle EventPolicies
+	if err := r.reconcileEventPolicies(ctx, s, channels, subs, featureFlags); err != nil {
+		return fmt.Errorf("failed to reconcile EventPolicies: %w", err)
+	}
+
 	err := auth.UpdateStatusWithEventPolicies(featureFlags, &s.Status.AppliedEventPoliciesStatus, &s.Status, r.eventPolicyLister, v1.SchemeGroupVersion.WithKind("Sequence"), s.ObjectMeta)
 	if err != nil {
 		return fmt.Errorf("could not update Sequence status with EventPolicies: %v", err)
@@ -186,7 +193,7 @@
 	subName := resources.SequenceSubscriptionName(p.Name, step)
 	sub, err := r.subscriptionLister.Subscriptions(p.Namespace).Get(subName)
 
-	// If the resource doesn't exist, we'll create it.
+	// If the resource doesn't exist, we'll create itF.
 	if apierrs.IsNotFound(err) {
 		sub = expected
 		logging.FromContext(ctx).Infof("Creating subscription: %+v", sub)
@@ -331,3 +338,99 @@
 
 	return nil
 }
+
+func (r *Reconciler) reconcileEventPolicies(ctx context.Context, s *v1.Sequence, channels []*eventingduckv1.Channelable, subs []*messagingv1.Subscription, featureFlags feature.Flags) error {
+	if featureFlags.IsOIDCAuthentication() {
+		// Create or update EventPolicies, and we skip the first channel as it's the input channel!
+		for i := 1; i < len(channels); i++ {
+			if err := r.reconcileChannelEventPolicy(ctx, s, channels[i], subs[i-1]); err != nil {
+				return err
+			}
+		}
+
+		// Handle input channel EventPolicy
+		if err := r.reconcileInputChannelEventPolicy(ctx, s, channels[0]); err != nil {
+			return err
+		}
+
+	} else {
+		// Clean up existing EventPolicies, as the authentication feature flag is disabled
+		if err := r.cleanupEventPolicies(ctx, s); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (r *Reconciler) reconcileChannelEventPolicy(ctx context.Context, s *v1.Sequence, channel *eventingduckv1.Channelable, subscription *messagingv1.Subscription) error {
+	expected := resources.MakeEventPolicyForSequenceChannel(s, channel, subscription)
+
+	return r.createOrUpdateEventPolicy(ctx, expected)
+}
+func (r *Reconciler) createOrUpdateEventPolicy(ctx context.Context, expected *eventingv1alpha1.EventPolicy) error {
+	existing, err := r.eventPolicyLister.EventPolicies(expected.Namespace).Get(expected.Name)
+	if apierrs.IsNotFound(err) {
+		_, err = r.eventingClientSet.EventingV1alpha1().EventPolicies(expected.Namespace).Create(ctx, expected, metav1.CreateOptions{})
+		return err
+	} else if err != nil {
+		return err
+	}
+
+	// Update if needed
+	if !equality.Semantic.DeepEqual(existing.Spec, expected.Spec) {
+		existing.Spec = expected.Spec
+		_, err = r.eventingClientSet.EventingV1alpha1().EventPolicies(existing.Namespace).Update(ctx, existing, metav1.UpdateOptions{})
+		return err
+	}
+
+	return nil
+}
+
+func (r *Reconciler) reconcileInputChannelEventPolicy(ctx context.Context, s *v1.Sequence, inputChannel *eventingduckv1.Channelable) error {
+	// Check if there's an EventPolicy for the Sequence
+	sequencePolicy, err := r.eventPolicyLister.EventPolicies(s.Namespace).Get(s.Name + "-ep")
+	if err != nil {
+		if apierrs.IsNotFound(err) {
+			// No EventPolicy for the Sequence, so we don't create one for the input channel
+			return nil
+		}
+		return err
+	}
+
+	expected := resources.MakeEventPolicyForSequenceInputChannel(s, inputChannel, sequencePolicy)
+
+	return r.createOrUpdateEventPolicy(ctx, expected)
+}
+
+func (r *Reconciler) cleanupEventPolicies(ctx context.Context, s *v1.Sequence) error {
+	policies, err := r.eventPolicyLister.EventPolicies(s.Namespace).List(labels.Everything())
+	if err != nil {
+		return err
+	}
+
+	for _, policy := range policies {
+		if metav1.IsControlledBy(policy, s) {
+			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) verifyEventPolicyCreation(ctx context.Context, expected *eventingv1alpha1.EventPolicy) error {
+	logging.FromContext(ctx).Infof("Verifying EventPolicy creation: %s", expected.Name)
+
+	// Try to get the EventPolicy using the client directly
+	policy, err := r.eventingClientSet.EventingV1alpha1().EventPolicies(expected.Namespace).Get(ctx, expected.Name, metav1.GetOptions{})
+	if err != nil {
+		logging.FromContext(ctx).Errorf("Failed to get EventPolicy after creation: %v", err)
+		return err
+	}
+
+	logging.FromContext(ctx).Infof("Successfully verified EventPolicy creation: %s", policy.Name)
+	return nil
+}