feat: support image rolling update (#8385)

go.mod

@@ -74,6 +74,7 @@ require (
 	k8s.io/kubectl v0.29.0
 	k8s.io/utils v0.0.0-20231127182322-b307cd553661
 	sigs.k8s.io/controller-runtime v0.17.2
+	sigs.k8s.io/kustomize/api v0.13.5-0.20230601165947-6ce0bf390ce3
 	sigs.k8s.io/yaml v1.4.0
 )
 
@@ -241,7 +242,6 @@ require (
 	k8s.io/metrics v0.29.0 // indirect
 	oras.land/oras-go v1.2.5 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
-	sigs.k8s.io/kustomize/api v0.13.5-0.20230601165947-6ce0bf390ce3 // indirect
 	sigs.k8s.io/kustomize/kyaml v0.14.3 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 )

pkg/controller/instanceset/in_place_update_util.go

@@ -144,10 +144,14 @@ func mergeInPlaceFields(src, dst *corev1.Pod) {
 		}
 	}
 	ignorePodVerticalScaling := viper.GetBool(FeatureGateIgnorePodVerticalScaling)
+	isImageUpdated := false
 	for _, container := range src.Spec.Containers {
 		for i, c := range dst.Spec.Containers {
 			if container.Name == c.Name {
-				dst.Spec.Containers[i].Image = container.Image
+				if dst.Spec.Containers[i].Image != container.Image {
+					dst.Spec.Containers[i].Image = container.Image
+					isImageUpdated = true
+				}
 				if !ignorePodVerticalScaling {
 					requests, limits := copyRequestsNLimitsFields(&container)
 					mergeResources(&requests, &dst.Spec.Containers[i].Resources.Requests)
@@ -157,6 +161,10 @@ func mergeInPlaceFields(src, dst *corev1.Pod) {
 			}
 		}
 	}
+	// remove role label if images are updated
+	if isImageUpdated {
+		delete(dst.Labels, constant.RoleLabelKey)
+	}
 }
 
 func equalField(old, new any) bool {

pkg/controller/instanceset/instance_util.go

@@ -38,6 +38,7 @@ import (
 	"k8s.io/kubectl/pkg/util/podutils"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/kustomize/api/image"
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
@@ -173,6 +174,46 @@ func isHealthy(pod *corev1.Pod) bool {
 	return isRunningAndReady(pod) && !isTerminating(pod)
 }
 
+// isRoleReady returns true if pod has role label
+func isRoleReady(pod *corev1.Pod, roles []workloads.ReplicaRole) bool {
+	if len(roles) == 0 {
+		return true
+	}
+	_, ok := pod.Labels[constant.RoleLabelKey]
+	return ok
+}
+
+// isImageMatched returns true if all container statuses have same image as defined in pod spec
+func isImageMatched(pod *corev1.Pod) bool {
+	for _, container := range pod.Spec.Containers {
+		index := slices.IndexFunc(pod.Status.ContainerStatuses, func(status corev1.ContainerStatus) bool {
+			return status.Name == container.Name
+		})
+		if index == -1 {
+			continue
+		}
+		specImage := container.Image
+		statusImage := pod.Status.ContainerStatuses[index].Image
+		// Image in status may not match the image used in the PodSpec.
+		// More info: https://kubernetes.io/docs/reference/kubernetes-api/workload-resources/pod-v1/#PodStatus
+		specName, specTag, specDigest := image.Split(specImage)
+		statusName, statusTag, statusDigest := image.Split(statusImage)
+		// if digest presents in spec, it must be same in status
+		if len(specDigest) != 0 && specDigest != statusDigest {
+			return false
+		}
+		// if tag presents in spec, it must be same in status
+		if len(specTag) != 0 && specTag != statusTag {
+			return false
+		}
+		// otherwise, statusName should be same as or has suffix of specName
+		if !strings.HasSuffix(statusName, specName) {
+			return false
+		}
+	}
+	return true
+}
+
 // getPodRevision gets the revision of Pod by inspecting the StatefulSetRevisionLabel. If pod has no revision the empty
 // string is returned.
 func getPodRevision(pod *corev1.Pod) string {

pkg/controller/instanceset/instance_util_test.go

@@ -1385,4 +1385,52 @@ var _ = Describe("instance util test", func() {
 			Expect(medianDuration).To(BeNumerically("<", time.Millisecond))
 		})
 	})
+
+	Context("isImageMatched", func() {
+		It("should work well", func() {
+			pod := builder.NewPodBuilder(namespace, name).GetObject()
+
+			By("spec: image name, status: hostname, image name, tag, digest")
+			pod.Spec.Containers = []corev1.Container{{
+				Name:  name,
+				Image: "nginx",
+			}}
+			pod.Status.ContainerStatuses = []corev1.ContainerStatus{{
+				Name:  name,
+				Image: "docker.io/nginx:latest@0f37a86c04f8",
+			}}
+			Expect(isImageMatched(pod)).Should(BeTrue())
+
+			By("digest not matches")
+			pod.Spec.Containers = []corev1.Container{{
+				Name:  name,
+				Image: "nginx:latest@xxxxxxxxx",
+			}}
+			Expect(isImageMatched(pod)).Should(BeFalse())
+
+			By("tag not matches")
+			pod.Spec.Containers = []corev1.Container{{
+				Name:  name,
+				Image: "nginx:xxxx@0f37a86c04f8",
+			}}
+			Expect(isImageMatched(pod)).Should(BeFalse())
+
+			By("hostname not matches")
+			pod.Spec.Containers = []corev1.Container{{
+				Name:  name,
+				Image: "apecloud.com/nginx",
+			}}
+			Expect(isImageMatched(pod)).Should(BeFalse())
+		})
+	})
+
+	Context("isRoleReady", func() {
+		It("should work well", func() {
+			pod := builder.NewPodBuilder(namespace, name).GetObject()
+			Expect(isRoleReady(pod, nil)).Should(BeTrue())
+			Expect(isRoleReady(pod, roles)).Should(BeFalse())
+			pod.Labels = map[string]string{constant.RoleLabelKey: "leader"}
+			Expect(isRoleReady(pod, roles)).Should(BeTrue())
+		})
+	})
 })

pkg/controller/instanceset/reconciler_update.go

@@ -135,8 +135,20 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 			break
 		}
 
+		if !isImageMatched(pod) {
+			tree.Logger.Info(fmt.Sprintf("InstanceSet %s/%s blocks on update as the image(s) of pod %s is not matched", its.Namespace, its.Name, pod.Name))
+			break
+		}
 		if !isHealthy(pod) {
-			tree.Logger.Info(fmt.Sprintf("InstanceSet %s/%s blocks on scale-in as the pod %s is not healthy", its.Namespace, its.Name, pod.Name))
+			tree.Logger.Info(fmt.Sprintf("InstanceSet %s/%s blocks on update as the pod %s is not healthy", its.Namespace, its.Name, pod.Name))
+			break
+		}
+		if !isRunningAndAvailable(pod, its.Spec.MinReadySeconds) {
+			tree.Logger.Info(fmt.Sprintf("InstanceSet %s/%s blocks on update as the pod %s is not available", its.Namespace, its.Name, pod.Name))
+			break
+		}
+		if !isRoleReady(pod, its.Spec.Roles) {
+			tree.Logger.Info(fmt.Sprintf("InstanceSet %s/%s blocks on update as the role of pod %s is not ready", its.Namespace, its.Name, pod.Name))
 			break
 		}
 
@@ -145,7 +157,7 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 			return kubebuilderx.Continue, err
 		}
 		if its.Spec.PodUpdatePolicy == workloads.StrictInPlacePodUpdatePolicyType && updatePolicy == RecreatePolicy {
-			message := fmt.Sprintf("InstanceSet %s/%s blocks on scale-in as the PodUpdatePolicy is %s and the pod %s can not inplace update",
+			message := fmt.Sprintf("InstanceSet %s/%s blocks on update as the PodUpdatePolicy is %s and the pod %s can not inplace update",
 				its.Namespace, its.Name, workloads.StrictInPlacePodUpdatePolicyType, pod.Name)
 			if tree != nil && tree.EventRecorder != nil {
 				tree.EventRecorder.Eventf(its, corev1.EventTypeWarning, EventReasonStrictInPlace, message)

pkg/controller/instanceset/reconciler_update_test.go

@@ -46,7 +46,6 @@ var _ = Describe("update reconciler test", func() {
 			SetTemplate(template).
 			SetVolumeClaimTemplates(volumeClaimTemplates...).
 			SetMinReadySeconds(minReadySeconds).
-			SetRoles(roles).
 			GetObject()
 	})