Add Fallback option UseCurrentReplicasAsMinimum

CHANGELOG.md

@@ -59,6 +59,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 
 ### New
 
+- **General**: Add fallback `behavior` with option `useCurrentReplicasAsMinimum` to use current number of replicas ([#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))

apis/keda/v1alpha1/scaledobject_types.go

@@ -56,6 +56,8 @@ 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 FallbackBehaviorUseCurrentReplicasAsMin = "useCurrentReplicasAsMinimum"
 
 // HealthStatus is the status for a ScaledObject's health
 type HealthStatus struct {
@@ -109,6 +111,10 @@ type ScaledObjectSpec struct {
 type Fallback struct {
 	FailureThreshold int32 `json:"failureThreshold"`
 	Replicas         int32 `json:"replicas"`
+	// +optional
+	// +kubebuilder:default=static
+	// +kubebuilder:validation:Enum=static;useCurrentReplicasAsMinimum
+	Behavior string `json:"behavior,omitempty"`
 }
 
 // AdvancedConfig specifies advance scaling options

config/crd/bases/keda.sh_scaledobjects.yaml

@@ -225,6 +225,12 @@ spec:
               fallback:
                 description: Fallback is the spec for fallback options
                 properties:
+                  behavior:
+                    default: static
+                    enum:
+                    - static
+                    - useCurrentReplicasAsMinimum
+                    type: string
                   failureThreshold:
                     format: int32
                     type: integer

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")
@@ -46,7 +48,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)
@@ -76,7 +78,11 @@ 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
+		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
 	}
@@ -103,8 +109,25 @@ 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.FallbackBehaviorUseCurrentReplicasAsMin:
+		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())
@@ -121,7 +144,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
 }