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

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>

pkg/controller/instanceset/reconciler_instance_alignment.go

@@ -21,6 +21,7 @@ package instanceset
 
 import (
 	"errors"
+	"slices"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -145,7 +146,7 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (
 		if err != nil {
 			return kubebuilderx.Continue, err
 		}
-		if err := tree.AddWithOption(newPod, r.joinMemberHook(tree, its, oldInstanceList, newPod)); err != nil {
+		if err := tree.AddWithOption(newPod, r.createInstance(tree, its, oldInstanceList, newPod)); err != nil {
 			return kubebuilderx.Continue, err
 		}
 		currentAlignedNameList = append(currentAlignedNameList, name)
@@ -198,7 +199,7 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (
 				its.Name,
 				pod.Name)
 		}
-		if err := tree.DeleteWithOption(pod, r.leaveMemberHook(tree, its, oldInstanceList, pod)); err != nil {
+		if err := tree.DeleteWithOption(pod, r.deleteInstance(tree, its, oldInstanceList, pod)); err != nil {
 			return kubebuilderx.Continue, err
 		}
 
@@ -224,37 +225,30 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (
 	return kubebuilderx.Continue, nil
 }
 
-func (r *instanceAlignmentReconciler) joinMemberHook(tree *kubebuilderx.ObjectTree,
-	its *workloads.InstanceSet, pods []client.Object, pod *corev1.Pod) kubebuilderx.WithHook {
+func (r *instanceAlignmentReconciler) createInstance(tree *kubebuilderx.ObjectTree,
+	its *workloads.InstanceSet, pods []client.Object, pod *corev1.Pod) kubebuilderx.WithPrevHook {
 	return func(obj client.Object) error {
-		if its.Status.InitReplicas == nil || its.Status.ReadyInitReplicas == nil ||
-			*its.Status.InitReplicas != *its.Status.ReadyInitReplicas {
-			return nil // init replicas
-		}
-		if its.Spec.LifecycleActions == nil || its.Spec.LifecycleActions.MemberJoin == nil {
-			return nil // member join not defined
-		}
-		joined := false
-		for _, inst := range its.Status.InstanceStatus {
-			if inst.PodName == pod.Name {
-				joined = ptr.Deref(inst.Joined, false)
+		joinMember := func(inst workloads.InstanceStatus) error {
+			if its.Spec.LifecycleActions == nil || its.Spec.LifecycleActions.MemberJoin == nil {
+				return nil
 			}
+			if ptr.Deref(inst.Joined, false) {
+				return nil
+			}
+			// TODO: should wait for the data to be loaded before joining the member?
+			return r.joinMember(tree, its, pods, pod)
 		}
-		if joined {
-			return nil
-		}
-
-		// TODO: should wait for the data to be loaded before joining the member?
-
-		if err := r.joinMember(tree, its, pods, pod); err != nil {
-			return err
+		idx := slices.IndexFunc(its.Status.InstanceStatus, func(status workloads.InstanceStatus) bool {
+			return status.PodName == pod.Name
+		})
+		var err error
+		if idx >= 0 {
+			err = joinMember(its.Status.InstanceStatus[idx])
 		}
-		for i, inst := range its.Status.InstanceStatus {
-			if inst.PodName == pod.Name {
-				its.Status.InstanceStatus[i].Joined = ptr.To(true)
-			}
+		if err == nil {
+			its.Status.InstanceStatus[idx].Joined = ptr.To(true)
 		}
-		return nil
+		return err
 	}
 }
 
@@ -273,22 +267,29 @@ func (r *instanceAlignmentReconciler) joinMember(tree *kubebuilderx.ObjectTree,
 	return nil
 }
 
-func (r *instanceAlignmentReconciler) leaveMemberHook(tree *kubebuilderx.ObjectTree,
-	its *workloads.InstanceSet, pods []client.Object, pod *corev1.Pod) kubebuilderx.WithHook {
+func (r *instanceAlignmentReconciler) deleteInstance(tree *kubebuilderx.ObjectTree,
+	its *workloads.InstanceSet, pods []client.Object, pod *corev1.Pod) kubebuilderx.WithPostHook {
 	return func(obj client.Object) error {
-		hasMemberLeaveDefined := its.Spec.LifecycleActions != nil && its.Spec.LifecycleActions.MemberLeave != nil
-		joined := func() bool {
-			for _, inst := range its.Status.InstanceStatus {
-				if inst.PodName == pod.Name {
-					return ptr.Deref(inst.Joined, true)
-				}
+		leaveMember := func(inst workloads.InstanceStatus) error {
+			if its.Spec.LifecycleActions == nil || its.Spec.LifecycleActions.MemberLeave == nil {
+				return nil
 			}
-			return false
-		}()
-		if !hasMemberLeaveDefined || !joined {
-			return nil
+			if !ptr.Deref(inst.Joined, false) {
+				return nil
+			}
+			return r.leaveMember(tree, its, pods, pod)
+		}
+		idx := slices.IndexFunc(its.Status.InstanceStatus, func(status workloads.InstanceStatus) bool {
+			return status.PodName == pod.Name
+		})
+		var err error
+		if idx >= 0 {
+			err = leaveMember(its.Status.InstanceStatus[idx])
 		}
-		return r.leaveMember(tree, its, pods, pod)
+		if err == nil {
+			its.Status.InstanceStatus = slices.Delete(its.Status.InstanceStatus, idx, idx+1)
+		}
+		return err
 	}
 }
 

pkg/controller/instanceset/reconciler_revision_update.go

@@ -20,7 +20,10 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package instanceset
 
 import (
+	"slices"
+
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
@@ -29,18 +32,20 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
 )
 
+func NewRevisionUpdateReconciler() kubebuilderx.Reconciler {
+	return &revisionUpdateReconciler{}
+}
+
 // revisionUpdateReconciler is responsible for updating the expected instance names and their corresponding revisions in the status when there are changes in the spec.
 type revisionUpdateReconciler struct{}
 
+var _ kubebuilderx.Reconciler = &revisionUpdateReconciler{}
+
 type instanceRevision struct {
 	name     string
 	revision string
 }
 
-func NewRevisionUpdateReconciler() kubebuilderx.Reconciler {
-	return &revisionUpdateReconciler{}
-}
-
 func (r *revisionUpdateReconciler) PreCondition(tree *kubebuilderx.ObjectTree) *kubebuilderx.CheckResult {
 	if tree.GetRoot() == nil || !model.IsObjectUpdating(tree.GetRoot()) {
 		return kubebuilderx.ConditionUnsatisfied
@@ -89,11 +94,14 @@ func (r *revisionUpdateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kub
 		updateRevision = instanceRevisionList[len(instanceRevisionList)-1].revision
 	}
 	its.Status.UpdateRevision = updateRevision
-	updatedReplicas, err := calculateUpdatedReplicas(its, tree.List(&corev1.Pod{}))
+
+	updatedReplicas, err := r.calculateUpdatedReplicas(its, tree.List(&corev1.Pod{}))
 	if err != nil {
 		return kubebuilderx.Continue, err
 	}
 	its.Status.UpdatedReplicas = updatedReplicas
+	r.initReplicasNInstanceStatus(its, instanceRevisionList)
+
 	// The 'ObservedGeneration' field is used to indicate whether the revisions have been updated.
 	// Computing these revisions in each reconciliation loop can be time-consuming, so we optimize it by
 	// performing the computation only when the 'spec' is updated.
@@ -102,7 +110,7 @@ func (r *revisionUpdateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kub
 	return kubebuilderx.Continue, nil
 }
 
-func calculateUpdatedReplicas(its *workloads.InstanceSet, pods []client.Object) (int32, error) {
+func (r *revisionUpdateReconciler) calculateUpdatedReplicas(its *workloads.InstanceSet, pods []client.Object) (int32, error) {
 	updatedReplicas := int32(0)
 	for i := range pods {
 		pod, _ := pods[i].(*corev1.Pod)
@@ -113,9 +121,38 @@ func calculateUpdatedReplicas(its *workloads.InstanceSet, pods []client.Object)
 		if updated {
 			updatedReplicas++
 		}
-
 	}
 	return updatedReplicas, nil
 }
 
-var _ kubebuilderx.Reconciler = &revisionUpdateReconciler{}
+func (r *revisionUpdateReconciler) initReplicasNInstanceStatus(its *workloads.InstanceSet, instances []instanceRevision) {
+	if its.Status.InitReplicas == nil && ptr.Deref(its.Spec.Replicas, 0) > 0 {
+		its.Status.InitReplicas = its.Spec.Replicas
+	}
+	if its.Status.InitReplicas == nil {
+		return // init replicas is not set or set to 0
+	}
+
+	init := false
+	if *its.Status.InitReplicas != ptr.Deref(its.Status.ReadyInitReplicas, 0) { // in init phase
+		init = true
+		// in case the replicas is changed in the middle of init phase
+		if ptr.Deref(its.Spec.Replicas, 0) == 0 {
+			its.Status.InitReplicas = nil
+		} else {
+			its.Status.InitReplicas = its.Spec.Replicas
+		}
+	}
+
+	for _, inst := range instances {
+		exist := slices.ContainsFunc(its.Status.InstanceStatus, func(status workloads.InstanceStatus) bool {
+			return status.PodName == inst.name
+		})
+		if !exist {
+			its.Status.InstanceStatus = append(its.Status.InstanceStatus, workloads.InstanceStatus{
+				PodName: inst.name,
+				Joined:  ptr.To(init),
+			})
+		}
+	}
+}