ScaledJob: introduce rolloutStrategy

CHANGELOG.md

@@ -23,6 +23,7 @@
 ### New
 
 - ScaledJob: introduce MultipleScalersCalculation ([#2016](https://github.com/kedacore/keda/pull/2016))
+- ScaledJob: introduce `RolloutStrategy` ([#2164](https://github.com/kedacore/keda/pull/2164))
 - Add Graphite Scaler ([#1628](https://github.com/kedacore/keda/pull/2092))
 - Add Cassandra Scaler ([#2211](https://github.com/kedacore/keda/pull/2211))
 

apis/keda/v1alpha1/scaledjob_types.go

@@ -51,6 +51,8 @@ type ScaledJobSpec struct {
  // +optional
  FailedJobsHistoryLimit *int32 `json:"failedJobsHistoryLimit,omitempty"`
  // +optional
+ RolloutStrategy string `json:"rolloutStrategy,omitempty"`
+ // +optional
  EnvSourceContainerName string `json:"envSourceContainerName,omitempty"`
  // +optional
  MaxReplicaCount *int32 `json:"maxReplicaCount,omitempty"`

config/crd/bases/keda.sh_scaledjobs.yaml

@@ -4,7 +4,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
  annotations:
- controller-gen.kubebuilder.io/version: v0.4.1
+ controller-gen.kubebuilder.io/version: v0.6.1
  creationTimestamp: null
  name: scaledjobs.keda.sh
 spec:
@@ -7354,6 +7354,8 @@ spec:
  pollingInterval:
  format: int32
  type: integer
+ rolloutStrategy:
+ type: string
  scalingStrategy:
  description: ScalingStrategy defines the strategy of Scaling
  properties:

config/crd/bases/keda.sh_triggerauthentications.yaml

@@ -4,7 +4,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
  annotations:
- controller-gen.kubebuilder.io/version: v0.4.1
+ controller-gen.kubebuilder.io/version: v0.6.1
  creationTimestamp: null
  name: triggerauthentications.keda.sh
 spec:
@@ -89,6 +89,8 @@ spec:
  type: object
  mount:
  type: string
+ namespace:
+ type: string
  role:
  type: string
  secrets:

controllers/keda/scaledjob_controller.go

@@ -134,7 +134,7 @@ func (r *ScaledJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 
 // reconcileScaledJob implements reconciler logic for K8s Jobs based ScaledJob
 func (r *ScaledJobReconciler) reconcileScaledJob(ctx context.Context, logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob) (string, error) {
- msg, err := r.deletePreviousVersionScaleJobs(logger, scaledJob)
+ msg, err := r.deletePreviousVersionScaleJobs(ctx, logger, scaledJob)
  if err != nil {
  return msg, err
  }
@@ -155,30 +155,35 @@ func (r *ScaledJobReconciler) reconcileScaledJob(ctx context.Context, logger log
  return "ScaledJob is defined correctly and is ready to scaling", nil
 }
 
-// Delete Jobs owned by the previous version of the scaledJob
-func (r *ScaledJobReconciler) deletePreviousVersionScaleJobs(logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob) (string, error) {
- opts := []client.ListOption{
- client.InNamespace(scaledJob.GetNamespace()),
- client.MatchingLabels(map[string]string{"scaledjob.keda.sh/name": scaledJob.GetName()}),
- }
- jobs := &batchv1.JobList{}
- err := r.Client.List(context.TODO(), jobs, opts...)
- if err != nil {
- return "Cannot get list of Jobs owned by this scaledJob", err
- }
-
- if jobs.Size() > 0 {
- logger.Info("Deleting jobs owned by the previous version of the scaledJob", "Number of jobs to delete", jobs.Size())
- }
- for _, job := range jobs.Items {
- job := job
- err = r.Client.Delete(context.TODO(), &job, client.PropagationPolicy(metav1.DeletePropagationBackground))
+// Delete Jobs owned by the previous version of the scaledJob based on the rolloutStrategy given for this scaledJob, if any
+func (r *ScaledJobReconciler) deletePreviousVersionScaleJobs(ctx context.Context, logger logr.Logger, scaledJob *kedav1alpha1.ScaledJob) (string, error) {
+ switch scaledJob.Spec.RolloutStrategy {
+ case "gradual":
+ logger.Info("RolloutStrategy: gradual, Not deleting jobs owned by the previous version of the scaleJob")
+ default:
+ opts := []client.ListOption{
+ client.InNamespace(scaledJob.GetNamespace()),
+ client.MatchingLabels(map[string]string{"scaledjob.keda.sh/name": scaledJob.GetName()}),
+ }
+ jobs := &batchv1.JobList{}
+ err := r.Client.List(ctx, jobs, opts...)
  if err != nil {
- return "Not able to delete job: " + job.Name, err
+ return "Cannot get list of Jobs owned by this scaledJob", err
  }
- }
 
- return fmt.Sprintf("Deleted jobs owned by the previous version of the scaleJob: %d jobs deleted", jobs.Size()), nil
+ if len(jobs.Items) > 0 {
+ logger.Info("RolloutStrategy: immediate, Deleting jobs owned by the previous version of the scaledJob", "numJobsToDelete", len(jobs.Items))
+ }
+ for _, job := range jobs.Items {
+ job := job
+ err = r.Client.Delete(ctx, &job, client.PropagationPolicy(metav1.DeletePropagationBackground))
+ if err != nil {
+ return "Not able to delete job: " + job.Name, err
+ }
+ }
+ return fmt.Sprintf("RolloutStrategy: immediate, deleted jobs owned by the previous version of the scaleJob: %d jobs deleted", len(jobs.Items)), nil
+ }
+ return fmt.Sprintf("RolloutStrategy: %s", scaledJob.Spec.RolloutStrategy), nil
 }
 
 // requestScaleLoop request ScaleLoop handler for the respective ScaledJob