update

Author: leon-inf

apis/workloads/v1/instanceset_types.go

@@ -533,20 +533,20 @@ type InstanceStatus struct {
 	// +optional
 	Role string `json:"role,omitempty"`
 
-	// Represents whether the instance is in volume expansion.
-	//
-	// +optional
-	VolumeExpansion bool `json:"volumeExpansion,omitempty"`
-
 	// The status of configs.
 	//
 	// +optional
 	Configs []InstanceConfigStatus `json:"configs,omitempty"`
 
-	// Joined indicates whether the instance is joined.
+	// Represents whether the instance is joined the cluster.
 	//
 	// +optional
 	Joined *bool `json:"joined,omitempty"`
+
+	// Represents whether the instance is in volume expansion.
+	//
+	// +optional
+	VolumeExpansion bool `json:"volumeExpansion,omitempty"`
 }
 
 type InstanceConfigStatus struct {

config/crd/bases/workloads.kubeblocks.io_instancesets.yaml

@@ -13439,7 +13439,7 @@ spec:
                         type: object
                       type: array
                     joined:
-                      description: Joined indicates whether the instance is joined.
+                      description: Represents whether the instance is joined the cluster.
                       type: boolean
                     podName:
                       default: Unknown

controllers/apps/component/transformer_component_workload.go

@@ -197,9 +197,6 @@ func (t *componentWorkloadTransformer) handleUpdate(transCtx *componentTransform
 		})
 	}
 
-	// if start {
-	//	return intctrlutil.NewDelayedRequeueError(time.Second, "workload is starting")
-	// }
 	return nil
 }
 

controllers/apps/component/transformer_component_workload_ops.go

@@ -21,7 +21,6 @@ package component
 
 import (
 	"crypto/sha256"
-	"errors"
 	"fmt"
 	"path/filepath"
 	"reflect"
@@ -42,7 +41,6 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/graph"
 	"github.com/apecloud/kubeblocks/pkg/controller/lifecycle"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
-	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
 type componentWorkloadOps struct {
@@ -95,162 +93,19 @@ func (r *componentWorkloadOps) horizontalScale() error {
 		in  = r.runningItsPodNameSet.Difference(r.desiredCompPodNameSet)
 		out = r.desiredCompPodNameSet.Difference(r.runningItsPodNameSet)
 	)
-	if in.Len() == 0 && out.Len() == 0 {
-		return r.postHorizontalScale() // TODO: how about consecutive horizontal scales?
-	}
-
-	if in.Len() > 0 {
-		if err := r.scaleIn(); err != nil {
-			return err
-		}
-	}
-
-	if out.Len() > 0 {
-		if err := r.scaleOut(); err != nil {
-			return err
-		}
-	}
-
-	r.transCtx.EventRecorder.Eventf(r.component,
-		corev1.EventTypeNormal,
-		"HorizontalScale",
-		"start horizontal scale component %s of cluster %s from %d to %d",
-		r.synthesizeComp.Name, r.synthesizeComp.ClusterName, int(*r.runningITS.Spec.Replicas), r.synthesizeComp.Replicas)
-
-	return nil
-}
-
-func (r *componentWorkloadOps) scaleIn() error {
-	if r.synthesizeComp.Replicas == 0 && len(r.synthesizeComp.VolumeClaimTemplates) > 0 {
-		if r.synthesizeComp.PVCRetentionPolicy.WhenScaled != appsv1.RetainPersistentVolumeClaimRetentionPolicyType {
-			return fmt.Errorf("when intending to scale-in to 0, only the \"Retain\" option is supported for the PVC retention policy")
-		}
-	}
-
-	deleteReplicas := r.runningItsPodNameSet.Difference(r.desiredCompPodNameSet).UnsortedList()
-	joinedReplicas := make([]string, 0)
-	err := component.DeleteReplicasStatus(r.protoITS, deleteReplicas, func(s component.ReplicaStatus) {
-		// has no member join defined or has joined successfully
-		if s.Provisioned && (s.MemberJoined == nil || *s.MemberJoined) {
-			joinedReplicas = append(joinedReplicas, s.Name)
-		}
-	})
-	if err != nil {
-		return err
-	}
-
-	// TODO: check the component definition to determine whether we need to call leave member before deleting replicas.
-	if err := r.leaveMember4ScaleIn(deleteReplicas, joinedReplicas); err != nil {
-		r.transCtx.Logger.Error(err, "leave member at scale-in error")
-		return err
-	}
-	return nil
-}
-
-func (r *componentWorkloadOps) leaveMember4ScaleIn(deleteReplicas, joinedReplicas []string) error {
-	pods, err := component.ListOwnedPods(r.transCtx.Context, r.cli,
-		r.synthesizeComp.Namespace, r.synthesizeComp.ClusterName, r.synthesizeComp.Name)
-	if err != nil {
-		return err
-	}
-
-	deleteReplicasSet := sets.New(deleteReplicas...)
-	joinedReplicasSet := sets.New(joinedReplicas...)
-	hasMemberLeaveDefined := r.synthesizeComp.LifecycleActions != nil && r.synthesizeComp.LifecycleActions.MemberLeave != nil
-	r.transCtx.Logger.Info("leave member at scaling-in", "delete replicas", deleteReplicas,
-		"joined replicas", joinedReplicas, "has member-leave action defined", hasMemberLeaveDefined)
-
-	leaveErrors := make([]error, 0)
-	for _, pod := range pods {
-		if deleteReplicasSet.Has(pod.Name) {
-			if joinedReplicasSet.Has(pod.Name) { // else: hasn't joined yet, no need to leave
-				if err = r.leaveMemberForPod(pod, pods); err != nil {
-					leaveErrors = append(leaveErrors, err)
-				}
-				joinedReplicasSet.Delete(pod.Name)
-			}
-			deleteReplicasSet.Delete(pod.Name)
-		}
-	}
-
-	if hasMemberLeaveDefined && len(joinedReplicasSet) > 0 {
-		leaveErrors = append(leaveErrors,
-			fmt.Errorf("some replicas have joined but not leaved since the Pod object is not exist: %v", sets.List(joinedReplicasSet)))
-	}
-	if len(leaveErrors) > 0 {
-		return intctrlutil.NewRequeueError(time.Second, fmt.Sprintf("%v", leaveErrors))
-	}
-	return nil
-}
-
-func (r *componentWorkloadOps) leaveMemberForPod(pod *corev1.Pod, pods []*corev1.Pod) error {
-	var (
-		synthesizedComp  = r.synthesizeComp
-		lifecycleActions = synthesizedComp.LifecycleActions
-	)
-
-	trySwitchover := func(lfa lifecycle.Lifecycle, pod *corev1.Pod) error {
-		if lifecycleActions.Switchover == nil {
-			return nil
-		}
-		err := lfa.Switchover(r.transCtx.Context, r.cli, nil, "")
-		if err != nil {
-			if errors.Is(err, lifecycle.ErrActionNotDefined) {
-				return nil
-			}
-			return err
-		}
-		r.transCtx.Logger.Info("successfully call switchover action for pod", "pod", pod.Name)
-		return nil
-	}
-
-	tryMemberLeave := func(lfa lifecycle.Lifecycle, pod *corev1.Pod) error {
-		if lifecycleActions.MemberLeave == nil {
-			return nil
-		}
-		err := lfa.MemberLeave(r.transCtx.Context, r.cli, nil)
-		if err != nil {
-			if errors.Is(err, lifecycle.ErrActionNotDefined) {
-				return nil
-			}
-			return err
-		}
-		r.transCtx.Logger.Info("successfully call leave member action for pod", "pod", pod.Name)
-		return nil
-	}
-
-	if lifecycleActions == nil || (lifecycleActions.Switchover == nil && lifecycleActions.MemberLeave == nil) {
-		return nil
-	}
-
-	lfa, err := lifecycle.New(synthesizedComp.Namespace, synthesizedComp.ClusterName, synthesizedComp.Name,
-		lifecycleActions, synthesizedComp.TemplateVars, pod, pods...)
-	if err != nil {
-		return err
-	}
-
-	if err := trySwitchover(lfa, pod); err != nil {
+	if err := r.buildDataReplicationTask(); err != nil {
 		return err
 	}
-
-	if err := tryMemberLeave(lfa, pod); err != nil {
-		return err
+	if in.Len() != 0 || out.Len() != 0 {
+		r.transCtx.EventRecorder.Eventf(r.component,
+			corev1.EventTypeNormal,
+			"HorizontalScale",
+			"start horizontal scale component %s of cluster %s from %d to %d",
+			r.synthesizeComp.Name, r.synthesizeComp.ClusterName, int(*r.runningITS.Spec.Replicas), r.synthesizeComp.Replicas)
 	}
-
 	return nil
 }
 
-func (r *componentWorkloadOps) scaleOut() error {
-	if err := r.buildDataReplicationTask(); err != nil {
-		return err
-	}
-
-	// replicas to be created
-	newReplicas := r.desiredCompPodNameSet.Difference(r.runningItsPodNameSet).UnsortedList()
-	hasMemberJoinDefined, hasDataActionDefined := hasMemberJoinNDataActionDefined(r.synthesizeComp.LifecycleActions)
-	return component.NewReplicasStatus(r.protoITS, newReplicas, hasMemberJoinDefined, hasDataActionDefined)
-}
-
 func (r *componentWorkloadOps) buildDataReplicationTask() error {
 	_, hasDataActionDefined := hasMemberJoinNDataActionDefined(r.synthesizeComp.LifecycleActions)
 	if !hasDataActionDefined {
@@ -271,7 +126,7 @@ func (r *componentWorkloadOps) buildDataReplicationTask() error {
 		return err
 	}
 
-	// the source replica
+	// choose the source replica
 	source, err := r.sourceReplica(r.synthesizeComp.LifecycleActions.DataDump, provisioningReplicas)
 	if err != nil {
 		return err
@@ -320,90 +175,6 @@ func (r *componentWorkloadOps) sourceReplica(dataDump *appsv1.Action, provisioni
 	return nil, fmt.Errorf("no available pod to dump data")
 }
 
-func (r *componentWorkloadOps) postHorizontalScale() error {
-	if err := r.postScaleOut(); err != nil {
-		return err
-	}
-	return nil
-}
-
-func (r *componentWorkloadOps) postScaleOut() error {
-	if err := r.buildDataReplicationTask(); err != nil {
-		return err
-	}
-	if err := r.joinMember4ScaleOut(); err != nil {
-		return err
-	}
-	return nil
-}
-
-func (r *componentWorkloadOps) joinMember4ScaleOut() error {
-	pods, err := component.ListOwnedPods(r.transCtx.Context, r.cli,
-		r.synthesizeComp.Namespace, r.synthesizeComp.ClusterName, r.synthesizeComp.Name)
-	if err != nil {
-		return err
-	}
-
-	joinErrors := make([]error, 0)
-	if err = component.UpdateReplicasStatusFunc(r.protoITS, func(replicas *component.ReplicasStatus) error {
-		for _, pod := range pods {
-			i := slices.IndexFunc(replicas.Status, func(r component.ReplicaStatus) bool {
-				return r.Name == pod.Name
-			})
-			if i < 0 {
-				continue // the pod is not in the replicas status?
-			}
-
-			status := replicas.Status[i]
-			if status.MemberJoined == nil || *status.MemberJoined {
-				continue // no need to join or already joined
-			}
-
-			// TODO: should wait for the data to be loaded before joining the member?
-
-			if err := r.joinMemberForPod(pod, pods); err != nil {
-				joinErrors = append(joinErrors, fmt.Errorf("pod %s: %w", pod.Name, err))
-			} else {
-				replicas.Status[i].MemberJoined = ptr.To(true)
-			}
-		}
-
-		notJoinedReplicas := make([]string, 0)
-		for _, r := range replicas.Status {
-			if r.MemberJoined != nil && !*r.MemberJoined {
-				notJoinedReplicas = append(notJoinedReplicas, r.Name)
-			}
-		}
-		if len(notJoinedReplicas) > 0 {
-			joinErrors = append(joinErrors, fmt.Errorf("some replicas have not joined: %v", notJoinedReplicas))
-		}
-		return nil
-	}); err != nil {
-		return err
-	}
-
-	if len(joinErrors) > 0 {
-		return intctrlutil.NewRequeueError(time.Second, fmt.Sprintf("%v", joinErrors))
-	}
-	return nil
-}
-
-func (r *componentWorkloadOps) joinMemberForPod(pod *corev1.Pod, pods []*corev1.Pod) error {
-	synthesizedComp := r.synthesizeComp
-	lfa, err := lifecycle.New(synthesizedComp.Namespace, synthesizedComp.ClusterName, synthesizedComp.Name,
-		synthesizedComp.LifecycleActions, synthesizedComp.TemplateVars, pod, pods...)
-	if err != nil {
-		return err
-	}
-	if err = lfa.MemberJoin(r.transCtx.Context, r.cli, nil); err != nil {
-		if !errors.Is(err, lifecycle.ErrActionNotDefined) {
-			return err
-		}
-	}
-	r.transCtx.Logger.Info("succeed to join member for pod", "pod", pod.Name)
-	return nil
-}
-
 func (r *componentWorkloadOps) reconfigure() error {
 	runningObjs, protoObjs, err := prepareFileTemplateObjects(r.transCtx)
 	if err != nil {

deploy/helm/crds/workloads.kubeblocks.io_instancesets.yaml

@@ -13439,7 +13439,7 @@ spec:
                         type: object
                       type: array
                     joined:
-                      description: Joined indicates whether the instance is joined.
+                      description: Represents whether the instance is joined the cluster.
                       type: boolean
                     podName:
                       default: Unknown

docs/developer_docs/api-reference/cluster.md

@@ -33335,40 +33335,40 @@ string
 </tr>
 <tr>
 <td>
-<code>volumeExpansion</code><br/>
+<code>configs</code><br/>
 <em>
-bool
+<a href="#workloads.kubeblocks.io/v1.InstanceConfigStatus">
+[]InstanceConfigStatus
+</a>
 </em>
 </td>
 <td>
 <em>(Optional)</em>
-<p>Represents whether the instance is in volume expansion.</p>
+<p>The status of configs.</p>
 </td>
 </tr>
 <tr>
 <td>
-<code>configs</code><br/>
+<code>joined</code><br/>
 <em>
-<a href="#workloads.kubeblocks.io/v1.InstanceConfigStatus">
-[]InstanceConfigStatus
-</a>
+bool
 </em>
 </td>
 <td>
 <em>(Optional)</em>
-<p>The status of configs.</p>
+<p>Represents whether the instance is joined the cluster.</p>
 </td>
 </tr>
 <tr>
 <td>
-<code>joined</code><br/>
+<code>volumeExpansion</code><br/>
 <em>
 bool
 </em>
 </td>
 <td>
 <em>(Optional)</em>
-<p>Joined indicates whether the instance is joined.</p>
+<p>Represents whether the instance is in volume expansion.</p>
 </td>
 </tr>
 </tbody>