feat: Provide support for explicitly pausing autoscaling of ScaledJobs

CHANGELOG.md

@@ -35,6 +35,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - [v1.2.0](#v120)
 - [v1.1.0](#v110)
 - [v1.0.0](#v100)
+- **General:** Introduce autoscaling is paused annotation for ScaledJobs ([#3303](https://github.com/kedacore/keda/issues/3303))
 
 ### New
 

apis/keda/v1alpha1/scaledjob_types.go

@@ -73,6 +73,8 @@ type ScaledJobStatus struct {
  LastActiveTime *metav1.Time `json:"lastActiveTime,omitempty"`
  // +optional
  Conditions Conditions `json:"conditions,omitempty"`
+ // +optional
+ Paused string `json:"isPaused,omitempty"`
 }
 
 // ScaledJobList contains a list of ScaledJob

config/crd/bases/keda.sh_scaledjobs.yaml

@@ -7743,6 +7743,8 @@ spec:
  lastActiveTime:
  format: date-time
  type: string
+ Paused:
+ type: string
  type: object
  type: object
  served: true

controllers/keda/scaledjob_controller.go

@@ -77,7 +77,11 @@ func (r *ScaledJobReconciler) SetupWithManager(mgr ctrl.Manager, options control
  WithOptions(options).
  // Ignore updates to ScaledJob Status (in this case metadata.Generation does not change)
  // so reconcile loop is not started on Status updates
- For(&kedav1alpha1.ScaledJob{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
+ For(&kedav1alpha1.ScaledJob{}, builder.WithPredicates(
+ predicate.Or(
+ kedacontrollerutil.PausedPredicate{},
+ predicate.GenerationChangedPredicate{},
+ ))).
  Complete(r)
 }
 
@@ -121,6 +125,11 @@ func (r *ScaledJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
  return ctrl.Result{}, err
  }
  }
+ // check if scaledJob.Status.IsPaused constant is not an empty string, if not an empty string then stopScaleLoop
+ if scaledJob.Status.Paused != "" {
+ reqLogger.Info("ScaledJob is paused, stopping scale loop")
+ return ctrl.Result{}, r.Pause(ctx, reqLogger, scaledJob)
+ }
 
  // Check jobTargetRef is specified
  if scaledJob.Spec.JobTargetRef == nil {
@@ -322,3 +331,13 @@ func (r *ScaledJobReconciler) updateTriggerTotalsOnDelete(namespacedName string)
 
  delete(scaledJobTriggers, namespacedName)
 }
+//stopScaleLoop when scaledJob.Status.IsPaused constant is set
+func (r *ScaledJobReconciler) Pause(ctx context.Context, logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob) error {
+ if scaledJob.Annotations["scaledjob.keda.sh/paused"] == "true" {
+ logger.V(1).Info("Stopping a ScaleLoop when scaledJob is paused")
+ // stopScaleLoop
+ return r.stopScaleLoop(ctx, logger, scaledJob)
+ //return r.scaleHandler.DeleteScalableObject(ctx, scaledJob)
+ }
+ return nil
+}

controllers/keda/util/predicate.go

@@ -9,6 +9,8 @@ import (
 
 const PausedReplicasAnnotation = "autoscaling.keda.sh/paused-replicas"
 
+const Paused = "autoscaling.keda.sh/paused"
+
 type PausedReplicasPredicate struct {
  predicate.Funcs
 }
@@ -61,3 +63,23 @@ func (ScaleObjectReadyConditionPredicate) Update(e event.UpdateEvent) bool {
 
  return false
 }
+type PausedPredicate struct {
+ predicate.Funcs
+}
+func (PausedPredicate) 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[Paused]; ok1 {
+ if oldVal, ok2 := oldAnnotations[Paused]; ok2 {
+ return newVal != oldVal
+ }
+ return true
+ }
+ }
+ return false
+}

pkg/scaling/executor/scale_jobs.go

@@ -428,3 +428,10 @@ func min(x, y int64) int64 {
  }
  return x
 }
+//check if scaledJob has Paused annotation, if it does, pause the scaledjob
+func (e *scaleExecutor) Pause(ctx context.Context, logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob) error {
+ if scaledJob.Annotations["scaledjob.keda.sh/paused"] == "true" {
+ return e.client.Status().Update(ctx, scaledJob)
+ }
+ return nil
+}