update

Author: leon-inf

apis/workloads/v1/instance_types.go

@@ -183,10 +183,25 @@ type InstanceStatus2 struct {
 	// +optional
 	Role string `json:"role,omitempty"`
 
+	// Represents whether the instance is provisioned.
+	//
+	// +optional
+	Provisioned bool `json:"provisioned,omitempty"`
+
+	// Represents whether the instance data is loaded.
+	//
+	// +optional
+	DataLoaded *bool `json:"dataLoaded,omitempty"`
+
+	// Represents whether the instance is joined the cluster.
+	//
+	// +optional
+	MemberJoined *bool `json:"memberJoined,omitempty"`
+
 	// Represents whether the instance is in volume expansion.
 	//
 	// +optional
-	VolumeExpansion bool `json:"volumeExpansion,omitempty"`
+	InVolumeExpansion bool `json:"inVolumeExpansion,omitempty"`
 }
 
 type InstanceAssistantObject struct {

apis/workloads/v1/instanceset_types.go

@@ -514,6 +514,11 @@ type LifecycleActions struct {
 	// +optional
 	MemberLeave *Action `json:"memberLeave,omitempty"`
 
+	// Defines the procedure for importing data into a replica.
+	//
+	// +optional
+	DataLoad *Action `json:"dataLoad,omitempty"`
+
 	// Defines the procedure that update a replica with new configuration.
 	//
 	// +optional
@@ -562,15 +567,25 @@ type InstanceStatus struct {
 	// +optional
 	Configs []InstanceConfigStatus `json:"configs,omitempty"`
 
+	// Represents whether the instance is provisioned.
+	//
+	// +optional
+	Provisioned bool `json:"provisioned,omitempty"`
+
+	// Represents whether the instance data is loaded.
+	//
+	// +optional
+	DataLoaded *bool `json:"dataLoaded,omitempty"`
+
 	// Represents whether the instance is joined the cluster.
 	//
 	// +optional
-	Joined *bool `json:"joined,omitempty"`
+	MemberJoined *bool `json:"memberJoined,omitempty"`
 
 	// Represents whether the instance is in volume expansion.
 	//
 	// +optional
-	VolumeExpansion bool `json:"volumeExpansion,omitempty"`
+	InVolumeExpansion bool `json:"inVolumeExpansion,omitempty"`
 }
 
 type InstanceConfigStatus struct {

apis/workloads/v1/zz_generated.deepcopy.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	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"
 
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
@@ -130,18 +131,15 @@ func (t *componentStatusTransformer) reconcileStatus(transCtx *componentTransfor
 	// check if the component has failed pod
 	hasFailedPod, messages := t.hasFailedPod()
 
-	// check if the component scale out failed
-	hasRunningScaleOut, hasFailedScaleOut, err := t.hasScaleOutRunning(transCtx)
-	if err != nil {
-		return err
-	}
+	// check if the component has scale out running
+	hasRunningScaleOut := t.hasScaleOutRunning()
 
 	// check if the volume expansion is running
 	hasRunningVolumeExpansion := t.hasVolumeExpansionRunning()
 
 	// check if the component has failure
 	hasFailure := func() bool {
-		return hasFailedPod || hasFailedScaleOut
+		return hasFailedPod
 	}()
 
 	// check if the component is in creating phase
@@ -219,35 +217,19 @@ func (t *componentStatusTransformer) isInstanceSetRunning() bool {
 	return t.runningITS.IsInstanceSetReady()
 }
 
-// hasScaleOutRunning checks if the scale out is running.
-func (t *componentStatusTransformer) hasScaleOutRunning(transCtx *componentTransformContext) (running bool, failed bool, err error) {
+func (t *componentStatusTransformer) hasScaleOutRunning() bool {
 	if t.runningITS == nil || t.runningITS.Spec.Replicas == nil {
-		return false, false, nil
+		return false
 	}
-
-	replicas, err := component.GetReplicasStatusFunc(t.protoITS, func(status component.ReplicaStatus) bool {
-		return status.DataLoaded != nil && !*status.DataLoaded ||
-			status.MemberJoined != nil && !*status.MemberJoined
+	return slices.ContainsFunc(t.runningITS.Status.InstanceStatus, func(inst workloads.InstanceStatus) bool {
+		return !ptr.Deref(inst.DataLoaded, true) || !ptr.Deref(inst.MemberJoined, true)
 	})
-	if err != nil {
-		return false, false, err
-	}
-	if len(replicas) == 0 {
-		return false, false, nil
-	}
-
-	// TODO: scale-out failed
-
-	return true, false, nil
 }
 
 func (t *componentStatusTransformer) hasVolumeExpansionRunning() bool {
-	for _, inst := range t.runningITS.Status.InstanceStatus {
-		if inst.VolumeExpansion {
-			return true
-		}
-	}
-	return false
+	return slices.ContainsFunc(t.runningITS.Status.InstanceStatus, func(inst workloads.InstanceStatus) bool {
+		return inst.InVolumeExpansion
+	})
 }
 
 // hasFailedPod checks if the instance set has failed pod.

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

@@ -50,9 +50,7 @@ type componentWorkloadOps struct {
 	synthesizeComp *component.SynthesizedComponent
 	dag            *graph.DAG
 
-	// runningITS is a snapshot of the InstanceSet that is already running
-	runningITS *workloads.InstanceSet
-	// protoITS is the InstanceSet object that is rebuilt from scratch during each reconcile process
+	runningITS            *workloads.InstanceSet
 	protoITS              *workloads.InstanceSet
 	desiredCompPodNameSet sets.Set[string]
 	runningItsPodNameSet  sets.Set[string]
@@ -93,7 +91,7 @@ func (r *componentWorkloadOps) horizontalScale() error {
 		in  = r.runningItsPodNameSet.Difference(r.desiredCompPodNameSet)
 		out = r.desiredCompPodNameSet.Difference(r.runningItsPodNameSet)
 	)
-	if err := r.buildDataReplicationTask(); err != nil {
+	if err := r.dataReplicationTask(); err != nil {
 		return err
 	}
 	if in.Len() != 0 || out.Len() != 0 {
@@ -106,7 +104,7 @@ func (r *componentWorkloadOps) horizontalScale() error {
 	return nil
 }
 
-func (r *componentWorkloadOps) buildDataReplicationTask() error {
+func (r *componentWorkloadOps) dataReplicationTask() error {
 	_, hasDataActionDefined := hasMemberJoinNDataActionDefined(r.synthesizeComp.LifecycleActions)
 	if !hasDataActionDefined {
 		return nil
@@ -119,11 +117,11 @@ func (r *componentWorkloadOps) buildDataReplicationTask() error {
 	}
 
 	// replicas in provisioning that the data has not been loaded
-	provisioningReplicas, err := component.GetReplicasStatusFunc(r.protoITS, func(s component.ReplicaStatus) bool {
-		return s.DataLoaded != nil && !*s.DataLoaded
-	})
-	if err != nil {
-		return err
+	var provisioningReplicas []string
+	for _, replica := range r.runningITS.Status.InstanceStatus {
+		if replica.DataLoaded != nil && !*replica.DataLoaded {
+			provisioningReplicas = append(provisioningReplicas, replica.PodName)
+		}
 	}
 
 	// choose the source replica

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

@@ -84,9 +84,10 @@ func (r *InstanceSetReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		Do(instanceset.NewDeletionReconciler()).
 		Do(instanceset.NewValidationReconciler()).
 		Do(instanceset.NewStatusReconciler()).
+		Do(instanceset.NewMembershipReconciler()).
 		Do(instanceset.NewRevisionUpdateReconciler()).
 		Do(instanceset.NewAssistantObjectReconciler()).
-		Do(instanceset.NewReplicasAlignmentReconciler()).
+		Do(instanceset.NewInstanceAlignmentReconciler()).
 		Do(instanceset.NewUpdateReconciler()).
 		Commit()