Bugfix: sync pod /status correctly in kwok-based simulator cluster

.gitignore

@@ -1 +1 @@
-bin/
+bin/

simulator/resourceapplier/resourceapplier.go

@@ -2,12 +2,14 @@ package resourceapplier
 
 import (
 	"context"
+	"encoding/json"
 
 	"golang.org/x/xerrors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic"
 )
 
@@ -124,6 +126,9 @@ func (s *Service) Create(ctx context.Context, resource *unstructured.Unstructure
 		return xerrors.Errorf("failed to create resource: %w", err)
 	}
 
+	if gvk.Kind == "Pod" {
+		return s.PatchPodStatus(ctx, resource)
+	}
 	return nil
 }
 
@@ -315,3 +320,32 @@ func (s *Service) addMutateBeforeUpdating(gvr schema.GroupVersionResource, fn []
 
 	s.mutateBeforeUpdating[gvr] = append(s.mutateBeforeUpdating[gvr], fn...)
 }
+
+func (s *Service) PatchPodStatus(ctx context.Context, resource *unstructured.Unstructured) error {
+	gvk := resource.GroupVersionKind()
+	gvr, err := s.findGVRForGVK(gvk)
+	if err != nil {
+		return err
+	}
+
+	namespace := resource.GetNamespace()
+	newStatus := resource.Object["status"]
+	patchData := map[string]interface{}{
+		"status": newStatus,
+	}
+	patchBytes, err := json.Marshal(patchData)
+	if err != nil {
+		return err
+	}
+	_, err = s.clients.DynamicClient.Resource(gvr).Namespace(namespace).Patch(
+		ctx,
+		resource.GetName(),
+		types.MergePatchType,
+		patchBytes, metav1.PatchOptions{},
+		"status",
+	)
+	if err != nil {
+		return xerrors.Errorf("failed to patch status: %w, gvr: %v, name: %v", err, gvr, resource.GetName())
+	}
+	return nil
+}

simulator/resourceapplier/resourceapplier_test.go

@@ -151,6 +151,54 @@ func TestResourceApplier_createPods(t *testing.T) {
 			filtered: false,
 			wantErr:  false,
 		},
+		{
+			name: "create a Pod have succeeded status",
+			podToCreate: &corev1.Pod{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Pod",
+					APIVersion: "v1",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "succeeded-pod",
+					Namespace: "default",
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  "container-1",
+							Image: "image-1",
+						},
+					},
+				},
+				Status: corev1.PodStatus{
+					Phase: corev1.PodSucceeded,
+				},
+			},
+			podAfterCreate: &corev1.Pod{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Pod",
+					APIVersion: "v1",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "succeeded-pod",
+					Namespace: "default",
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  "container-1",
+							Image: "image-1",
+						},
+					},
+				},
+				Status: corev1.PodStatus{
+					Phase: corev1.PodSucceeded,
+				},
+			},
+			filter:   nil,
+			filtered: false,
+			wantErr:  false,
+		},
 	}
 
 	for _, tt := range tests {
@@ -337,6 +385,54 @@ func TestResourceApplier_updatePods(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "update an Pod have succeeded status",
+			originalPod: &corev1.Pod{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Pod",
+					APIVersion: "v1",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "pod-1",
+					Namespace: "default",
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  "container-1",
+							Image: "image-1",
+						},
+					},
+				},
+			},
+			updatePod: func(pod *corev1.Pod) {
+				pod.Status = corev1.PodStatus{
+					Phase: corev1.PodSucceeded,
+				}
+			},
+			podAfterUpdate: &corev1.Pod{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Pod",
+					APIVersion: "v1",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "pod-1",
+					Namespace: "default",
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  "container-1",
+							Image: "image-1",
+						},
+					},
+				},
+				Status: corev1.PodStatus{
+					Phase: corev1.PodSucceeded,
+				},
+			},
+			wantErr: false,
+		},
 	}
 
 	for _, tt := range tests {

simulator/syncer/syncer.go

@@ -87,15 +87,27 @@ func (s *Service) addFunc(obj interface{}) {
 	}
 }
 
-func (s *Service) updateFunc(_, newObj interface{}) {
+func (s *Service) updateFunc(oldObj, newObj interface{}) {
 	ctx := context.Background()
-	unstructObj, ok := newObj.(*unstructured.Unstructured)
+	newUnstructuredObj, ok := newObj.(*unstructured.Unstructured)
 	if !ok {
 		klog.Error("Failed to convert runtime.Object to *unstructured.Unstructured")
 		return
 	}
 
-	err := s.resourceApplierService.Update(ctx, unstructObj)
+	oldUnstructuredObj, ok := oldObj.(*unstructured.Unstructured)
+	if !ok {
+		klog.Error("Failed to convert runtime.Object to *unstructured.Unstructured")
+		return
+	}
+
+	if newUnstructuredObj.GetKind() == "Pod" && !podStatusEqual(oldUnstructuredObj, newUnstructuredObj) {
+		if patchErr := s.resourceApplierService.PatchPodStatus(ctx, newUnstructuredObj); patchErr != nil {
+			klog.Errorf("Failed to patch pod status: %v", patchErr)
+		}
+	}
+
+	err := s.resourceApplierService.Update(ctx, newUnstructuredObj)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			// We just ignore the not found error because the scheduler may preempt the Pods, or users may remove the resources for debugging.
@@ -124,3 +136,9 @@ func (s *Service) deleteFunc(obj interface{}) {
 		}
 	}
 }
+
+func podStatusEqual(oldPod, newPod *unstructured.Unstructured) bool {
+	oldPhase, _, _ := unstructured.NestedString(oldPod.Object, "status", "phase")
+	newPhase, _, _ := unstructured.NestedString(newPod.Object, "status", "phase")
+	return oldPhase == newPhase
+}