coordinate its update process

Author: cjc7373

apis/workloads/v1/instanceset_types.go

@@ -352,6 +352,11 @@ type InstanceSetStatus struct {
 	// +optional
 	UpdateRevisions map[string]string `json:"updateRevisions,omitempty"`
 
+	// revisions that will be applied during next pod update process
+	//
+	// +optional
+	ProposedRevisions map[string]string `json:"proposedRevisions,omitempty"`
+
 	// TemplatesStatus represents status of each instance generated by InstanceTemplates.
 	//
 	// +optional

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

@@ -11794,6 +11794,12 @@ spec:
                   InstanceSet's generation, which is updated on mutation by the API Server.
                 format: int64
                 type: integer
+              proposedRevisions:
+                additionalProperties:
+                  type: string
+                description: revisions that will be applied during next pod update
+                  process
+                type: object
               readyInitReplicas:
                 description: |-
                   Represents the number of instances that have already reached the InstanceStatus during the cluster initialization stage.

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

@@ -11794,6 +11794,12 @@ spec:
                   InstanceSet's generation, which is updated on mutation by the API Server.
                 format: int64
                 type: integer
+              proposedRevisions:
+                additionalProperties:
+                  type: string
+                description: revisions that will be applied during next pod update
+                  process
+                type: object
               readyInitReplicas:
                 description: |-
                   Represents the number of instances that have already reached the InstanceStatus during the cluster initialization stage.

docs/developer_docs/api-reference/cluster.md

@@ -33632,6 +33632,18 @@ key is the pod name, value is the revision.</p>
 </tr>
 <tr>
 <td>
+<code>proposedRevisions</code><br/>
+<em>
+map[string]string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>revisions that will be applied during next pod update process</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>templatesStatus</code><br/>
 <em>
 <a href="#workloads.kubeblocks.io/v1.InstanceTemplateStatus">

pkg/controller/instanceset/in_place_update_util.go

@@ -296,6 +296,10 @@ func getPodUpdatePolicy(its *workloads.InstanceSet, pod *corev1.Pod) (podUpdateP
 	if err != nil {
 		return noOpsPolicy, "", err
 	}
+	proposedRevisions, err := GetRevisions(its.Status.ProposedRevisions)
+	if err != nil {
+		return noOpsPolicy, "", err
+	}
 
 	// In case of the ITS is stopping and replicas is 0, we can't compose the instance template and instance
 	// to choose the update policy, so we use the pod update policy directly.
@@ -327,7 +331,7 @@ func getPodUpdatePolicy(its *workloads.InstanceSet, pod *corev1.Pod) (podUpdateP
 	}
 
 	specUpdatePolicy := getPodUpdatePolicyInSpec(its, pod, newPod)
-	if getPodRevision(pod) != updateRevisions[pod.Name] {
+	if getPodRevision(pod) != updateRevisions[pod.Name] && getPodRevision(pod) != proposedRevisions[pod.Name] {
 		return recreatePolicy, specUpdatePolicy, nil
 	}
 

pkg/controller/instanceset/instance_util.go

@@ -425,14 +425,23 @@ func MergeNodeSelectorOnceAnnotation(its *workloads.InstanceSet, podToNodeMappin
 }
 
 func buildInstancePodByTemplate(name string, template *instancetemplate.InstanceTemplateExt, parent *workloads.InstanceSet, revision string) (*corev1.Pod, error) {
+	injectNewServiceAccount := func(template *corev1.PodTemplateSpec) {
+		newSAName, ok := parent.Annotations[constant.ProposedServiceAccountNameAnnotationKey]
+		if ok {
+			template.Spec.ServiceAccountName = newSAName
+		}
+	}
+
 	// 1. build a pod from template
 	var err error
 	if len(revision) == 0 {
-		revision, err = buildInstanceTemplateRevision(&template.PodTemplateSpec, parent)
+		revision, err = buildInstanceTemplateRevision(&template.PodTemplateSpec, parent, injectNewServiceAccount)
 		if err != nil {
 			return nil, err
 		}
 	}
+	podTemplate := template.PodTemplateSpec.DeepCopy()
+	injectNewServiceAccount(podTemplate)
 	labels := getMatchLabels(parent.Name)
 	pod := builder.NewPodBuilder(parent.Namespace, name).
 		AddAnnotationsInMap(template.Annotations).
@@ -442,7 +451,7 @@ func buildInstancePodByTemplate(name string, template *instancetemplate.Instance
 		AddLabels(instancetemplate.TemplateNameLabelKey, template.Name). // TODO: remove this label later
 		AddLabels(constant.KBAppInstanceTemplateLabelKey, template.Name).
 		AddControllerRevisionHashLabel(revision).
-		SetPodSpec(*template.Spec.DeepCopy()).
+		SetPodSpec(podTemplate.Spec).
 		GetObject()
 	// Set these immutable fields only on initial Pod creation, not updates.
 	pod.Spec.Hostname = pod.Name
@@ -612,8 +621,12 @@ func copyAndMerge(oldObj, newObj client.Object) client.Object {
 	}
 }
 
-func buildInstanceTemplateRevision(template *corev1.PodTemplateSpec, parent *workloads.InstanceSet) (string, error) {
-	podTemplate := filterInPlaceFields(template)
+func buildInstanceTemplateRevision(template *corev1.PodTemplateSpec, parent *workloads.InstanceSet, mutateTemplateFn func(template *corev1.PodTemplateSpec)) (string, error) {
+	templateCopy := template.DeepCopy()
+	if mutateTemplateFn != nil {
+		mutateTemplateFn(templateCopy)
+	}
+	podTemplate := filterInPlaceFields(templateCopy)
 	its := builder.NewInstanceSetBuilder(parent.Namespace, parent.Name).
 		SetUID(parent.UID).
 		AddAnnotationsInMap(parent.Annotations).

pkg/controller/instanceset/reconciler_instance_alignment.go

@@ -143,15 +143,6 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (
 			return kubebuilderx.Continue, err
 		}
 
-		// HACK: inject new serviceaccount name if possible
-		if its.Annotations != nil {
-			newName, ok := its.Annotations[constant.ProposedServiceAccountNameAnnotationKey]
-			if ok {
-				its.Annotations[constant.ServiceAccountInUseAnnotationKey] = newName
-				newPod.Spec.ServiceAccountName = newName
-			}
-		}
-
 		if err := tree.Add(newPod); err != nil {
 			return kubebuilderx.Continue, err
 		}

pkg/controller/instanceset/reconciler_revision_update.go

@@ -27,6 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
+	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/instancetemplate"
 	"github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
@@ -74,12 +75,26 @@ func (r *revisionUpdateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kub
 
 	// build instance revision list from instance templates
 	var instanceRevisionList []instanceRevision
+	proposedRevisions := make(map[string]string)
 	for instanceName, templateExt := range nameMap {
-		revision, err := buildInstanceTemplateRevision(&templateExt.PodTemplateSpec, its)
+		updatedRevision, err := buildInstanceTemplateRevision(&templateExt.PodTemplateSpec, its, nil)
 		if err != nil {
 			return kubebuilderx.Continue, err
 		}
-		instanceRevisionList = append(instanceRevisionList, instanceRevision{name: instanceName, revision: revision})
+
+		proposedRevision, err := buildInstanceTemplateRevision(&templateExt.PodTemplateSpec, its, func(template *corev1.PodTemplateSpec) {
+			newSAName, ok := its.Annotations[constant.ProposedServiceAccountNameAnnotationKey]
+			if ok {
+				template.Spec.ServiceAccountName = newSAName
+			}
+		})
+		if err != nil {
+			return kubebuilderx.Continue, err
+		}
+		if proposedRevision != updatedRevision {
+			proposedRevisions[instanceName] = proposedRevision
+		}
+		instanceRevisionList = append(instanceRevisionList, instanceRevision{name: instanceName, revision: updatedRevision})
 	}
 	updatedRevisions := make(map[string]string, len(instanceRevisionList))
 	for _, r := range instanceRevisionList {
@@ -92,6 +107,11 @@ func (r *revisionUpdateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kub
 		return kubebuilderx.Continue, err
 	}
 	its.Status.UpdateRevisions = revisions
+	proposedRevisions, err = buildRevisions(proposedRevisions)
+	if err != nil {
+		return kubebuilderx.Continue, err
+	}
+	its.Status.ProposedRevisions = proposedRevisions
 	updateRevision := ""
 	if len(instanceRevisionList) > 0 {
 		updateRevision = instanceRevisionList[len(instanceRevisionList)-1].revision

pkg/controller/instanceset/reconciler_status.go

@@ -191,6 +191,28 @@ func (r *statusReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 	if its.Spec.MinReadySeconds > 0 && availableReplicas != readyReplicas {
 		return kubebuilderx.RetryAfter(time.Second), nil
 	}
+
+	// serviceaccount name migration process
+	proposedRevisions, err := GetRevisions(its.Status.ProposedRevisions)
+	if err != nil {
+		return kubebuilderx.Continue, err
+	}
+
+	updateDone := true
+	for podName, revision := range proposedRevisions {
+		if currentRevisions[podName] != revision {
+			updateDone = false
+			break
+		}
+	}
+	if updateDone {
+		_, ok1 := its.Annotations[constant.ServiceAccountInUseAnnotationKey]
+		proposedSA, ok2 := its.Annotations[constant.ProposedServiceAccountNameAnnotationKey]
+		if !ok1 && ok2 {
+			its.Annotations[constant.ServiceAccountInUseAnnotationKey] = proposedSA
+		}
+	}
+
 	return kubebuilderx.Continue, nil
 }