refactor: clean up component h-scale orchestration

Author: leon-ape

controllers/apps/component/transformer_component_pre_terminate.go

@@ -107,13 +107,12 @@ func (t *componentPreTerminateTransformer) provisioned(transCtx *componentTransf
 		return false, client.IgnoreNotFound(err)
 	}
 
-	provisioned, err := component.GetReplicasStatusFunc(its, func(s component.ReplicaStatus) bool {
-		return s.Provisioned
-	})
-	if err != nil {
-		return false, err
+	for _, status := range its.Status.InstanceStatus {
+		if status.Provisioned {
+			return true, nil
+		}
 	}
-	return len(provisioned) > 0, nil
+	return false, nil
 }
 
 func (t *componentPreTerminateTransformer) checkPreTerminateDone(transCtx *componentTransformContext, dag *graph.DAG) bool {

controllers/apps/component/transformer_component_pre_terminate_test.go

@@ -37,7 +37,6 @@ import (
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
 	appsutil "github.com/apecloud/kubeblocks/controllers/apps/util"
 	"github.com/apecloud/kubeblocks/pkg/constant"
-	"github.com/apecloud/kubeblocks/pkg/controller/component"
 	"github.com/apecloud/kubeblocks/pkg/controller/graph"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
 	kbacli "github.com/apecloud/kubeblocks/pkg/kbagent/client"
@@ -65,10 +64,12 @@ var _ = Describe("pre-terminate transformer test", func() {
 	}
 
 	provisioned := func(its *workloads.InstanceSet) {
-		replicas := []string{
-			fmt.Sprintf("%s-0", its.Name),
+		its.Status.InstanceStatus = []workloads.InstanceStatus{
+			{
+				PodName:     fmt.Sprintf("%s-0", its.Name),
+				Provisioned: true,
+			},
 		}
-		Expect(component.StatusReplicasStatus(its, replicas, false, false)).Should(Succeed())
 	}
 
 	BeforeEach(func() {
@@ -199,12 +200,9 @@ var _ = Describe("pre-terminate transformer test", func() {
 
 		It("not provisioned", func() {
 			its := reader.Objects[1].(*workloads.InstanceSet)
-			Expect(component.UpdateReplicasStatusFunc(its, func(r *component.ReplicasStatus) error {
-				for i := range r.Status {
-					r.Status[i].Provisioned = false
-				}
-				return nil
-			})).Should(Succeed())
+			for i := range its.Status.InstanceStatus {
+				its.Status.InstanceStatus[i].Provisioned = false
+			}
 
 			transformer := &componentPreTerminateTransformer{}
 			err := transformer.Transform(transCtx, dag)

controllers/apps/component/transformer_component_status.go

@@ -225,20 +225,18 @@ func (t *componentStatusTransformer) hasScaleOutRunning(transCtx *componentTrans
 		return false, false, nil
 	}
 
-	replicas, err := component.GetReplicasStatusFunc(t.protoITS, func(status component.ReplicaStatus) bool {
-		return status.DataLoaded != nil && !*status.DataLoaded ||
-			status.MemberJoined != nil && !*status.MemberJoined
-	})
-	if err != nil {
-		return false, false, err
-	}
-	if len(replicas) == 0 {
-		return false, false, nil
+	for _, status := range t.runningITS.Status.InstanceStatus {
+		if status.DataLoaded != nil && !*status.DataLoaded {
+			return true, false, nil
+		}
+		if status.MemberJoined != nil && !*status.MemberJoined {
+			return true, false, nil
+		}
 	}
 
 	// TODO: scale-out failed
 
-	return true, false, nil
+	return false, false, nil
 }
 
 func (t *componentStatusTransformer) hasVolumeExpansionRunning() bool {

controllers/apps/component/transformer_component_workload.go

@@ -27,7 +27,6 @@ import (
 	"golang.org/x/exp/maps"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -102,10 +101,6 @@ func (t *componentWorkloadTransformer) reconcileWorkload(ctx context.Context, cl
 
 	t.buildInstanceSetPlacementAnnotation(comp, protoITS)
 
-	if err := t.reconcileReplicasStatus(ctx, cli, synthesizedComp, runningITS, protoITS); err != nil {
-		return err
-	}
-
 	return nil
 }
 
@@ -121,43 +116,6 @@ func (t *componentWorkloadTransformer) buildInstanceSetPlacementAnnotation(comp
 	}
 }
 
-func (t *componentWorkloadTransformer) reconcileReplicasStatus(ctx context.Context, cli client.Reader,
-	synthesizedComp *component.SynthesizedComponent, runningITS, protoITS *workloads.InstanceSet) error {
-	var (
-		namespace   = synthesizedComp.Namespace
-		clusterName = synthesizedComp.ClusterName
-		compName    = synthesizedComp.Name
-	)
-
-	// HACK: sync replicas status from runningITS to protoITS
-	component.BuildReplicasStatus(runningITS, protoITS)
-
-	replicas, err := func() ([]string, error) {
-		pods, err := component.ListOwnedPods(ctx, cli, namespace, clusterName, compName)
-		if err != nil {
-			return nil, err
-		}
-		podNameSet := sets.New[string]()
-		for _, pod := range pods {
-			podNameSet.Insert(pod.Name)
-		}
-
-		desiredPodNames, err := component.GetDesiredPodNamesByITS(runningITS, protoITS)
-		if err != nil {
-			return nil, err
-		}
-		desiredPodNameSet := sets.New(desiredPodNames...)
-
-		return desiredPodNameSet.Intersection(podNameSet).UnsortedList(), nil
-	}()
-	if err != nil {
-		return err
-	}
-
-	hasMemberJoinDefined, hasDataActionDefined := hasMemberJoinNDataActionDefined(synthesizedComp.LifecycleActions.ComponentLifecycleActions)
-	return component.StatusReplicasStatus(protoITS, replicas, hasMemberJoinDefined, hasDataActionDefined)
-}
-
 func (t *componentWorkloadTransformer) handleUpdate(transCtx *componentTransformContext, cli model.GraphClient, dag *graph.DAG,
 	synthesizedComp *component.SynthesizedComponent, comp *appsv1.Component, runningITS, protoITS *workloads.InstanceSet) error {
 	start, stop := t.handleWorkloadStartNStop(transCtx, synthesizedComp, runningITS, &protoITS)
@@ -206,13 +164,13 @@ func isCompStopped(synthesizedComp *component.SynthesizedComponent) bool {
 	return ptr.Deref(synthesizedComp.Stop, false)
 }
 
-func (t *componentWorkloadTransformer) handleWorkloadUpdate(transCtx *componentTransformContext, dag *graph.DAG,
-	synthesizeComp *component.SynthesizedComponent, comp *appsv1.Component, obj, its *workloads.InstanceSet) error {
-	cwo, err := newComponentWorkloadOps(transCtx, t.Client, synthesizeComp, comp, obj, its, dag)
+func (t *componentWorkloadTransformer) handleWorkloadUpdate(_ *componentTransformContext, _ *graph.DAG,
+	synthesizeComp *component.SynthesizedComponent, _ *appsv1.Component, obj, its *workloads.InstanceSet) error {
+	cwo, err := newComponentWorkloadOps(synthesizeComp, obj, its)
 	if err != nil {
 		return err
 	}
-	if err := cwo.horizontalScale(); err != nil {
+	if err := cwo.validateHorizontalScale(); err != nil {
 		return err
 	}
 	return nil