add annotation to pause autoscaling at a fixed count (#2765)

Signed-off-by: Sanskar Jaiswal <jaiswalsanskar078@gmail.com>

CHANGELOG.md

@@ -30,11 +30,12 @@
 
 ### New
 
+- **General:** Introduce annotation `"autoscaling.keda.sh/paused-replicas"` for ScaledObjects to pause scaling at a fixed replica count. ([#944](https://github.com/kedacore/keda/issues/944))
+- **General:** Introduce ARM-based container image for KEDA ([#2263](https://github.com/kedacore/keda/issues/2263)|[#2262](https://github.com/kedacore/keda/issues/2262))
 - **General:** Introduce new AWS DynamoDB Scaler ([#2486](https://github.com/kedacore/keda/issues/2482))
 - **General:** Introduce new Azure Data Explorer Scaler ([#1488](https://github.com/kedacore/keda/issues/1488)|[#2734](https://github.com/kedacore/keda/issues/2734))
 - **General:** Introduce new GCP Stackdriver Scaler ([#2661](https://github.com/kedacore/keda/issues/2661))
 - **General:** Introduce new GCP Storage Scaler ([#2628](https://github.com/kedacore/keda/issues/2628))
-- **General:** Introduce ARM-based container image for KEDA ([#2263](https://github.com/kedacore/keda/issues/2263)|[#2262](https://github.com/kedacore/keda/issues/2262))
 - **General:** Provide support for authentication via Azure Key Vault ([#900](https://github.com/kedacore/keda/issues/900)|[#2733](https://github.com/kedacore/keda/issues/2733))
 - **General**: Support for `ValueMetricType` in `ScaledObject` for all scalers except CPU/Memory ([#2030](https://github.com/kedacore/keda/issues/2030))
 

apis/keda/v1alpha1/scaledobject_types.go

@@ -150,6 +150,8 @@ type ScaledObjectStatus struct {
 	Conditions Conditions `json:"conditions,omitempty"`
 	// +optional
 	Health map[string]HealthStatus `json:"health,omitempty"`
+	// +optional
+	PausedReplicaCount *int32 `json:"pausedReplicaCount,omitempty"`
 }
 
 // +kubebuilder:object:root=true

config/crd/bases/keda.sh_scaledobjects.yaml

@@ -333,6 +333,9 @@ spec:
               originalReplicaCount:
                 format: int32
                 type: integer
+              pausedReplicaCount:
+                format: int32
+                type: integer
               resourceMetricNames:
                 items:
                   type: string

controllers/keda/hpa.go

@@ -31,6 +31,7 @@ import (
 
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
 	kedacontrollerutil "github.com/kedacore/keda/v2/controllers/keda/util"
+	"github.com/kedacore/keda/v2/pkg/scaling/executor"
 	version "github.com/kedacore/keda/v2/version"
 )
 
@@ -90,10 +91,26 @@ func (r *ScaledObjectReconciler) newHPAForScaledObject(ctx context.Context, logg
 		labels[key] = value
 	}
 
+	minReplicas := getHPAMinReplicas(scaledObject)
+	maxReplicas := getHPAMaxReplicas(scaledObject)
+
+	pausedCount, err := executor.GetPausedReplicaCount(scaledObject)
+	if err != nil {
+		return nil, err
+	}
+	if pausedCount != nil {
+		// MinReplicas on HPA can't be 0
+		if *pausedCount == 0 {
+			*pausedCount = 1
+		}
+		minReplicas = pausedCount
+		maxReplicas = *pausedCount
+	}
+
 	hpa := &autoscalingv2beta2.HorizontalPodAutoscaler{
 		Spec: autoscalingv2beta2.HorizontalPodAutoscalerSpec{
-			MinReplicas: getHPAMinReplicas(scaledObject),
-			MaxReplicas: getHPAMaxReplicas(scaledObject),
+			MinReplicas: minReplicas,
+			MaxReplicas: maxReplicas,
 			Metrics:     scaledObjectMetricSpecs,
 			Behavior:    behavior,
 			ScaleTargetRef: autoscalingv2beta2.CrossVersionObjectReference{

controllers/keda/scaledobject_controller.go

@@ -122,7 +122,9 @@ func (r *ScaledObjectReconciler) SetupWithManager(mgr ctrl.Manager, options cont
 		// predicate.GenerationChangedPredicate{} ignore updates to ScaledObject Status
 		// (in this case metadata.Generation does not change)
 		// so reconcile loop is not started on Status updates
-		For(&kedav1alpha1.ScaledObject{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
+		For(&kedav1alpha1.ScaledObject{}, builder.WithPredicates(
+			predicate.Or(kedacontrollerutil.PausedReplicasPredicate{}, predicate.GenerationChangedPredicate{}),
+		)).
 		Owns(&autoscalingv2beta2.HorizontalPodAutoscaler{}).
 		Complete(r)
 }
@@ -281,7 +283,9 @@ func (r *ScaledObjectReconciler) checkTargetResourceIsScalable(ctx context.Conte
 	logger.V(1).Info("Parsed Group, Version, Kind, Resource", "GVK", gvkString, "Resource", gvkr.Resource)
 
 	// do we need the scale to update the status later?
-	wantStatusUpdate := scaledObject.Status.ScaleTargetKind != gvkString || scaledObject.Status.OriginalReplicaCount == nil
+	_, present := scaledObject.GetAnnotations()[kedacontrollerutil.PausedReplicasAnnotation]
+	removePausedStatus := scaledObject.Status.PausedReplicaCount != nil && !present
+	wantStatusUpdate := scaledObject.Status.ScaleTargetKind != gvkString || scaledObject.Status.OriginalReplicaCount == nil || removePausedStatus
 
 	// check if we already know.
 	var scale *autoscalingv1.Scale
@@ -321,6 +325,10 @@ func (r *ScaledObjectReconciler) checkTargetResourceIsScalable(ctx context.Conte
 			status.OriginalReplicaCount = &scale.Spec.Replicas
 		}
 
+		if removePausedStatus {
+			status.PausedReplicaCount = nil
+		}
+
 		if err := kedacontrollerutil.UpdateScaledObjectStatus(ctx, r.Client, logger, scaledObject, status); err != nil {
 			return gvkr, err
 		}

controllers/keda/util/predicate.go

@@ -0,0 +1,30 @@
+package util
+
+import (
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+const PausedReplicasAnnotation = "autoscaling.keda.sh/paused-replicas"
+
+type PausedReplicasPredicate struct {
+	predicate.Funcs
+}
+
+func (PausedReplicasPredicate) Update(e event.UpdateEvent) bool {
+	if e.ObjectOld == nil || e.ObjectNew == nil {
+		return false
+	}
+
+	newAnnotations := e.ObjectNew.GetAnnotations()
+	oldAnnotations := e.ObjectOld.GetAnnotations()
+	if newAnnotations != nil && oldAnnotations != nil {
+		if newVal, ok1 := newAnnotations[PausedReplicasAnnotation]; ok1 {
+			if oldVal, ok2 := oldAnnotations[PausedReplicasAnnotation]; ok2 {
+				return newVal != oldVal
+			}
+			return true
+		}
+	}
+	return false
+}

pkg/scaling/executor/scale_scaledobjects.go

@@ -18,6 +18,7 @@ package executor
 
 import (
 	"context"
+	"strconv"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -28,6 +29,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+	kedacontrollerutil "github.com/kedacore/keda/v2/controllers/keda/util"
+
 	"github.com/kedacore/keda/v2/pkg/eventreason"
 )
 
@@ -69,12 +72,6 @@ func (e *scaleExecutor) RequestScale(ctx context.Context, scaledObject *kedav1al
 		currentReplicas = currentScale.Spec.Replicas
 	}
 
-	// if scaledObject.Spec.MinReplicaCount is not set, then set the default value (0)
-	minReplicas := int32(0)
-	if scaledObject.Spec.MinReplicaCount != nil {
-		minReplicas = *scaledObject.Spec.MinReplicaCount
-	}
-
 	// if the ScaledObject's triggers aren't in the error state,
 	// but ScaledObject.Status.ReadyCondition is set not set to 'true' -> set it back to 'true'
 	readyCondition := scaledObject.Status.Conditions.GetReadyCondition()
@@ -85,6 +82,44 @@ func (e *scaleExecutor) RequestScale(ctx context.Context, scaledObject *kedav1al
 		}
 	}
 
+	// Check if we are paused, and if we are then update the scale to the desired count.
+	pausedCount, err := GetPausedReplicaCount(scaledObject)
+	if err != nil {
+		if err := e.setReadyCondition(ctx, logger, scaledObject, metav1.ConditionFalse,
+			kedav1alpha1.ScaledObjectConditionReadySucccesReason, kedav1alpha1.ScaledObjectConditionReadySuccessMessage); err != nil {
+			logger.Error(err, "error setting ready condition")
+		}
+		logger.Error(err, "error getting the paused replica count on the current ScaledObject.")
+		return
+	}
+
+	status := scaledObject.Status.DeepCopy()
+	if pausedCount != nil && *pausedCount != currentReplicas && status.PausedReplicaCount == nil {
+		_, err := e.updateScaleOnScaleTarget(ctx, scaledObject, currentScale, *pausedCount)
+		if err != nil {
+			logger.Error(err, "error scaling target to paused replicas count", "paused replicas", *pausedCount)
+			if err := e.setReadyCondition(ctx, logger, scaledObject, metav1.ConditionUnknown,
+				kedav1alpha1.ScaledObjectConditionReadySucccesReason, kedav1alpha1.ScaledObjectConditionReadySuccessMessage); err != nil {
+				logger.Error(err, "error setting ready condition")
+			}
+			return
+		}
+		status.PausedReplicaCount = pausedCount
+		err = kedacontrollerutil.UpdateScaledObjectStatus(ctx, e.client, logger, scaledObject, status)
+		if err != nil {
+			logger.Error(err, "error updating status paused replica count")
+			return
+		}
+		logger.Info("Successfully scaled target to paused replicas count", "paused replicas", *pausedCount)
+		return
+	}
+
+	// if scaledObject.Spec.MinReplicaCount is not set, then set the default value (0)
+	minReplicas := int32(0)
+	if scaledObject.Spec.MinReplicaCount != nil {
+		minReplicas = *scaledObject.Spec.MinReplicaCount
+	}
+
 	if isActive {
 		switch {
 		case scaledObject.Spec.IdleReplicaCount != nil && currentReplicas < minReplicas,
@@ -320,3 +355,17 @@ func getIdleOrMinimumReplicaCount(scaledObject *kedav1alpha1.ScaledObject) (bool
 
 	return false, *scaledObject.Spec.MinReplicaCount
 }
+
+func GetPausedReplicaCount(scaledObject *kedav1alpha1.ScaledObject) (*int32, error) {
+	if scaledObject.Annotations != nil {
+		if val, ok := scaledObject.Annotations[kedacontrollerutil.PausedReplicasAnnotation]; ok {
+			conv, err := strconv.ParseInt(val, 10, 32)
+			if err != nil {
+				return nil, err
+			}
+			count := int32(conv)
+			return &count, nil
+		}
+	}
+	return nil, nil
+}

pkg/scaling/executor/scale_scaledobjects_test.go

@@ -398,3 +398,65 @@ func TestScaleFromIdleToMinReplicasWhenActive(t *testing.T) {
 	condition := scaledObject.Status.Conditions.GetActiveCondition()
 	assert.Equal(t, true, condition.IsTrue())
 }
+
+func TestScaleToPausedReplicasCount(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	client := mock_client.NewMockClient(ctrl)
+	recorder := record.NewFakeRecorder(1)
+	mockScaleClient := mock_scale.NewMockScalesGetter(ctrl)
+	mockScaleInterface := mock_scale.NewMockScaleInterface(ctrl)
+	statusWriter := mock_client.NewMockStatusWriter(ctrl)
+
+	scaleExecutor := NewScaleExecutor(client, mockScaleClient, nil, recorder)
+
+	scaledObject := v1alpha1.ScaledObject{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "name",
+			Namespace: "namespace",
+			Annotations: map[string]string{
+				"autoscaling.keda.sh/paused-replicas": "0",
+			},
+		},
+		Spec: v1alpha1.ScaledObjectSpec{
+			ScaleTargetRef: &v1alpha1.ScaleTarget{
+				Name: "name",
+			},
+		},
+		Status: v1alpha1.ScaledObjectStatus{
+			ScaleTargetGVKR: &v1alpha1.GroupVersionKindResource{
+				Group: "apps",
+				Kind:  "Deployment",
+			},
+		},
+	}
+
+	scaledObject.Status.Conditions = *v1alpha1.GetInitializedConditions()
+
+	pausedReplicaCount := int32(0)
+	replicaCount := int32(2)
+
+	client.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).SetArg(2, appsv1.Deployment{
+		Spec: appsv1.DeploymentSpec{
+			Replicas: &replicaCount,
+		},
+	})
+
+	scale := &autoscalingv1.Scale{
+		Spec: autoscalingv1.ScaleSpec{
+			Replicas: replicaCount,
+		},
+	}
+
+	mockScaleClient.EXPECT().Scales(gomock.Any()).Return(mockScaleInterface).Times(2)
+	mockScaleInterface.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(scale, nil)
+	mockScaleInterface.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Eq(scale), gomock.Any())
+
+	client.EXPECT().Status().Return(statusWriter).Times(2)
+	statusWriter.EXPECT().Patch(gomock.Any(), gomock.Any(), gomock.Any()).Times(2)
+
+	scaleExecutor.RequestScale(context.TODO(), &scaledObject, true, false)
+
+	assert.Equal(t, pausedReplicaCount, scale.Spec.Replicas)
+	condition := scaledObject.Status.Conditions.GetActiveCondition()
+	assert.Equal(t, false, condition.IsTrue())
+}