diff --git a/apis/rollout/v1alpha1/validation/rolloutrun.go b/apis/rollout/v1alpha1/validation/rolloutrun.go index 95cebf5..624759b 100644 --- a/apis/rollout/v1alpha1/validation/rolloutrun.go +++ b/apis/rollout/v1alpha1/validation/rolloutrun.go @@ -108,6 +108,7 @@ func validateRolloutRunStepTargets(targets []rolloutv1alpha1.RolloutRunStepTarge allErrs = append(allErrs, field.Duplicate(fldPath.Index(i).Child("name"), target.CrossClusterObjectNameReference)) } targetMap[target.CrossClusterObjectNameReference] = true + allErrs = append(allErrs, validateToleration(target.Toleration, fldPath.Index(i).Child("toleration"))...) } return allErrs diff --git a/apis/rollout/v1alpha1/validation/rolloutstrategy.go b/apis/rollout/v1alpha1/validation/rolloutstrategy.go index e2187b4..f263098 100644 --- a/apis/rollout/v1alpha1/validation/rolloutstrategy.go +++ b/apis/rollout/v1alpha1/validation/rolloutstrategy.go @@ -188,6 +188,24 @@ func ValidateRolloutStrategyTargets(targets *rolloutv1alpha1.RolloutStrategyTarg allErrs = append(allErrs, appsvalidation.ValidatePositiveIntOrPercent(targets.Replicas, fldPath.Child("replicas"))...) allErrs = append(allErrs, ValidateResourceMatch(targets.Match, fldPath.Child("matchTargets"))...) + allErrs = append(allErrs, validateToleration(targets.Toleration, fldPath.Child("toleration"))...) + + return allErrs +} + +func validateToleration(toleration *rolloutv1alpha1.RolloutStepTargetToleration, fldPath *field.Path) field.ErrorList { + if toleration == nil { + return nil + } + + allErrs := field.ErrorList{} + + if toleration.FailureThreshold == nil { + allErrs = append(allErrs, field.Required(fldPath.Child("failureThreshold"), "must be set when toleration is configured")) + } + if toleration.InitialDelaySeconds == nil { + allErrs = append(allErrs, field.Required(fldPath.Child("initialDelaySeconds"), "must be set when toleration is configured")) + } return allErrs } diff --git a/apis/rollout/v1alpha1/validation/rolloutstrategy_test.go b/apis/rollout/v1alpha1/validation/rolloutstrategy_test.go index d24e802..9ecaf57 100644 --- a/apis/rollout/v1alpha1/validation/rolloutstrategy_test.go +++ b/apis/rollout/v1alpha1/validation/rolloutstrategy_test.go @@ -311,6 +311,52 @@ func TestValidateRolloutStrategy_V2(t *testing.T) { }(), wantErr: false, }, + { + name: "toleration without failureThreshold", + obj: func() *rolloutv1alpha1.RolloutStrategy { + obj := validV2Strategy.DeepCopy() + obj.BatchV2.Batches[0].Targets[0].Toleration = &rolloutv1alpha1.RolloutStepTargetToleration{ + InitialDelaySeconds: ptr.To[int32](300), + } + return obj + }(), + wantErr: true, + errLen: 1, + }, + { + name: "toleration without initialDelaySeconds", + obj: func() *rolloutv1alpha1.RolloutStrategy { + obj := validV2Strategy.DeepCopy() + obj.BatchV2.Batches[0].Targets[0].Toleration = &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](2), + } + return obj + }(), + wantErr: true, + errLen: 1, + }, + { + name: "toleration with both fields set is valid", + obj: func() *rolloutv1alpha1.RolloutStrategy { + obj := validV2Strategy.DeepCopy() + obj.BatchV2.Batches[0].Targets[0].Toleration = &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](2), + InitialDelaySeconds: ptr.To[int32](300), + } + return obj + }(), + wantErr: false, + }, + { + name: "toleration with both fields nil", + obj: func() *rolloutv1alpha1.RolloutStrategy { + obj := validV2Strategy.DeepCopy() + obj.BatchV2.Batches[0].Targets[0].Toleration = &rolloutv1alpha1.RolloutStepTargetToleration{} + return obj + }(), + wantErr: true, + errLen: 2, + }, } for i := range tests { diff --git a/config/crd/bases/rollout.kusionstack.io_rolloutruns.yaml b/config/crd/bases/rollout.kusionstack.io_rolloutruns.yaml index f6ddbe0..687323d 100644 --- a/config/crd/bases/rollout.kusionstack.io_rolloutruns.yaml +++ b/config/crd/bases/rollout.kusionstack.io_rolloutruns.yaml @@ -2264,32 +2264,30 @@ spec: - targets type: object type: array - toleration: - description: Toleration is the toleration policy of the canary strategy - properties: - initialDelaySeconds: - description: Number of seconds after the toleration check has started before the task are initiated. - format: int32 - type: integer - taskFailureThreshold: - anyOf: - - type: integer - - type: string - description: |- - FailureThreshold indicates how many failed pods can be tolerated before marking the rollout task as success - If not set, the default value is 0, which means no failed pods can be tolerated - This is a task level threshold. - x-kubernetes-int-or-string: true - workloadTotalFailureThreshold: - anyOf: - - type: integer - - type: string - description: |- - WorkloadFailureThreshold indicates how many failed pods can be tolerated in all upgraded pods of one workload. - The default value is 0, which means no failed pods can be tolerated. - This is a workload level threshold. - x-kubernetes-int-or-string: true - type: object + tolerations: + description: |- + Tolerations records the accumulated toleration from skipped batches per workload. + When a batch is skipped, the gap between expected and actual replicas for each workload + is accumulated into this field, allowing subsequent batches to tolerate the deficit. + items: + description: RolloutRunTolerationTarget records the toleration value accumulated from skipped batches for a specific workload. + properties: + cluster: + description: Cluster defines which cluster the workload is in. + type: string + name: + description: Name is the workload name. + type: string + toleration: + description: |- + Toleration is the accumulated toleration value from skipped batches. + It represents how many replicas the workload is allowed to be short of. + format: int32 + type: integer + required: + - toleration + type: object + type: array type: object canary: description: Canary defines the canary strategy diff --git a/go.mod b/go.mod index 9d5fce4..4813499 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( k8s.io/klog/v2 v2.130.1 k8s.io/kubernetes v1.22.2 k8s.io/utils v0.0.0-20241210054802-24370beab758 - kusionstack.io/kube-api v0.7.5-0.20260512114711-9570d38337c2 + kusionstack.io/kube-api v0.7.5-0.20260901080654-b15c6deabdf8 kusionstack.io/kube-utils v0.2.1-0.20251125083928-1134a582b341 kusionstack.io/resourceconsist v0.0.4 sigs.k8s.io/controller-runtime v0.21.0 diff --git a/go.sum b/go.sum index d2f1479..6f9ba8e 100644 --- a/go.sum +++ b/go.sum @@ -1021,8 +1021,10 @@ k8s.io/sample-apiserver v0.22.2/go.mod h1:h+/DIV5EmuNq4vfPr5TSXy9mIBVXXlPAKQMPbj k8s.io/system-validators v1.5.0/go.mod h1:bPldcLgkIUK22ALflnsXk8pvkTEndYdNuaHH6gRrl0Q= k8s.io/utils v0.0.0-20240102154912-e7106e64919e h1:eQ/4ljkx21sObifjzXwlPKpdGLrCfRziVtos3ofG/sQ= k8s.io/utils v0.0.0-20240102154912-e7106e64919e/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= -kusionstack.io/kube-api v0.7.5-0.20260512114711-9570d38337c2 h1:jrUVO6a6/fuwdfF8NnehVXGbiYV57AfIJvqnsYRB3sI= -kusionstack.io/kube-api v0.7.5-0.20260512114711-9570d38337c2/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk= +kusionstack.io/kube-api v0.7.5-0.20260629032255-be28009453cb h1:7lrtNodnXf7rsm3NzVhGZ4TwrM2ekJ656FN0qT9fHaI= +kusionstack.io/kube-api v0.7.5-0.20260629032255-be28009453cb/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk= +kusionstack.io/kube-api v0.7.5-0.20260901080654-b15c6deabdf8 h1:4deV4LV41/ecKS9KgUYRnQgLeDBacy9lRjBkxW2+vzo= +kusionstack.io/kube-api v0.7.5-0.20260901080654-b15c6deabdf8/go.mod h1:e1jtrQH2LK5fD2nTyfIXG6nYrYbU8VXShRxTRwVPaLk= kusionstack.io/kube-utils v0.2.1-0.20251125083928-1134a582b341 h1:dnMtHJvIpU3338WpqGiNN2qXWZFiXaoiuzR9jwhvWpg= kusionstack.io/kube-utils v0.2.1-0.20251125083928-1134a582b341/go.mod h1:Lz5SBYWg9+jw+kP0CAyf/b62D5DeUPf6+jE1d0WC4cI= kusionstack.io/resourceconsist v0.0.4 h1:wRqLJuNh8O4TT6p0uOklFpHUKiRdRxcAH71Sw/q9LhE= diff --git a/pkg/controllers/rollout/rollout_controller.go b/pkg/controllers/rollout/rollout_controller.go index fd599a6..066e70e 100644 --- a/pkg/controllers/rollout/rollout_controller.go +++ b/pkg/controllers/rollout/rollout_controller.go @@ -677,7 +677,6 @@ func (r *RolloutReconciler) applyOneTimeStrategy(ctx context.Context, obj *rollo var batch *rolloutv1alpha1.RolloutRunBatchStrategy // Check if using BatchV2 (V2 strategy scenario) - // Note: BatchStrategyV2 does not support Toleration, only V1 BatchStrategy has it if strategy.BatchV2 != nil { batch = &rolloutv1alpha1.RolloutRunBatchStrategy{ Batches: constructRolloutRunBatchesV2(strategy.BatchV2, workloads), @@ -685,13 +684,13 @@ func (r *RolloutReconciler) applyOneTimeStrategy(ctx context.Context, obj *rollo } else { // Use original Batch field (V1 scenario) batch = &rolloutv1alpha1.RolloutRunBatchStrategy{ - Batches: constructRolloutRunBatches(&strategy.Batch, workloads), - Toleration: strategy.Batch.Toleration, + Batches: constructRolloutRunBatches(&strategy.Batch, workloads), } } - if batch.Toleration == nil && run.Spec.Batch != nil { - batch.Toleration = run.Spec.Batch.Toleration + // Inherit Tolerations from the existing run spec if not changed + if run.Spec.Batch != nil { + batch.Tolerations = run.Spec.Batch.Tolerations } if equality.Semantic.DeepEqual(batch, run.Spec.Batch) { diff --git a/pkg/controllers/rollout/utils.go b/pkg/controllers/rollout/utils.go index eb85881..819e9ab 100644 --- a/pkg/controllers/rollout/utils.go +++ b/pkg/controllers/rollout/utils.go @@ -92,7 +92,6 @@ func constructRolloutRun(obj *rolloutv1alpha1.Rollout, strategy *rolloutv1alpha1 // Determine V1 vs V2 path based on BatchV2 presence if strategy.BatchV2 != nil { // V2 path: BatchV2 + optional CanaryV2 - // Note: BatchStrategyV2 does not support Toleration, only V1 BatchStrategy has it if strategy.CanaryV2 != nil { run.Spec.Canary = constructRolloutRunCanaryV2(strategy.CanaryV2, workloadWrappers) } @@ -105,8 +104,7 @@ func constructRolloutRun(obj *rolloutv1alpha1.Rollout, strategy *rolloutv1alpha1 run.Spec.Canary = constructRolloutRunCanary(strategy.Canary, workloadWrappers) } run.Spec.Batch = &rolloutv1alpha1.RolloutRunBatchStrategy{ - Toleration: strategy.Batch.Toleration, - Batches: constructRolloutRunBatches(strategy.Batch, workloadWrappers), + Batches: constructRolloutRunBatches(strategy.Batch, workloadWrappers), } } @@ -238,6 +236,7 @@ func resolveRolloutTargets(targets []rolloutv1alpha1.RolloutStrategyTargets, wor }, Replicas: t.Replicas, ReplicaSlidingWindow: t.ReplicaSlidingWindow, + Toleration: t.Toleration, } result = append(result, target) } diff --git a/pkg/controllers/rolloutrun/executor/batch.go b/pkg/controllers/rolloutrun/executor/batch.go index 680e96c..a777202 100644 --- a/pkg/controllers/rolloutrun/executor/batch.go +++ b/pkg/controllers/rolloutrun/executor/batch.go @@ -205,6 +205,8 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat batchTargetStatuses := make([]rolloutv1alpha1.RolloutWorkloadStatus, 0) allWorkloadReady := true + allWorkloadsAutoSkippable := true + for _, item := range currentBatch.Targets { info := ctx.Workloads.Get(item.Cluster, item.Name) if info == nil { @@ -227,6 +229,11 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat allWorkloadReady = false logger.V(3).Info("still waiting for target to be ready", "target", item.CrossClusterObjectNameReference, "reason", reason) + // Check auto-skip toleration for this workload + if !e.canAutoSkipTarget(item, info, currentBatchExpectedReplicas, isLastBatch, newStatus) { + allWorkloadsAutoSkippable = false + } + expectedReplicas, err := e.calculateExpectedReplicasBySlidingWindow(status, currentBatchExpectedReplicas, item.ReplicaSlidingWindow) if err != nil { return false, retryStop, err @@ -250,10 +257,64 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat return true, retryImmediately, nil } + if allWorkloadsAutoSkippable { + logger.Info("auto-skipping batch due to toleration") + newStatus.BatchStatus.Records[currentBatchIndex].State = StepSkipped + recordRolloutRunTolerations(&newStatus.BatchStatus.Tolerations, rolloutRun.Spec.Batch.Batches, ctx.Workloads, currentBatchIndex) + return true, retryImmediately, nil + } + // wait for next reconcile return false, retryDefault, nil } +// canAutoSkipTarget checks if the workload target meets the auto-skip toleration conditions. +// Returns true only when the workload is not ready due to a real deficit (gap > 0) within +// the toleration threshold and the initial delay has elapsed. +// Transient states (Generation mismatch, terminating replicas, last-batch overscaling) +// are NOT auto-skippable because the gap is unreliable until the workload stabilizes. +func (e *batchExecutor) canAutoSkipTarget(item rolloutv1alpha1.RolloutRunStepTarget, info *workload.Info, currentBatchExpectedReplicas int32, isLastBatch bool, newStatus *rolloutv1alpha1.RolloutRunStatus) bool { + if item.Toleration == nil || item.Toleration.FailureThreshold == nil { + return false + } + + // Not skippable while workload has not been reconciled yet (Generation mismatch). + // UpdatedAvailableReplicas may be stale from the previous generation. + if info.Generation != info.Status.ObservedGeneration { + return false + } + + // On last batch, not skippable if observed replicas exceed desired or terminating replicas exist(strict check). + if isLastBatch && info.Status.ObservedReplicas > info.Status.DesiredReplicas || info.Status.TerminatingReplicas != 0 { + return false + } + + // Only evaluate toleration on a real deficit. + gap := currentBatchExpectedReplicas - info.Status.UpdatedAvailableReplicas + if gap <= 0 { + return false + } + + if gap > *item.Toleration.FailureThreshold { + return false + } + + // gap is within threshold, check timeout + if item.Toleration.InitialDelaySeconds != nil { + currentBatchIndex := newStatus.BatchStatus.CurrentBatchIndex + startTime := newStatus.BatchStatus.Records[currentBatchIndex].StartTime + if startTime == nil { + return false + } + elapsed := time.Since(startTime.Time) + if elapsed < time.Duration(*item.Toleration.InitialDelaySeconds)*time.Second { + return false + } + } + + return true +} + // calculateExpectedReplicasBySlidingWindow calculate expected replicas by sliding window // if window is nil, return currentBatchExpectedReplicas // if window is not nil, return min(currentBatchExpectedReplicas, updatedAvailableReplicas + increment) diff --git a/pkg/controllers/rolloutrun/executor/batch_test.go b/pkg/controllers/rolloutrun/executor/batch_test.go index 69ca2b6..6158f56 100644 --- a/pkg/controllers/rolloutrun/executor/batch_test.go +++ b/pkg/controllers/rolloutrun/executor/batch_test.go @@ -318,6 +318,216 @@ func (s *batchExecutorTestSuite) Test_BatchExecutor_Do() { s.runBatchTestCases(tests) } +func (s *batchExecutorTestSuite) Test_BatchExecutor_Do_SkipToleration() { + tests := []batchExectorTestCase{ + { + name: "auto-skip applies in middle batch when gap within threshold and delay elapsed", + getObjects: func() (*rolloutv1alpha1.Rollout, *rolloutv1alpha1.RolloutRun) { + rollout := s.rollout.DeepCopy() + rolloutRun := s.rolloutRun.DeepCopy() + + // 3 batches, currently on batch 2 (index 1), not the last batch + rolloutRun.Spec.Batch.Batches = []rolloutv1alpha1.RolloutRunStep{ + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(30)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTargetWithToleration("cluster-a", "test-a", intstr.FromInt(60), &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(100)), + }}, + } + rolloutRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + rolloutRun.Status.BatchStatus = &rolloutv1alpha1.RolloutRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 1, + CurrentBatchState: StepRunning, + }, + Records: []rolloutv1alpha1.RolloutRunStepStatus{ + {Index: ptr.To[int32](0), State: StepSkipped}, + {Index: ptr.To[int32](1), State: StepRunning, StartTime: ptr.To(metav1.Now())}, + {Index: ptr.To[int32](2), State: StepNone}, + }, + } + return rollout, rolloutRun + }, + getWorkloads: func() []client.Object { + // UpdatedAvailableReplicas = 55, expected = 60, gap = 5 <= FailureThreshold(5) + // InitialDelaySeconds = 0 means elapsed(>=0s) >= 0s -> auto-skip + return []client.Object{ + newFakeObject("cluster-a", "default", "test-a", 100, 55, 55), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) // not all done, move to next batch + s.Equal(reconcile.Result{Requeue: true}, result) + }, + assertStatus: func(status *rolloutv1alpha1.RolloutRunStatus) { + s.Equal(StepPostBatchStepHook, status.BatchStatus.CurrentBatchState) + // StepSkipped set in doBatchUpgrading is overwritten by state engine's MoveToNextState; + // the durable observable for auto-skip is the Tolerations field below. + // Tolerations should record gap = 5 for the skipped workload + s.Len(status.BatchStatus.Tolerations, 1) + s.Equal(int32(5), status.BatchStatus.Tolerations[0].Toleration) + }, + }, + { + name: "auto-skip does not apply when gap exceeds threshold, batch stays running", + getObjects: func() (*rolloutv1alpha1.Rollout, *rolloutv1alpha1.RolloutRun) { + rollout := s.rollout.DeepCopy() + rolloutRun := s.rolloutRun.DeepCopy() + + rolloutRun.Spec.Batch.Batches = []rolloutv1alpha1.RolloutRunStep{ + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(30)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTargetWithToleration("cluster-a", "test-a", intstr.FromInt(60), &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(100)), + }}, + } + rolloutRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + rolloutRun.Status.BatchStatus = &rolloutv1alpha1.RolloutRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 1, + CurrentBatchState: StepRunning, + }, + Records: []rolloutv1alpha1.RolloutRunStepStatus{ + {Index: ptr.To[int32](0), State: StepSkipped}, + {Index: ptr.To[int32](1), State: StepRunning, StartTime: ptr.To(metav1.Now())}, + {Index: ptr.To[int32](2), State: StepNone}, + }, + } + return rollout, rolloutRun + }, + getWorkloads: func() []client.Object { + // gap = 60 - 52 = 8 > FailureThreshold(5) -> not auto-skippable, keep waiting + return []client.Object{ + newFakeObject("cluster-a", "default", "test-a", 100, 52, 52), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.RolloutRunStatus) { + s.Equal(StepRunning, status.BatchStatus.CurrentBatchState) + s.Empty(status.BatchStatus.Tolerations) + }, + }, + { + name: "auto-skip on last batch transitions toward success", + getObjects: func() (*rolloutv1alpha1.Rollout, *rolloutv1alpha1.RolloutRun) { + rollout := s.rollout.DeepCopy() + rolloutRun := s.rolloutRun.DeepCopy() + + rolloutRun.Spec.Batch.Batches = []rolloutv1alpha1.RolloutRunStep{ + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(30)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(60)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTargetWithToleration("cluster-a", "test-a", intstr.FromInt(100), &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + } + rolloutRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + rolloutRun.Status.BatchStatus = &rolloutv1alpha1.RolloutRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 2, + CurrentBatchState: StepRunning, + }, + Records: []rolloutv1alpha1.RolloutRunStepStatus{ + {Index: ptr.To[int32](0), State: StepSkipped}, + {Index: ptr.To[int32](1), State: StepSucceeded}, + {Index: ptr.To[int32](2), State: StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + return rollout, rolloutRun + }, + getWorkloads: func() []client.Object { + // Last batch: gap = 100 - 96 = 4 <= FailureThreshold(5) -> auto-skippable on last batch too + return []client.Object{ + newFakeObject("cluster-a", "default", "test-a", 100, 96, 96), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) // still need to go through PostBatchStepHook and Recycle + s.Equal(reconcile.Result{Requeue: true}, result) + }, + assertStatus: func(status *rolloutv1alpha1.RolloutRunStatus) { + s.Equal(StepPostBatchStepHook, status.BatchStatus.CurrentBatchState) + s.Len(status.BatchStatus.Tolerations, 1) + s.Equal(int32(4), status.BatchStatus.Tolerations[0].Toleration) + }, + }, + { + name: "no skip toleration, behavior unchanged", + getObjects: func() (*rolloutv1alpha1.Rollout, *rolloutv1alpha1.RolloutRun) { + rollout := s.rollout.DeepCopy() + rolloutRun := s.rolloutRun.DeepCopy() + + rolloutRun.Spec.Batch.Batches = []rolloutv1alpha1.RolloutRunStep{ + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(10)), + }}, + } + rolloutRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + rolloutRun.Status.BatchStatus = &rolloutv1alpha1.RolloutRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: StepRunning, + }, + Records: []rolloutv1alpha1.RolloutRunStepStatus{ + {Index: ptr.To[int32](0), State: StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + // No skip tolerations + return rollout, rolloutRun + }, + getWorkloads: func() []client.Object { + // UpdatedAvailableReplicas=8 < expected=10, no toleration -> not ready + return []client.Object{ + newFakeObject("cluster-a", "default", "test-a", 100, 8, 8), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.RolloutRunStatus) { + s.Equal(StepRunning, status.BatchStatus.CurrentBatchState) + s.Empty(status.BatchStatus.Tolerations) + }, + }, + } + + s.runBatchTestCases(tests) +} + +func newRunStepTargetWithToleration(cluster, name string, replicas intstr.IntOrString, toleration *rolloutv1alpha1.RolloutStepTargetToleration) rolloutv1alpha1.RolloutRunStepTarget { + target := newRunStepTarget(cluster, name, replicas) + target.Toleration = toleration + return target +} + func newRunStepTarget(cluster, name string, replicas intstr.IntOrString) rolloutv1alpha1.RolloutRunStepTarget { return newRunStepTargetWithSlidingWindow(cluster, name, replicas, nil) } diff --git a/pkg/controllers/rolloutrun/executor/can_auto_skip_target_test.go b/pkg/controllers/rolloutrun/executor/can_auto_skip_target_test.go new file mode 100644 index 0000000..8b7ebe8 --- /dev/null +++ b/pkg/controllers/rolloutrun/executor/can_auto_skip_target_test.go @@ -0,0 +1,259 @@ +package executor + +import ( + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + + "kusionstack.io/rollout/pkg/workload" +) + +// TestCanAutoSkipTargetRolloutRun directly exercises batchExecutor.canAutoSkipTarget +// to cover the transient-state guards (problem #4) that are difficult to trigger +// through the full Reconcile loop: +// - Generation mismatch +// - TerminatingReplicas != 0 +// - Last-batch strict check (ObservedReplicas > DesiredReplicas) +// - gap <= 0 while not ready +// - Gap > FailureThreshold +// - InitialDelay not yet elapsed +// - nil Toleration / nil FailureThreshold +// - Healthy deficit with delay elapsed -> skippable (auto-skip scenario 1) +func TestCanAutoSkipTargetRolloutRun(t *testing.T) { + e := &batchExecutor{} + + now := ptr.To(metav1.Now()) + pastTime := ptr.To(metav1.Time{Time: time.Now().Add(-10 * time.Minute)}) + + type args struct { + item rolloutv1alpha1.RolloutRunStepTarget + info *workload.Info + currentBatchExpectedReplicas int32 + isLastBatch bool + newStatus *rolloutv1alpha1.RolloutRunStatus + } + + tests := []struct { + name string + args args + want bool + }{ + { + name: "nil Toleration returns false", + args: args{ + item: rolloutv1alpha1.RolloutRunStepTarget{Toleration: nil}, + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(5).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "nil FailureThreshold returns false", + args: args{ + item: rolloutv1alpha1.RolloutRunStepTarget{ + Toleration: &rolloutv1alpha1.RolloutStepTargetToleration{InitialDelaySeconds: ptr.To[int32](0)}, + }, + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(5).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "Generation mismatch returns false (transient state, gap unreliable)", + args: args{ + item: tolerationTarget(2, ptr.To[int32](0)), + info: newInfoBuilder().generation(2).observedGen(1).updatedAvailable(5).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "TerminatingReplicas != 0 returns false (non-last batch, still converging)", + args: args{ + item: tolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(5).desired(10).terminating(3).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "Last batch strict check: ObservedReplicas > DesiredReplicas returns false", + args: args{ + item: tolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(5).desired(10).observed(11).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: true, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "gap <= 0 with healthy replicas returns false (already satisfied, no deficit to tolerate)", + args: args{ + item: tolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(10).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "gap > FailureThreshold returns false (scenario 2: keep waiting)", + args: args{ + item: tolerationTarget(2, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(7).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "gap == FailureThreshold and InitialDelay not yet elapsed returns false", + args: args{ + item: tolerationTarget(5, ptr.To[int32](300)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(5).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), // started now, 300s not elapsed + }, + want: false, + }, + { + name: "gap == FailureThreshold and InitialDelay elapsed returns true (scenario 1: auto-skip middle batch)", + args: args{ + item: tolerationTarget(5, ptr.To[int32](300)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(5).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, pastTime), // started 10 min ago, 300s elapsed + }, + want: true, + }, + { + name: "gap within threshold on last batch, no strict check violation, delay elapsed -> auto-skip last batch", + args: args{ + item: tolerationTarget(5, ptr.To[int32](300)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(6).desired(10).observed(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: true, + newStatus: newRolloutRunStatusWithStart(0, pastTime), + }, + want: true, + }, + { + name: "last batch with TerminatingReplicas != 0 returns false (strict check blocks auto-skip)", + args: args{ + item: tolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(6).desired(10).observed(10).terminating(1).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: true, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "InitialDelaySeconds nil treats delay as already elapsed", + args: args{ + item: tolerationTarget(5, nil), + info: newInfoBuilder().generation(1).observedGen(1).updatedAvailable(5).desired(10).build(), + currentBatchExpectedReplicas: 10, + isLastBatch: false, + newStatus: newRolloutRunStatusWithStart(0, now), + }, + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := e.canAutoSkipTarget(tt.args.item, tt.args.info, tt.args.currentBatchExpectedReplicas, tt.args.isLastBatch, tt.args.newStatus) + if got != tt.want { + t.Errorf("canAutoSkipTarget() = %v, want %v", got, tt.want) + } + }) + } +} + +// tolerationTarget builds a RolloutRunStepTarget with a Toleration. +// failureThreshold is required; initialDelay may be nil to indicate "no delay". +func tolerationTarget(failureThreshold int32, initialDelay *int32) rolloutv1alpha1.RolloutRunStepTarget { + tol := &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](failureThreshold), + } + if initialDelay != nil { + tol.InitialDelaySeconds = ptr.To[int32](*initialDelay) + } + return rolloutv1alpha1.RolloutRunStepTarget{Toleration: tol} +} + +func newRolloutRunStatusWithStart(currentBatchIndex int32, startTime *metav1.Time) *rolloutv1alpha1.RolloutRunStatus { + records := []rolloutv1alpha1.RolloutRunStepStatus{ + {Index: ptr.To[int32](0), State: StepRunning, StartTime: startTime}, + {Index: ptr.To[int32](1), State: StepNone}, + {Index: ptr.To[int32](2), State: StepNone}, + } + return &rolloutv1alpha1.RolloutRunStatus{ + BatchStatus: &rolloutv1alpha1.RolloutRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: currentBatchIndex, + }, + Records: records, + }, + } +} + +// infoBuilder is a fluent builder for workload.Info to keep test cases concise. +type infoBuilder struct { + info *workload.Info +} + +func newInfoBuilder() *infoBuilder { + return &infoBuilder{info: &workload.Info{}} +} + +func (b *infoBuilder) generation(g int64) *infoBuilder { + b.info.Generation = g + return b +} + +func (b *infoBuilder) observedGen(g int64) *infoBuilder { + b.info.Status.ObservedGeneration = g + return b +} + +func (b *infoBuilder) updatedAvailable(n int32) *infoBuilder { + b.info.Status.UpdatedAvailableReplicas = n + return b +} + +func (b *infoBuilder) desired(n int32) *infoBuilder { + b.info.Status.DesiredReplicas = n + return b +} + +func (b *infoBuilder) observed(n int32) *infoBuilder { + b.info.Status.ObservedReplicas = n + return b +} + +func (b *infoBuilder) terminating(n int32) *infoBuilder { + b.info.Status.TerminatingReplicas = n + return b +} + +func (b *infoBuilder) build() *workload.Info { + return b.info +} diff --git a/pkg/controllers/rolloutrun/executor/default_test.go b/pkg/controllers/rolloutrun/executor/default_test.go index e765250..a7aa872 100644 --- a/pkg/controllers/rolloutrun/executor/default_test.go +++ b/pkg/controllers/rolloutrun/executor/default_test.go @@ -50,8 +50,7 @@ var ( }, Webhooks: []rolloutv1alpha1.RolloutWebhook{}, Batch: &rolloutv1alpha1.RolloutRunBatchStrategy{ - Toleration: &rolloutv1alpha1.TolerationStrategy{}, - Batches: []rolloutv1alpha1.RolloutRunStep{}, + Batches: []rolloutv1alpha1.RolloutRunStep{}, }, }, Status: rolloutv1alpha1.RolloutRunStatus{ @@ -76,8 +75,7 @@ var ( Targets: []rolloutv1alpha1.RolloutRunStepTarget{}, }, Batch: &rolloutv1alpha1.RolloutRunBatchStrategy{ - Toleration: &rolloutv1alpha1.TolerationStrategy{}, - Batches: []rolloutv1alpha1.RolloutRunStep{}, + Batches: []rolloutv1alpha1.RolloutRunStep{}, }, }, Status: rolloutv1alpha1.RolloutRunStatus{ diff --git a/pkg/controllers/rolloutrun/executor/do_command.go b/pkg/controllers/rolloutrun/executor/do_command.go index 08a4123..d0ecc14 100644 --- a/pkg/controllers/rolloutrun/executor/do_command.go +++ b/pkg/controllers/rolloutrun/executor/do_command.go @@ -4,6 +4,8 @@ import ( rolloutapis "kusionstack.io/kube-api/rollout" rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" ctrl "sigs.k8s.io/controller-runtime" + + "kusionstack.io/rollout/pkg/workload" ) // doCommand @@ -28,27 +30,73 @@ func (r *Executor) doCommand(ctx *ExecutorContext) ctrl.Result { } case rolloutapis.AnnoManualCommandSkip: if batchError != nil { - handleBatchStatusWhenSkipped(newStatus, len(rolloutRun.Spec.Batch.Batches)) + handleBatchStatusWhenSkipped(newStatus, len(rolloutRun.Spec.Batch.Batches), rolloutRun.Spec.Batch.Batches, ctx.Workloads) } case rolloutapis.AnnoManualCommandCancel: newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseCanceling case rolloutapis.AnnoManualCommandForceSkipCurrentBatch: - handleBatchStatusWhenSkipped(newStatus, len(rolloutRun.Spec.Batch.Batches)) + handleBatchStatusWhenSkipped(newStatus, len(rolloutRun.Spec.Batch.Batches), rolloutRun.Spec.Batch.Batches, ctx.Workloads) } return ctrl.Result{Requeue: true} } -func handleBatchStatusWhenSkipped(newStatus *rolloutv1alpha1.RolloutRunStatus, batchSize int) { +func handleBatchStatusWhenSkipped(newStatus *rolloutv1alpha1.RolloutRunStatus, batchSize int, batches []rolloutv1alpha1.RolloutRunStep, workloads *workload.Set) { currentBatchIndex := newStatus.BatchStatus.CurrentBatchIndex if newStatus.Error != nil { newStatus.Error = nil } - // only skip when current batch is not the last batch - if int(currentBatchIndex) < (batchSize - 1) { - newStatus.BatchStatus.Records[currentBatchIndex].State = StepSkipped - newStatus.BatchStatus.CurrentBatchIndex = currentBatchIndex + 1 - newStatus.BatchStatus.CurrentBatchState = StepNone + newStatus.BatchStatus.Records[currentBatchIndex].State = StepSkipped + + // Record tolerations for each workload in the current batch + recordRolloutRunTolerations(&newStatus.BatchStatus.Tolerations, batches, workloads, currentBatchIndex) + + if int(currentBatchIndex) >= (batchSize - 1) { + // Last batch: advance to PostRollout phase (will transition to Succeeded) + newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePostRollout + return + } + + // Not the last batch: advance to the next batch + newStatus.BatchStatus.CurrentBatchIndex = currentBatchIndex + 1 + newStatus.BatchStatus.CurrentBatchState = StepNone +} + +// recordRolloutRunTolerations upserts tolerations for each workload in the current batch. +// For each target, it calculates the gap between expected and actual updated available replicas, +// and updates (or inserts) the toleration value into the tolerations slice. +func recordRolloutRunTolerations(tolerations *[]rolloutv1alpha1.RolloutRunTolerationTarget, batches []rolloutv1alpha1.RolloutRunStep, workloads *workload.Set, currentBatchIndex int32) { + if workloads == nil || int(currentBatchIndex) >= len(batches) { + return + } + currentBatch := batches[currentBatchIndex] + for _, target := range currentBatch.Targets { + info := workloads.Get(target.Cluster, target.Name) + if info == nil { + continue + } + status := info.APIStatus() + currentBatchExpectedReplicas, _ := workload.CalculateUpdatedReplicas(&status.Replicas, target.Replicas) + gap := currentBatchExpectedReplicas - status.UpdatedAvailableReplicas + if gap < 0 { + gap = 0 + } + upsertToleration(tolerations, target.CrossClusterObjectNameReference, gap) + } +} + +// upsertToleration updates the toleration value for the given workload reference, +// or appends a new entry if the workload is not yet present. +func upsertToleration(tolerations *[]rolloutv1alpha1.RolloutRunTolerationTarget, ref rolloutv1alpha1.CrossClusterObjectNameReference, gap int32) { + for i, t := range *tolerations { + if t.CrossClusterObjectNameReference == ref { + (*tolerations)[i].Toleration = gap + return + } } + *tolerations = append(*tolerations, rolloutv1alpha1.RolloutRunTolerationTarget{ + CrossClusterObjectNameReference: ref, + Toleration: gap, + }) } diff --git a/pkg/controllers/rolloutrun/executor/do_command_test.go b/pkg/controllers/rolloutrun/executor/do_command_test.go new file mode 100644 index 0000000..6e9419c --- /dev/null +++ b/pkg/controllers/rolloutrun/executor/do_command_test.go @@ -0,0 +1,144 @@ +package executor + +import ( + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + + "kusionstack.io/rollout/pkg/workload" +) + +func TestHandleBatchStatusWhenSkipped(t *testing.T) { + tests := []struct { + name string + batchIndex int32 + batchSize int + batches []rolloutv1alpha1.RolloutRunStep + workloads *workload.Set + expectedCurrentBatchIndex int32 + expectedCurrentBatchState rolloutv1alpha1.RolloutStepState + expectedPhase rolloutv1alpha1.RolloutRunPhase + expectedRecordState rolloutv1alpha1.RolloutStepState + expectedToleration *int32 + }{ + { + name: "skip advances to next batch", + batchIndex: 0, + batchSize: 3, + batches: []rolloutv1alpha1.RolloutRunStep{ + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(30)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(60)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(100)), + }}, + }, + workloads: newTestWorkloadSet("cluster-a", "test-a", 1, 100, 25), + expectedCurrentBatchIndex: 1, + expectedCurrentBatchState: rolloutv1alpha1.RolloutStepNone, + expectedRecordState: rolloutv1alpha1.RolloutStepSkipped, + expectedToleration: ptr.To[int32](5), // 30 - 25 = 5 + }, + { + name: "skip advances to next batch from middle", + batchIndex: 1, + batchSize: 3, + batches: []rolloutv1alpha1.RolloutRunStep{ + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(30)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(60)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(100)), + }}, + }, + workloads: newTestWorkloadSet("cluster-a", "test-a", 1, 100, 52), + expectedCurrentBatchIndex: 2, + expectedCurrentBatchState: rolloutv1alpha1.RolloutStepNone, + expectedRecordState: rolloutv1alpha1.RolloutStepSkipped, + expectedToleration: ptr.To[int32](8), // 60 - 52 = 8 + }, + { + name: "skip last batch transitions to PostRollout phase", + batchIndex: 2, + batchSize: 3, + batches: []rolloutv1alpha1.RolloutRunStep{ + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(30)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(60)), + }}, + {Targets: []rolloutv1alpha1.RolloutRunStepTarget{ + newRunStepTarget("cluster-a", "test-a", intstr.FromInt(100)), + }}, + }, + workloads: newTestWorkloadSet("cluster-a", "test-a", 1, 100, 93), + expectedCurrentBatchIndex: 2, // unchanged - last batch does not advance index + expectedCurrentBatchState: rolloutv1alpha1.RolloutStepNone, + expectedPhase: rolloutv1alpha1.RolloutRunPhasePostRollout, + expectedRecordState: rolloutv1alpha1.RolloutStepSkipped, + expectedToleration: ptr.To[int32](7), // 100 - 93 = 7 + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + newStatus := &rolloutv1alpha1.RolloutRunStatus{ + BatchStatus: &rolloutv1alpha1.RolloutRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: tt.batchIndex, + }, + Records: make([]rolloutv1alpha1.RolloutRunStepStatus, tt.batchSize), + }, + } + + handleBatchStatusWhenSkipped(newStatus, tt.batchSize, tt.batches, tt.workloads) + + if newStatus.BatchStatus.CurrentBatchIndex != tt.expectedCurrentBatchIndex { + t.Errorf("CurrentBatchIndex = %d, want %d", newStatus.BatchStatus.CurrentBatchIndex, tt.expectedCurrentBatchIndex) + } + if newStatus.BatchStatus.CurrentBatchState != tt.expectedCurrentBatchState { + t.Errorf("CurrentBatchState = %v, want %v", newStatus.BatchStatus.CurrentBatchState, tt.expectedCurrentBatchState) + } + if tt.expectedPhase != "" && newStatus.Phase != tt.expectedPhase { + t.Errorf("Phase = %v, want %v", newStatus.Phase, tt.expectedPhase) + } + if tt.expectedRecordState != "" && newStatus.BatchStatus.Records[tt.batchIndex].State != tt.expectedRecordState { + t.Errorf("Records[%d].State = %v, want %v", tt.batchIndex, newStatus.BatchStatus.Records[tt.batchIndex].State, tt.expectedRecordState) + } + if tt.expectedToleration != nil { + if len(newStatus.BatchStatus.Tolerations) != 1 { + t.Errorf("expected 1 toleration entry, got %d", len(newStatus.BatchStatus.Tolerations)) + } else if newStatus.BatchStatus.Tolerations[0].Toleration != *tt.expectedToleration { + t.Errorf("Tolerations[0].Toleration = %d, want %d", newStatus.BatchStatus.Tolerations[0].Toleration, *tt.expectedToleration) + } + } + }) + } +} + +// newTestWorkloadSet creates a workload.Set with a single workload for testing +func newTestWorkloadSet(cluster, name string, generation int64, desiredReplicas, updatedAvailableReplicas int32) *workload.Set { + return workload.NewSet(&workload.Info{ + ClusterName: cluster, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Generation: generation, + }, + Status: workload.InfoStatus{ + ObservedGeneration: generation, + DesiredReplicas: desiredReplicas, + UpdatedAvailableReplicas: updatedAvailableReplicas, + }, + }) +} diff --git a/pkg/controllers/rolloutrun/rolloutrun_controller.go b/pkg/controllers/rolloutrun/rolloutrun_controller.go index 83751b9..964c32e 100644 --- a/pkg/controllers/rolloutrun/rolloutrun_controller.go +++ b/pkg/controllers/rolloutrun/rolloutrun_controller.go @@ -129,6 +129,10 @@ func (r *RolloutRunReconciler) Reconcile(ctx context.Context, req ctrl.Request) newStatus := obj.Status.DeepCopy() + // Save original Tolerations before executor runs, as executor may modify + // RolloutRun.Spec.Batch.Tolerations when a batch is skipped. + originalTolerations := saveTolerations(obj) + accessor, workloads, canaryWorkloads, err := r.findWorkloadsCrossCluster(ctx, obj) if err != nil { return reconcile.Result{}, err @@ -141,6 +145,12 @@ func (r *RolloutRunReconciler) Reconcile(ctx context.Context, req ctrl.Request) logger.Error(tempErr, "failed to clean up annotation") } + // Persist spec changes if Tolerations was modified by the executor + if tempErr := r.updateTolerations(ctx, obj, originalTolerations); tempErr != nil { + logger.Error(tempErr, "failed to update tolerations") + return reconcile.Result{}, tempErr + } + updateStatus := r.updateStatusOnly(ctx, obj, newStatus, workloads, canaryWorkloads) if updateStatus != nil { logger.Error(updateStatus, "failed to update status") @@ -367,3 +377,38 @@ func (r *RolloutRunReconciler) updateStatusOnly(ctx context.Context, obj *rollou r.rvExpectation.ExpectUpdate(key, obj.ResourceVersion) // nolint return nil } + +// saveTolerations saves a deep copy of the current Tolerations before the executor runs. +// Returns nil if the batch strategy is not set or Tolerations is empty. +func saveTolerations(obj *rolloutv1alpha1.RolloutRun) []rolloutv1alpha1.RolloutRunTolerationTarget { + if obj.Spec.Batch == nil || len(obj.Spec.Batch.Tolerations) == 0 { + return nil + } + saved := make([]rolloutv1alpha1.RolloutRunTolerationTarget, len(obj.Spec.Batch.Tolerations)) + copy(saved, obj.Spec.Batch.Tolerations) + return saved +} + +// updateTolerations persists spec changes to Tolerations if the executor modified them +// (e.g., when a batch is skipped). +func (r *RolloutRunReconciler) updateTolerations(ctx context.Context, obj *rolloutv1alpha1.RolloutRun, original []rolloutv1alpha1.RolloutRunTolerationTarget) error { + if obj.Spec.Batch == nil { + return nil + } + if equality.Semantic.DeepEqual(original, obj.Spec.Batch.Tolerations) { + return nil + } + + _, err := kubeutilclient.UpdateOnConflict(clusterinfo.WithCluster(ctx, clusterinfo.Fed), r.Client, r.Client, obj, func(in *rolloutv1alpha1.RolloutRun) error { + if in.Spec.Batch != nil { + in.Spec.Batch.Tolerations = obj.Spec.Batch.Tolerations + } + return nil + }) + if err != nil { + return err + } + key := utils.ObjectKeyString(obj) + r.rvExpectation.ExpectUpdate(key, obj.ResourceVersion) // nolint + return nil +} diff --git a/pkg/controllers/scalerun/executor/batch.go b/pkg/controllers/scalerun/executor/batch.go index cab476a..f790342 100644 --- a/pkg/controllers/scalerun/executor/batch.go +++ b/pkg/controllers/scalerun/executor/batch.go @@ -203,6 +203,8 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat batchTargetStatuses := make([]rolloutv1alpha1.ScaleWorkloadStatus, 0) allWorkloadReady := true + allWorkloadsAutoSkippable := true + for _, item := range currentBatch.Targets { info := ctx.Workloads.Get(item.Cluster, item.Name) if info == nil { @@ -230,6 +232,12 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat } allWorkloadReady = false + + // Check auto-skip toleration for scale-up scenarios + if !e.canAutoSkipTarget(item, info, workloadStatus.ScaleFrom, workloadStatus.ScaleTo, newStatus) { + allWorkloadsAutoSkippable = false + } + if !needApplyReplicas { // if the target's replicas has been updated, we will not change replicas continue @@ -252,10 +260,62 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat return true, retryImmediately, nil } + if allWorkloadsAutoSkippable { + logger.Info("auto-skipping batch due to toleration") + newStatus.Batches.Records[currentBatchIndex].State = rorexecutor.StepSkipped + recordScaleRunTolerations(&newStatus.Batches.Tolerations, scaleRun.Spec.Batch.Batches, ctx.Workloads, currentBatchIndex, newStatus.Batches.Records[currentBatchIndex].Targets) + return true, retryImmediately, nil + } + // wait for next reconcile return false, retryDefault, nil } +// canAutoSkipTarget checks if the scale target meets the auto-skip toleration conditions. +// Toleration only applies for scale-up scenarios (ScaleFrom < ScaleTo). +// Returns true only when the workload is not ready due to a real deficit (gap > 0) within +// the toleration threshold and the initial delay has elapsed. +// Transient states (Generation mismatch) are NOT auto-skippable because the +// AvailableReplicas may be stale. +func (e *batchExecutor) canAutoSkipTarget(item rolloutv1alpha1.ScaleRunStepTarget, info *workload.Info, scaledFrom, scaledTo int32, newStatus *rolloutv1alpha1.ScaleRunStatus) bool { + // Toleration only applies for scale-up scenarios + if scaledFrom >= scaledTo { + return false + } + if item.Toleration == nil || item.Toleration.FailureThreshold == nil { + return false + } + + // Not skippable while workload has not been reconciled yet (Generation mismatch). + if info.Generation != info.Status.ObservedGeneration { + return false + } + + // Only evaluate toleration on a real deficit. + gap := scaledTo - info.Status.AvailableReplicas + if gap <= 0 { + return false + } + if gap > *item.Toleration.FailureThreshold { + return false + } + + // gap is within threshold, check timeout + if item.Toleration.InitialDelaySeconds != nil { + currentBatchIndex := newStatus.Batches.CurrentBatchIndex + startTime := newStatus.Batches.Records[currentBatchIndex].StartTime + if startTime == nil { + return false + } + elapsed := time.Since(startTime.Time) + if elapsed < time.Duration(*item.Toleration.InitialDelaySeconds)*time.Second { + return false + } + } + + return true +} + func (e *batchExecutor) checkScaledReady(info *workload.Info, scaledFrom, scaledTo int32) bool { if info.Status.ObservedGeneration != info.Generation { return false diff --git a/pkg/controllers/scalerun/executor/batch_test.go b/pkg/controllers/scalerun/executor/batch_test.go new file mode 100644 index 0000000..984376d --- /dev/null +++ b/pkg/controllers/scalerun/executor/batch_test.go @@ -0,0 +1,696 @@ +/** + * Copyright 2024 The KusionStack Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package executor + +import ( + "time" + + "github.com/stretchr/testify/suite" + appsv1 "k8s.io/api/apps/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" + "kusionstack.io/rollout/pkg/workload" +) + +// fakeWebhookExecutor is a no-op webhook executor used by the batch test suite. +// It reports every hook as immediately completed so the state machine can +// advance through PreBatchStepHook / PostBatchStepHook transitions. +type fakeWebhookExecutor struct{} + +func (e *fakeWebhookExecutor) Do(ctx *ExecutorContext, hookType rolloutv1alpha1.HookType) (bool, time.Duration, error) { + return true, retryImmediately, nil +} + +func (e *fakeWebhookExecutor) Cancel(ctx *ExecutorContext) {} + +type batchExecutorTestSuite struct { + suite.Suite + + executor *batchExecutor + scaleRun *rolloutv1alpha1.ScaleRun +} + +func (s *batchExecutorTestSuite) SetupSuite() { + s.executor = newBatchExecutor(&fakeWebhookExecutor{}) +} + +func (s *batchExecutorTestSuite) SetupTest() { + s.scaleRun = testScaleRun.DeepCopy() +} + +// runBatchTestCases executes each table-driven batch executor testcase and +// invokes the provided assert hooks for done/result/err, NewStatus, and +// (optionally) the workload objects post-reconcile. +func (s *batchExecutorTestSuite) runBatchTestCases(tests []scaleBatchTestCase) { + for i := range tests { + tt := tests[i] + s.Run(tt.name, func() { + scaleRun := tt.getObjects() + var objs []client.Object + if tt.getWorkloads != nil { + objs = tt.getWorkloads() + } + ctx := createTestScaleExecutorContext(scaleRun, objs...) + done, got, err := s.executor.Do(ctx) + tt.assertResult(done, got, err) + + if tt.assertStatus != nil { + tt.assertStatus(ctx.NewStatus) + } + if len(objs) > 0 && tt.assertWorkloads != nil { + newObjs := []client.Object{} + for _, info := range ctx.Workloads.ToSlice() { + newObjs = append(newObjs, info.Object) + } + tt.assertWorkloads(newObjs) + } + }) + } +} + +type scaleBatchTestCase struct { + name string + getObjects func() *rolloutv1alpha1.ScaleRun + getWorkloads func() []client.Object + assertResult func(done bool, result reconcile.Result, err error) + assertStatus func(status *rolloutv1alpha1.ScaleRunStatus) + assertWorkloads func(objects []client.Object) +} + +// newScaleRunStepTarget builds a ScaleRunStepTarget with the given replicas and +// no toleration. +func newScaleRunStepTarget(cluster, name string, replicas int32) rolloutv1alpha1.ScaleRunStepTarget { + return rolloutv1alpha1.ScaleRunStepTarget{ + CrossClusterObjectNameReference: rolloutv1alpha1.CrossClusterObjectNameReference{ + Cluster: cluster, + Name: name, + }, + Replicas: replicas, + } +} + +// newScaleRunStepTargetWithToleration builds a ScaleRunStepTarget with a +// toleration. Toleration only applies for scale-up scenarios (ScaleFrom < ScaleTo). +func newScaleRunStepTargetWithToleration(cluster, name string, replicas int32, toleration *rolloutv1alpha1.RolloutStepTargetToleration) rolloutv1alpha1.ScaleRunStepTarget { + target := newScaleRunStepTarget(cluster, name, replicas) + target.Toleration = toleration + return target +} + +// Test_BatchExecutor_Do_SkipToleration covers the auto-skip toleration +// scenarios for ScaleRun (design scenarios 6 & 7), plus the negative cases that +// must NOT trigger auto-skip: +// - scale-down: toleration does not apply (scenario 6: no skip) +// - no toleration / nil FailureThreshold: behavior unchanged +// - gap exceeds FailureThreshold: keep waiting (scenario 2) +// - Generation mismatch: transient state, not skippable +// - middle-batch auto-skip when gap within threshold & delay elapsed (scenario 7) +// - last-batch auto-skip transitions toward success (scenario 7 on last batch) +func (s *batchExecutorTestSuite) Test_BatchExecutor_Do_SkipToleration() { + tests := []scaleBatchTestCase{ + { + name: "auto-skip applies in middle batch when gap within threshold and delay elapsed", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + // 3 batches, currently on batch index 1 (middle, scale-up) + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTargetWithToleration("cluster-a", "test-a", 20, &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 10), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 1, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepSkipped}, + {Index: ptr.To[int32](1), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + {Index: ptr.To[int32](2), State: rorexecutor.StepNone}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // ScaleFrom = Spec.Replicas(10), ScaleTo = item.Replicas(20) + // gap = 20 - 15 = 5 <= FailureThreshold(5) -> auto-skippable + // InitialDelaySeconds = 0 means no wait + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 10, 15, 15), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) // not done yet; state machine advanced to PostBatchStepHook + s.Equal(reconcile.Result{Requeue: true}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepPostBatchStepHook, status.Batches.CurrentBatchState) + // StepSkipped set in doBatchUpgrading is overwritten by MoveToNextState, + // so the durable observable for auto-skip is the Tolerations field. + // gap = 20 - 15 = 5 + s.Len(status.Batches.Tolerations, 1) + s.Equal(int32(5), status.Batches.Tolerations[0].Toleration) + }, + }, + { + name: "auto-skip does not apply when gap exceeds threshold, batch stays running", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTargetWithToleration("cluster-a", "test-a", 20, &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 10), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 1, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepSkipped}, + {Index: ptr.To[int32](1), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + {Index: ptr.To[int32](2), State: rorexecutor.StepNone}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // gap = 20 - 12 = 8 > FailureThreshold(5) -> not auto-skippable, keep waiting + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 10, 12, 12), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepRunning, status.Batches.CurrentBatchState) + s.Empty(status.Batches.Tolerations) + }, + }, + { + name: "auto-skip on last batch transitions toward success", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 30), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTargetWithToleration("cluster-a", "test-a", 20, &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 2, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepSucceeded}, + {Index: ptr.To[int32](1), State: rorexecutor.StepSucceeded}, + {Index: ptr.To[int32](2), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // Last batch: scale-up 10 -> 20, gap = 20 - 16 = 4 <= 5 + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 10, 16, 16), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) // still need to go through PostBatchStepHook and Recycle + s.Equal(reconcile.Result{Requeue: true}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepPostBatchStepHook, status.Batches.CurrentBatchState) + s.Len(status.Batches.Tolerations, 1) + s.Equal(int32(4), status.Batches.Tolerations[0].Toleration) + }, + }, + { + name: "scale-down does not skip: toleration only applies for scale-up (scenario 6)", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + // 1 batch: scale-down from 10 to 5 + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTargetWithToleration("cluster-a", "test-a", 5, &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // Workload still has Spec.Replicas=10, ObservedReplicas=10 (not yet scaled down) + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 10, 10, 10), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepRunning, status.Batches.CurrentBatchState) + // Scale-down: toleration should not be recorded + s.Empty(status.Batches.Tolerations) + }, + assertWorkloads: func(objs []client.Object) { + s.Require().Len(objs, 1) + sts := objs[0].(*appsv1.StatefulSet) + // Scale was applied: Spec.Replicas changed 10 -> 5 + s.NotNil(sts.Spec.Replicas) + s.Equal(int32(5), *sts.Spec.Replicas) + }, + }, + { + name: "no skip toleration, behavior unchanged", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + // 1 batch: scale-up with no Toleration configured + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // AvailableReplicas=8 < ScaleTo=20 -> not ready; no toleration -> keep waiting + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 10, 8, 8), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepRunning, status.Batches.CurrentBatchState) + s.Empty(status.Batches.Tolerations) + }, + }, + { + name: "Generation mismatch blocks auto-skip (transient state, gap unreliable)", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTargetWithToleration("cluster-a", "test-a", 20, &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](5), + InitialDelaySeconds: ptr.To[int32](0), + }), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // Generation mismatch: spec updated but status hasn't caught up + obj := newFakeScaleObject("cluster-a", "default", "test-a", 10, 15, 15) + obj.Generation = 2 + obj.Status.ObservedGeneration = 1 + return []client.Object{obj} + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepRunning, status.Batches.CurrentBatchState) + // Gap within threshold, but Generation mismatch blocks auto-skip + s.Empty(status.Batches.Tolerations) + }, + }, + } + + s.runBatchTestCases(tests) +} + +// Test_BatchExecutor_Do_Running covers doBatchUpgrading behavior outside the +// auto-skip path: applying Scale on first reconcile (scale-up & scale-down) and +// the all-ready transition to PostBatchStepHook. +func (s *batchExecutorTestSuite) Test_BatchExecutor_Do_Running() { + tests := []scaleBatchTestCase{ + { + name: "scale-up not ready: apply replicas and requeue", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // Spec.Replicas=10, Available=10 -> not yet at ScaleTo=20 + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 10, 10, 10), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepRunning, status.Batches.CurrentBatchState) + s.Len(status.Batches.Records, 1) + s.Len(status.Batches.Records[0].Targets, 1) + // ScaleFrom/ScaleTo captured for the target + s.Equal(int32(10), status.Batches.Records[0].Targets[0].ScaleFrom) + s.Equal(int32(20), status.Batches.Records[0].Targets[0].ScaleTo) + }, + assertWorkloads: func(objs []client.Object) { + s.Require().Len(objs, 1) + sts := objs[0].(*appsv1.StatefulSet) + s.NotNil(sts.Spec.Replicas) + s.Equal(int32(20), *sts.Spec.Replicas) // applied + }, + }, + { + // On the second reconcile after replicas were applied, Records[0].Targets + // already carries ScaleFrom/ScaleTo, so findCurrentWorkloadStatus returns + // a match and needApplyReplicas=false. With AvailableReplicas>=ScaleTo, + // checkScaledReady returns true and the all-ready path advances the state + // machine to PostBatchStepHook. + name: "scale-up all ready (second reconcile): move to PostBatchStepHook", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + { + Index: ptr.To[int32](0), + State: rorexecutor.StepRunning, + StartTime: ptr.To(metav1.Now()), + // Pre-populated to simulate a prior reconcile that + // already recorded ScaleFrom/ScaleTo. Without this, + // needApplyReplicas is forced true and the all-ready + // branch is bypassed on the first reconcile. + Targets: []rolloutv1alpha1.ScaleWorkloadStatus{ + {Cluster: "cluster-a", Name: "test-a", ScaleFrom: 10, ScaleTo: 20}, + }, + }, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // Workload already scaled to 20 and Available=20 >= ScaleTo=20 -> ready. + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 20, 20, 20), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{Requeue: true}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepPostBatchStepHook, status.Batches.CurrentBatchState) + s.Empty(status.Batches.Tolerations) + }, + }, + { + name: "scale-down not ready: apply replicas and requeue", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 5), + }}, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: rorexecutor.StepRunning, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepRunning, StartTime: ptr.To(metav1.Now())}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + // Workload still at Spec=10, Observed=10 (not yet scaled to 5) + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-a", 10, 10, 10), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepRunning, status.Batches.CurrentBatchState) + s.Len(status.Batches.Records, 1) + s.Len(status.Batches.Records[0].Targets, 1) + s.Equal(int32(10), status.Batches.Records[0].Targets[0].ScaleFrom) + s.Equal(int32(5), status.Batches.Records[0].Targets[0].ScaleTo) + }, + assertWorkloads: func(objs []client.Object) { + s.Require().Len(objs, 1) + sts := objs[0].(*appsv1.StatefulSet) + s.NotNil(sts.Spec.Replicas) + s.Equal(int32(5), *sts.Spec.Replicas) // applied + }, + }, + } + + s.runBatchTestCases(tests) +} + +// Test_BatchExecutor_Do_Pending_Paused covers the StepNone -> StepPending +// transition. With Breakpoint=true, doPausing sets Phase=Paused and the state +// engine advances to StepPending. +func (s *batchExecutorTestSuite) Test_BatchExecutor_Do_Pending_Paused() { + tests := []scaleBatchTestCase{ + { + name: "None to Pending(Paused) with breakpoint batch", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + { + Breakpoint: true, + Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-0", 10), + }, + }, + { + Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-1", 10), + }, + }, + } + scaleRun.Status.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 0, + CurrentBatchState: rorexecutor.StepNone, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepNone}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + return []client.Object{ + newFakeScaleObject("cluster-a", "default", "test-0", 10, 10, 10), + newFakeScaleObject("cluster-a", "default", "test-1", 10, 10, 10), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) + s.Equal(reconcile.Result{Requeue: true}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rolloutv1alpha1.RolloutRunPhasePaused, status.Phase) + s.Equal(rorexecutor.StepPending, status.Batches.CurrentBatchState) + s.Equal(rorexecutor.StepPending, status.Batches.Records[0].State) + }, + assertWorkloads: func(objs []client.Object) { + // Only the current batch's targets get the progressing + // annotation from BatchScaleControl.Initialize. test-0 lives + // in the current batch (index 0); test-1 is in the next batch + // and is NOT initialized yet. + s.Require().Len(objs, 2) + progressingByName := map[string]bool{} + for _, obj := range objs { + progressingByName[obj.GetName()] = workload.IsProgressing(obj) + } + s.True(progressingByName["test-0"], "test-0 should be progressing (current batch target)") + s.False(progressingByName["test-1"], "test-1 should NOT be progressing (next batch)") + }, + }, + } + + s.runBatchTestCases(tests) +} + +// Test_BatchExecutor_Do_Recycling covers the StepResourceRecycling state on the +// last batch, where release() finalizes all workloads (removes progressing +// annotation) and the state advances to StepSucceeded. +func (s *batchExecutorTestSuite) Test_BatchExecutor_Do_Recycling() { + tests := []scaleBatchTestCase{ + { + name: "Recycling on last batch finalizes workloads and moves to Succeeded", + getObjects: func() *rolloutv1alpha1.ScaleRun { + scaleRun := s.scaleRun.DeepCopy() + scaleRun.Spec.Batch.Batches = []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-0", 10), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-1", 10), + }}, + } + scaleRun.Status.Batches = &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: 1, // last batch + CurrentBatchState: rorexecutor.StepResourceRecycling, + }, + Records: []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepSucceeded}, + {Index: ptr.To[int32](1), State: rorexecutor.StepResourceRecycling}, + }, + } + return scaleRun + }, + getWorkloads: func() []client.Object { + return []client.Object{ + withProgressingInfo(newFakeScaleObject("cluster-a", "default", "test-0", 10, 10, 10)), + withProgressingInfo(newFakeScaleObject("cluster-a", "default", "test-1", 10, 10, 10)), + } + }, + assertResult: func(done bool, result reconcile.Result, err error) { + s.Require().NoError(err) + s.False(done) // not yet done at batch level (still moves to next state) + // doRecycle -> release on last batch returns (true, retryImmediately, nil); + // the state engine converts retryImmediately to reconcile.Result{Requeue: true}. + s.Equal(reconcile.Result{Requeue: true}, result) + }, + assertStatus: func(status *rolloutv1alpha1.ScaleRunStatus) { + s.Equal(rorexecutor.StepSucceeded, status.Batches.CurrentBatchState) + s.Equal(rorexecutor.StepSucceeded, status.Batches.Records[1].State) + }, + assertWorkloads: func(objs []client.Object) { + for _, obj := range objs { + s.False(workload.IsProgressing(obj)) + } + }, + }, + } + + s.runBatchTestCases(tests) +} diff --git a/pkg/controllers/scalerun/executor/can_auto_skip_target_test.go b/pkg/controllers/scalerun/executor/can_auto_skip_target_test.go new file mode 100644 index 0000000..0c313e5 --- /dev/null +++ b/pkg/controllers/scalerun/executor/can_auto_skip_target_test.go @@ -0,0 +1,224 @@ +package executor + +import ( + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + + rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" + "kusionstack.io/rollout/pkg/workload" +) + +// TestCanAutoSkipTargetScaleRun directly exercises scalerun.batchExecutor.canAutoSkipTarget +// to cover scale-up/scale-down and transient-state guards (problem #4, design scenarios 6 & 7): +// - Scale-down (ScaleFrom >= ScaleTo) returns false (scenario 6: skip toleration entirely) +// - Scale-up with Generation mismatch returns false +// - Scale-up with gap <= 0 returns false +// - Scale-up with gap > FailureThreshold returns false +// - Scale-up with InitialDelay not elapsed returns false +// - Scale-up with gap within threshold and delay elapsed returns true (scenario 7: auto-skip) +// - Scale-up with nil Toleration / nil FailureThreshold / nil InitialDelaySeconds +func TestCanAutoSkipTargetScaleRun(t *testing.T) { + e := &batchExecutor{} + + now := ptr.To(metav1.Now()) + pastTime := ptr.To(metav1.Time{Time: time.Now().Add(-10 * time.Minute)}) + + type args struct { + item rolloutv1alpha1.ScaleRunStepTarget + info *workload.Info + scaledFrom int32 + scaledTo int32 + newStatus *rolloutv1alpha1.ScaleRunStatus + } + + tests := []struct { + name string + args args + want bool + }{ + { + name: "scale-down (ScaleFrom >= ScaleTo) returns false (scenario 6: toleration skipped)", + args: args{ + item: scaleTolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).available(8).desired(10).build(), + scaledFrom: 10, + scaledTo: 5, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "scale-up with nil Toleration returns false", + args: args{ + item: rolloutv1alpha1.ScaleRunStepTarget{Toleration: nil}, + info: newInfoBuilder().generation(1).observedGen(1).available(15).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "scale-up with nil FailureThreshold returns false", + args: args{ + item: rolloutv1alpha1.ScaleRunStepTarget{ + Toleration: &rolloutv1alpha1.RolloutStepTargetToleration{InitialDelaySeconds: ptr.To[int32](0)}, + }, + info: newInfoBuilder().generation(1).observedGen(1).available(15).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "scale-up with Generation mismatch returns false", + args: args{ + item: scaleTolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(2).observedGen(1).available(15).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "scale-up with gap <= 0 returns false (already satisfied)", + args: args{ + item: scaleTolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).available(20).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "scale-up with gap > FailureThreshold returns false", + args: args{ + item: scaleTolerationTarget(2, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).available(15).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "scale-up with InitialDelay not yet elapsed returns false", + args: args{ + item: scaleTolerationTarget(5, ptr.To[int32](300)), + info: newInfoBuilder().generation(1).observedGen(1).available(15).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + { + name: "scale-up with gap within threshold and delay elapsed returns true (scenario 7: auto-skip)", + args: args{ + item: scaleTolerationTarget(5, ptr.To[int32](300)), + info: newInfoBuilder().generation(1).observedGen(1).available(15).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, pastTime), + }, + want: true, + }, + { + name: "scale-up with InitialDelaySeconds nil treats delay as already elapsed", + args: args{ + item: scaleTolerationTarget(5, nil), + info: newInfoBuilder().generation(1).observedGen(1).available(18).desired(20).build(), + scaledFrom: 10, + scaledTo: 20, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: true, + }, + { + name: "scale-up with equal ScaleFrom and ScaleTo returns false (degenerate, not scale-up)", + args: args{ + item: scaleTolerationTarget(5, ptr.To[int32](0)), + info: newInfoBuilder().generation(1).observedGen(1).available(10).desired(10).build(), + scaledFrom: 10, + scaledTo: 10, + newStatus: newScaleRunStatusWithStart(0, now), + }, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := e.canAutoSkipTarget(tt.args.item, tt.args.info, tt.args.scaledFrom, tt.args.scaledTo, tt.args.newStatus) + if got != tt.want { + t.Errorf("canAutoSkipTarget() = %v, want %v", got, tt.want) + } + }) + } +} + +func scaleTolerationTarget(failureThreshold int32, initialDelay *int32) rolloutv1alpha1.ScaleRunStepTarget { + tol := &rolloutv1alpha1.RolloutStepTargetToleration{ + FailureThreshold: ptr.To[int32](failureThreshold), + } + if initialDelay != nil { + tol.InitialDelaySeconds = ptr.To[int32](*initialDelay) + } + return rolloutv1alpha1.ScaleRunStepTarget{Toleration: tol} +} + +func newScaleRunStatusWithStart(currentBatchIndex int32, startTime *metav1.Time) *rolloutv1alpha1.ScaleRunStatus { + records := []rolloutv1alpha1.ScaleRunStepStatus{ + {Index: ptr.To[int32](0), State: rorexecutor.StepRunning, StartTime: startTime}, + {Index: ptr.To[int32](1), State: rorexecutor.StepNone}, + {Index: ptr.To[int32](2), State: rorexecutor.StepNone}, + } + return &rolloutv1alpha1.ScaleRunStatus{ + Batches: &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: currentBatchIndex, + }, + Records: records, + }, + } +} + +// infoBuilder is a fluent builder for workload.Info to keep test cases concise. +type infoBuilder struct { + info *workload.Info +} + +func newInfoBuilder() *infoBuilder { + return &infoBuilder{info: &workload.Info{}} +} + +func (b *infoBuilder) generation(g int64) *infoBuilder { + b.info.Generation = g + return b +} + +func (b *infoBuilder) observedGen(g int64) *infoBuilder { + b.info.Status.ObservedGeneration = g + return b +} + +func (b *infoBuilder) available(n int32) *infoBuilder { + b.info.Status.AvailableReplicas = n + return b +} + +func (b *infoBuilder) desired(n int32) *infoBuilder { + b.info.Status.DesiredReplicas = n + return b +} + +func (b *infoBuilder) build() *workload.Info { + return b.info +} diff --git a/pkg/controllers/scalerun/executor/default_test.go b/pkg/controllers/scalerun/executor/default_test.go new file mode 100644 index 0000000..b756f6c --- /dev/null +++ b/pkg/controllers/scalerun/executor/default_test.go @@ -0,0 +1,152 @@ +/** + * Copyright 2024 The KusionStack Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package executor + +import ( + "context" + + "github.com/go-logr/logr" + "github.com/google/uuid" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + fakeclientset "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/kubernetes/scheme" + corev1client "k8s.io/client-go/kubernetes/typed/core/v1" + "k8s.io/client-go/tools/record" + rolloutapi "kusionstack.io/kube-api/rollout" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/log/zap" + + "kusionstack.io/rollout/pkg/workload" + "kusionstack.io/rollout/pkg/workload/statefulset" +) + +// testScaleRun is the template used as the base for each test case. +// Tests deep-copy and then mutate Spec/Status as needed. +var testScaleRun = rolloutv1alpha1.ScaleRun{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-scalerun", + Namespace: metav1.NamespaceDefault, + UID: types.UID(uuid.New().String()), + Labels: make(map[string]string), + Annotations: make(map[string]string), + }, + Spec: rolloutv1alpha1.ScaleRunSpec{ + TargetType: rolloutv1alpha1.ObjectTypeRef{ + APIVersion: statefulset.GVK.GroupVersion().String(), + Kind: statefulset.GVK.Kind, + }, + Batch: &rolloutv1alpha1.ScaleRunBatchStrategy{}, + }, + Status: rolloutv1alpha1.ScaleRunStatus{ + Conditions: []rolloutv1alpha1.Condition{}, + }, +} + +func newTestScaleLogger() logr.Logger { + return zap.New(zap.UseDevMode(true), zap.ConsoleEncoder()) +} + +func createTestScaleExecutorContext(scaleRun *rolloutv1alpha1.ScaleRun, objs ...client.Object) *ExecutorContext { + infos := []*workload.Info{} + inter := newTestScaleWorkloadInterface() + rolloutv1alpha1.AddToScheme(scheme.Scheme) + clientbuilder := fake.NewClientBuilder().WithScheme(scheme.Scheme) + for i := range objs { + obj := objs[i] + clientbuilder.WithObjects(obj) + w, _ := inter.GetInfo(obj.GetLabels()["kusionstack.io/cluster"], obj) + infos = append(infos, w) + } + + kubeClient := fakeclientset.NewSimpleClientset() + broadcaster := record.NewBroadcaster() + broadcaster.StartStructuredLogging(0) + broadcaster.StartRecordingToSink(&corev1client.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")}) + recorder := broadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "test"}) + + workloads := workload.NewSet(infos...) + c := clientbuilder.Build() + ctx := &ExecutorContext{ + Context: context.TODO(), + Client: c, + Recorder: recorder, + Accessor: inter, + ScaleRun: scaleRun, + Workloads: workloads, + NewStatus: scaleRun.Status.DeepCopy(), + } + ctx.Initialize() + ctx.WithLogger(newTestScaleLogger()) + return ctx +} + +func newTestScaleWorkloadInterface() workload.Accessor { + return statefulset.New() +} + +// newFakeScaleObject builds a StatefulSet suitable for scale-run tests. +// +// specReplicas: the workload's current Spec.Replicas (maps to info.Status.DesiredReplicas). +// availableReplicas: the workload's Status.AvailableReplicas (governs scale-up readiness and gap calc). +// observedReplicas: the workload's Status.Replicas (maps to info.Status.ObservedReplicas). +// +// Generation and ObservedGeneration are both 1 by default; tests can mutate +// the returned object to simulate a generation mismatch. +func newFakeScaleObject(cluster, namespace, name string, specReplicas, availableReplicas, observedReplicas int32) *appsv1.StatefulSet { + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Generation: 1, + Labels: map[string]string{ + "kusionstack.io/cluster": cluster, + }, + }, + Spec: appsv1.StatefulSetSpec{ + Replicas: &specReplicas, + UpdateStrategy: appsv1.StatefulSetUpdateStrategy{ + Type: appsv1.RollingUpdateStatefulSetStrategyType, + }, + }, + Status: appsv1.StatefulSetStatus{ + ObservedGeneration: 1, + Replicas: observedReplicas, + ReadyReplicas: availableReplicas, + AvailableReplicas: availableReplicas, + CurrentReplicas: observedReplicas, + UpdatedReplicas: observedReplicas, + CurrentRevision: "v1", + UpdateRevision: "v1", + }, + } +} + +// withProgressingInfo adds the rollout progressing annotation to a StatefulSet, +// mirroring what BatchScaleControl.Initialize would write. Used to simulate +// workloads that already have progressing info from a prior reconcile. +func withProgressingInfo(obj *appsv1.StatefulSet) *appsv1.StatefulSet { + if obj.Annotations == nil { + obj.Annotations = map[string]string{} + } + obj.Annotations[rolloutapi.AnnoRolloutProgressingInfo] = "" + return obj +} diff --git a/pkg/controllers/scalerun/executor/do_command.go b/pkg/controllers/scalerun/executor/do_command.go index 793d17a..2756e55 100644 --- a/pkg/controllers/scalerun/executor/do_command.go +++ b/pkg/controllers/scalerun/executor/do_command.go @@ -6,6 +6,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" + "kusionstack.io/rollout/pkg/workload" ) // doCommand @@ -16,10 +17,7 @@ func (r *Executor) doCommand(ctx *ExecutorContext) ctrl.Result { logger.Info("processing manual command", "command", cmd) newStatus := ctx.NewStatus - newBatchStatus := ctx.NewStatus.Batches - batchError := newStatus.Error - currentBatchIndex := newBatchStatus.CurrentBatchIndex switch cmd { case rolloutapis.AnnoManualCommandPause: newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePausing @@ -28,35 +26,102 @@ func (r *Executor) doCommand(ctx *ExecutorContext) ctrl.Result { newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseProgressing } case rolloutapis.AnnoManualCommandRetry: - if batchError != nil { + if newStatus.Error != nil { newStatus.Error = nil } case rolloutapis.AnnoManualCommandSkip: - if batchError != nil { - newStatus.Error = nil - - if int(currentBatchIndex) < (len(scaleRun.Spec.Batch.Batches) - 1) { - currentBatchIndex++ - newBatchStatus.CurrentBatchIndex = currentBatchIndex - newBatchStatus.CurrentBatchState = rorexecutor.StepNone - } else { - newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePostRollout - } + if newStatus.Error != nil { + handleBatchStatusWhenSkipped(newStatus, len(scaleRun.Spec.Batch.Batches), scaleRun.Spec.Batch.Batches, ctx.Workloads) } case rolloutapis.AnnoManualCommandCancel: newStatus.Phase = rolloutv1alpha1.RolloutRunPhaseCanceling case rolloutapis.AnnoManualCommandForceSkipCurrentBatch: - if batchError != nil { - newStatus.Error = nil + handleBatchStatusWhenSkipped(newStatus, len(scaleRun.Spec.Batch.Batches), scaleRun.Spec.Batch.Batches, ctx.Workloads) + } + + return ctrl.Result{Requeue: true} +} + +// handleBatchStatusWhenSkipped advances the batch state when the current batch is manually skipped. +// - Marks the current batch record as StepSkipped. +// - Records tolerations for each workload in the current batch (scale-up scenarios only). +// - On the last batch, transitions the phase to PostRollout (which will then become Succeeded). +// - On non-last batch, advances CurrentBatchIndex and resets CurrentBatchState. +func handleBatchStatusWhenSkipped(newStatus *rolloutv1alpha1.ScaleRunStatus, batchSize int, batches []rolloutv1alpha1.ScaleRunStep, workloads *workload.Set) { + currentBatchIndex := newStatus.Batches.CurrentBatchIndex + if newStatus.Error != nil { + newStatus.Error = nil + } + + newStatus.Batches.Records[currentBatchIndex].State = rorexecutor.StepSkipped + + // Record tolerations for each workload in the current batch (scale-up only) + recordScaleRunTolerations(&newStatus.Batches.Tolerations, batches, workloads, currentBatchIndex, newStatus.Batches.Records[currentBatchIndex].Targets) + + if int(currentBatchIndex) >= (batchSize - 1) { + // Last batch: advance to PostRollout phase (will transition to Succeeded) + newStatus.Phase = rolloutv1alpha1.RolloutRunPhasePostRollout + return + } + + // Not the last batch: advance to the next batch + newStatus.Batches.CurrentBatchIndex = currentBatchIndex + 1 + newStatus.Batches.CurrentBatchState = rorexecutor.StepNone +} + +// recordScaleRunTolerations upserts tolerations for each workload in the current batch. +// Toleration only applies for scale-up scenarios (ScaleFrom < ScaleTo). +// For each target, it computes the gap between ScaleTo and the workload's AvailableReplicas, +// and upserts (insert or update) the toleration value into the tolerations slice. +func recordScaleRunTolerations(tolerations *[]rolloutv1alpha1.RolloutRunTolerationTarget, batches []rolloutv1alpha1.ScaleRunStep, workloads *workload.Set, currentBatchIndex int32, currentTargets []rolloutv1alpha1.ScaleWorkloadStatus) { + if workloads == nil || int(currentBatchIndex) >= len(batches) { + return + } + currentBatch := batches[currentBatchIndex] + for _, target := range currentBatch.Targets { + info := workloads.Get(target.Cluster, target.Name) + if info == nil { + continue + } + // Locate ScaleFrom/ScaleTo recorded for this workload in the current batch status + var scaledFrom, scaledTo int32 + found := false + for _, st := range currentTargets { + if st.Cluster == target.Cluster && st.Name == target.Name { + scaledFrom = st.ScaleFrom + scaledTo = st.ScaleTo + found = true + break + } } - if int(currentBatchIndex) < (len(scaleRun.Spec.Batch.Batches) - 1) { - currentBatchIndex++ - newBatchStatus.CurrentBatchIndex = currentBatchIndex - newBatchStatus.CurrentBatchState = rorexecutor.StepNone - } else { - newBatchStatus.CurrentBatchState = rorexecutor.StepPostBatchStepHook + if !found { + // No status yet; fall back to spec values + scaledFrom = info.Status.DesiredReplicas + scaledTo = target.Replicas } + // Toleration only applies for scale-up scenarios + if scaledFrom >= scaledTo { + continue + } + gap := scaledTo - info.Status.AvailableReplicas + if gap < 0 { + gap = 0 + } + upsertToleration(tolerations, target.CrossClusterObjectNameReference, gap) } +} - return ctrl.Result{Requeue: true} +// upsertToleration updates the toleration value for the given workload reference, +// or appends a new entry if the workload is not yet present. +func upsertToleration(tolerations *[]rolloutv1alpha1.RolloutRunTolerationTarget, ref rolloutv1alpha1.CrossClusterObjectNameReference, gap int32) { + for i, t := range *tolerations { + if t.CrossClusterObjectNameReference == ref { + (*tolerations)[i].Toleration = gap + return + } + } + *tolerations = append(*tolerations, rolloutv1alpha1.RolloutRunTolerationTarget{ + CrossClusterObjectNameReference: ref, + Toleration: gap, + }) } diff --git a/pkg/controllers/scalerun/executor/do_command_test.go b/pkg/controllers/scalerun/executor/do_command_test.go new file mode 100644 index 0000000..82a64b6 --- /dev/null +++ b/pkg/controllers/scalerun/executor/do_command_test.go @@ -0,0 +1,334 @@ +/** + * Copyright 2024 The KusionStack Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package executor + +import ( + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + rolloutv1alpha1 "kusionstack.io/kube-api/rollout/v1alpha1" + + rorexecutor "kusionstack.io/rollout/pkg/controllers/rolloutrun/executor" + "kusionstack.io/rollout/pkg/workload" +) + +// TestHandleBatchStatusWhenSkipped exercises scalerun.handleBatchStatusWhenSkipped +// to cover manual-skip toleration recording (design scenario 5) plus the scale-down +// negative case (scenario 6). ScaleRun's toleration logic differs from RolloutRun +// in two key ways: +// - gap uses info.Status.AvailableReplicas (NOT UpdatedAvailableReplicas) +// - ScaleFrom/ScaleTo are sourced from Records[currentBatchIndex].Targets, with +// fallback to info.Status.DesiredReplicas / target.Replicas when no entry yet. +// - Toleration only applies for scale-up (ScaleFrom < ScaleTo); scale-down is +// skipped entirely (no entry appended). +// - Last batch advances Phase to PostRollout; non-last batch advances the +// CurrentBatchIndex and resets CurrentBatchState to StepNone. +func TestHandleBatchStatusWhenSkipped(t *testing.T) { + tests := []struct { + name string + batchIndex int32 + batchSize int + batches []rolloutv1alpha1.ScaleRunStep + currentTargets []rolloutv1alpha1.ScaleWorkloadStatus + workloads *workload.Set + expectedCurrentBatchIndex int32 + expectedCurrentBatchState rolloutv1alpha1.RolloutStepState + expectedPhase rolloutv1alpha1.RolloutRunPhase + expectedRecordState rolloutv1alpha1.RolloutStepState + expectedToleration *int32 + expectNoToleration bool + }{ + { + // Scenario 5: manual skip on first batch (scale-up 10 -> 20). + // available=15, so gap = 20 - 15 = 5, recorded as toleration. + // Non-last batch: advance to index 1, state -> StepNone. + name: "skip advances to next batch (scale-up with toleration recorded)", + batchIndex: 0, + batchSize: 3, + batches: []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 30), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 40), + }}, + }, + currentTargets: []rolloutv1alpha1.ScaleWorkloadStatus{ + {Cluster: "cluster-a", Name: "test-a", ScaleFrom: 10, ScaleTo: 20}, + }, + workloads: newTestScaleWorkloadSet("cluster-a", "test-a", 1, 10, 15), + expectedCurrentBatchIndex: 1, + expectedCurrentBatchState: rorexecutor.StepNone, + expectedRecordState: rorexecutor.StepSkipped, + expectedToleration: ptr.To[int32](5), // 20 - 15 = 5 + }, + { + // Scenario 5 (middle): manual skip on middle batch (scale-up 20 -> 30). + // available=22, so gap = 30 - 22 = 8, recorded as toleration. + // Non-last batch: advance to index 2, state -> StepNone. + name: "skip advances to next batch from middle", + batchIndex: 1, + batchSize: 3, + batches: []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 30), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 40), + }}, + }, + currentTargets: []rolloutv1alpha1.ScaleWorkloadStatus{ + {Cluster: "cluster-a", Name: "test-a", ScaleFrom: 20, ScaleTo: 30}, + }, + workloads: newTestScaleWorkloadSet("cluster-a", "test-a", 1, 20, 22), + expectedCurrentBatchIndex: 2, + expectedCurrentBatchState: rorexecutor.StepNone, + expectedRecordState: rorexecutor.StepSkipped, + expectedToleration: ptr.To[int32](8), // 30 - 22 = 8 + }, + { + // Scenario 5 (last): manual skip on last batch (scale-up 30 -> 40). + // available=33, so gap = 40 - 33 = 7, recorded as toleration. + // Last batch: Phase -> PostRollout, CurrentBatchIndex unchanged. + name: "skip last batch transitions to PostRollout phase (scale-up)", + batchIndex: 2, + batchSize: 3, + batches: []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 30), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 40), + }}, + }, + currentTargets: []rolloutv1alpha1.ScaleWorkloadStatus{ + {Cluster: "cluster-a", Name: "test-a", ScaleFrom: 30, ScaleTo: 40}, + }, + workloads: newTestScaleWorkloadSet("cluster-a", "test-a", 1, 30, 33), + expectedCurrentBatchIndex: 2, // unchanged - last batch does not advance index + expectedCurrentBatchState: rorexecutor.StepNone, + expectedPhase: rolloutv1alpha1.RolloutRunPhasePostRollout, + expectedRecordState: rorexecutor.StepSkipped, + expectedToleration: ptr.To[int32](7), // 40 - 33 = 7 + }, + { + // Scenario 6: scale-down (ScaleFrom=10 >= ScaleTo=5). + // recordScaleRunTolerations skips scale-down entirely, so no + // toleration is appended. Still advances to next batch (index 1). + name: "scale-down does not record toleration", + batchIndex: 0, + batchSize: 2, + batches: []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 5), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 10), + }}, + }, + currentTargets: []rolloutv1alpha1.ScaleWorkloadStatus{ + {Cluster: "cluster-a", Name: "test-a", ScaleFrom: 10, ScaleTo: 5}, + }, + workloads: newTestScaleWorkloadSet("cluster-a", "test-a", 1, 10, 10), + expectedCurrentBatchIndex: 1, + expectedCurrentBatchState: rorexecutor.StepNone, + expectedRecordState: rorexecutor.StepSkipped, + expectNoToleration: true, + }, + { + // Fallback path: when Records[currentBatchIndex].Targets has no + // matching entry, ScaleFrom falls back to info.Status.DesiredReplicas + // and ScaleTo falls back to target.Replicas. Here DesiredReplicas=10 + // and target.Replicas=20 -> scale-up -> gap = 20 - 15 = 5. + name: "fallback to DesiredReplicas/Replicas when currentTargets empty", + batchIndex: 0, + batchSize: 2, + batches: []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 30), + }}, + }, + currentTargets: nil, // no Targets recorded yet -> fallback path + workloads: newTestScaleWorkloadSet("cluster-a", "test-a", 1, 10, 15), + expectedCurrentBatchIndex: 1, + expectedCurrentBatchState: rorexecutor.StepNone, + expectedRecordState: rorexecutor.StepSkipped, + expectedToleration: ptr.To[int32](5), // 20 - 15 = 5 + }, + { + // Gap clamping: when available exceeds ScaleTo, gap is clamped to 0. + // ScaleTo=20, Available=25 -> raw gap = -5 -> clamped to 0. + name: "gap clamped to zero when available exceeds ScaleTo", + batchIndex: 0, + batchSize: 2, + batches: []rolloutv1alpha1.ScaleRunStep{ + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 20), + }}, + {Targets: []rolloutv1alpha1.ScaleRunStepTarget{ + newScaleRunStepTarget("cluster-a", "test-a", 30), + }}, + }, + currentTargets: []rolloutv1alpha1.ScaleWorkloadStatus{ + {Cluster: "cluster-a", Name: "test-a", ScaleFrom: 10, ScaleTo: 20}, + }, + workloads: newTestScaleWorkloadSet("cluster-a", "test-a", 1, 10, 25), + expectedCurrentBatchIndex: 1, + expectedCurrentBatchState: rorexecutor.StepNone, + expectedRecordState: rorexecutor.StepSkipped, + expectedToleration: ptr.To[int32](0), // 20 - 25 = -5 -> clamped to 0 + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + newStatus := &rolloutv1alpha1.ScaleRunStatus{ + Batches: &rolloutv1alpha1.ScaleRunBatchStatus{ + RolloutBatchStatus: rolloutv1alpha1.RolloutBatchStatus{ + CurrentBatchIndex: tt.batchIndex, + }, + Records: make([]rolloutv1alpha1.ScaleRunStepStatus, tt.batchSize), + }, + } + // Inject the current targets for the active batch (used by + // recordScaleRunTolerations to look up ScaleFrom/ScaleTo). + newStatus.Batches.Records[tt.batchIndex].Targets = tt.currentTargets + + handleBatchStatusWhenSkipped(newStatus, tt.batchSize, tt.batches, tt.workloads) + + if newStatus.Batches.CurrentBatchIndex != tt.expectedCurrentBatchIndex { + t.Errorf("CurrentBatchIndex = %d, want %d", newStatus.Batches.CurrentBatchIndex, tt.expectedCurrentBatchIndex) + } + if newStatus.Batches.CurrentBatchState != tt.expectedCurrentBatchState { + t.Errorf("CurrentBatchState = %v, want %v", newStatus.Batches.CurrentBatchState, tt.expectedCurrentBatchState) + } + if tt.expectedPhase != "" && newStatus.Phase != tt.expectedPhase { + t.Errorf("Phase = %v, want %v", newStatus.Phase, tt.expectedPhase) + } + if tt.expectedRecordState != "" && newStatus.Batches.Records[tt.batchIndex].State != tt.expectedRecordState { + t.Errorf("Records[%d].State = %v, want %v", tt.batchIndex, newStatus.Batches.Records[tt.batchIndex].State, tt.expectedRecordState) + } + if tt.expectNoToleration { + if len(newStatus.Batches.Tolerations) != 0 { + t.Errorf("expected 0 toleration entries for scale-down, got %d: %v", len(newStatus.Batches.Tolerations), newStatus.Batches.Tolerations) + } + } else if tt.expectedToleration != nil { + if len(newStatus.Batches.Tolerations) != 1 { + t.Errorf("expected 1 toleration entry, got %d", len(newStatus.Batches.Tolerations)) + } else if newStatus.Batches.Tolerations[0].Toleration != *tt.expectedToleration { + t.Errorf("Tolerations[0].Toleration = %d, want %d", newStatus.Batches.Tolerations[0].Toleration, *tt.expectedToleration) + } + } + }) + } +} + +// TestUpsertToleration_ScaleRun directly exercises scalerun.upsertToleration +// (shared with RolloutRun but living in the scalerun package namespace via +// re-declaration is not the case here; this function is package-local). +// It verifies: +// - append when ref not present +// - overwrite when ref already exists (no duplicate entries) +// - multiple distinct refs each get their own entry +func TestUpsertToleration_ScaleRun(t *testing.T) { + refA := rolloutv1alpha1.CrossClusterObjectNameReference{Cluster: "cluster-a", Name: "test-a"} + refB := rolloutv1alpha1.CrossClusterObjectNameReference{Cluster: "cluster-b", Name: "test-b"} + + t.Run("appends new entry when ref absent", func(t *testing.T) { + tolerations := []rolloutv1alpha1.RolloutRunTolerationTarget{} + upsertToleration(&tolerations, refA, 5) + if len(tolerations) != 1 { + t.Fatalf("expected 1 entry, got %d", len(tolerations)) + } + if tolerations[0].Toleration != 5 { + t.Errorf("Toleration = %d, want 5", tolerations[0].Toleration) + } + if tolerations[0].CrossClusterObjectNameReference != refA { + t.Errorf("ref mismatch: got %+v, want %+v", tolerations[0].CrossClusterObjectNameReference, refA) + } + }) + + t.Run("overwrites existing entry when ref present", func(t *testing.T) { + tolerations := []rolloutv1alpha1.RolloutRunTolerationTarget{ + {CrossClusterObjectNameReference: refA, Toleration: 5}, + } + upsertToleration(&tolerations, refA, 9) + if len(tolerations) != 1 { + t.Fatalf("expected 1 entry (no duplicates), got %d", len(tolerations)) + } + if tolerations[0].Toleration != 9 { + t.Errorf("Toleration = %d, want 9 (overwritten)", tolerations[0].Toleration) + } + }) + + t.Run("multiple distinct refs each get their own entry", func(t *testing.T) { + tolerations := []rolloutv1alpha1.RolloutRunTolerationTarget{ + {CrossClusterObjectNameReference: refA, Toleration: 5}, + } + upsertToleration(&tolerations, refB, 3) + if len(tolerations) != 2 { + t.Fatalf("expected 2 entries, got %d", len(tolerations)) + } + // Update A again to ensure B is not disturbed and A is overwritten + upsertToleration(&tolerations, refA, 7) + if len(tolerations) != 2 { + t.Fatalf("expected 2 entries (no dup), got %d", len(tolerations)) + } + got := map[string]int32{} + for _, tr := range tolerations { + got[tr.Cluster+"/"+tr.Name] = tr.Toleration + } + if got["cluster-a/test-a"] != 7 { + t.Errorf("cluster-a/test-a Toleration = %d, want 7", got["cluster-a/test-a"]) + } + if got["cluster-b/test-b"] != 3 { + t.Errorf("cluster-b/test-b Toleration = %d, want 3", got["cluster-b/test-b"]) + } + }) +} + +// newTestScaleWorkloadSet creates a workload.Set with a single workload for ScaleRun tests. +// ScaleRun uses info.Status.AvailableReplicas (NOT UpdatedAvailableReplicas like RolloutRun), +// so the helper populates AvailableReplicas rather than UpdatedAvailableReplicas. +func newTestScaleWorkloadSet(cluster, name string, generation int64, desiredReplicas, availableReplicas int32) *workload.Set { + return workload.NewSet(&workload.Info{ + ClusterName: cluster, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Generation: generation, + }, + Status: workload.InfoStatus{ + ObservedGeneration: generation, + DesiredReplicas: desiredReplicas, + AvailableReplicas: availableReplicas, + }, + }) +} diff --git a/pkg/controllers/scalerun/executor/suit_test.go b/pkg/controllers/scalerun/executor/suit_test.go new file mode 100644 index 0000000..1917855 --- /dev/null +++ b/pkg/controllers/scalerun/executor/suit_test.go @@ -0,0 +1,29 @@ +/** + * Copyright 2024 The KusionStack Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package executor + +import ( + "testing" + + "github.com/stretchr/testify/suite" +) + +// In order for 'go test' to run this suite, we need to create +// a normal test function and pass our suite to suite.Run +func TestBatchExecutorTestSuite(t *testing.T) { + suite.Run(t, new(batchExecutorTestSuite)) +} diff --git a/pkg/workload/info.go b/pkg/workload/info.go index 3cc29db..be0722a 100644 --- a/pkg/workload/info.go +++ b/pkg/workload/info.go @@ -107,10 +107,12 @@ func (o *Info) CheckUpdatedReady(replicas int32, strictCheck bool) (bool, string if o.Generation != o.Status.ObservedGeneration { return false, "workload Generation and ObservedGeneration are mismatched" } + if o.Status.UpdatedAvailableReplicas < replicas { return false, "workload updated available replicas is not satisfied" } - if strictCheck && (o.Status.ObservedReplicas > o.Status.DesiredReplicas || o.Status.TerminatingReplicas != 0) { + + if strictCheck && o.Status.ObservedReplicas > o.Status.DesiredReplicas || o.Status.TerminatingReplicas != 0 { return false, "workload observed replicas is more than desiredReplicas" } return true, "" diff --git a/pkg/workload/info_test.go b/pkg/workload/info_test.go new file mode 100644 index 0000000..f7d2675 --- /dev/null +++ b/pkg/workload/info_test.go @@ -0,0 +1,154 @@ +package workload + +import ( + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestCheckUpdatedReady(t *testing.T) { + tests := []struct { + name string + generation int64 + observedGen int64 + updatedAvailable int32 + desiredReplicas int32 + observedReplicas int32 + replicas int32 + strictCheck bool + expectedReady bool + expectedReason string + }{ + { + name: "generation mismatch returns not ready", + generation: 2, + observedGen: 1, + updatedAvailable: 10, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 10, + strictCheck: false, + expectedReady: false, + expectedReason: "workload Generation and ObservedGeneration are mismatched", + }, + { + name: "no toleration, replicas satisfied, returns ready", + generation: 1, + observedGen: 1, + updatedAvailable: 10, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 10, + strictCheck: false, + expectedReady: true, + expectedReason: "", + }, + { + name: "no toleration, replicas not satisfied, returns not ready", + generation: 1, + observedGen: 1, + updatedAvailable: 8, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 10, + strictCheck: false, + expectedReady: false, + expectedReason: "workload updated available replicas is not satisfied", + }, + { + name: "replicas not satisfied without toleration, returns not ready", + generation: 1, + observedGen: 1, + updatedAvailable: 8, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 10, + strictCheck: false, + expectedReady: false, + expectedReason: "workload updated available replicas is not satisfied", + }, + { + name: "replicas not satisfied (gap), returns not ready", + generation: 1, + observedGen: 1, + updatedAvailable: 8, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 10, + strictCheck: false, + expectedReady: false, + expectedReason: "workload updated available replicas is not satisfied", + }, + { + name: "gap exists, returns not ready", + generation: 1, + observedGen: 1, + updatedAvailable: 8, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 10, + strictCheck: false, + expectedReady: false, + expectedReason: "workload updated available replicas is not satisfied", + }, + { + name: "last batch gap exists, returns not ready", + generation: 1, + observedGen: 1, + updatedAvailable: 96, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 100, + strictCheck: true, + expectedReady: false, + expectedReason: "workload updated available replicas is not satisfied", + }, + { + name: "last batch replicas satisfied, returns ready", + generation: 1, + observedGen: 1, + updatedAvailable: 100, + desiredReplicas: 100, + observedReplicas: 100, + replicas: 100, + strictCheck: true, + expectedReady: true, + expectedReason: "", + }, + { + name: "last batch observed replicas exceeds desired", + generation: 1, + observedGen: 1, + updatedAvailable: 100, + desiredReplicas: 100, + observedReplicas: 105, + replicas: 100, + strictCheck: true, + expectedReady: false, + expectedReason: "workload observed replicas is more than desiredReplicas", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + info := &Info{ + ObjectMeta: metav1.ObjectMeta{ + Generation: tt.generation, + }, + Status: InfoStatus{ + ObservedGeneration: tt.observedGen, + UpdatedAvailableReplicas: tt.updatedAvailable, + DesiredReplicas: tt.desiredReplicas, + ObservedReplicas: tt.observedReplicas, + }, + } + ready, reason := info.CheckUpdatedReady(tt.replicas, tt.strictCheck) + if ready != tt.expectedReady { + t.Errorf("CheckUpdatedReady() ready = %v, want %v", ready, tt.expectedReady) + } + if reason != tt.expectedReason { + t.Errorf("CheckUpdatedReady() reason = %v, want %v", reason, tt.expectedReason) + } + }) + } +}