rfac: handle paused scaling directly in reconciler

CHANGELOG.md

@@ -87,6 +87,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - **General**: Check updated status for Fallback condition instead of ScaledObject ([#7488](https://github.com/kedacore/keda/issues/7488))
 - **General**: Fix int64 overflow in milli-quantity conversion for very large metric values ([#7441](https://github.com/kedacore/keda/issues/7441))
 - **General**: Fix ScaledObject admission webhook to return validation error from `verifyReplicaCount`, preventing invalid ScaledObjects from being created ([#5954](https://github.com/kedacore/keda/issues/5954))
+- **General**: Handle paused scaling directly in reconciler ([#7663](https://github.com/kedacore/keda/issues/7663))
 - **Azure Data Explorer Scaler**: Remove clientSecretFromEnv support ([#7554](https://github.com/kedacore/keda/pull/7554))
 - **Cron Scaler**: Fix metric name generation so cron expressions with comma-separated values no longer produce invalid metric names ([#7448](https://github.com/kedacore/keda/issues/7448))
 - **Forgejo Scaler**: Limit HTTP error response logging ([#7469](https://github.com/kedacore/keda/pull/7469))

apis/keda/v1alpha1/scaledjob_types.go

@@ -17,6 +17,8 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"strconv"
+
 	batchv1 "k8s.io/api/batch/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -169,6 +171,19 @@ func (s *ScaledJob) GenerateIdentifier() string {
 	return GenerateIdentifier("ScaledJob", s.Namespace, s.Name)
 }
 
+// NeedToBePausedByAnnotation checks whether this ScaledJob should be paused based on the PausedAnnotation.
+func (s *ScaledJob) NeedToBePausedByAnnotation() bool {
+	value, found := s.GetAnnotations()[PausedAnnotation]
+	if !found {
+		return false
+	}
+	boolVal, err := strconv.ParseBool(value)
+	if err != nil {
+		return true
+	}
+	return boolVal
+}
+
 // GetStatusConditions returns a pointer to the status conditions for in-place modification.
 func (s *ScaledJob) GetStatusConditions() *Conditions { return &s.Status.Conditions }
 

apis/keda/v1alpha1/scaledobject_types.go

@@ -278,6 +278,21 @@ func getBoolAnnotation(so *ScaledObject, annotation string) bool {
 	return boolVal
 }
 
+// GetPausedReplicaCount returns the paused replica count from the annotation, nil if not present.
+func (so *ScaledObject) GetPausedReplicaCount() (*int32, error) {
+	if so.Annotations != nil {
+		if val, ok := so.Annotations[PausedReplicasAnnotation]; ok {
+			conv, err := strconv.ParseInt(val, 10, 32)
+			if err != nil {
+				return nil, err
+			}
+			count := int32(conv)
+			return &count, nil
+		}
+	}
+	return nil, nil
+}
+
 // IsUsingModifiers determines whether scalingModifiers are defined or not
 func (so *ScaledObject) IsUsingModifiers() bool {
 	return so.Spec.Advanced != nil && !reflect.DeepEqual(so.Spec.Advanced.ScalingModifiers, ScalingModifiers{})

apis/keda/v1alpha1/scaledobject_types_test.go

@@ -643,6 +643,92 @@ func TestGetHPAReplicas(t *testing.T) {
 	}
 }
 
+func TestGetPausedReplicaCount(t *testing.T) {
+	tests := []struct {
+		name      string
+		so        *ScaledObject
+		wantNil   bool
+		wantValue int32
+		wantErr   bool
+	}{
+		{
+			name:    "nil annotations returns nil",
+			so:      &ScaledObject{},
+			wantNil: true,
+		},
+		{
+			name: "annotation absent returns nil",
+			so: func() *ScaledObject {
+				so := &ScaledObject{}
+				so.Annotations = map[string]string{"other-annotation": "value"}
+				return so
+			}(),
+			wantNil: true,
+		},
+		{
+			name: "valid integer annotation returns pointer to value",
+			so: func() *ScaledObject {
+				so := &ScaledObject{}
+				so.Annotations = map[string]string{
+					PausedReplicasAnnotation: "5",
+				}
+				return so
+			}(),
+			wantNil:   false,
+			wantValue: 5,
+		},
+		{
+			name: "zero value annotation returns pointer to zero",
+			so: func() *ScaledObject {
+				so := &ScaledObject{}
+				so.Annotations = map[string]string{
+					PausedReplicasAnnotation: "0",
+				}
+				return so
+			}(),
+			wantNil:   false,
+			wantValue: 0,
+		},
+		{
+			name: "invalid string annotation returns error",
+			so: func() *ScaledObject {
+				so := &ScaledObject{}
+				so.Annotations = map[string]string{
+					PausedReplicasAnnotation: "not-a-number",
+				}
+				return so
+			}(),
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := tt.so.GetPausedReplicaCount()
+			if tt.wantErr {
+				if err == nil {
+					t.Error("Expected error but got nil")
+				}
+				return
+			}
+			if err != nil {
+				t.Errorf("Expected no error but got: %v", err)
+				return
+			}
+			if tt.wantNil {
+				if result != nil {
+					t.Errorf("Expected nil but got %d", *result)
+				}
+			} else {
+				if result == nil {
+					t.Error("Expected non-nil result but got nil")
+				} else if *result != tt.wantValue {
+					t.Errorf("Expected %d but got %d", tt.wantValue, *result)
+				}
+			}
+		})
+	}
+}
+
 // Helper function to check if a string contains a substring
 func contains(s, substr string) bool {
 	return strings.Contains(s, substr)

controllers/keda/hpa.go

@@ -33,7 +33,6 @@ 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"
 	kedastatus "github.com/kedacore/keda/v2/pkg/status"
 	version "github.com/kedacore/keda/v2/version"
 )
@@ -162,19 +161,6 @@ func (r *ScaledObjectReconciler) newHPAForScaledObject(ctx context.Context, logg
 	minReplicas := scaledObject.GetHPAMinReplicas()
 	maxReplicas := scaledObject.GetHPAMaxReplicas()
 
-	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 := &autoscalingv2.HorizontalPodAutoscaler{
 		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
 			MinReplicas: minReplicas,

controllers/keda/scaledjob_controller.go

@@ -227,35 +227,29 @@ func (r *ScaledJobReconciler) reconcileScaledJob(ctx context.Context, logger log
 	return "ScaledJob is defined correctly and is ready to scaling", nil
 }
 
-// checkIfPaused checks the presence of "autoscaling.keda.sh/paused" annotation on the scaledJob and stop the scale loop.
+// checkIfPaused checks the presence of "autoscaling.keda.sh/paused" annotation on the scaledJob and stops the scale loop.
 func (r *ScaledJobReconciler) checkIfPaused(ctx context.Context, logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob, conditions *kedav1alpha1.Conditions) (bool, error) {
-	pausedAnnotationValue, pausedAnnotation := scaledJob.GetAnnotations()[kedav1alpha1.PausedAnnotation]
-	pausedStatus := conditions.GetPausedCondition().Status == metav1.ConditionTrue
-	shouldPause := false
-	if pausedAnnotation {
-		var err error
-		shouldPause, err = strconv.ParseBool(pausedAnnotationValue)
-		if err != nil {
-			shouldPause = true
-		}
-	}
+	shouldPause := scaledJob.NeedToBePausedByAnnotation()
+	isPausedInStatus := conditions.GetPausedCondition().Status == metav1.ConditionTrue
+
 	if shouldPause {
-		if !pausedStatus {
-			logger.Info("ScaledJob is paused, stopping scaling loop.")
+		// Set Paused condition before stopping scale loop to prevent races with new reconciles
+		if !isPausedInStatus {
 			msg := kedav1alpha1.ScaledJobConditionPausedMessage
-			if err := r.stopScaleLoop(ctx, logger, scaledJob); err != nil {
-				msg = "failed to stop the scale loop for paused ScaledJob"
-				conditions.SetPausedCondition(metav1.ConditionFalse, "ScaledJobStopScaleLoopFailed", msg)
-				r.EventEmitter.Emit(scaledJob, scaledJob.Namespace, corev1.EventTypeWarning, eventingv1alpha1.ScaledJobFailedType, eventreason.ScaledJobPauseFailed, msg)
+			conditions.SetPausedCondition(metav1.ConditionTrue, kedav1alpha1.ScaledJobConditionPausedReason, msg)
+			if err := kedastatus.SetStatusConditions(ctx, r.Client, logger, scaledJob, conditions); err != nil {
 				return false, err
 			}
-			conditions.SetPausedCondition(metav1.ConditionTrue, kedav1alpha1.ScaledJobConditionPausedReason, msg)
 			r.EventEmitter.Emit(scaledJob, scaledJob.Namespace, corev1.EventTypeNormal, eventingv1alpha1.ScaledJobPausedType, eventreason.ScaledJobPaused, msg)
 		}
+
+		if err := r.stopScaleLoop(ctx, logger, scaledJob); err != nil {
+			return false, err
+		}
 		return true, nil
 	}
-	if pausedStatus {
-		logger.Info("Unpausing ScaledJob.")
+
+	if isPausedInStatus {
 		msg := kedav1alpha1.ScaledJobConditionUnpausedMessage
 		conditions.SetPausedCondition(metav1.ConditionFalse, kedav1alpha1.ScaledJobConditionUnpausedReason, msg)
 		r.EventEmitter.Emit(scaledJob, scaledJob.Namespace, corev1.EventTypeNormal, eventingv1alpha1.ScaledJobUnpausedType, eventreason.ScaledJobUnpaused, msg)