update logic to support identity owner at ta level

Signed-off-by: Jorge Turrado <[email protected]>

apis/keda/v1alpha1/triggerauthentication_types.go

@@ -136,6 +136,9 @@ type AuthPodIdentity struct {
 	IdentityID *string `json:"identityId"`
 	// +optional
 	RoleArn string `json:"roleArn"`
+	// +kubebuilder:validation:Enum=keda;workload
+	// +optional
+	IdentityOwner *string `json:"identityOwner"`
 }
 
 func (a *AuthPodIdentity) GetIdentityID() string {
@@ -145,6 +148,13 @@ func (a *AuthPodIdentity) GetIdentityID() string {
 	return *a.IdentityID
 }
 
+func (a *AuthPodIdentity) IsKedaIdentityOwner() bool {
+	if a.IdentityOwner == nil {
+		return true
+	}
+	return *a.IdentityOwner == "keda"
+}
+
 // AuthSecretTargetRef is used to authenticate using a reference to a secret
 type AuthSecretTargetRef struct {
 	Parameter string `json:"parameter"`

apis/keda/v1alpha1/triggerauthentication_webhook.go

@@ -113,6 +113,10 @@ func validateSpec(spec *TriggerAuthenticationSpec) (admission.Warnings, error) {
 			if spec.PodIdentity.IdentityID != nil && *spec.PodIdentity.IdentityID == "" {
 				return nil, fmt.Errorf("identityid of PodIdentity should not be empty. If it's set, identityId has to be different than \"\"")
 			}
+		case PodIdentityProviderAws:
+			if spec.PodIdentity.RoleArn != "" && !spec.PodIdentity.IsKedaIdentityOwner() {
+				return nil, fmt.Errorf("identityid of PodIdentity should not be empty. If it's set, identityId has to be different than \"\"")
+			}
 		default:
 			return nil, nil
 		}

apis/keda/v1alpha1/triggerauthentication_webhook_test.go

@@ -24,13 +24,13 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-var _ = It("validate triggerauthentication when IdentityID is nil", func() {
+var _ = It("validate triggerauthentication when IdentityID is nil, roleArn is empty and identityOwner is nil", func() {
 	namespaceName := "nilidentityid"
 	namespace := createNamespace(namespaceName)
 	err := k8sClient.Create(context.Background(), namespace)
 	Expect(err).ToNot(HaveOccurred())
 
-	spec := createTriggerAuthenticationSpecWithPodIdentity(nil)
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", nil, nil)
 	ta := createTriggerAuthentication("nilidentityidta", namespaceName, "TriggerAuthentication", spec)
 	Eventually(func() error {
 		return k8sClient.Create(context.Background(), ta)
@@ -44,7 +44,7 @@ var _ = It("validate triggerauthentication when IdentityID is empty", func() {
 	Expect(err).ToNot(HaveOccurred())
 
 	identityID := ""
-	spec := createTriggerAuthenticationSpecWithPodIdentity(&identityID)
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", &identityID, nil)
 	ta := createTriggerAuthentication("emptyidentityidta", namespaceName, "TriggerAuthentication", spec)
 	Eventually(func() error {
 		return k8sClient.Create(context.Background(), ta)
@@ -58,7 +58,76 @@ var _ = It("validate triggerauthentication when IdentityID is not empty", func()
 	Expect(err).ToNot(HaveOccurred())
 
 	identityID := "12345"
-	spec := createTriggerAuthenticationSpecWithPodIdentity(&identityID)
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", &identityID, nil)
+	ta := createTriggerAuthentication("identityidta", namespaceName, "TriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+var _ = It("validate triggerauthentication when RoleArn is not empty and IdentityOwner is nil", func() {
+	namespaceName := "rolearn"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "Helo", nil, nil)
+	ta := createTriggerAuthentication("identityidta", namespaceName, "TriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+var _ = It("validate triggerauthentication when RoleArn is not empty and IdentityOwner is keda", func() {
+	namespaceName := "rolearnandkedaowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "keda"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "Helo", nil, &identityOwner)
+	ta := createTriggerAuthentication("identityidta", namespaceName, "TriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+var _ = It("validate triggerauthentication when RoleArn is not empty and IdentityOwner is workload", func() {
+	namespaceName := "rolearnandworkloadowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "workload"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "Helo", nil, &identityOwner)
+	ta := createTriggerAuthentication("identityidta", namespaceName, "TriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).Should(HaveOccurred())
+})
+
+var _ = It("validate triggerauthentication when RoleArn is empty and IdentityOwner is keda", func() {
+	namespaceName := "kedaowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "keda"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", nil, &identityOwner)
+	ta := createTriggerAuthentication("identityidta", namespaceName, "TriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+var _ = It("validate triggerauthentication when RoleArn is not empty and IdentityOwner is workload", func() {
+	namespaceName := "workloadowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "workload"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", nil, &identityOwner)
 	ta := createTriggerAuthentication("identityidta", namespaceName, "TriggerAuthentication", spec)
 	Eventually(func() error {
 		return k8sClient.Create(context.Background(), ta)
@@ -71,7 +140,7 @@ var _ = It("validate clustertriggerauthentication when IdentityID is nil", func(
 	err := k8sClient.Create(context.Background(), namespace)
 	Expect(err).ToNot(HaveOccurred())
 
-	spec := createTriggerAuthenticationSpecWithPodIdentity(nil)
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", nil, nil)
 	ta := createTriggerAuthentication("clusternilidentityidta", namespaceName, "ClusterTriggerAuthentication", spec)
 	Eventually(func() error {
 		return k8sClient.Create(context.Background(), ta)
@@ -85,7 +154,7 @@ var _ = It("validate clustertriggerauthentication when IdentityID is empty", fun
 	Expect(err).ToNot(HaveOccurred())
 
 	identityID := ""
-	spec := createTriggerAuthenticationSpecWithPodIdentity(&identityID)
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", &identityID, nil)
 	ta := createTriggerAuthentication("clusteremptyidentityidta", namespaceName, "ClusterTriggerAuthentication", spec)
 	Eventually(func() error {
 		return k8sClient.Create(context.Background(), ta)
@@ -99,18 +168,89 @@ var _ = It("validate clustertriggerauthentication when IdentityID is not empty",
 	Expect(err).ToNot(HaveOccurred())
 
 	identityID := "12345"
-	spec := createTriggerAuthenticationSpecWithPodIdentity(&identityID)
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", &identityID, nil)
 	ta := createTriggerAuthentication("clusteridentityidta", namespaceName, "ClusterTriggerAuthentication", spec)
 	Eventually(func() error {
 		return k8sClient.Create(context.Background(), ta)
 	}).ShouldNot(HaveOccurred())
 })
 
-func createTriggerAuthenticationSpecWithPodIdentity(identityID *string) TriggerAuthenticationSpec {
+var _ = It("validate clustertriggerauthentication when RoleArn is not empty and IdentityOwner is nil", func() {
+	namespaceName := "clusterrolearn"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "Helo", nil, nil)
+	ta := createTriggerAuthentication("clusteridentityidta", namespaceName, "ClusterTriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+var _ = It("validate clustertriggerauthentication when RoleArn is not empty and IdentityOwner is keda", func() {
+	namespaceName := "clusterrolearnandkedaowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "keda"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "Helo", nil, &identityOwner)
+	ta := createTriggerAuthentication("clusteridentityidta", namespaceName, "ClusterTriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+var _ = It("validate clustertriggerauthentication when RoleArn is not empty and IdentityOwner is workload", func() {
+	namespaceName := "clusterrolearnandworkloadowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "workload"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "Helo", nil, &identityOwner)
+	ta := createTriggerAuthentication("clusteridentityidta", namespaceName, "ClusterTriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).Should(HaveOccurred())
+})
+
+var _ = It("validate clustertriggerauthentication when RoleArn is empty and IdentityOwner is keda", func() {
+	namespaceName := "clusterandkedaowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "keda"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", nil, &identityOwner)
+	ta := createTriggerAuthentication("clusteridentityidta", namespaceName, "ClusterTriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+var _ = It("validate clustertriggerauthentication when RoleArn is not empty and IdentityOwner is workload", func() {
+	namespaceName := "clusterandworkloadowner"
+	namespace := createNamespace(namespaceName)
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	identityOwner := "workload"
+	spec := createTriggerAuthenticationSpecWithPodIdentity(PodIdentityProviderAzure, "", nil, &identityOwner)
+	ta := createTriggerAuthentication("clusteridentityidta", namespaceName, "TriggerAuthentication", spec)
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), ta)
+	}).ShouldNot(HaveOccurred())
+})
+
+func createTriggerAuthenticationSpecWithPodIdentity(provider PodIdentityProvider, roleArn string, identityID, identityOwner *string) TriggerAuthenticationSpec {
 	return TriggerAuthenticationSpec{
 		PodIdentity: &AuthPodIdentity{
-			Provider:   PodIdentityProviderAzure,
-			IdentityID: identityID,
+			Provider:      PodIdentityProviderAzure,
+			IdentityID:    identityID,
+			RoleArn:       roleArn,
+			IdentityOwner: identityOwner,
 		},
 	}
 }

config/crd/bases/keda.sh_clustertriggerauthentications.yaml

@@ -111,6 +111,11 @@ spec:
                     properties:
                       identityId:
                         type: string
+                      identityOwner:
+                        enum:
+                        - keda
+                        - workload
+                        type: string
                       provider:
                         description: PodIdentityProvider contains the list of providers
                         enum:
@@ -235,6 +240,11 @@ spec:
                 properties:
                   identityId:
                     type: string
+                  identityOwner:
+                    enum:
+                    - keda
+                    - workload
+                    type: string
                   provider:
                     description: PodIdentityProvider contains the list of providers
                     enum:

config/crd/bases/keda.sh_triggerauthentications.yaml

@@ -110,6 +110,11 @@ spec:
                     properties:
                       identityId:
                         type: string
+                      identityOwner:
+                        enum:
+                        - keda
+                        - workload
+                        type: string
                       provider:
                         description: PodIdentityProvider contains the list of providers
                         enum:
@@ -234,6 +239,11 @@ spec:
                 properties:
                   identityId:
                     type: string
+                  identityOwner:
+                    enum:
+                    - keda
+                    - workload
+                    type: string
                   provider:
                     description: PodIdentityProvider contains the list of providers
                     enum:

pkg/scaling/resolver/scale_resolvers.go

@@ -185,21 +185,24 @@ func ResolveAuthRefAndPodIdentity(ctx context.Context, client client.Client, log
 		switch podIdentity.Provider {
 		case kedav1alpha1.PodIdentityProviderAws:
 			if podIdentity.RoleArn != "" {
-				if podIdentity.RoleArn == "workload" {
-					serviceAccountName := defaultServiceAccount
-					if podTemplateSpec.Spec.ServiceAccountName != "" {
-						serviceAccountName = podTemplateSpec.Spec.ServiceAccountName
-					}
-					serviceAccount := &corev1.ServiceAccount{}
-					err := client.Get(ctx, types.NamespacedName{Name: serviceAccountName, Namespace: namespace}, serviceAccount)
-					if err != nil {
-						return nil, kedav1alpha1.AuthPodIdentity{Provider: kedav1alpha1.PodIdentityProviderNone},
-							fmt.Errorf("error getting service account: '%s', error: %w", serviceAccountName, err)
-					}
-					authParams["awsRoleArn"] = serviceAccount.Annotations[kedav1alpha1.PodIdentityAnnotationEKS]
-				} else {
-					authParams["awsRoleArn"] = podIdentity.RoleArn
+				if !podIdentity.IsKedaIdentityOwner() {
+					return nil, kedav1alpha1.AuthPodIdentity{Provider: kedav1alpha1.PodIdentityProviderNone},
+						fmt.Errorf("roleArn can't be set if KEDA isn't identity owner, current value: '%s'", *podIdentity.IdentityOwner)
+				}
+				authParams["awsRoleArn"] = podIdentity.RoleArn
+			}
+			if !podIdentity.IsKedaIdentityOwner() {
+				serviceAccountName := defaultServiceAccount
+				if podTemplateSpec.Spec.ServiceAccountName != "" {
+					serviceAccountName = podTemplateSpec.Spec.ServiceAccountName
+				}
+				serviceAccount := &corev1.ServiceAccount{}
+				err := client.Get(ctx, types.NamespacedName{Name: serviceAccountName, Namespace: namespace}, serviceAccount)
+				if err != nil {
+					return nil, kedav1alpha1.AuthPodIdentity{Provider: kedav1alpha1.PodIdentityProviderNone},
+						fmt.Errorf("error getting service account: '%s', error: %w", serviceAccountName, err)
 				}
+				authParams["awsRoleArn"] = serviceAccount.Annotations[kedav1alpha1.PodIdentityAnnotationEKS]
 			}
 		case kedav1alpha1.PodIdentityProviderAwsEKS:
 			serviceAccountName := defaultServiceAccount