feat: support rolling by sliding window in batch release

Makefile

@@ -52,7 +52,7 @@ fmt: golangci
 	$(GOLANGCI) fmt
 
 .PHONY: lint
-lint: golangci
+lint: fmt
 	$(GOLANGCI) run 
 
 .PHONY: test

apis/rollout/v1alpha1/rolloutrun_types.go

@@ -120,6 +120,11 @@ type RolloutRunStepTarget struct {
 
 	// Replicas is the replicas of the rollout task, which represents the number of pods to be upgraded
 	Replicas intstr.IntOrString `json:"replicas"`
+
+	// ReplicaSlidingWindow used to control the number of pods that are allowed to be upgraded in
+	// a sliding window for progressive rollout smoothly.
+	// +optional
+	ReplicaSlidingWindow *intstr.IntOrString `json:"replicaSlidingWindow,omitempty"`
 }
 
 type RolloutRunStatus struct {

apis/rollout/v1alpha1/rolloutstrategy_types.go

@@ -90,6 +90,11 @@ type RolloutStep struct {
 	// Replicas is the replicas of the rollout task, which represents the number of pods to be upgraded
 	Replicas intstr.IntOrString `json:"replicas"`
 
+	// ReplicaSlidingWindow used to control the number of pods that are allowed to be upgraded in
+	// a sliding window for progressive rollout smoothly.
+	// +optional
+	ReplicaSlidingWindow *intstr.IntOrString `json:"replicaSlidingWindow,omitempty"`
+
 	// traffic strategy
 	// +optional
 	Traffic *TrafficStrategy `json:"traffic,omitempty"`

config/crd/bases/rollout.kusionstack.io_rolloutruns.yaml

@@ -90,6 +90,14 @@ spec:
                               name:
                                 description: Name is the resource name
                                 type: string
+                              replicaSlidingWindow:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                description: |-
+                                  ReplicaSlidingWindow used to control the number of pods that are allowed to be upgraded in
+                                  a sliding window for progressive rollout smoothly.
+                                x-kubernetes-int-or-string: true
                               replicas:
                                 anyOf:
                                 - type: integer
@@ -485,6 +493,14 @@ spec:
                         name:
                           description: Name is the resource name
                           type: string
+                        replicaSlidingWindow:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          description: |-
+                            ReplicaSlidingWindow used to control the number of pods that are allowed to be upgraded in
+                            a sliding window for progressive rollout smoothly.
+                          x-kubernetes-int-or-string: true
                         replicas:
                           anyOf:
                           - type: integer

config/crd/bases/rollout.kusionstack.io_rolloutstrategies.yaml

@@ -113,6 +113,14 @@ spec:
                       description: Properties contains additional information for
                         step
                       type: object
+                    replicaSlidingWindow:
+                      anyOf:
+                      - type: integer
+                      - type: string
+                      description: |-
+                        ReplicaSlidingWindow used to control the number of pods that are allowed to be upgraded in
+                        a sliding window for progressive rollout smoothly.
+                      x-kubernetes-int-or-string: true
                     replicas:
                       anyOf:
                       - type: integer

config/kind/workload/bases/rollout.yaml

@@ -44,5 +44,6 @@ batch:
   batches:
   - replicas: 1
   - replicas: 3
+    replicaSlidingWindow: 1
     breakpoint: true
   - replicas: 100%

pkg/controllers/rollout/utils.go

@@ -155,7 +155,8 @@ func constructRolloutRunBatches(strategy *rolloutv1alpha1.BatchStrategy, workloa
 					Cluster: info.ClusterName,
 					Name:    info.Name,
 				},
-				Replicas: b.Replicas,
+				Replicas:             b.Replicas,
+				ReplicaSlidingWindow: b.ReplicaSlidingWindow,
 			}
 			targets = append(targets, target)
 		}

pkg/controllers/rollout/utils_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/davecgh/go-spew/spew"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/utils/ptr"
 
 	rolloutv1alpha1 "kusionstack.io/rollout/apis/rollout/v1alpha1"
 	"kusionstack.io/rollout/pkg/workload"
@@ -61,8 +62,9 @@ func Test_constructRolloutRunBatches(t *testing.T) {
 							},
 						},
 						{
-							Breakpoint: true,
-							Replicas:   intstr.FromString("50%"),
+							Breakpoint:           true,
+							Replicas:             intstr.FromString("50%"),
+							ReplicaSlidingWindow: ptr.To(intstr.FromString("10%")),
 							Match: &rolloutv1alpha1.ResourceMatch{
 								Names: []rolloutv1alpha1.CrossClusterObjectNameReference{
 									{
@@ -128,14 +130,16 @@ func Test_constructRolloutRunBatches(t *testing.T) {
 								Cluster: "cluster-a",
 								Name:    "test-1",
 							},
-							Replicas: intstr.FromString("50%"),
+							Replicas:             intstr.FromString("50%"),
+							ReplicaSlidingWindow: ptr.To(intstr.FromString("10%")),
 						},
 						{
 							CrossClusterObjectNameReference: rolloutv1alpha1.CrossClusterObjectNameReference{
 								Cluster: "cluster-b",
 								Name:    "test-1",
 							},
-							Replicas: intstr.FromString("50%"),
+							Replicas:             intstr.FromString("50%"),
+							ReplicaSlidingWindow: ptr.To(intstr.FromString("10%")),
 						},
 					},
 				},

pkg/controllers/rolloutrun/control/control.go

@@ -78,7 +78,7 @@ func (c *BatchReleaseControl) Initialize(info *workload.Info, ownerKind, ownerNa
 	return err
 }
 
-func (c *BatchReleaseControl) UpdatePartition(info *workload.Info, expectedUpdated intstr.IntOrString) (bool, error) {
+func (c *BatchReleaseControl) UpdatePartition(info *workload.Info, expectedUpdated int32) (bool, error) {
 	ctx := clusterinfo.WithCluster(context.Background(), info.ClusterName)
 	obj := info.Object
 	return utils.UpdateOnConflict(ctx, c.client, c.client, obj, func() error {

pkg/controllers/rolloutrun/executor/batch.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	ctrl "sigs.k8s.io/controller-runtime"
 
 	rolloutv1alpha1 "kusionstack.io/rollout/apis/rollout/v1alpha1"
@@ -210,23 +211,28 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat
 		status := info.APIStatus()
 		batchTargetStatuses = append(batchTargetStatuses, info.APIStatus())
 
-		expectedUpdatedReplicas, _ := workload.CalculateUpdatedReplicas(&status.Replicas, item.Replicas)
+		currentBatchExpectedReplicas, _ := workload.CalculateUpdatedReplicas(&status.Replicas, item.Replicas)
 
-		if info.CheckUpdatedReady(expectedUpdatedReplicas) {
+		if info.CheckUpdatedReady(currentBatchExpectedReplicas) {
 			// if the target is ready, we will not change partition
 			continue
 		}
 
 		allWorkloadReady = false
 		logger.V(3).Info("still waiting for target to be ready", "target", item.CrossClusterObjectNameReference)
 
+		expectedReplicas, err := e.calculateExpectedReplicasBySlidingWindow(status, currentBatchExpectedReplicas, item.ReplicaSlidingWindow)
+		if err != nil {
+			return false, retryStop, err
+		}
+
 		// ensure partition: upgradePartition is an idempotent function
-		changed, err := batchControl.UpdatePartition(info, item.Replicas)
+		changed, err := batchControl.UpdatePartition(info, expectedReplicas)
 		if err != nil {
 			return false, retryStop, err
 		}
 		if changed {
-			logger.V(2).Info("upgrade target partition", "target", item.CrossClusterObjectNameReference, "partition", expectedUpdatedReplicas)
+			logger.V(2).Info("upgrade target partition", "target", item.CrossClusterObjectNameReference, "partition", expectedReplicas)
 		}
 	}
 
@@ -240,3 +246,20 @@ func (e *batchExecutor) doBatchUpgrading(ctx *ExecutorContext) (bool, time.Durat
 	// wait for next reconcile
 	return false, retryDefault, nil
 }
+
+// calculateExpectedReplicasBySlidingWindow calculate expected replicas by sliding window
+// if window is nil, return currentBatchExpectedReplicas
+// if window is not nil, return min(currentBatchExpectedReplicas, updatedAvailableReplicas + increment)
+func (e *batchExecutor) calculateExpectedReplicasBySlidingWindow(status rolloutv1alpha1.RolloutWorkloadStatus, currentBatchExpectedReplicas int32, window *intstr.IntOrString) (int32, error) {
+	if window == nil {
+		return currentBatchExpectedReplicas, nil
+	}
+	increment, err := workload.CalculateUpdatedReplicas(&status.Replicas, *window)
+	if err != nil {
+		return currentBatchExpectedReplicas, err
+	}
+	expected := status.UpdatedAvailableReplicas + increment
+	// limit expected replicas to currentBatchExpectedReplicas
+	expected = min(currentBatchExpectedReplicas, expected)
+	return expected, nil
+}

pkg/controllers/rolloutrun/executor/batch_test.go

@@ -315,12 +315,17 @@ func (s *batchExecutorTestSuite) Test_BatchExecutor_Do() {
 }
 
 func newRunStepTarget(cluster, name string, replicas intstr.IntOrString) rolloutv1alpha1.RolloutRunStepTarget {
+	return newRunStepTargetWithSlidingWindow(cluster, name, replicas, nil)
+}
+
+func newRunStepTargetWithSlidingWindow(cluster, name string, replicas intstr.IntOrString, window *intstr.IntOrString) rolloutv1alpha1.RolloutRunStepTarget {
 	return rolloutv1alpha1.RolloutRunStepTarget{
 		CrossClusterObjectNameReference: rolloutv1alpha1.CrossClusterObjectNameReference{
 			Cluster: cluster,
 			Name:    name,
 		},
-		Replicas: replicas,
+		Replicas:             replicas,
+		ReplicaSlidingWindow: window,
 	}
 }
 
@@ -482,7 +487,7 @@ func (s *batchExecutorTestSuite) Test_BatchExecutor_Do_Running() {
 			},
 		},
 		{
-			name: "workflow instance not found",
+			name: "workload instance not found",
 			getObjects: func() (*rolloutv1alpha1.Rollout, *rolloutv1alpha1.RolloutRun) {
 				rollout := s.rollout.DeepCopy()
 				rolloutRun := s.rolloutRun.DeepCopy()
@@ -569,13 +574,90 @@ func (s *batchExecutorTestSuite) Test_BatchExecutor_Do_Running() {
 				for _, obj := range objs {
 					if s.IsType(&appsv1.StatefulSet{}, obj) {
 						sts := obj.(*appsv1.StatefulSet)
-						s.NotNil(sts.Spec.UpdateStrategy.RollingUpdate)
-						s.NotNil(sts.Spec.UpdateStrategy.RollingUpdate.Partition)
+						s.Require().NotNil(sts.Spec.UpdateStrategy.RollingUpdate)
+						s.Require().NotNil(sts.Spec.UpdateStrategy.RollingUpdate.Partition)
 						s.EqualValues(90, *sts.Spec.UpdateStrategy.RollingUpdate.Partition)
 					}
 				}
 			},
 		},
+		{
+			name: "upgrade workload partition by sliding window",
+			getObjects: func() (*rolloutv1alpha1.Rollout, *rolloutv1alpha1.RolloutRun) {
+				rollout := s.rollout.DeepCopy()
+				rolloutRun := s.rolloutRun.DeepCopy()
+
+				// setup rolloutRun
+				rolloutRun.Spec.Batch.Batches = []rolloutv1alpha1.RolloutRunStep{{
+					Targets: []rolloutv1alpha1.RolloutRunStepTarget{
+						// test-a with normal sliding window
+						newRunStepTargetWithSlidingWindow("cluster-a", "test-a", intstr.FromInt(50), ptr.To(intstr.FromInt(10))),
+						// test-b with a too big sliding window
+						newRunStepTargetWithSlidingWindow("cluster-a", "test-b", intstr.FromInt(10), ptr.To(intstr.FromInt(50))),
+					},
+				}}
+				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()),
+						},
+					},
+				}
+				return rollout, rolloutRun
+			},
+			getWorkloads: func() []client.Object {
+				return []client.Object{
+					newFakeObject("cluster-a", "default", "test-a", 100, 20, 15),
+					newFakeObject("cluster-a", "default", "test-b", 100, 0, 0),
+				}
+			},
+			assertResult: func(done bool, result reconcile.Result, err error) {
+				s.Require().NoError(err)
+				s.Equal(reconcile.Result{RequeueAfter: retryDefault}, result)
+				s.False(done)
+			},
+			assertStatus: func(status *rolloutv1alpha1.RolloutRunStatus) {
+				s.Len(status.BatchStatus.Records, 1)
+				s.Len(status.BatchStatus.Records[0].Targets, 2)
+
+				for _, target := range status.BatchStatus.Records[0].Targets {
+					s.EqualValues(100, target.Replicas)
+					switch target.Name {
+					case "test-a":
+						s.EqualValues(15, target.UpdatedReplicas)
+						s.EqualValues(15, target.UpdatedReadyReplicas)
+						s.EqualValues(15, target.UpdatedAvailableReplicas)
+					case "test-b":
+						s.EqualValues(0, target.UpdatedReplicas)
+						s.EqualValues(0, target.UpdatedReadyReplicas)
+						s.EqualValues(0, target.UpdatedAvailableReplicas)
+					}
+				}
+			},
+			assertWorkloads: func(objs []client.Object) {
+				for _, obj := range objs {
+					if s.IsType(&appsv1.StatefulSet{}, obj) {
+						sts := obj.(*appsv1.StatefulSet)
+						s.Require().NotNil(sts.Spec.UpdateStrategy.RollingUpdate)
+						s.Require().NotNil(sts.Spec.UpdateStrategy.RollingUpdate.Partition)
+						switch sts.Name {
+						case "test-a":
+							// partition = total(100) - (updatedReplicas(15) + slidingWindow(10) ) = 75
+							s.EqualValues(75, *sts.Spec.UpdateStrategy.RollingUpdate.Partition)
+						case "test-b":
+							s.EqualValues(90, *sts.Spec.UpdateStrategy.RollingUpdate.Partition)
+						}
+					}
+				}
+			},
+		},
 		{
 			name: "waiting for workload ready",
 			getObjects: func() (*rolloutv1alpha1.Rollout, *rolloutv1alpha1.RolloutRun) {