Merge branch 'kedacore:main' into main

.devcontainer/Dockerfile

@@ -37,7 +37,6 @@ RUN apt-get update \
         github.com/ramya-rao-a/go-outline \
         github.com/acroca/go-symbols \
         github.com/godoctor/godoctor \
-        golang.org/x/tools/cmd/guru \
         golang.org/x/tools/cmd/gorename \
         github.com/rogpeppe/godef \
         github.com/zmb3/gogetdoc \

CHANGELOG.md

@@ -65,10 +65,12 @@ Here is an overview of all new **experimental** features:
 
 - **General**: Add command-line flag in Adapter to allow override of gRPC Authority Header ([#5449](https://github.com/kedacore/keda/issues/5449))
 - **General**: Add OPENTELEMETRY flag in e2e test YAML ([#5375](https://github.com/kedacore/keda/issues/5375))
+- **General**: Add support for cross tenant/cloud authentication when using Azure Workload Identity for TriggerAuthentication ([#5441](https://github.com/kedacore/keda/issues/5441))
+- **MongoDB Scaler**: Add scheme field support srv record ([#5544](https://github.com/kedacore/keda/issues/5544))
 
 ### Fixes
 
-- TODO ([#XXX](https://github.com/kedacore/keda/issues/XXX))
+- **General**: Validate empty array value of triggers in ScaledObject/ScaledJob creation ([#5520](https://github.com/kedacore/keda/issues/5520))
 
 ### Deprecations
 
@@ -84,10 +86,12 @@ New deprecation(s):
 
 ### Other
 
+- **General**: Allow E2E tests to be run against existing KEDA and/or Kafka installation ([#5595](https://github.com/kedacore/keda/pull/5595))
 - **General**: Improve readability of utility function getParameterFromConfigV2 ([#5037](https://github.com/kedacore/keda/issues/5037))
-- **General**: Introduce ENABLE_OPENTELEMETRY in deploying/testing process  ([#5375](https://github.com/kedacore/keda/issues/5375))
+- **General**: Introduce ENABLE_OPENTELEMETRY in deploying/testing process  ([#5375](https://github.com/kedacore/keda/issues/5375)|[#5578](https://github.com/kedacore/keda/issues/5578))
 - **General**: Migrate away from unmaintained golang/mock and use uber/gomock ([#5440](https://github.com/kedacore/keda/issues/5440))
 - **General**: Minor refactor to reduce copy/paste code in ScaledObject webhook ([#5397](https://github.com/kedacore/keda/issues/5397))
+- **Kafka**: Expose GSSAPI service name  ([#5474](https://github.com/kedacore/keda/issues/5474))
 
 ## v2.13.1
 

apis/keda/v1alpha1/scaledjob_webhook.go

@@ -0,0 +1,73 @@
+/*
+Copyright 2024 The KEDA 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 v1alpha1
+
+import (
+	"encoding/json"
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+)
+
+var scaledjoblog = logf.Log.WithName("scaledjob-validation-webhook")
+
+func (s *ScaledJob) SetupWebhookWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(s).
+		Complete()
+}
+
+// +kubebuilder:webhook:path=/validate-keda-sh-v1alpha1-scaledjob,mutating=false,failurePolicy=ignore,sideEffects=None,groups=keda.sh,resources=scaledjobs,verbs=create;update,versions=v1alpha1,name=vscaledjob.kb.io,admissionReviewVersions=v1
+
+var _ webhook.Validator = &ScaledJob{}
+
+// ValidateCreate implements webhook.Validator so a webhook will be registered for the type
+func (s *ScaledJob) ValidateCreate() (admission.Warnings, error) {
+	val, _ := json.MarshalIndent(s, "", "  ")
+	scaledjoblog.Info(fmt.Sprintf("validating scaledjob creation for %s", string(val)))
+	return nil, verifyTriggers(s, "create", false)
+}
+
+func (s *ScaledJob) ValidateUpdate(old runtime.Object) (admission.Warnings, error) {
+	val, _ := json.MarshalIndent(s, "", "  ")
+	scaledobjectlog.V(1).Info(fmt.Sprintf("validating scaledjob update for %s", string(val)))
+
+	oldTa := old.(*ScaledJob)
+	if isScaledJobRemovingFinalizer(s.ObjectMeta, oldTa.ObjectMeta, s.Spec, oldTa.Spec) {
+		scaledjoblog.V(1).Info("finalizer removal, skipping validation")
+		return nil, nil
+	}
+	return nil, verifyTriggers(s, "update", false)
+}
+
+func (s *ScaledJob) ValidateDelete() (admission.Warnings, error) {
+	return nil, nil
+}
+
+func isScaledJobRemovingFinalizer(om metav1.ObjectMeta, oldOm metav1.ObjectMeta, spec ScaledJobSpec, oldSpec ScaledJobSpec) bool {
+	taSpec, _ := json.MarshalIndent(spec, "", "  ")
+	oldTaSpec, _ := json.MarshalIndent(oldSpec, "", "  ")
+	taSpecString := string(taSpec)
+	oldTaSpecString := string(oldTaSpec)
+
+	return len(om.Finalizers) == 0 && len(oldOm.Finalizers) == 1 && taSpecString == oldTaSpecString
+}

apis/keda/v1alpha1/scaledjob_webhook_test.go

@@ -0,0 +1,83 @@
+/*
+Copyright 2024 The KEDA 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 v1alpha1
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var _ = It("should validate empty triggers in ScaledJob", func() {
+
+	namespaceName := "scaledjob-empty-triggers-set"
+	namespace := createNamespace(namespaceName)
+
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	sj := createScaledJob(sjName, namespaceName, []ScaleTriggers{})
+
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), sj)
+	}).Should(HaveOccurred())
+})
+
+// -------------------------------------------------------------------------- //
+// ----------------------------- HELP FUNCTIONS ----------------------------- //
+// -------------------------------------------------------------------------- //
+func createScaledJob(name string, namespace string, triggers []ScaleTriggers) *ScaledJob {
+	return &ScaledJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ScaledJob",
+			APIVersion: "keda.sh",
+		},
+		Spec: ScaledJobSpec{
+			JobTargetRef: &batchv1.JobSpec{
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"app": name,
+					},
+				},
+				Template: corev1.PodTemplateSpec{
+					ObjectMeta: metav1.ObjectMeta{
+						Labels: map[string]string{
+							"app": name,
+						},
+					},
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:  name,
+								Image: name,
+							},
+						},
+					},
+				},
+			},
+			Triggers: triggers,
+		},
+	}
+}

apis/keda/v1alpha1/scaledobject_webhook.go

@@ -132,7 +132,6 @@ func validateWorkload(so *ScaledObject, action string, dryRun bool) (admission.W
 
 	verifyFunctions := []func(*ScaledObject, string, bool) error{
 		verifyCPUMemoryScalers,
-		verifyTriggers,
 		verifyScaledObjects,
 		verifyHpas,
 		verifyReplicaCount,
@@ -145,6 +144,17 @@ func validateWorkload(so *ScaledObject, action string, dryRun bool) (admission.W
 		}
 	}
 
+	verifyCommonFunctions := []func(interface{}, string, bool) error{
+		verifyTriggers,
+	}
+
+	for i := range verifyCommonFunctions {
+		err := verifyCommonFunctions[i](so, action, dryRun)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	scaledobjectlog.V(1).Info(fmt.Sprintf("scaledobject %s is valid", so.Name))
 	return nil, nil
 }
@@ -158,11 +168,27 @@ func verifyReplicaCount(incomingSo *ScaledObject, action string, _ bool) error {
 	return nil
 }
 
-func verifyTriggers(incomingSo *ScaledObject, action string, _ bool) error {
-	err := ValidateTriggers(incomingSo.Spec.Triggers)
+func verifyTriggers(incomingObject interface{}, action string, _ bool) error {
+	var triggers []ScaleTriggers
+	var name string
+	var namespace string
+	switch obj := incomingObject.(type) {
+	case *ScaledObject:
+		triggers = obj.Spec.Triggers
+		name = obj.Name
+		namespace = obj.Namespace
+	case *ScaledJob:
+		triggers = obj.Spec.Triggers
+		name = obj.Name
+		namespace = obj.Namespace
+	default:
+		return fmt.Errorf("unknown scalable object type %v", incomingObject)
+	}
+
+	err := ValidateTriggers(triggers)
 	if err != nil {
-		scaledobjectlog.WithValues("name", incomingSo.Name).Error(err, "validation error")
-		metricscollector.RecordScaledObjectValidatingErrors(incomingSo.Namespace, action, "incorrect-triggers")
+		scaledobjectlog.WithValues("name", name).Error(err, "validation error")
+		metricscollector.RecordScaledObjectValidatingErrors(namespace, action, "incorrect-triggers")
 	}
 	return err
 }

apis/keda/v1alpha1/scaledobject_webhook_test.go

@@ -583,6 +583,30 @@ var _ = It("shouldn't create so when stabilizationWindowSeconds exceeds 3600", f
 		Should(HaveOccurred())
 })
 
+var _ = It("should validate empty triggers in ScaledObject", func() {
+
+	namespaceName := "empty-triggers-set"
+	namespace := createNamespace(namespaceName)
+
+	err := k8sClient.Create(context.Background(), namespace)
+	Expect(err).ToNot(HaveOccurred())
+
+	workload := createDeployment(namespaceName, false, false)
+	workload.Spec.Template.Spec.Containers[0].Resources.Limits = v1.ResourceList{
+		v1.ResourceCPU: *resource.NewMilliQuantity(100, resource.DecimalSI),
+	}
+
+	err = k8sClient.Create(context.Background(), workload)
+	Expect(err).ToNot(HaveOccurred())
+
+	so := createScaledObject(soName, namespaceName, workloadName, "apps/v1", "Deployment", false, map[string]string{}, "")
+	so.Spec.Triggers = []ScaleTriggers{}
+
+	Eventually(func() error {
+		return k8sClient.Create(context.Background(), so)
+	}).Should(HaveOccurred())
+})
+
 // ============================ SCALING MODIFIERS ============================ \\
 // =========================================================================== \\
 

apis/keda/v1alpha1/scaletriggers_types.go

@@ -51,6 +51,11 @@ type AuthenticationRef struct {
 // - useCachedMetrics is defined only for a supported triggers
 func ValidateTriggers(triggers []ScaleTriggers) error {
 	triggersCount := len(triggers)
+
+	if triggersCount == 0 {
+		return fmt.Errorf("no triggers defined in the ScaledObject/ScaledJob")
+	}
+
 	if triggers != nil && triggersCount > 0 {
 		triggerNames := make(map[string]bool, triggersCount)
 		for i := 0; i < triggersCount; i++ {
@@ -66,7 +71,7 @@ func ValidateTriggers(triggers []ScaleTriggers) error {
 			if name != "" {
 				if _, found := triggerNames[name]; found {
 					// found duplicate name
-					return fmt.Errorf("triggerName %q is defined multiple times in the ScaledObject, but it must be unique", name)
+					return fmt.Errorf("triggerName %q is defined multiple times in the ScaledObject/ScaledJob, but it must be unique", name)
 				}
 				triggerNames[name] = true
 			}

apis/keda/v1alpha1/scaletriggers_types_test.go

@@ -38,7 +38,7 @@ func TestValidateTriggers(t *testing.T) {
 					Type: "prometheus",
 				},
 			},
-			expectedErrMsg: "triggerName \"trigger1\" is defined multiple times in the ScaledObject, but it must be unique",
+			expectedErrMsg: "triggerName \"trigger1\" is defined multiple times in the ScaledObject/ScaledJob, but it must be unique",
 		},
 		{
 			name: "unsupported useCachedMetrics property for cpu scaler",
@@ -84,6 +84,11 @@ func TestValidateTriggers(t *testing.T) {
 			},
 			expectedErrMsg: "",
 		},
+		{
+			name:           "empty triggers array should be blocked",
+			triggers:       []ScaleTriggers{},
+			expectedErrMsg: "no triggers defined in the ScaledObject/ScaledJob",
+		},
 	}
 
 	for _, test := range tests {

apis/keda/v1alpha1/suite_test.go

@@ -52,6 +52,7 @@ var cancel context.CancelFunc
 const (
 	workloadName = "deployment-name"
 	soName       = "test-so"
+	sjName       = "test-sj"
 )
 
 func TestAPIs(t *testing.T) {
@@ -119,6 +120,8 @@ var _ = BeforeSuite(func() {
 
 	err = (&ScaledObject{}).SetupWebhookWithManager(mgr)
 	Expect(err).NotTo(HaveOccurred())
+	err = (&ScaledJob{}).SetupWebhookWithManager(mgr)
+	Expect(err).NotTo(HaveOccurred())
 	err = (&TriggerAuthentication{}).SetupWebhookWithManager(mgr)
 	Expect(err).NotTo(HaveOccurred())
 	err = (&ClusterTriggerAuthentication{}).SetupWebhookWithManager(mgr)

apis/keda/v1alpha1/triggerauthentication_types.go

@@ -139,13 +139,24 @@ const (
 // AuthPodIdentity allows users to select the platform native identity
 // mechanism
 type AuthPodIdentity struct {
-	// +kubebuilder:validation:Enum=azure;azure-workload;gcp;aws;aws-eks;aws-kiam
+	// +kubebuilder:validation:Enum=azure;azure-workload;gcp;aws;aws-eks;aws-kiam;none
 	Provider PodIdentityProvider `json:"provider"`
+
 	// +optional
 	IdentityID *string `json:"identityId"`
+
+	// +optional
+	// Set identityTenantId to override the default Azure tenant id. If this is set, then the IdentityID must also be set
+	IdentityTenantID *string `json:"identityTenantId"`
+
 	// +optional
+	// Set identityAuthorityHost to override the default Azure authority host. If this is set, then the IdentityTenantID must also be set
+	IdentityAuthorityHost *string `json:"identityAuthorityHost"`
+
+	// +kubebuilder:validation:Optional
 	// RoleArn sets the AWS RoleArn to be used. Mutually exclusive with IdentityOwner
-	RoleArn string `json:"roleArn"`
+	RoleArn *string `json:"roleArn"`
+
 	// +kubebuilder:validation:Enum=keda;workload
 	// +optional
 	// IdentityOwner configures which identity has to be used during auto discovery, keda or the scaled workload. Mutually exclusive with roleArn
@@ -159,6 +170,20 @@ func (a *AuthPodIdentity) GetIdentityID() string {
 	return *a.IdentityID
 }
 
+func (a *AuthPodIdentity) GetIdentityTenantID() string {
+	if a.IdentityTenantID == nil {
+		return ""
+	}
+	return *a.IdentityTenantID
+}
+
+func (a *AuthPodIdentity) GetIdentityAuthorityHost() string {
+	if a.IdentityAuthorityHost == nil {
+		return ""
+	}
+	return *a.IdentityAuthorityHost
+}
+
 func (a *AuthPodIdentity) IsWorkloadIdentityOwner() bool {
 	if a.IdentityOwner == nil {
 		return false