fix: Fix status-writer in the Updater target

Author: rg0now

pkg/pipeline/default_engine.go

@@ -273,7 +273,7 @@ func (eng *defaultEngine) EvaluateStage(s *Stage, delta cache.Delta) ([]cache.De
 
 		vs := []unstruct{}
 		for i, elem := range list {
-			a := deepCopy(u)
+			a := object.DeepCopyAny(u)
 			v, ok := a.(unstruct)
 			if !ok {
 				return nil, errors.New("could not deepcopy object content")
@@ -730,27 +730,6 @@ func containsDelta(ds []cache.Delta, delta cache.Delta) bool {
 	})
 }
 
-func deepCopy(value any) any {
-	switch v := value.(type) {
-	case bool, int64, float64, string:
-		return v
-	case []any:
-		newList := make([]any, len(v))
-		for i, item := range v {
-			newList[i] = deepCopy(item)
-		}
-		return newList
-	case map[string]any:
-		newMap := make(map[string]any)
-		for k, item := range v {
-			newMap[k] = deepCopy(item)
-		}
-		return newMap
-	default:
-		return v
-	}
-}
-
 func packDeltas(targetView string, deltaType cache.DeltaType, vs []unstruct) []cache.Delta {
 	ret := []cache.Delta{}
 	for _, v := range vs {

pkg/reconciler/reconciler_test.go

@@ -341,6 +341,7 @@ var _ = Describe("Reconciler", func() {
 			// object.SetContent(view2, map[string]any{"a": int64(1), "b": int64(2)})
 			object.SetContent(res, map[string]any{"b": int64(2)})
 			Expect(object.DeepEqual(res, event.Object.(object.Object))).To(BeTrue())
+			// Expect(res).To(Equal(event.Object.(object.Object)))
 
 			// Push a delete to the target
 			err = target.Write(ctx, cache.Delta{Type: cache.Deleted, Object: view2})
@@ -361,7 +362,7 @@ var _ = Describe("Reconciler", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(mgr).NotTo(BeNil())
 
-			// Register target
+			// Register an updater target
 			group, version := "", "v1"
 			target := NewTarget(mgr, opv1a1.Target{
 				Resource: opv1a1.Resource{
@@ -391,10 +392,6 @@ var _ = Describe("Reconciler", func() {
 				Version: "v1",
 				Kind:    "Pod",
 			}))
-			// make a Get so that we get a full object (tracker returns only a runtime.Object)
-
-			// getFromClient := &corev1.Pod{}
-			// Expect(c.Get(ctx, client.ObjectKeyFromObject(pod), getFromClient)).NotTo(HaveOccurred())
 			getFromClient, err := object.ConvertRuntimeObjectToClientObject(getFromTracker)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(getFromClient.GetObjectKind().GroupVersionKind()).To(Equal(schema.GroupVersionKind{
@@ -413,8 +410,8 @@ var _ = Describe("Reconciler", func() {
 			// Push update to the target
 			newPod := object.DeepCopy(pod2)
 			unstructured.RemoveNestedField(newPod.UnstructuredContent(), "spec", "containers")
-
 			unstructured.SetNestedField(newPod.UnstructuredContent(), "Always", "spec", "restartPolicy")
+
 			err = target.Write(ctx, cache.Delta{Type: cache.Updated, Object: newPod})
 			Expect(err).NotTo(HaveOccurred())
 
@@ -426,6 +423,7 @@ var _ = Describe("Reconciler", func() {
 				Version: "v1",
 				Kind:    "Pod",
 			}))
+
 			getFromClient, err = object.ConvertRuntimeObjectToClientObject(getFromTracker)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(getFromClient.GetObjectKind().GroupVersionKind()).To(Equal(schema.GroupVersionKind{
@@ -439,6 +437,34 @@ var _ = Describe("Reconciler", func() {
 			Expect(p.Spec.Containers).To(BeEmpty())
 			Expect(p.Spec.RestartPolicy).To(Equal(corev1.RestartPolicy("Always")))
 
+			// Set the status
+			unstructured.SetNestedField(newPod.UnstructuredContent(), map[string]any{
+				"message": "testmessage",
+				"reason":  "testreason",
+			}, "status")
+
+			err = target.Write(ctx, cache.Delta{Type: cache.Updated, Object: newPod})
+			Expect(err).NotTo(HaveOccurred())
+
+			getFromTracker, err = tracker.Get(gvr, "testns", "testpod")
+			Expect(err).NotTo(HaveOccurred())
+			getFromClient, err = object.ConvertRuntimeObjectToClientObject(getFromTracker)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(getFromClient.GetObjectKind().GroupVersionKind()).To(Equal(schema.GroupVersionKind{
+				Group:   "",
+				Version: "v1",
+				Kind:    "Pod",
+			}))
+			p = getFromClient.(*corev1.Pod)
+			Expect(p.GetName()).To(Equal("testpod"))
+			Expect(p.GetNamespace()).To(Equal("testns"))
+			Expect(p.Spec.Containers).To(BeEmpty())
+			Expect(p.Spec.RestartPolicy).To(Equal(corev1.RestartPolicy("Always")))
+			Expect(p.Status).To(Equal(corev1.PodStatus{
+				Message: "testmessage",
+				Reason:  "testreason",
+			}))
+
 			// Delete from the target
 			err = target.Write(ctx, cache.Delta{Type: cache.Deleted, Object: newPod})
 			Expect(err).NotTo(HaveOccurred())

pkg/reconciler/resource.go

@@ -12,7 +12,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/json"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	runtimeManager "sigs.k8s.io/controller-runtime/pkg/manager"
 	runtimePredicate "sigs.k8s.io/controller-runtime/pkg/predicate"
 	runtimeSource "sigs.k8s.io/controller-runtime/pkg/source"
@@ -232,7 +231,7 @@ func (t *target) update(ctx context.Context, delta cache.Delta) error {
 
 	//nolint:nolintlint
 	switch delta.Type { //nolint:exhaustive
-	case cache.Added, cache.Upserted:
+	case cache.Added, cache.Upserted, cache.Updated, cache.Replaced:
 		t.log.V(2).Info("add/upsert", "event-type", delta.Type, "object", client.ObjectKeyFromObject(delta.Object))
 
 		gvk, err := t.Resource.GetGVK()
@@ -244,24 +243,53 @@ func (t *target) update(ctx context.Context, delta cache.Delta) error {
 		obj.SetName(delta.Object.GetName())
 		obj.SetNamespace(delta.Object.GetNamespace())
 
-		if res, err := controllerutil.CreateOrUpdate(context.TODO(), c, obj, func() error {
-			obj.SetUnstructuredContent(delta.Object.UnstructuredContent())
+		// WARNING: the Update target cannot be used to delete labels and annotations, use
+		// the Patcher target for that (this is because we don't want the user to remove
+		// important labels/annotations accidentally and taking care of each in the
+		// pipeline may be too difficult)
+		//
+		// Use our own CreateOrUpdate that will also update the status
+		res, err := CreateOrUpdate(context.TODO(), c, obj, func() error {
+			// remove stuff that's no longer there
+			for k := range obj.UnstructuredContent() {
+				if k == "metadata" {
+					continue
+				}
+				if _, ok, _ := unstructured.NestedFieldNoCopy(delta.Object.UnstructuredContent(), k); !ok {
+					unstructured.RemoveNestedField(obj.UnstructuredContent(), k)
+				}
+			}
+
+			// then update the content with new keys: metadata and status will be handled separately
+			for k, v := range delta.Object.UnstructuredContent() {
+				if k == "metadata" {
+					continue
+				}
+
+				if err := unstructured.SetNestedField(obj.UnstructuredContent(), v, k); err != nil {
+					t.log.Error(err, "failed to update object field during update",
+						"object", client.ObjectKeyFromObject(obj).String(), "key", k)
+					continue
+				}
+			}
+
+			mergeMetadata(obj, delta.Object)
+
+			// restore metadata
 			obj.SetGroupVersionKind(gvk)
 			obj.SetName(delta.Object.GetName())
 			obj.SetNamespace(delta.Object.GetNamespace())
+
 			return nil
-		}); err != nil {
-			return fmt.Errorf("create/update resource %s/%s failed with operation code %s: %w",
-				delta.Object.GetNamespace(), delta.Object.GetName(), res, err)
+		})
+
+		if err != nil {
+			return fmt.Errorf("create/update resource %s failed with operation code %s: %w",
+				client.ObjectKeyFromObject(delta.Object).String(), res, err)
 		}
 
 		return nil
 
-	case cache.Updated, cache.Replaced:
-		t.log.V(2).Info("update", "event-type", delta.Type, "object", client.ObjectKeyFromObject(delta.Object))
-
-		return c.Update(ctx, delta.Object)
-
 	case cache.Deleted:
 		t.log.V(2).Info("delete", "event-type", delta.Type, "object", client.ObjectKeyFromObject(delta.Object))
 
@@ -353,3 +381,29 @@ func removeNested(m map[string]any) map[string]any {
 	}
 	return result
 }
+
+func mergeMetadata(obj, new object.Object) {
+	labels := obj.GetLabels()
+	newLabels := new.GetLabels()
+	if newLabels != nil {
+		if labels == nil {
+			labels = map[string]string{}
+		}
+		for k, v := range newLabels {
+			labels[k] = v
+		}
+		obj.SetLabels(labels)
+	}
+
+	annotations := obj.GetAnnotations()
+	newAnnotations := new.GetAnnotations()
+	if newAnnotations != nil {
+		if annotations == nil {
+			annotations = map[string]string{}
+		}
+		for k, v := range newAnnotations {
+			annotations[k] = v
+		}
+		obj.SetAnnotations(annotations)
+	}
+}

pkg/reconciler/utils.go

@@ -0,0 +1,76 @@
+package reconciler
+
+import (
+	"context"
+	"fmt"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+
+	"github.com/hsnlab/dcontroller/pkg/object"
+)
+
+// CreateOrUpdate creates or updates the given object in the Kubernetes cluster. The object's
+// desired state must be reconciled with the existing state inside the passed in callback MutateFn.
+//
+// The MutateFn is called regardless of creating or updating an object.
+//
+// It returns the executed operation and an error.
+//
+// Note: this version differs from default controllerutil.CreateOrUpdate in two subtle ways
+//   - it uses the unstructured API via object.Object
+//   - changes made by MutateFn to the status subresource will be handled, changes to any other
+//     sub-resource will be discarded
+//   - errors produced by the `Create` branch (after a failed `Get`) will be ignored: this is to
+//     make the function usable with spl// it clients in tests where updates to dot appear in Get
+func CreateOrUpdate(ctx context.Context, c client.Client, obj object.Object, f controllerutil.MutateFn) (controllerutil.OperationResult, error) {
+	key := client.ObjectKeyFromObject(obj)
+	if err := c.Get(ctx, key, obj); err != nil {
+		if !apierrors.IsNotFound(err) {
+			return controllerutil.OperationResultNone, err
+		}
+		if err := mutate(f, key, obj); err != nil {
+			return controllerutil.OperationResultNone, err
+		}
+		if err := c.Create(ctx, obj); err != nil {
+			// this is not an error: default to the update branch
+			goto update
+		}
+		return controllerutil.OperationResultCreated, nil
+	}
+
+update:
+	if err := mutate(f, key, obj); err != nil {
+		return controllerutil.OperationResultNone, err
+	}
+
+	// Update may rewrite out status
+	newStatus, hasStatus, _ := unstructured.NestedMap(obj.UnstructuredContent(), "status")
+
+	if err := c.Update(ctx, obj); err != nil {
+		return controllerutil.OperationResultNone, err
+	}
+
+	// take care of the status just now
+	if hasStatus {
+		if err := unstructured.SetNestedMap(obj.UnstructuredContent(), newStatus, "status"); err == nil {
+			if err := c.Status().Update(ctx, obj); err != nil {
+				return controllerutil.OperationResultNone, err
+			}
+		}
+	}
+
+	return controllerutil.OperationResultUpdated, nil
+}
+
+func mutate(f controllerutil.MutateFn, key client.ObjectKey, obj client.Object) error {
+	if err := f(); err != nil {
+		return err
+	}
+	if newKey := client.ObjectKeyFromObject(obj); key != newKey {
+		return fmt.Errorf("MutateFn cannot mutate object name and/or object namespace")
+	}
+	return nil
+}