Support scale controller for StatefulSet (#1901)

pkg/controller/common/rollout/rollout_plan_controller.go

@@ -336,6 +336,7 @@ func (r *Controller) GetWorkloadController() (workloads.WorkloadController, erro
 	}
 
 	if r.targetWorkload.GroupVersionKind().Group == kruisev1.GroupVersion.Group {
+		// check if the target workload is CloneSet
 		if r.targetWorkload.GetKind() == reflect.TypeOf(kruisev1.CloneSet{}).Name() {
 			// check whether current rollout plan is for workload rolling or scaling
 			if r.sourceWorkload != nil {
@@ -348,6 +349,7 @@ func (r *Controller) GetWorkloadController() (workloads.WorkloadController, erro
 	}
 
 	if r.targetWorkload.GroupVersionKind().Group == apps.GroupName {
+		// check if the target workload is Deployment
 		if r.targetWorkload.GetKind() == reflect.TypeOf(apps.Deployment{}).Name() {
 			// check whether current rollout plan is for workload rolling or scaling
 			if r.sourceWorkload != nil {
@@ -358,5 +360,6 @@ func (r *Controller) GetWorkloadController() (workloads.WorkloadController, erro
 				r.rolloutSpec, r.rolloutStatus, target), nil
 		}
 	}
+
 	return nil, fmt.Errorf("the workload kind `%s` is not supported", kind)
 }

pkg/controller/common/rollout/workloads/common.go

@@ -110,14 +110,14 @@ func verifyBatchesWithScale(rolloutSpec *v1alpha1.RolloutPlan, originalSize, tar
 }
 
 func calculateNewBatchTarget(rolloutSpec *v1alpha1.RolloutPlan, originalSize, targetSize, currentBatch int) int {
-	newPodTarget := originalSize
 	if currentBatch == len(rolloutSpec.RolloutBatches)-1 {
-		newPodTarget = targetSize
 		// special handle the last batch, we ignore the rest of the batch in case there are rounding errors
 		klog.InfoS("use the target size as the total pod target for the last rolling batch",
-			"current batch", currentBatch, "new  pod target", newPodTarget)
-		return newPodTarget
+			"current batch", currentBatch, "new pod target", targetSize)
+		return targetSize
 	}
+
+	newPodTarget := originalSize
 	for i := 0; i <= currentBatch && i < len(rolloutSpec.RolloutBatches); i++ {
 		if targetSize > originalSize {
 			batchSize, _ := intstr.GetValueFromIntOrPercent(&rolloutSpec.RolloutBatches[i].Replicas, targetSize-originalSize,
@@ -129,15 +129,24 @@ func calculateNewBatchTarget(rolloutSpec *v1alpha1.RolloutPlan, originalSize, ta
 			newPodTarget -= batchSize
 		}
 	}
+
 	klog.InfoS("calculated the number of new pod size", "current batch", currentBatch,
 		"new pod target", newPodTarget)
 	return newPodTarget
 }
 
-func getDeployReplicaSize(deploy *apps.Deployment) int32 {
+func getDeploymentReplicas(deploy *apps.Deployment) int32 {
 	// replicas default is 1
 	if deploy.Spec.Replicas != nil {
 		return *deploy.Spec.Replicas
 	}
 	return 1
 }
+
+func getStatefulSetReplicas(statefulSet *apps.StatefulSet) int32 {
+	// replicas default is 1
+	if statefulSet.Spec.Replicas != nil {
+		return *statefulSet.Spec.Replicas
+	}
+	return 1
+}

pkg/controller/common/rollout/workloads/deployment_rollout_controller.go

@@ -107,12 +107,12 @@ func (c *DeploymentRolloutController) VerifySpec(ctx context.Context) (bool, err
 		return false, verifyErr
 	}
 
-	if !c.sourceDeploy.Spec.Paused && getDeployReplicaSize(&c.sourceDeploy) != c.sourceDeploy.Status.Replicas {
+	if !c.sourceDeploy.Spec.Paused && getDeploymentReplicas(&c.sourceDeploy) != c.sourceDeploy.Status.Replicas {
 		return false, fmt.Errorf("the source deployment %s is still being reconciled, need to be paused or stable",
 			c.sourceDeploy.GetName())
 	}
 
-	if !c.targetDeploy.Spec.Paused && getDeployReplicaSize(&c.targetDeploy) != c.targetDeploy.Status.Replicas {
+	if !c.targetDeploy.Spec.Paused && getDeploymentReplicas(&c.targetDeploy) != c.targetDeploy.Status.Replicas {
 		return false, fmt.Errorf("the target deployment %s is still being reconciled, need to be paused or stable",
 			c.targetDeploy.GetName())
 	}
@@ -152,7 +152,7 @@ func (c *DeploymentRolloutController) Initialize(ctx context.Context) (bool, err
 
 	// claim target deployment
 	// make sure we start with the matching replicas and target
-	targetInitSize := pointer.Int32Ptr(c.rolloutStatus.RolloutTargetSize - getDeployReplicaSize(&c.sourceDeploy))
+	targetInitSize := pointer.Int32Ptr(c.rolloutStatus.RolloutTargetSize - getDeploymentReplicas(&c.sourceDeploy))
 	if _, err := c.claimDeployment(ctx, &c.targetDeploy, targetInitSize); err != nil {
 		// nolint:nilerr
 		return false, nil
@@ -257,8 +257,8 @@ func (c *DeploymentRolloutController) FinalizeOneBatch(ctx context.Context) (boo
 		return false, nil
 	}
 
-	sourceTarget := getDeployReplicaSize(&c.sourceDeploy)
-	targetTarget := getDeployReplicaSize(&c.targetDeploy)
+	sourceTarget := getDeploymentReplicas(&c.sourceDeploy)
+	targetTarget := getDeploymentReplicas(&c.targetDeploy)
 	if sourceTarget+targetTarget != c.rolloutStatus.RolloutTargetSize {
 		err := fmt.Errorf("deployment targets don't match total rollout, sourceTarget = %d, targetTarget = %d, "+
 			"rolloutTargetSize = %d", sourceTarget, targetTarget, c.rolloutStatus.RolloutTargetSize)
@@ -315,7 +315,7 @@ func (c *DeploymentRolloutController) fetchDeployments(ctx context.Context) erro
 
 // calculateRolloutTotalSize fetches the Deployment and returns the replicas (not the actual number of pods)
 func (c *DeploymentRolloutController) calculateRolloutTotalSize() (int32, error) {
-	sourceSize := getDeployReplicaSize(&c.sourceDeploy)
+	sourceSize := getDeploymentReplicas(&c.sourceDeploy)
 	// the spec target size is the truth if it's set
 	if c.rolloutSpec.TargetSize != nil {
 		targetSize := *c.rolloutSpec.TargetSize
@@ -366,7 +366,7 @@ func (c *DeploymentRolloutController) rolloutBatchFirstHalf(ctx context.Context,
 
 	if rolloutStrategy == v1alpha1.IncreaseFirstRolloutStrategyType {
 		// set the target replica first which should increase its size
-		if getDeployReplicaSize(&c.targetDeploy) < targetSize {
+		if getDeploymentReplicas(&c.targetDeploy) < targetSize {
 			klog.InfoS("set target deployment replicas", "deploy", c.targetDeploy.Name, "targetSize", targetSize)
 			_ = c.scaleDeployment(ctx, &c.targetDeploy, targetSize)
 			c.recorder.Event(c.parentController, event.Normal("Batch Rollout",
@@ -377,14 +377,14 @@ func (c *DeploymentRolloutController) rolloutBatchFirstHalf(ctx context.Context,
 
 		// do nothing if the target is already reached
 		klog.InfoS("target deployment replicas overshoot the size already", "deploy", c.targetDeploy.Name,
-			"deployment size", getDeployReplicaSize(&c.targetDeploy), "targetSize", targetSize)
+			"deployment size", getDeploymentReplicas(&c.targetDeploy), "targetSize", targetSize)
 		return true, nil
 	}
 
 	if rolloutStrategy == v1alpha1.DecreaseFirstRolloutStrategyType {
 		// set the source replicas first which should shrink its size
 		sourceSize := c.calculateCurrentSource(c.rolloutStatus.RolloutTargetSize)
-		if getDeployReplicaSize(&c.sourceDeploy) > sourceSize {
+		if getDeploymentReplicas(&c.sourceDeploy) > sourceSize {
 			klog.InfoS("set source deployment replicas", "source deploy", c.sourceDeploy.Name, "sourceSize", sourceSize)
 			_ = c.scaleDeployment(ctx, &c.sourceDeploy, sourceSize)
 			c.recorder.Event(c.parentController, event.Normal("Batch Rollout",
@@ -395,7 +395,7 @@ func (c *DeploymentRolloutController) rolloutBatchFirstHalf(ctx context.Context,
 
 		// do nothing if the reduce target is already reached
 		klog.InfoS("source deployment replicas overshoot the size already", "source deploy", c.sourceDeploy.Name,
-			"deployment size", getDeployReplicaSize(&c.sourceDeploy), "sourceSize", sourceSize)
+			"deployment size", getDeploymentReplicas(&c.sourceDeploy), "sourceSize", sourceSize)
 		return true, nil
 	}
 

pkg/controller/common/rollout/workloads/deployment_scale_controller.go

@@ -279,7 +279,7 @@ func (s *DeploymentScaleController) size(ctx context.Context) (int32, error) {
 			return 0, err
 		}
 	}
-	return getDeployReplicaSize(s.deploy), nil
+	return getDeploymentReplicas(s.deploy), nil
 }
 
 func (s *DeploymentScaleController) fetchDeployment(ctx context.Context) error {

pkg/controller/common/rollout/workloads/statefulset_controller.go

@@ -0,0 +1,111 @@
+/*
+Copyright 2021 The KubeVela 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 workloads
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/crossplane/crossplane-runtime/pkg/event"
+	apps "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/pointer"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+)
+
+type statefulSetController struct {
+	workloadController
+	targetNamespacedName types.NamespacedName
+}
+
+// add the parent controller to the owner of the StatefulSet, and initialize the size
+// before kicking start the update and start from every pod in the old version
+func (c *statefulSetController) claimStatefulSet(ctx context.Context, statefulSet *apps.StatefulSet, initSize *int32) (bool, error) {
+	if controller := metav1.GetControllerOf(statefulSet); controller != nil &&
+		controller.Kind == v1beta1.AppRolloutKind && controller.APIVersion == v1beta1.SchemeGroupVersion.String() {
+		// it's already there
+		return true, nil
+	}
+
+	statefulSetPatch := client.MergeFrom(statefulSet.DeepCopyObject())
+
+	// add the parent controller to the owner of the StatefulSet
+	ref := metav1.NewControllerRef(c.parentController, v1beta1.AppRolloutKindVersionKind)
+	statefulSet.SetOwnerReferences(append(statefulSet.GetOwnerReferences(), *ref))
+
+	if initSize != nil {
+		statefulSet.Spec.Replicas = initSize
+	}
+
+	// patch the StatefulSet
+	if err := c.client.Patch(ctx, statefulSet, statefulSetPatch, client.FieldOwner(c.parentController.GetUID())); err != nil {
+		c.recorder.Event(c.parentController, event.Warning("Failed to the start the StatefulSet update", err))
+		c.rolloutStatus.RolloutRetry(err.Error())
+		return false, err
+	}
+	return false, nil
+}
+
+// scale the StatefulSet
+func (c *statefulSetController) scaleStatefulSet(ctx context.Context, statefulSet *apps.StatefulSet, size int32) error {
+	statefulSetPatch := client.MergeFrom(statefulSet.DeepCopyObject())
+	statefulSet.Spec.Replicas = pointer.Int32Ptr(size)
+
+	// patch the StatefulSet
+	if err := c.client.Patch(ctx, statefulSet, statefulSetPatch, client.FieldOwner(c.parentController.GetUID())); err != nil {
+		c.recorder.Event(c.parentController, event.Warning(event.Reason(fmt.Sprintf(
+			"Failed to update the StatefulSet %s to the correct target %d", statefulSet.GetName(), size)), err))
+		c.rolloutStatus.RolloutRetry(err.Error())
+		return err
+	}
+
+	klog.InfoS("Submitted upgrade quest for StatefulSet", "StatefulSet",
+		statefulSet.GetName(), "target replica size", size, "batch", c.rolloutStatus.CurrentBatch)
+	return nil
+}
+
+// remove the parent controller from the StatefulSet's owner list
+func (c *statefulSetController) releaseStatefulSet(ctx context.Context, statefulSet *apps.StatefulSet) (bool, error) {
+	statefulSetPatch := client.MergeFrom(statefulSet.DeepCopyObject())
+
+	var newOwnerList []metav1.OwnerReference
+	found := false
+	for _, owner := range statefulSet.GetOwnerReferences() {
+		if owner.Kind == v1beta1.AppRolloutKind && owner.APIVersion == v1beta1.SchemeGroupVersion.String() {
+			found = true
+			continue
+		}
+		newOwnerList = append(newOwnerList, owner)
+	}
+	if !found {
+		klog.InfoS("the StatefulSet is already released", "StatefulSet", statefulSet.Name)
+		return true, nil
+	}
+	statefulSet.SetOwnerReferences(newOwnerList)
+
+	// patch the StatefulSet
+	if err := c.client.Patch(ctx, statefulSet, statefulSetPatch, client.FieldOwner(c.parentController.GetUID())); err != nil {
+		c.recorder.Event(c.parentController, event.Warning("Failed to the release the StatefulSet", err))
+		c.rolloutStatus.RolloutRetry(err.Error())
+		return false, err
+	}
+	return false, nil
+}

pkg/controller/common/rollout/workloads/statefulset_scale_controller.go

@@ -0,0 +1,284 @@
+/*
+Copyright 2021 The KubeVela 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 workloads
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/crossplane/crossplane-runtime/pkg/event"
+	appsv1 "k8s.io/api/apps/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/oam-dev/kubevela/apis/standard.oam.dev/v1alpha1"
+	"github.com/oam-dev/kubevela/pkg/oam"
+	"github.com/oam-dev/kubevela/pkg/oam/util"
+)
+
+// StatefulSetScaleController is responsible for handle scale StatefulSet type of workloads
+type StatefulSetScaleController struct {
+	statefulSetController
+	statefulSet *appsv1.StatefulSet
+}
+
+// NewStatefulSetScaleController creates StatefulSet scale controller
+func NewStatefulSetScaleController(client client.Client, recorder event.Recorder, parentController oam.Object, rolloutSpec *v1alpha1.RolloutPlan, rolloutStatus *v1alpha1.RolloutStatus, workloadName types.NamespacedName) *StatefulSetScaleController {
+	return &StatefulSetScaleController{
+		statefulSetController: statefulSetController{
+			workloadController: workloadController{
+				client:           client,
+				recorder:         recorder,
+				parentController: parentController,
+				rolloutSpec:      rolloutSpec,
+				rolloutStatus:    rolloutStatus,
+			},
+			targetNamespacedName: workloadName,
+		},
+	}
+}
+
+// VerifySpec verifies that the StatefulSet is stable and can be scaled
+func (s *StatefulSetScaleController) VerifySpec(ctx context.Context) (bool, error) {
+	var verifyErr error
+	defer func() {
+		if verifyErr != nil {
+			klog.Error(verifyErr)
+			s.recorder.Event(s.parentController, event.Warning("VerifyFailed", verifyErr))
+		}
+	}()
+
+	// the rollout has to have a target size in the scale case
+	if s.rolloutSpec.TargetSize == nil {
+		return false, fmt.Errorf("the rollout plan is attempting to scale the StatefulSet %s without a target",
+			s.targetNamespacedName.Name)
+	}
+	s.rolloutStatus.RolloutTargetSize = *s.rolloutSpec.TargetSize
+	klog.InfoS("record the target size", "target size", *s.rolloutSpec.TargetSize)
+
+	// fetch the StatefulSet and get its current size
+	originalSize, verifyErr := s.size(ctx)
+	if verifyErr != nil {
+		s.rolloutStatus.RolloutRetry(verifyErr.Error())
+		// nolint: nilerr
+		return false, nil
+	}
+	s.rolloutStatus.RolloutOriginalSize = originalSize
+	klog.InfoS("record the original size", "original size", originalSize)
+
+	// check if the rollout batch replicas scale up/down to the replicas target
+	if verifyErr = verifyBatchesWithScale(s.rolloutSpec, int(originalSize),
+		int(s.rolloutStatus.RolloutTargetSize)); verifyErr != nil {
+		return false, verifyErr
+	}
+
+	// check if the StatefulSet is scaling
+	if s.statefulSet.Status.Replicas != originalSize {
+		verifyErr = fmt.Errorf("the StatefulSet %s is in the middle of scaling, target size = %d, real size = %d",
+			s.statefulSet.GetName(), originalSize, s.statefulSet.Status.Replicas)
+		s.rolloutStatus.RolloutRetry(verifyErr.Error())
+		return false, nil
+	}
+
+	// check if the StatefulSet is upgrading
+	if s.statefulSet.Status.UpdatedReplicas != originalSize {
+		verifyErr = fmt.Errorf("the StatefulSet %s is in the middle of updating, target size = %d, updated pod = %d",
+			s.statefulSet.GetName(), originalSize, s.statefulSet.Status.UpdatedReplicas)
+		s.rolloutStatus.RolloutRetry(verifyErr.Error())
+		return false, nil
+	}
+
+	// check if the StatefulSet has any controller
+	if controller := metav1.GetControllerOf(s.statefulSet); controller != nil {
+		return false, fmt.Errorf("the statefulSet %s has a controller owner %s",
+			s.statefulSet.GetName(), controller.String())
+	}
+
+	// mark the scale verified
+	s.recorder.Event(s.parentController, event.Normal("Scale Verified",
+		"Rollout spec and the StatefulSet resource are verified"))
+	return true, nil
+}
+
+// Initialize makes sure that the StatefulSet is under our control
+func (s *StatefulSetScaleController) Initialize(ctx context.Context) (bool, error) {
+	if err := s.fetchStatefulSet(ctx); err != nil {
+		s.rolloutStatus.RolloutRetry(err.Error())
+		// nolint: nilerr
+		return false, nil
+	}
+
+	claimedBefore, err := s.claimStatefulSet(ctx, s.statefulSet, nil)
+	if err != nil {
+		// nolint:nilerr
+		return false, nil
+	}
+	if !claimedBefore {
+		// mark the rollout initialized
+		s.recorder.Event(s.parentController, event.Normal("Scale Initialized", "StatefulSet is initialized"))
+	}
+	return true, nil
+}
+
+// RolloutOneBatchPods calculates the number of pods we can scale to according to the rollout spec
+func (s *StatefulSetScaleController) RolloutOneBatchPods(ctx context.Context) (bool, error) {
+	if err := s.fetchStatefulSet(ctx); err != nil {
+		s.rolloutStatus.RolloutRetry(err.Error())
+		// nolint: nilerr
+		return false, nil
+	}
+
+	// set the replica according to the batch
+	newPodTarget := calculateNewBatchTarget(s.rolloutSpec, int(s.rolloutStatus.RolloutOriginalSize),
+		int(s.rolloutStatus.RolloutTargetSize), int(s.rolloutStatus.CurrentBatch))
+
+	if err := s.scaleStatefulSet(ctx, s.statefulSet, int32(newPodTarget)); err != nil {
+		// nolint:nilerr
+		return false, nil
+	}
+
+	// record the scale
+	klog.InfoS("scale one batch", "current batch", s.rolloutStatus.CurrentBatch)
+	s.recorder.Event(s.parentController, event.Normal("Batch Rollout",
+		fmt.Sprintf("Submitted scale quest for batch %d", s.rolloutStatus.CurrentBatch)))
+	s.rolloutStatus.UpgradedReplicas = int32(newPodTarget)
+	return true, nil
+}
+
+// CheckOneBatchPods checks to see if the pods are scaled according to the rollout plan
+func (s *StatefulSetScaleController) CheckOneBatchPods(ctx context.Context) (bool, error) {
+	if err := s.fetchStatefulSet(ctx); err != nil {
+		s.rolloutStatus.RolloutRetry(err.Error())
+		// nolint:nilerr
+		return false, nil
+	}
+
+	newPodTarget := calculateNewBatchTarget(s.rolloutSpec, int(s.rolloutStatus.RolloutOriginalSize),
+		int(s.rolloutStatus.RolloutTargetSize), int(s.rolloutStatus.CurrentBatch))
+	readyPodCount := int(s.statefulSet.Status.ReadyReplicas)
+	currentBatch := s.rolloutSpec.RolloutBatches[s.rolloutStatus.CurrentBatch]
+	unavail := 0
+	if currentBatch.MaxUnavailable != nil {
+		unavail, _ = intstr.GetValueFromIntOrPercent(currentBatch.MaxUnavailable,
+			util.Abs(int(s.rolloutStatus.RolloutTargetSize-s.rolloutStatus.RolloutOriginalSize)), true)
+	}
+	klog.InfoS("checking the scaling progress", "current batch", s.rolloutStatus.CurrentBatch,
+		"new pod count target", newPodTarget, "new ready pod count", readyPodCount,
+		"max unavailable pod allowed", unavail)
+	s.rolloutStatus.UpgradedReadyReplicas = int32(readyPodCount)
+	isScaleDown := s.rolloutStatus.RolloutTargetSize < s.rolloutStatus.RolloutOriginalSize
+	targetReached := (isScaleDown && readyPodCount <= newPodTarget) || (!isScaleDown && unavail+readyPodCount >= newPodTarget)
+
+	if targetReached {
+		// record the successful upgrade
+		klog.InfoS("the current batch is ready", "current batch", s.rolloutStatus.CurrentBatch,
+			"target", newPodTarget, "readyPodCount", readyPodCount, "max unavailable allowed", unavail)
+		s.recorder.Event(s.parentController, event.Normal("Batch Available",
+			fmt.Sprintf("Batch %d is available", s.rolloutStatus.CurrentBatch)))
+		return true, nil
+	}
+
+	// continue to verify
+	klog.InfoS("the batch is not ready yet", "current batch", s.rolloutStatus.CurrentBatch,
+		"target", newPodTarget, "readyPodCount", readyPodCount, "max unavailable allowed", unavail)
+	s.rolloutStatus.RolloutRetry("the batch is not ready yet")
+	return false, nil
+}
+
+// FinalizeOneBatch makes sure that the current batch and replica count in the status are validate
+func (s *StatefulSetScaleController) FinalizeOneBatch(ctx context.Context) (bool, error) {
+	if s.rolloutSpec.BatchPartition != nil && s.rolloutStatus.CurrentBatch > *s.rolloutSpec.BatchPartition {
+		err := fmt.Errorf("the current batch value in the status is greater than the batch partition")
+		klog.ErrorS(err, "we have moved past the user defined partition", "user specified batch partition",
+			*s.rolloutSpec.BatchPartition, "current batch we are working on", s.rolloutStatus.CurrentBatch)
+		return false, err
+	}
+
+	if s.rolloutStatus.RolloutOriginalSize == s.rolloutStatus.RolloutTargetSize {
+		return true, nil
+	}
+
+	finishedPodCount := int(s.rolloutStatus.UpgradedReplicas)
+	currentBatch := int(s.rolloutStatus.CurrentBatch)
+
+	// calculate the pod target just before the current batch
+	preBatchTarget := calculateNewBatchTarget(s.rolloutSpec, int(s.rolloutStatus.RolloutOriginalSize),
+		int(s.rolloutStatus.RolloutTargetSize), currentBatch-1)
+	// calculate the pod target with the current batch
+	curBatchTarget := calculateNewBatchTarget(s.rolloutSpec, int(s.rolloutStatus.RolloutOriginalSize),
+		int(s.rolloutStatus.RolloutTargetSize), currentBatch)
+
+	if finishedPodCount < util.Min(preBatchTarget, curBatchTarget) {
+		err := fmt.Errorf("the upgraded replica in the status is less than the lower bound")
+		klog.ErrorS(err, "rollout status inconsistent", "existing pod target", finishedPodCount,
+			"the lower bound", util.Min(preBatchTarget, curBatchTarget))
+		return false, err
+	}
+
+	if finishedPodCount > util.Max(preBatchTarget, curBatchTarget) {
+		err := fmt.Errorf("the upgraded replica in the status is greater than the upper bound")
+		klog.ErrorS(err, "rollout status inconsistent", "existing pod target", finishedPodCount,
+			"the upper bound", util.Max(preBatchTarget, curBatchTarget))
+		return false, err
+	}
+
+	return true, nil
+}
+
+// Finalize makes sure the StatefulSet is scaled and ready to use
+func (s *StatefulSetScaleController) Finalize(ctx context.Context, succeed bool) bool {
+	if err := s.fetchStatefulSet(ctx); err != nil {
+		s.rolloutStatus.RolloutRetry(err.Error())
+		return false
+	}
+
+	releasedBefore, err := s.releaseStatefulSet(ctx, s.statefulSet)
+	if err != nil {
+		return false
+	}
+	if !releasedBefore {
+		// mark the resource finalized
+		s.recorder.Event(s.parentController, event.Normal("Scale Finalized",
+			fmt.Sprintf("Scale resource are finalized, succeed := %t", succeed)))
+	}
+	return true
+}
+
+func (s *StatefulSetScaleController) size(ctx context.Context) (int32, error) {
+	if s.statefulSet == nil {
+		if err := s.fetchStatefulSet(ctx); err != nil {
+			return 0, err
+		}
+	}
+	return getStatefulSetReplicas(s.statefulSet), nil
+}
+
+func (s *StatefulSetScaleController) fetchStatefulSet(ctx context.Context) error {
+	workload := appsv1.StatefulSet{}
+	if err := s.client.Get(ctx, s.targetNamespacedName, &workload); err != nil {
+		if !apierrors.IsNotFound(err) {
+			s.recorder.Event(s.parentController, event.Warning("Failed to get the StatefulSet", err))
+		}
+		return err
+	}
+	s.statefulSet = &workload
+	return nil
+}

pkg/controller/common/rollout/workloads/statefulset_scale_integration_test.go

@@ -0,0 +1,499 @@
+/*
+Copyright 2021 The KubeVela 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 workloads
+
+import (
+	"github.com/crossplane/crossplane-runtime/pkg/event"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
+	"github.com/oam-dev/kubevela/apis/standard.oam.dev/v1alpha1"
+	"github.com/oam-dev/kubevela/pkg/oam/util"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/utils/pointer"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+var _ = Describe("StatefulSet controller", func() {
+	var (
+		s              StatefulSetScaleController
+		ns             corev1.Namespace
+		name           string
+		namespace      string
+		statefulSet    appsv1.StatefulSet
+		namespacedName client.ObjectKey
+	)
+
+	BeforeEach(func() {
+		namespace = "rollout-ns"
+		name = "rollout1"
+		appRollout := v1beta1.AppRollout{ObjectMeta: metav1.ObjectMeta{Name: name}}
+		namespacedName = client.ObjectKey{Name: name, Namespace: namespace}
+
+		s = StatefulSetScaleController{
+			statefulSetController: statefulSetController{
+				workloadController: workloadController{
+					client: k8sClient,
+					rolloutSpec: &v1alpha1.RolloutPlan{
+						TargetSize: pointer.Int32Ptr(10),
+						RolloutBatches: []v1alpha1.RolloutBatch{
+							{
+								Replicas: intstr.FromInt(1),
+							},
+							{
+								Replicas: intstr.FromString("20%"),
+							},
+							{
+								Replicas: intstr.FromString("80%"),
+							},
+						},
+					},
+					rolloutStatus:    &v1alpha1.RolloutStatus{RollingState: v1alpha1.RolloutSucceedState},
+					parentController: &appRollout,
+					recorder: event.NewAPIRecorder(mgr.GetEventRecorderFor("AppRollout")).
+						WithAnnotations("controller", "AppRollout"),
+				},
+				targetNamespacedName: namespacedName,
+			},
+		}
+
+		statefulSet = appsv1.StatefulSet{
+			TypeMeta:   metav1.TypeMeta{APIVersion: appsv1.SchemeGroupVersion.String(), Kind: "StatefulSet"},
+			ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name},
+			Spec: appsv1.StatefulSetSpec{
+				Replicas: pointer.Int32Ptr(1),
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{"env": "staging"},
+				},
+				Template: corev1.PodTemplateSpec{
+					ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"env": "staging"}},
+					Spec:       corev1.PodSpec{Containers: []corev1.Container{{Name: name, Image: "nginx"}}},
+				},
+			},
+		}
+
+		ns = corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: namespace,
+			},
+		}
+		By("Create a namespace")
+		Expect(k8sClient.Create(ctx, &ns)).Should(SatisfyAny(Succeed(), &util.AlreadyExistMatcher{}))
+	})
+
+	AfterEach(func() {
+		By("clean up")
+		k8sClient.Delete(ctx, &statefulSet)
+	})
+
+	Context("TestNewStatefulSetScaleController", func() {
+		It("init a StatefulSet Scale Controller", func() {
+			recorder := event.NewAPIRecorder(mgr.GetEventRecorderFor("AppRollout")).
+				WithAnnotations("controller", "AppRollout")
+			parentController := &v1beta1.AppRollout{ObjectMeta: metav1.ObjectMeta{Name: name}}
+			rolloutSpec := &v1alpha1.RolloutPlan{
+				RolloutBatches: []v1alpha1.RolloutBatch{{
+					Replicas: intstr.FromInt(1),
+				}},
+			}
+			rolloutStatus := &v1alpha1.RolloutStatus{RollingState: v1alpha1.RolloutSucceedState}
+			workloadNamespacedName := client.ObjectKey{Name: name, Namespace: namespace}
+			got := NewStatefulSetScaleController(k8sClient, recorder, parentController, rolloutSpec, rolloutStatus, workloadNamespacedName)
+			controller := &StatefulSetScaleController{
+				statefulSetController: statefulSetController{
+					workloadController: workloadController{
+						client:           k8sClient,
+						recorder:         recorder,
+						parentController: parentController,
+						rolloutSpec:      rolloutSpec,
+						rolloutStatus:    rolloutStatus,
+					},
+					targetNamespacedName: workloadNamespacedName,
+				},
+			}
+			Expect(got).Should(Equal(controller))
+		})
+	})
+
+	Context("TestVerifySpec", func() {
+		It("rollout need a target size", func() {
+			s.rolloutSpec.TargetSize = nil
+			ligit, err := s.VerifySpec(ctx)
+			Expect(ligit).Should(BeFalse())
+			Expect(err.Error()).Should(ContainSubstring("without a target"))
+		})
+
+		It("could not fetch StatefulSet workload", func() {
+			ligit, err := s.VerifySpec(ctx)
+			Expect(ligit).Should(BeFalse())
+			Expect(err).Should(BeNil())
+		})
+
+		It("rollout batch doesn't fit scale target", func() {
+			By("Create a StatefulSet")
+			statefulSet.Spec.Replicas = pointer.Int32Ptr(15)
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("Verify should fail as the scale batches don't match")
+			s.rolloutSpec.RolloutBatches[2].Replicas = intstr.FromInt(10)
+			consistent, err := s.VerifySpec(ctx)
+			Expect(consistent).Should(BeFalse())
+			Expect(err).ShouldNot(BeNil())
+		})
+
+		It("the StatefulSet is in the middle of scaling", func() {
+			By("Create a StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("verify should fail because replica does not match")
+			consistent, err := s.VerifySpec(ctx)
+			Expect(consistent).Should(BeFalse())
+			Expect(err).Should(BeNil())
+		})
+
+		It("the StatefulSet is in the middle of updating", func() {
+			By("Create a StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+			By("Update the StatefulSet status")
+			statefulSet.Status.Replicas = 1
+			Expect(k8sClient.Status().Update(ctx, &statefulSet)).Should(Succeed())
+
+			By("verify should fail because replica are not upgraded")
+			consistent, err := s.VerifySpec(ctx)
+			Expect(consistent).Should(BeFalse())
+			Expect(err).Should(BeNil())
+		})
+
+		It("spec is valid", func() {
+			By("Create a StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+			By("Update the StatefulSet status")
+			statefulSet.Status.Replicas = 1
+			statefulSet.Status.UpdatedReplicas = 1
+			statefulSet.Status.ReadyReplicas = 1
+			Expect(k8sClient.Status().Update(ctx, &statefulSet)).Should(Succeed())
+
+			By("verify should pass and record the size")
+			consistent, err := s.VerifySpec(ctx)
+			Expect(consistent).Should(BeTrue())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.RolloutTargetSize).Should(BeEquivalentTo(10))
+			Expect(s.rolloutStatus.RolloutOriginalSize).Should(BeEquivalentTo(1))
+		})
+	})
+
+	Context("TestInitialize", func() {
+		It("could not fetch StatefulSet workload", func() {
+			consistent, err := s.Initialize(ctx)
+			Expect(consistent).Should(BeFalse())
+			Expect(err).Should(BeNil())
+		})
+
+		It("failed to patch the owner of StatefulSet", func() {
+			By("Create a StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("initialize will fail because StatefulSet has wrong owner reference")
+			initialized, err := s.Initialize(ctx)
+			Expect(initialized).Should(BeFalse())
+			Expect(err).Should(BeNil())
+		})
+
+		It("workload StatefulSet is controlled by appRollout already", func() {
+			By("Create a StatefulSet")
+			statefulSet.SetOwnerReferences([]metav1.OwnerReference{{
+				APIVersion: v1beta1.SchemeGroupVersion.String(),
+				Kind:       v1beta1.AppRolloutKind,
+				Name:       "def",
+				UID:        "123456",
+				Controller: pointer.BoolPtr(true),
+			}})
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("initialize succeed without patching")
+			initialized, err := s.Initialize(ctx)
+			Expect(initialized).Should(BeTrue())
+			Expect(err).Should(BeNil())
+		})
+
+		It("successfully initialized StatefulSet", func() {
+			By("Create a StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("initialize succeeds")
+			s.parentController.SetUID("1231586900")
+			initialized, err := s.Initialize(ctx)
+			Expect(initialized).Should(BeTrue())
+			Expect(err).Should(BeNil())
+		})
+	})
+
+	Context("TestRolloutOneBatchPods", func() {
+		It("could not fetch StatefulSet workload", func() {
+			consistent, err := s.RolloutOneBatchPods(ctx)
+			Expect(consistent).Should(BeFalse())
+			Expect(err).Should(BeNil())
+		})
+
+		It("successfully rollout, current batch number is not equal to the expected one", func() {
+			By("Create a StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("rollout the second batch of current StatefulSet")
+			s.rolloutStatus.CurrentBatch = 1
+			s.rolloutStatus.RolloutOriginalSize = 0
+			s.rolloutStatus.RolloutTargetSize = 10
+			done, err := s.RolloutOneBatchPods(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.UpgradedReplicas).Should(BeEquivalentTo(3))
+			Expect(k8sClient.Get(ctx, s.targetNamespacedName, &statefulSet)).Should(Succeed())
+			Expect(*statefulSet.Spec.Replicas).Should(BeEquivalentTo(3))
+		})
+	})
+
+	Context("TestCheckOneBatchPods", func() {
+		It("could not fetch StatefulSet workload", func() {
+			consistent, err := s.CheckOneBatchPods(ctx)
+			Expect(consistent).Should(BeFalse())
+			Expect(err).Should(BeNil())
+		})
+
+		It("current ready Pods are less than expected during scale-up", func() {
+			By("Create the StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+			By("Update the StatefulSet status")
+			statefulSet.Status.Replicas = 3
+			statefulSet.Status.ReadyReplicas = 3
+			Expect(k8sClient.Status().Update(ctx, &statefulSet)).Should(Succeed())
+
+			By("checking should fail as not enough pod ready")
+			s.rolloutStatus.CurrentBatch = 1
+			s.rolloutStatus.RolloutOriginalSize = 2
+			s.rolloutStatus.RolloutTargetSize = 10
+			done, err := s.CheckOneBatchPods(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.UpgradedReadyReplicas).Should(BeEquivalentTo(statefulSet.Status.ReadyReplicas))
+
+			perc := intstr.FromString("20%")
+			s.rolloutSpec.RolloutBatches[1] = v1alpha1.RolloutBatch{
+				Replicas:       perc,
+				MaxUnavailable: &perc,
+			}
+			By("checking one batch should succeed with unavailble allowed")
+			done, err = s.CheckOneBatchPods(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.UpgradedReadyReplicas).Should(BeEquivalentTo(statefulSet.Status.ReadyReplicas))
+		})
+
+		It("current ready Pods are more than expected during scale-down", func() {
+			By("Create the StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+			By("Update the StatefulSet status")
+			statefulSet.Status.Replicas = 10
+			statefulSet.Status.ReadyReplicas = 10
+			Expect(k8sClient.Status().Update(ctx, &statefulSet)).Should(Succeed())
+
+			By("checking should fail as not enough pod ready")
+			s.rolloutStatus.CurrentBatch = 1
+			s.rolloutStatus.RolloutOriginalSize = 12
+			s.rolloutStatus.RolloutTargetSize = 5
+			done, err := s.CheckOneBatchPods(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.UpgradedReadyReplicas).Should(BeEquivalentTo(statefulSet.Status.ReadyReplicas))
+
+			perc := intstr.FromString("20%")
+			s.rolloutSpec.RolloutBatches[1] = v1alpha1.RolloutBatch{
+				Replicas:       perc,
+				MaxUnavailable: &perc,
+			}
+			By("checking one batch should still fail even with unavailble allowed")
+			done, err = s.CheckOneBatchPods(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.UpgradedReadyReplicas).Should(BeEquivalentTo(statefulSet.Status.ReadyReplicas))
+		})
+
+		It("there are more pods increased during scale-up", func() {
+			By("Create the StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+			By("Update the StatefulSet status")
+			statefulSet.Status.Replicas = 9
+			statefulSet.Status.ReadyReplicas = 9
+			Expect(k8sClient.Status().Update(ctx, &statefulSet)).Should(Succeed())
+
+			By("checking should pass when ready Pods are more than expected during scale-up")
+			s.rolloutStatus.CurrentBatch = 1
+			s.rolloutStatus.RolloutOriginalSize = 2
+			s.rolloutStatus.RolloutTargetSize = 10
+			done, err := s.CheckOneBatchPods(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.UpgradedReadyReplicas).Should(BeEquivalentTo(statefulSet.Status.ReadyReplicas))
+		})
+
+		It("there are more pods decreased during scale-down", func() {
+			By("Create the StatefulSet")
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+			By("Update the StatefulSet status")
+			statefulSet.Status.Replicas = 6
+			statefulSet.Status.ReadyReplicas = 6
+			Expect(k8sClient.Status().Update(ctx, &statefulSet)).Should(Succeed())
+
+			By("checking should pass when ready Pods are less than expected during scale-down")
+			s.rolloutStatus.CurrentBatch = 1
+			s.rolloutStatus.RolloutOriginalSize = 12
+			s.rolloutStatus.RolloutTargetSize = 5
+			done, err := s.CheckOneBatchPods(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+			Expect(s.rolloutStatus.UpgradedReadyReplicas).Should(BeEquivalentTo(statefulSet.Status.ReadyReplicas))
+		})
+	})
+
+	Context("TestFinalizeOneBatch", func() {
+		BeforeEach(func() {
+			s.rolloutSpec.RolloutBatches[0] = v1alpha1.RolloutBatch{
+				Replicas: intstr.FromInt(2),
+			}
+		})
+
+		It("test illegal batch partition", func() {
+			By("finalizing one batch")
+			s.rolloutSpec.BatchPartition = pointer.Int32Ptr(2)
+			s.rolloutStatus.CurrentBatch = 3
+			done, err := s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err.Error()).Should(ContainSubstring("the current batch value in the status is greater than the batch partition"))
+		})
+
+		It("test finalize during scale-up", func() {
+			By("finalizing one batch when there're too few upgraded Pods")
+			s.rolloutStatus.UpgradedReplicas = 6
+			s.rolloutStatus.CurrentBatch = 1
+			s.rolloutStatus.RolloutOriginalSize = 5
+			s.rolloutStatus.RolloutTargetSize = 12
+			done, err := s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err.Error()).Should(ContainSubstring("upgraded replica in the status is less than the lower bound"))
+
+			By("finalizing one batch with enough upgraded Pods (lower bound)")
+			s.rolloutStatus.UpgradedReplicas = 7
+			done, err = s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+
+			By("finalizing one batch with enough upgraded Pods (upper bound)")
+			s.rolloutStatus.UpgradedReplicas = 9
+			done, err = s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+
+			By("finalizing one batch when there're too many upgraded Pods")
+			s.rolloutStatus.UpgradedReplicas = 10
+			done, err = s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err.Error()).Should(ContainSubstring("upgraded replica in the status is greater than the upper bound"))
+		})
+
+		It("test finalize during scale-down", func() {
+			By("finalizing one batch when there're too many upgraded Pods")
+			s.rolloutStatus.UpgradedReplicas = 13
+			s.rolloutStatus.CurrentBatch = 1
+			s.rolloutStatus.RolloutOriginalSize = 14
+			s.rolloutStatus.RolloutTargetSize = 2
+			done, err := s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err.Error()).Should(ContainSubstring("upgraded replica in the status is greater than the upper bound"))
+
+			By("finalizing one batch with enough upgraded Pods (upper bound)")
+			s.rolloutStatus.UpgradedReplicas = 12
+			done, err = s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+
+			By("finalizing one batch with enough upgraded Pods (lower bound)")
+			s.rolloutStatus.UpgradedReplicas = 9
+			done, err = s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeTrue())
+			Expect(err).Should(BeNil())
+
+			By("finalizing one batch when there're too few upgraded Pods")
+			s.rolloutStatus.UpgradedReplicas = 8
+			done, err = s.FinalizeOneBatch(ctx)
+			Expect(done).Should(BeFalse())
+			Expect(err.Error()).Should(ContainSubstring("upgraded replica in the status is less than the lower bound"))
+		})
+	})
+
+	Context("TestFinalize", func() {
+		It("failed to fetch StatefulSet workload", func() {
+			By("finalizing")
+			finalized := s.Finalize(ctx, true)
+			Expect(finalized).Should(BeFalse())
+		})
+
+		It("Already finalize StatefulSet", func() {
+			By("Create a StatefulSet")
+			statefulSet.SetOwnerReferences([]metav1.OwnerReference{{
+				APIVersion: v1beta1.SchemeGroupVersion.String(),
+				Kind:       "notRollout",
+				Name:       "def",
+				UID:        "123456",
+			}})
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("finalizing without patch")
+			finalized := s.Finalize(ctx, true)
+			Expect(finalized).Should(BeTrue())
+		})
+
+		It("successfully to finalize StatefulSet", func() {
+			By("Create a StatefulSet")
+			statefulSet.SetOwnerReferences([]metav1.OwnerReference{
+				{
+					APIVersion: v1beta1.SchemeGroupVersion.String(),
+					Kind:       v1beta1.AppRolloutKind,
+					Name:       "def",
+					UID:        "123456",
+				},
+				{
+					APIVersion: corev1.SchemeGroupVersion.String(),
+					Kind:       "StatefulSet",
+					Name:       "def",
+					UID:        "654321",
+				},
+			})
+			Expect(k8sClient.Create(ctx, &statefulSet)).Should(Succeed())
+
+			By("finalizing with patch")
+			finalized := s.Finalize(ctx, false)
+			Expect(finalized).Should(BeTrue())
+			Expect(k8sClient.Get(ctx, s.targetNamespacedName, &statefulSet)).Should(Succeed())
+			Expect(len(statefulSet.GetOwnerReferences())).Should(BeEquivalentTo(1))
+			Expect(statefulSet.GetOwnerReferences()[0].Kind).Should(Equal("StatefulSet"))
+		})
+	})
+})