test: add some tests

pkg/controllers/collaset/collaset_controller.go

@@ -210,7 +210,7 @@ func calculateStatus(instance *appsv1alpha1.CollaSet, newStatus *appsv1alpha1.Co
 		replicas++
 
 		isUpdated := false
-		if isUpdated = controllerutils.IsPodUpdatedRevision(podWrapper.Pod, updatedRevision.Name); isUpdated {
+		if isUpdated = utils.IsPodUpdatedRevision(podWrapper.Pod, updatedRevision.Name); isUpdated {
 			updatedReplicas++
 		}
 
@@ -229,7 +229,7 @@ func calculateStatus(instance *appsv1alpha1.CollaSet, newStatus *appsv1alpha1.Co
 			}
 		}
 
-		if controllerutils.IsServiceAvailable(podWrapper.Pod) {
+		if controllerutils.IsPodServiceAvailable(podWrapper.Pod) {
 			availableReplicas++
 			if isUpdated {
 				updatedAvailableReplicas++

pkg/controllers/collaset/collaset_controller_test.go

@@ -117,7 +117,7 @@ var _ = Describe("collaset controller", func() {
 		Eventually(func() bool {
 			Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: podToDelete.Namespace, Name: podToDelete.Name}, podToDelete)).Should(BeNil())
 			return podToDelete.DeletionTimestamp != nil
-		}, 500*time.Second, 1*time.Second).Should(BeTrue())
+		}, 5*time.Second, 1*time.Second).Should(BeTrue())
 
 		// there should be 3 pods and one of them is terminating
 		Eventually(func() bool {

pkg/controllers/collaset/revision.go

@@ -22,6 +22,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	appsalphav1 "kusionstack.io/operating/apis/apps/v1alpha1"
+	"kusionstack.io/operating/pkg/controllers/utils/revision"
 )
 
 func getCollaSetPatch(cls *appsalphav1.CollaSet) ([]byte, error) {
@@ -52,6 +53,8 @@ func getCollaSetPatch(cls *appsalphav1.CollaSet) ([]byte, error) {
 	return patch, err
 }
 
+var _ revision.OwnerAdapter = &revisionOwnerAdapter{}
+
 type revisionOwnerAdapter struct {
 }
 

pkg/controllers/collaset/synccontrol/scale.go

@@ -20,7 +20,6 @@ import (
 	"sort"
 
 	collasetutils "kusionstack.io/operating/pkg/controllers/collaset/utils"
-	controllerutils "kusionstack.io/operating/pkg/controllers/utils"
 	"kusionstack.io/operating/pkg/controllers/utils/podopslifecycle"
 )
 
@@ -52,5 +51,5 @@ func (s ActivePodsForDeletion) Less(i, j int) bool {
 		return false
 	}
 
-	return controllerutils.ComparePod(l.Pod, r.Pod)
+	return collasetutils.ComparePod(l.Pod, r.Pod)
 }

pkg/controllers/collaset/synccontrol/update.go

@@ -25,10 +25,10 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	appsv1alpha1 "kusionstack.io/operating/apis/apps/v1alpha1"
 	"kusionstack.io/operating/pkg/controllers/collaset/utils"
 	collasetutils "kusionstack.io/operating/pkg/controllers/collaset/utils"
-	controllerutils "kusionstack.io/operating/pkg/controllers/utils"
 	"kusionstack.io/operating/pkg/controllers/utils/podopslifecycle"
 )
 
@@ -134,7 +134,7 @@ func (o orderByDefault) Less(i, j int) bool {
 		return false
 	}
 
-	return controllerutils.ComparePod(l.Pod, r.Pod)
+	return utils.ComparePod(l.Pod, r.Pod)
 }
 
 type PodUpdater interface {
@@ -191,7 +191,7 @@ func (u *InPlaceIfPossibleUpdater) AnalyseAndGetUpdatedPod(cls *appsv1alpha1.Col
 	}
 
 	inPlaceUpdateSupport = true
-	updatedPod, err = controllerutils.PatchToPod(currentPod, updatedPod, podUpdateInfo.Pod)
+	updatedPod, err = utils.PatchToPod(currentPod, updatedPod, podUpdateInfo.Pod)
 
 	if onlyMetadataChanged {
 		if updatedPod.Annotations != nil {

pkg/controllers/collaset/utils/pod.go

@@ -1,4 +1,5 @@
 /*
+Copyright 2014 The Kubernetes Authors.
 Copyright 2023 The KusionStack Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -17,15 +18,20 @@ limitations under the License.
 package utils
 
 import (
+	"encoding/json"
 	"fmt"
 	"strconv"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apimachineryvalidation "k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/strategicpatch"
 
 	appsv1alpha1 "kusionstack.io/operating/apis/apps/v1alpha1"
 	controllerutils "kusionstack.io/operating/pkg/controllers/utils"
+	revisionutils "kusionstack.io/operating/pkg/controllers/utils/revision"
 	"kusionstack.io/operating/pkg/utils"
 )
 
@@ -64,11 +70,172 @@ func GetPodInstanceID(pod *corev1.Pod) (int, error) {
 }
 
 func NewPodFrom(owner metav1.Object, ownerRef *metav1.OwnerReference, revision *appsv1.ControllerRevision) (*corev1.Pod, error) {
-	pod, err := controllerutils.NewPodFrom(owner, ownerRef, revision)
+	pod, err := GetPodFromRevision(revision)
 	if err != nil {
-		return nil, err
+		return pod, err
 	}
 
+	pod.Namespace = owner.GetNamespace()
+	pod.GenerateName = GetPodsPrefix(owner.GetName())
+	pod.OwnerReferences = append(pod.OwnerReferences, *ownerRef)
+
+	pod.Labels[appsv1.ControllerRevisionHashLabelKey] = revision.Name
+
 	utils.ControllByKusionStack(pod)
 	return pod, nil
 }
+
+func GetPodRevisionPatch(revision *appsv1.ControllerRevision) ([]byte, error) {
+	var raw map[string]interface{}
+	if err := json.Unmarshal([]byte(revision.Data.Raw), &raw); err != nil {
+		return nil, err
+	}
+
+	spec := raw["spec"].(map[string]interface{})
+	template := spec["template"].(map[string]interface{})
+	patch, err := json.Marshal(template)
+	return patch, err
+}
+
+func ApplyPatchFromRevision(pod *corev1.Pod, revision *appsv1.ControllerRevision) (*corev1.Pod, error) {
+	patch, err := GetPodRevisionPatch(revision)
+	if err != nil {
+		return nil, err
+	}
+
+	clone := pod.DeepCopy()
+	patched, err := strategicpatch.StrategicMergePatch([]byte(runtime.EncodeOrDie(revisionutils.PodCodec, clone)), patch, clone)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(patched, clone)
+	if err != nil {
+		return nil, err
+	}
+	return clone, nil
+}
+
+// PatchToPod Use three way merge to get a updated pod.
+func PatchToPod(currentRevisionPod, updateRevisionPod, currentPod *corev1.Pod) (*corev1.Pod, error) {
+	currentRevisionPodBytes, err := json.Marshal(currentRevisionPod)
+	if err != nil {
+		return nil, err
+	}
+	updateRevisionPodBytes, err := json.Marshal(updateRevisionPod)
+
+	if err != nil {
+		return nil, err
+	}
+
+	// 1. find the extra changes based on current revision
+	patch, err := strategicpatch.CreateTwoWayMergePatch(currentRevisionPodBytes, updateRevisionPodBytes, &corev1.Pod{})
+	if err != nil {
+		return nil, err
+	}
+
+	// 2. apply above changes to current pod
+	// We don't apply the diff between currentPod and currentRevisionPod to updateRevisionPod,
+	// because the PodTemplate changes should have the highest priority.
+	currentPodBytes, err := json.Marshal(currentPod)
+	if err != nil {
+		return nil, err
+	}
+	if updateRevisionPodBytes, err = strategicpatch.StrategicMergePatch(currentPodBytes, patch, &corev1.Pod{}); err != nil {
+		return nil, err
+	}
+
+	newPod := &corev1.Pod{}
+	err = json.Unmarshal(updateRevisionPodBytes, newPod)
+	return newPod, err
+}
+
+func GetPodFromRevision(revision *appsv1.ControllerRevision) (*corev1.Pod, error) {
+	pod, err := ApplyPatchFromRevision(&corev1.Pod{}, revision)
+	if err != nil {
+		return nil, err
+	}
+
+	return pod, nil
+}
+
+func GetPodsPrefix(controllerName string) string {
+	// use the dash (if the name isn't too long) to make the pod name a bit prettier
+	prefix := fmt.Sprintf("%s-", controllerName)
+	if len(apimachineryvalidation.NameIsDNSSubdomain(prefix, true)) != 0 {
+		prefix = controllerName
+	}
+	return prefix
+}
+
+func ComparePod(l, r *corev1.Pod) bool {
+	// 1. Unassigned < assigned
+	// If only one of the pods is unassigned, the unassigned one is smaller
+	if l.Spec.NodeName != r.Spec.NodeName && (len(l.Spec.NodeName) == 0 || len(r.Spec.NodeName) == 0) {
+		return len(l.Spec.NodeName) == 0
+	}
+	// 2. PodPending < PodUnknown < PodRunning
+	m := map[corev1.PodPhase]int{corev1.PodPending: 0, corev1.PodUnknown: 1, corev1.PodRunning: 2}
+	if m[l.Status.Phase] != m[r.Status.Phase] {
+		return m[l.Status.Phase] < m[r.Status.Phase]
+	}
+	// 3. Not ready < ready
+	// If only one of the pods is not ready, the not ready one is smaller
+	if controllerutils.IsPodReady(l) != controllerutils.IsPodReady(r) {
+		return !controllerutils.IsPodReady(l)
+	}
+	// TODO: take availability into account when we push minReadySeconds information from deployment into pods,
+	//       see https://github.com/kubernetes/kubernetes/issues/22065
+	// 4. Been ready for empty time < less time < more time
+	// If both pods are ready, the latest ready one is smaller
+	if controllerutils.IsPodReady(l) && controllerutils.IsPodReady(r) && !podReadyTime(l).Equal(podReadyTime(r)) {
+		return afterOrZero(podReadyTime(l), podReadyTime(r))
+	}
+	// 5. Pods with containers with higher restart counts < lower restart counts
+	if maxContainerRestarts(l) != maxContainerRestarts(r) {
+		return maxContainerRestarts(l) > maxContainerRestarts(r)
+	}
+	// 6. Empty creation time pods < newer pods < older pods
+	if !l.CreationTimestamp.Equal(&r.CreationTimestamp) {
+		return afterOrZero(&l.CreationTimestamp, &r.CreationTimestamp)
+	}
+	return false
+}
+
+func maxContainerRestarts(pod *corev1.Pod) int {
+	var maxRestarts int32
+	for _, c := range pod.Status.ContainerStatuses {
+		if c.RestartCount > maxRestarts {
+			maxRestarts = c.RestartCount
+		}
+	}
+	return int(maxRestarts)
+}
+
+func podReadyTime(pod *corev1.Pod) *metav1.Time {
+	if controllerutils.IsPodReady(pod) {
+		for _, c := range pod.Status.Conditions {
+			// we only care about pod ready conditions
+			if c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue {
+				return &c.LastTransitionTime
+			}
+		}
+	}
+	return &metav1.Time{}
+}
+
+// afterOrZero checks if time t1 is after time t2; if one of them
+// is zero, the zero time is seen as after non-zero time.
+func afterOrZero(t1, t2 *metav1.Time) bool {
+	if t1.Time.IsZero() || t2.Time.IsZero() {
+		return t1.Time.IsZero()
+	}
+	return t1.After(t2.Time)
+}
+
+func IsPodUpdatedRevision(pod *corev1.Pod, revision string) bool {
+	if pod.Labels == nil {
+		return false
+	}
+
+	return pod.Labels[appsv1.ControllerRevisionHashLabelKey] == revision
+}

pkg/controllers/podopslifecycle/podopslifecycle_controller.go

@@ -193,7 +193,7 @@ func (r *ReconcilePodOpsLifecycle) addServiceAvailable(pod *corev1.Pod) (bool, e
 		return false, nil
 	}
 
-	satisfied, _, err := controllerutils.SatisfyExpectedFinalizers(pod) // whether all expected finalizers are satisfied
+	satisfied, _, err := controllerutils.IsExpectedFinalizerSatisfied(pod) // whether all expected finalizers are satisfied
 	if err != nil || !satisfied {
 		return false, err
 	}
@@ -362,7 +362,7 @@ func (r *ReconcilePodOpsLifecycle) initPodTransitionRuleManager() {
 		return labels != nil && labelHasPrefix(labels, v1alpha1.PodPostCheckLabelPrefix)
 	})
 	podtransitionrule.AddUnAvailableFunc(func(po *corev1.Pod) (bool, *int64) {
-		return !controllerutils.IsServiceAvailable(po), nil
+		return !controllerutils.IsPodServiceAvailable(po), nil
 	})
 }
 

pkg/controllers/utils/patch_test.go

@@ -17,11 +17,12 @@
 package utils
 
 import (
-	"fmt"
 	"testing"
 )
 
 func TestPatch(t *testing.T) {
 	val := GetLabelAnnoPatchBytes(nil, nil, nil, map[string]string{"detailAnno": "newDetail"})
-	fmt.Println(string(val))
+	if "{\"metadata\":{\"annotations\":{\"detailAnno\":\"newDetail\"}}}" != string(val) {
+		t.Fatalf("get unexpected patch: %s", val)
+	}
 }