Add Fallback option behavior for dynamic fallback calculation (keda…

…core#6464) Signed-off-by: rickbrouwer <[email protected]> Signed-off-by: Rick Brouwer <[email protected]> Co-authored-by: Jan Wozniak <[email protected]>
rickbrouwer · Feb 22, 2025 · 4cd84f3 · 4cd84f3
1 parent 767eb3d
commit 4cd84f3
Show file tree

Hide file tree

Showing 9 changed files with 598 additions and 52 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -63,6 +63,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 
 ### New
 
+- **General**: Add Fallback option `behavior` for dynamic fallback calculation ([#6450](https://github.com/kedacore/keda/issues/6450))
 - **General**: Enable OpenSSF Scorecard to enhance security practices across the project ([#5913](https://github.com/kedacore/keda/issues/5913))
 - **General**: Introduce new NSQ scaler ([#3281](https://github.com/kedacore/keda/issues/3281))
 - **General**: Operator flag to control patching of webhook resources certificates ([#6184](https://github.com/kedacore/keda/issues/6184))

diff --git a/apis/keda/v1alpha1/scaledobject_types.go b/apis/keda/v1alpha1/scaledobject_types.go
@@ -59,6 +59,10 @@ const ScaledObjectTransferHpaOwnershipAnnotation = "scaledobject.keda.sh/transfe
 const ValidationsHpaOwnershipAnnotation = "validations.keda.sh/hpa-ownership"
 const PausedReplicasAnnotation = "autoscaling.keda.sh/paused-replicas"
 const PausedAnnotation = "autoscaling.keda.sh/paused"
+const FallbackBehaviorStatic = "static"
+const FallbackBehaviorCurrentReplicas = "currentReplicas"
+const FallbackBehaviorCurrentReplicasIfHigher = "currentReplicasIfHigher"
+const FallbackBehaviorCurrentReplicasIfLower = "currentReplicasIfLower"
 
 // HealthStatus is the status for a ScaledObject's health
 type HealthStatus struct {
@@ -112,6 +116,10 @@ type ScaledObjectSpec struct {
 type Fallback struct {
 	FailureThreshold int32 `json:"failureThreshold"`
 	Replicas         int32 `json:"replicas"`
+	// +optional
+	// +kubebuilder:default=static
+	// +kubebuilder:validation:Enum=static;currentReplicas;currentReplicasIfHigher;currentReplicasIfLower
+	Behavior string `json:"behavior,omitempty"`
 }
 
 // AdvancedConfig specifies advance scaling options

diff --git a/config/crd/bases/keda.sh_scaledobjects.yaml b/config/crd/bases/keda.sh_scaledobjects.yaml
@@ -225,6 +225,14 @@ spec:
               fallback:
                 description: Fallback is the spec for fallback options
                 properties:
+                  behavior:
+                    default: static
+                    enum:
+                    - static
+                    - currentReplicas
+                    - currentReplicasIfHigher
+                    - currentReplicasIfLower
+                    type: string
                   failureThreshold:
                     format: int32
                     type: integer

diff --git a/pkg/fallback/fallback.go b/pkg/fallback/fallback.go
@@ -24,11 +24,13 @@ import (
 	v2 "k8s.io/api/autoscaling/v2"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/scale"
 	"k8s.io/metrics/pkg/apis/external_metrics"
 	runtimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+	"github.com/kedacore/keda/v2/pkg/scaling/resolver"
 )
 
 var log = logf.Log.WithName("fallback")
@@ -51,7 +53,7 @@ func isFallbackEnabled(scaledObject *kedav1alpha1.ScaledObject, metricSpec v2.Me
 	return true
 }
 
-func GetMetricsWithFallback(ctx context.Context, client runtimeclient.Client, metrics []external_metrics.ExternalMetricValue, suppressedError error, metricName string, scaledObject *kedav1alpha1.ScaledObject, metricSpec v2.MetricSpec) ([]external_metrics.ExternalMetricValue, bool, error) {
+func GetMetricsWithFallback(ctx context.Context, client runtimeclient.Client, scaleClient scale.ScalesGetter, metrics []external_metrics.ExternalMetricValue, suppressedError error, metricName string, scaledObject *kedav1alpha1.ScaledObject, metricSpec v2.MetricSpec) ([]external_metrics.ExternalMetricValue, bool, error) {
 	status := scaledObject.Status.DeepCopy()
 
 	initHealthStatus(status)
@@ -81,7 +83,16 @@ func GetMetricsWithFallback(ctx context.Context, client runtimeclient.Client, me
 		log.Info("Failed to validate ScaledObject Spec. Please check that parameters are positive integers", "scaledObject.Namespace", scaledObject.Namespace, "scaledObject.Name", scaledObject.Name)
 		return nil, false, suppressedError
 	case *healthStatus.NumberOfFailures > scaledObject.Spec.Fallback.FailureThreshold:
-		return doFallback(scaledObject, metricSpec, metricName, suppressedError), true, nil
+		var currentReplicas int32
+		var err error
+
+		if scaledObject.Spec.Fallback.Behavior != kedav1alpha1.FallbackBehaviorStatic {
+			currentReplicas, err = resolver.GetCurrentReplicas(ctx, client, scaleClient, scaledObject)
+			if err != nil {
+				return nil, false, suppressedError
+			}
+		}
+		return doFallback(scaledObject, metricSpec, metricName, currentReplicas, suppressedError), true, nil
 	default:
 		return nil, false, suppressedError
 	}
@@ -108,8 +119,34 @@ func HasValidFallback(scaledObject *kedav1alpha1.ScaledObject) bool {
 		modifierChecking
 }
 
-func doFallback(scaledObject *kedav1alpha1.ScaledObject, metricSpec v2.MetricSpec, metricName string, suppressedError error) []external_metrics.ExternalMetricValue {
-	replicas := int64(scaledObject.Spec.Fallback.Replicas)
+func doFallback(scaledObject *kedav1alpha1.ScaledObject, metricSpec v2.MetricSpec, metricName string, currentReplicas int32, suppressedError error) []external_metrics.ExternalMetricValue {
+	fallbackBehavior := scaledObject.Spec.Fallback.Behavior
+	fallbackReplicas := int64(scaledObject.Spec.Fallback.Replicas)
+	var replicas int64
+
+	switch fallbackBehavior {
+	case kedav1alpha1.FallbackBehaviorStatic:
+		replicas = fallbackReplicas
+	case kedav1alpha1.FallbackBehaviorCurrentReplicas:
+		replicas = int64(currentReplicas)
+	case kedav1alpha1.FallbackBehaviorCurrentReplicasIfHigher:
+		currentReplicasCount := int64(currentReplicas)
+		if currentReplicasCount > fallbackReplicas {
+			replicas = currentReplicasCount
+		} else {
+			replicas = fallbackReplicas
+		}
+	case kedav1alpha1.FallbackBehaviorCurrentReplicasIfLower:
+		currentReplicasCount := int64(currentReplicas)
+		if currentReplicasCount < fallbackReplicas {
+			replicas = currentReplicasCount
+		} else {
+			replicas = fallbackReplicas
+		}
+	default:
+		replicas = fallbackReplicas
+	}
+
 	var normalisationValue int64
 	if !scaledObject.IsUsingModifiers() {
 		normalisationValue = int64(metricSpec.External.Target.AverageValue.AsApproximateFloat64())
@@ -126,7 +163,13 @@ func doFallback(scaledObject *kedav1alpha1.ScaledObject, metricSpec v2.MetricSpe
 	}
 	fallbackMetrics := []external_metrics.ExternalMetricValue{metric}
 
-	log.Info("Suppressing error, falling back to fallback.replicas", "scaledObject.Namespace", scaledObject.Namespace, "scaledObject.Name", scaledObject.Name, "suppressedError", suppressedError, "fallback.replicas", replicas)
+	log.Info("Suppressing error, using fallback metrics",
+		"scaledObject.Namespace", scaledObject.Namespace,
+		"scaledObject.Name", scaledObject.Name,
+		"suppressedError", suppressedError,
+		"fallback.behavior", fallbackBehavior,
+		"fallback.replicas", fallbackReplicas,
+		"workload.currentReplicas", currentReplicas)
 	return fallbackMetrics
 }