Refactor the AuthStatus Logic in Eventing OIDC Feature Track

pkg/auth/serviceaccount.go

@@ -28,8 +28,11 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes"
 	corev1listers "k8s.io/client-go/listers/core/v1"
+	"knative.dev/eventing/pkg/apis/feature"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/ptr"
+	pkgreconciler "knative.dev/pkg/reconciler"
 )
 
 // GetOIDCServiceAccountNameForResource returns the service account name to use
@@ -94,3 +97,27 @@ func EnsureOIDCServiceAccountExistsForResource(ctx context.Context, serviceAccou
 
 	return nil
 }
+
+type OIDCIdentityStatusMarker interface {
+	MarkOIDCIdentityCreatedSucceeded()
+	MarkOIDCIdentityCreatedSucceededWithReason(reason, messageFormat string, messageA ...interface{})
+	MarkOIDCIdentityCreatedFailed(reason, messageFormat string, messageA ...interface{})
+}
+
+func SetupOIDCServiceAccount(ctx context.Context, flags feature.Flags, serviceAccountLister corev1listers.ServiceAccountLister, kubeclient kubernetes.Interface, gvk schema.GroupVersionKind, objectMeta metav1.ObjectMeta, marker OIDCIdentityStatusMarker, setAuthStatus func(a *duckv1.AuthStatus)) pkgreconciler.Event {
+	if flags.IsOIDCAuthentication() {
+		saName := GetOIDCServiceAccountNameForResource(gvk, objectMeta)
+		setAuthStatus(&duckv1.AuthStatus{
+			ServiceAccountName: &saName,
+		})
+		if err := EnsureOIDCServiceAccountExistsForResource(ctx, serviceAccountLister, kubeclient, gvk, objectMeta); err != nil {
+			marker.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
+			return err
+		}
+		marker.MarkOIDCIdentityCreatedSucceeded()
+	} else {
+		setAuthStatus(nil)
+		marker.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	}
+	return nil
+}

pkg/auth/serviceaccount_test.go

@@ -17,14 +17,22 @@ limitations under the License.
 package auth
 
 import (
+	"context"
 	"testing"
 
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+
 	"github.com/google/go-cmp/cmp"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/pkg/apis/feature"
+	rttestingv1 "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/ptr"
+	rectesting "knative.dev/pkg/reconciler/testing"
 )
 
 func TestGetOIDCServiceAccountNameForResource(t *testing.T) {
@@ -100,3 +108,97 @@ func TestGetOIDCServiceAccountForResource(t *testing.T) {
 		t.Errorf("GetServiceAccount() = %+v, want %+v - diff %s", got, want, diff)
 	}
 }
+
+func TestEnsureOIDCServiceAccountExistsForResource(t *testing.T) {
+	ctx, _ := rectesting.SetupFakeContext(t)
+	gvk := eventingv1.SchemeGroupVersion.WithKind("Broker")
+	objectMeta := metav1.ObjectMeta{
+		Name:      "my-broker",
+		Namespace: "my-namespace",
+		UID:       "my-uuid",
+	}
+
+	eventtypes := make([]runtime.Object, 0, 10)
+	listers := rttestingv1.NewListers(eventtypes)
+
+	err := EnsureOIDCServiceAccountExistsForResource(ctx, listers.GetServiceAccountLister(), kubeclient.Get(ctx), gvk, objectMeta)
+	if err != nil {
+		t.Errorf("EnsureOIDCServiceAccountExistsForResource failed: %s", err)
+
+	}
+	expected := GetOIDCServiceAccountForResource(gvk, objectMeta)
+	sa, err := kubeclient.Get(ctx).CoreV1().ServiceAccounts(objectMeta.Namespace).Get(context.TODO(), expected.Name, metav1.GetOptions{})
+	if err != nil {
+		t.Errorf("get ServiceAccounts  failed: %s", err)
+	}
+	if sa == nil || sa.Name != expected.Name {
+		t.Errorf("EnsureOIDCServiceAccountExistsForResource create ServiceAccounts  failed: %s", err)
+	}
+}
+
+func TestSetupOIDCServiceAccount(t *testing.T) {
+	ctx, _ := rectesting.SetupFakeContext(t)
+	gvk := eventingv1.SchemeGroupVersion.WithKind("Trigger")
+	objectMeta := metav1.ObjectMeta{
+		Name:      "my-trigger",
+		Namespace: "my-namespace",
+		UID:       "my-uuid",
+	}
+	eventtypes := make([]runtime.Object, 0, 10)
+	listers := rttestingv1.NewListers(eventtypes)
+	trigger := rttestingv1.NewTrigger("my-trigger", "my-namespace", "my-broker")
+	expected := GetOIDCServiceAccountForResource(gvk, objectMeta)
+	err := SetupOIDCServiceAccount(ctx, feature.Flags{
+		feature.OIDCAuthentication: feature.Enabled,
+	}, listers.GetServiceAccountLister(), kubeclient.Get(ctx), gvk, objectMeta, &trigger.Status, func(as *duckv1.AuthStatus) {
+		trigger.Status.Auth = as
+	})
+
+	if err != nil {
+		t.Errorf("SetupOIDCServiceAccount  failed: %s", err)
+	}
+	if trigger.Status.Auth == nil || *trigger.Status.Auth.ServiceAccountName != expected.Name {
+		t.Errorf("SetupOIDCServiceAccount setAuthStatus  failed")
+	}
+
+	// match OIDCIdentityCreated condition
+	matched := false
+	for _, condition := range trigger.Status.Conditions {
+		if condition.Type == eventingv1.TriggerConditionOIDCIdentityCreated {
+			if condition.Reason == "" {
+				matched = true
+			}
+		}
+	}
+	if !matched {
+		t.Errorf("SetupOIDCServiceAccount didn't set TriggerConditionOIDCIdentityCreated Status")
+	}
+
+	err = SetupOIDCServiceAccount(ctx, feature.Flags{
+		feature.OIDCAuthentication: feature.Disabled,
+	}, listers.GetServiceAccountLister(), kubeclient.Get(ctx), gvk, objectMeta, &trigger.Status, func(as *duckv1.AuthStatus) {
+		trigger.Status.Auth = as
+	})
+
+	if err != nil {
+		t.Errorf("SetupOIDCServiceAccount  failed: %s", err)
+	}
+
+	if trigger.Status.Auth != nil {
+		t.Errorf("SetupOIDCServiceAccount setAuthStatus  failed")
+	}
+
+	// match OIDCIdentityCreated condition
+	matched = false
+	for _, condition := range trigger.Status.Conditions {
+		if condition.Type == eventingv1.TriggerConditionOIDCIdentityCreated {
+			if condition.Reason == "authentication-oidc feature disabled" {
+				matched = true
+			}
+		}
+	}
+
+	if !matched {
+		t.Errorf("SetupOIDCServiceAccount didn't set TriggerConditionOIDCIdentityCreated Status")
+	}
+}

pkg/reconciler/apiserversource/apiserversource.go

@@ -100,20 +100,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1.ApiServerSour
 
 	// OIDC authentication
 	featureFlags := feature.FromContext(ctx)
-	if featureFlags.IsOIDCAuthentication() {
-		saName := auth.GetOIDCServiceAccountNameForResource(v1.SchemeGroupVersion.WithKind("ApiServerSource"), source.ObjectMeta)
-		source.Status.Auth = &duckv1.AuthStatus{
-			ServiceAccountName: &saName,
-		}
-
-		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeClientSet, v1.SchemeGroupVersion.WithKind("ApiServerSource"), source.ObjectMeta); err != nil {
-			source.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
-			return err
-		}
-		source.Status.MarkOIDCIdentityCreatedSucceeded()
-	} else {
-		source.Status.Auth = nil
-		source.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	if err := auth.SetupOIDCServiceAccount(ctx, featureFlags, r.serviceAccountLister, r.kubeClientSet, v1.SchemeGroupVersion.WithKind("ApiServerSource"), source.ObjectMeta, &source.Status, func(as *duckv1.AuthStatus) {
+		source.Status.Auth = as
+	}); err != nil {
+		return err
 	}
 
 	sinkAddr, err := r.sinkResolver.AddressableFromDestinationV1(ctx, *dest, source)

pkg/reconciler/broker/trigger/trigger.go

@@ -144,20 +144,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, t *eventingv1.Trigger) p
 	}
 
 	featureFlags := feature.FromContext(ctx)
-	if featureFlags.IsOIDCAuthentication() {
-		saName := auth.GetOIDCServiceAccountNameForResource(eventingv1.SchemeGroupVersion.WithKind("Trigger"), t.ObjectMeta)
-		t.Status.Auth = &duckv1.AuthStatus{
-			ServiceAccountName: &saName,
-		}
-
-		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeclient, eventingv1.SchemeGroupVersion.WithKind("Trigger"), t.ObjectMeta); err != nil {
-			t.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
-			return err
-		}
-		t.Status.MarkOIDCIdentityCreatedSucceeded()
-	} else {
-		t.Status.Auth = nil
-		t.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	if err = auth.SetupOIDCServiceAccount(ctx, featureFlags, r.serviceAccountLister, r.kubeclient, eventingv1.SchemeGroupVersion.WithKind("Trigger"), t.ObjectMeta, &t.Status, func(as *duckv1.AuthStatus) {
+		t.Status.Auth = as
+	}); err != nil {
+		return err
 	}
 
 	sub, err := r.subscribeToBrokerChannel(ctx, b, t, brokerTrigger)

pkg/reconciler/parallel/parallel.go

@@ -80,19 +80,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, p *v1.Parallel) pkgrecon
 	// 3. Rinse and repeat step #2 above for each branch in the list
 	// OIDC authentication
 	featureFlags := feature.FromContext(ctx)
-	if featureFlags.IsOIDCAuthentication() {
-		saName := auth.GetOIDCServiceAccountNameForResource(v1.SchemeGroupVersion.WithKind("Parallel"), p.ObjectMeta)
-		p.Status.Auth = &duckv1knative.AuthStatus{
-			ServiceAccountName: &saName,
-		}
-		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeclient, v1.SchemeGroupVersion.WithKind("Parallel"), p.ObjectMeta); err != nil {
-			p.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
-			return err
-		}
-		p.Status.MarkOIDCIdentityCreatedSucceeded()
-	} else {
-		p.Status.Auth = nil
-		p.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	if err := auth.SetupOIDCServiceAccount(ctx, featureFlags, r.serviceAccountLister, r.kubeclient, v1.SchemeGroupVersion.WithKind("Parallel"), p.ObjectMeta, &p.Status, func(as *duckv1knative.AuthStatus) {
+		p.Status.Auth = as
+	}); err != nil {
+		return err
 	}
 
 	if p.Status.BranchStatuses == nil {

pkg/reconciler/pingsource/pingsource.go

@@ -22,8 +22,6 @@ import (
 	"fmt"
 
 	v1 "k8s.io/client-go/listers/core/v1"
-	"knative.dev/eventing/pkg/apis/feature"
-	"knative.dev/eventing/pkg/auth"
 
 	"go.uber.org/zap"
 
@@ -45,7 +43,9 @@ import (
 
 	"knative.dev/eventing/pkg/adapter/mtping"
 	"knative.dev/eventing/pkg/adapter/v2"
+	"knative.dev/eventing/pkg/apis/feature"
 	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
+	"knative.dev/eventing/pkg/auth"
 	pingsourcereconciler "knative.dev/eventing/pkg/client/injection/reconciler/sources/v1/pingsource"
 	"knative.dev/eventing/pkg/reconciler/pingsource/resources"
 	reconcilersource "knative.dev/eventing/pkg/reconciler/source"
@@ -107,20 +107,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *sourcesv1.PingSo
 
 	// OIDC authentication
 	featureFlags := feature.FromContext(ctx)
-	if featureFlags.IsOIDCAuthentication() {
-		saName := auth.GetOIDCServiceAccountNameForResource(sourcesv1.SchemeGroupVersion.WithKind("PingSource"), source.ObjectMeta)
-		source.Status.Auth = &duckv1.AuthStatus{
-			ServiceAccountName: &saName,
-		}
-
-		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeClientSet, sourcesv1.SchemeGroupVersion.WithKind("PingSource"), source.ObjectMeta); err != nil {
-			source.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
-			return err
-		}
-		source.Status.MarkOIDCIdentityCreatedSucceeded()
-	} else {
-		source.Status.Auth = nil
-		source.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	if err := auth.SetupOIDCServiceAccount(ctx, featureFlags, r.serviceAccountLister, r.kubeClientSet, sourcesv1.SchemeGroupVersion.WithKind("PingSource"), source.ObjectMeta, &source.Status, func(as *duckv1.AuthStatus) {
+		source.Status.Auth = as
+	}); err != nil {
+		return err
 	}
 
 	sinkAddr, err := r.sinkResolver.AddressableFromDestinationV1(ctx, *dest, source)

pkg/reconciler/sequence/sequence.go

@@ -130,20 +130,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, s *v1.Sequence) pkgrecon
 	}
 
 	featureFlags := feature.FromContext(ctx)
-	if featureFlags.IsOIDCAuthentication() {
-		saName := auth.GetOIDCServiceAccountNameForResource(v1.SchemeGroupVersion.WithKind("Sequence"), s.ObjectMeta)
-		s.Status.Auth = &duckv1.AuthStatus{
-			ServiceAccountName: &saName,
-		}
-
-		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeclient, v1.SchemeGroupVersion.WithKind("Sequence"), s.ObjectMeta); err != nil {
-			s.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
-			return err
-		}
-		s.Status.MarkOIDCIdentityCreatedSucceeded()
-	} else {
-		s.Status.Auth = nil
-		s.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	if err := auth.SetupOIDCServiceAccount(ctx, featureFlags, r.serviceAccountLister, r.kubeclient, v1.SchemeGroupVersion.WithKind("Sequence"), s.ObjectMeta, &s.Status, func(as *duckv1.AuthStatus) {
+		s.Status.Auth = as
+	}); err != nil {
+		return err
 	}
 
 	return r.removeUnwantedSubscriptions(ctx, s, subs)

pkg/reconciler/subscription/subscription.go

@@ -93,20 +93,10 @@ var _ subscriptionreconciler.Finalizer = (*Reconciler)(nil)
 func (r *Reconciler) ReconcileKind(ctx context.Context, subscription *v1.Subscription) pkgreconciler.Event {
 	// OIDC authentication
 	featureFlags := feature.FromContext(ctx)
-	if featureFlags.IsOIDCAuthentication() {
-		saName := auth.GetOIDCServiceAccountNameForResource(v1.SchemeGroupVersion.WithKind("Subscription"), subscription.ObjectMeta)
-		subscription.Status.Auth = &duckv1.AuthStatus{
-			ServiceAccountName: &saName,
-		}
-
-		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeclient, v1.SchemeGroupVersion.WithKind("Subscription"), subscription.ObjectMeta); err != nil {
-			subscription.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
-			return err
-		}
-		subscription.Status.MarkOIDCIdentityCreatedSucceeded()
-	} else {
-		subscription.Status.Auth = nil
-		subscription.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	if err := auth.SetupOIDCServiceAccount(ctx, featureFlags, r.serviceAccountLister, r.kubeclient, v1.SchemeGroupVersion.WithKind("Subscription"), subscription.ObjectMeta, &subscription.Status, func(as *duckv1.AuthStatus) {
+		subscription.Status.Auth = as
+	}); err != nil {
+		return err
 	}
 
 	// Find the channel for this subscription.