Move server-side apply function to a shared package

Multiple controllers had a method for this, but the implementations
differed slightly. This combines their fixes and tests them in a single
place.

Author: cbandy

internal/bridge/crunchybridgecluster/apply.go

@@ -15,6 +15,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/crunchydata/postgres-operator/internal/bridge"
+	"github.com/crunchydata/postgres-operator/internal/controller/runtime"
 	"github.com/crunchydata/postgres-operator/internal/naming"
 	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
 )
@@ -152,7 +153,7 @@ func (r *CrunchyBridgeClusterReconciler) reconcilePostgresRoleSecrets(
 			roleSecrets[roleName], err = r.generatePostgresRoleSecret(cluster, role, clusterRole)
 		}
 		if err == nil {
-			err = errors.WithStack(r.apply(ctx, roleSecrets[roleName]))
+			err = errors.WithStack(runtime.Apply(ctx, r.Writer, roleSecrets[roleName]))
 		}
 		if err != nil {
 			log.Error(err, "Issue creating role secret.")

internal/bridge/crunchybridgecluster/postgres.go

@@ -437,7 +437,7 @@ func (r *PGUpgradeReconciler) Reconcile(ctx context.Context, upgrade *v1beta1.PG
 
 	// TODO: error from apply could mean that the job exists with a different spec.
 	if err == nil && !upgradeJobComplete {
-		err = errors.WithStack(r.apply(ctx,
+		err = errors.WithStack(runtime.Apply(ctx, r.Writer,
 			r.generateUpgradeJob(ctx, upgrade, world.ClusterPrimary, config.FetchKeyCommand(&world.Cluster.Spec))))
 	}
 
@@ -448,7 +448,7 @@ func (r *PGUpgradeReconciler) Reconcile(ctx context.Context, upgrade *v1beta1.PG
 	if err == nil && upgradeJobComplete && !removeDataJobsComplete {
 		for _, sts := range world.ClusterReplicas {
 			if err == nil {
-				err = r.apply(ctx, r.generateRemoveDataJob(ctx, upgrade, sts))
+				err = runtime.Apply(ctx, r.Writer, r.generateRemoveDataJob(ctx, upgrade, sts))
 			}
 		}
 	}

internal/controller/pgupgrade/apply.go

@@ -6,13 +6,10 @@ package postgrescluster
 
 import (
 	"context"
-	"reflect"
 
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/equality"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/crunchydata/postgres-operator/internal/kubeapi"
+	"github.com/crunchydata/postgres-operator/internal/controller/runtime"
 )
 
 // apply sends an apply patch to object's endpoint in the Kubernetes API and
@@ -21,41 +18,5 @@ import (
 // - https://docs.k8s.io/reference/using-api/server-side-apply/#managers
 // - https://docs.k8s.io/reference/using-api/server-side-apply/#conflicts
 func (r *Reconciler) apply(ctx context.Context, object client.Object) error {
-	// Generate an apply-patch by comparing the object to its zero value.
-	zero := reflect.New(reflect.TypeOf(object).Elem()).Interface()
-	data, err := client.MergeFrom(zero.(client.Object)).Data(object)
-	apply := client.RawPatch(client.Apply.Type(), data)
-
-	// Keep a copy of the object before any API calls.
-	intent := object.DeepCopyObject()
-	patch := kubeapi.NewJSONPatch()
-
-	// Send the apply-patch with force=true.
-	if err == nil {
-		err = r.Writer.Patch(ctx, object, apply, client.ForceOwnership)
-	}
-
-	// Some fields cannot be server-side applied correctly. When their outcome
-	// does not match the intent, send a json-patch to get really specific.
-	switch actual := object.(type) {
-	case *corev1.Service:
-		applyServiceSpec(patch, actual.Spec, intent.(*corev1.Service).Spec, "spec")
-	}
-
-	// Send the json-patch when necessary.
-	if err == nil && !patch.IsEmpty() {
-		err = r.Writer.Patch(ctx, object, patch)
-	}
-	return err
-}
-
-// applyServiceSpec is called by Reconciler.apply to work around issues
-// with server-side apply.
-func applyServiceSpec(
-	patch *kubeapi.JSON6902, actual, intent corev1.ServiceSpec, path ...string,
-) {
-	// Service.Spec.Selector cannot be unset; perhaps https://issue.k8s.io/117447
-	if !equality.Semantic.DeepEqual(actual.Selector, intent.Selector) {
-		patch.Replace(append(path, "selector")...)(intent.Selector)
-	}
+	return runtime.Apply(ctx, r.Writer, object)
 }

internal/controller/pgupgrade/pgupgrade_controller.go

@@ -16,7 +16,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
-	"github.com/crunchydata/postgres-operator/internal/kubeapi"
+	"github.com/crunchydata/postgres-operator/internal/controller/runtime"
 	"github.com/crunchydata/postgres-operator/internal/logging"
 	"github.com/crunchydata/postgres-operator/internal/naming"
 	"github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
@@ -31,7 +31,7 @@ func (r *Reconciler) adoptObject(ctx context.Context, postgresCluster *v1beta1.P
 		return err
 	}
 
-	patchBytes, err := kubeapi.NewMergePatch().
+	patchBytes, err := runtime.NewMergePatch().
 		Add("metadata", "ownerReferences")(obj.GetOwnerReferences()).Bytes()
 	if err != nil {
 		return err
@@ -160,8 +160,8 @@ func (r *Reconciler) manageControllerRefs(ctx context.Context,
 func (r *Reconciler) releaseObject(ctx context.Context,
 	postgresCluster *v1beta1.PostgresCluster, obj client.Object) error {
 
-	// TODO create a strategic merge type in kubeapi instead of using Merge7386
-	patch, err := kubeapi.NewMergePatch().
+	// TODO create a strategic merge type instead of using Merge7386
+	patch, err := runtime.NewMergePatch().
 		Add("metadata", "ownerReferences")([]map[string]string{{
 		"$patch": "delete",
 		"uid":    string(postgresCluster.GetUID()),

internal/controller/postgrescluster/apply.go

@@ -0,0 +1,58 @@
+// Copyright 2021 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package runtime
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// Apply sends an apply patch with force=true using cc and updates object with any returned content.
+// The client is responsible for setting fieldManager; see [client.WithFieldOwner].
+//
+// - https://docs.k8s.io/reference/using-api/server-side-apply#managers
+// - https://docs.k8s.io/reference/using-api/server-side-apply#conflicts
+func Apply[
+	// NOTE: This interface can go away following https://go.dev/issue/47487.
+	ClientPatch interface {
+		Patch(context.Context, client.Object, client.Patch, ...client.PatchOption) error
+	},
+	T interface{ client.Object },
+](ctx context.Context, cc ClientPatch, object T) error {
+	// Generate an apply-patch by comparing the object to its zero value.
+	data, err := client.MergeFrom(*new(T)).Data(object)
+	apply := client.RawPatch(client.Apply.Type(), data)
+
+	// Keep a copy of the object before any API calls.
+	intent := object.DeepCopyObject()
+
+	// Send the apply-patch with force=true.
+	if err == nil {
+		err = cc.Patch(ctx, object, apply, client.ForceOwnership)
+	}
+
+	// Some fields cannot be server-side applied correctly.
+	// When their outcome does not match the intent, send a json-patch to get really specific.
+	patch := NewJSONPatch()
+
+	switch actual := any(object).(type) {
+	case *corev1.Service:
+		intent := intent.(*corev1.Service)
+
+		// Service.Spec.Selector cannot be unset; perhaps https://issue.k8s.io/117447
+		if !equality.Semantic.DeepEqual(actual.Spec.Selector, intent.Spec.Selector) {
+			patch.Replace("spec", "selector")(intent.Spec.Selector)
+		}
+	}
+
+	// Send the json-patch when necessary.
+	if err == nil && !patch.IsEmpty() {
+		err = cc.Patch(ctx, object, patch)
+	}
+	return err
+}

internal/controller/postgrescluster/controller_ref_manager.go

@@ -2,10 +2,9 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 
-package postgrescluster
+package runtime_test
 
 import (
-	"context"
 	"errors"
 	"regexp"
 	"strings"
@@ -23,17 +22,18 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
+	"github.com/crunchydata/postgres-operator/internal/controller/runtime"
 	"github.com/crunchydata/postgres-operator/internal/testing/require"
 )
 
 func TestServerSideApply(t *testing.T) {
-	ctx := context.Background()
-	cfg, cc := setupKubernetes(t)
+	ctx := t.Context()
+	config, base := require.Kubernetes2(t)
 	require.ParallelCapacity(t, 0)
 
-	ns := setupNamespace(t, cc)
+	ns := require.Namespace(t, base)
 
-	dc, err := discovery.NewDiscoveryClientForConfig(cfg)
+	dc, err := discovery.NewDiscoveryClientForConfig(config)
 	assert.NilError(t, err)
 
 	server, err := dc.ServerVersion()
@@ -43,8 +43,7 @@ func TestServerSideApply(t *testing.T) {
 	assert.NilError(t, err)
 
 	t.Run("ObjectMeta", func(t *testing.T) {
-		cc := client.WithFieldOwner(cc, t.Name())
-		reconciler := Reconciler{Writer: cc}
+		cc := client.WithFieldOwner(base, t.Name())
 		constructor := func() *corev1.ConfigMap {
 			var cm corev1.ConfigMap
 			cm.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("ConfigMap"))
@@ -78,17 +77,16 @@ func TestServerSideApply(t *testing.T) {
 			assert.Assert(t, after.GetResourceVersion() == before.GetResourceVersion())
 		}
 
-		// Our apply method generates the correct apply-patch.
+		// Our [runtime.Apply] generates the correct apply-patch.
 		again := constructor()
-		assert.NilError(t, reconciler.apply(ctx, again))
+		assert.NilError(t, runtime.Apply(ctx, cc, again))
 		assert.Assert(t, again.GetResourceVersion() != "")
 		assert.Assert(t, again.GetResourceVersion() == after.GetResourceVersion(),
 			"expected to correctly no-op")
 	})
 
 	t.Run("ControllerReference", func(t *testing.T) {
-		cc := client.WithFieldOwner(cc, t.Name())
-		reconciler := Reconciler{Writer: cc}
+		cc := client.WithFieldOwner(base, t.Name())
 
 		// Setup two possible controllers.
 		controller1 := new(corev1.ConfigMap)
@@ -128,8 +126,8 @@ func TestServerSideApply(t *testing.T) {
 		assert.Assert(t, len(status.ErrStatus.Details.Causes) != 0)
 		assert.Equal(t, status.ErrStatus.Details.Causes[0].Field, "metadata.ownerReferences")
 
-		// Try to change the controller using our apply method.
-		err2 := reconciler.apply(ctx, applied)
+		// Try to change the controller using our [runtime.Apply].
+		err2 := runtime.Apply(ctx, cc, applied)
 
 		// Same result; patch not accepted.
 		assert.DeepEqual(t, err1, err2,
@@ -162,8 +160,7 @@ func TestServerSideApply(t *testing.T) {
 			{"empty", make(map[string]string)},
 		} {
 			t.Run(tt.name, func(t *testing.T) {
-				cc := client.WithFieldOwner(cc, t.Name())
-				reconciler := Reconciler{Writer: cc}
+				cc := client.WithFieldOwner(base, t.Name())
 
 				intent := constructor(tt.name + "-selector")
 				intent.Spec.Selector = tt.selector
@@ -190,9 +187,9 @@ func TestServerSideApply(t *testing.T) {
 				assert.Assert(t, len(after.Spec.Selector) != len(intent.Spec.Selector),
 					"got %v", after.Spec.Selector)
 
-				// Our apply method corrects it.
+				// Our [runtime.Apply] corrects it.
 				again := intent.DeepCopy()
-				assert.NilError(t, reconciler.apply(ctx, again))
+				assert.NilError(t, runtime.Apply(ctx, cc, again))
 				assert.Assert(t,
 					equality.Semantic.DeepEqual(again.Spec.Selector, intent.Spec.Selector),
 					"\n--- again.Spec.Selector\n+++ intent.Spec.Selector\n%v",

internal/controller/runtime/apply.go

@@ -76,6 +76,7 @@ func (fn ClientUpdate) Update(ctx context.Context, obj client.Object, opts ...cl
 	return fn(ctx, obj, opts...)
 }
 
+// WarningHandler implements [rest.WarningHandler] and [rest.WarningHandlerWithContext] as a single function.
 type WarningHandler func(ctx context.Context, code int, agent string, text string)
 
 func (fn WarningHandler) HandleWarningHeader(code int, agent string, text string) {

internal/controller/postgrescluster/apply_test.go renamed to internal/controller/runtime/apply_test.go

@@ -50,7 +50,7 @@ func FromUnstructuredObject[
 		FromUnstructured(object.UnstructuredContent(), result)
 }
 
-// ToUnstructuredList returns a copy of list by marshaling through JSON.
+// ToUnstructuredList returns a copy of list using reflection.
 func ToUnstructuredList(list client.ObjectList) (*unstructured.UnstructuredList, error) {
 	content, err := runtime.
 		DefaultUnstructuredConverter.
@@ -61,7 +61,7 @@ func ToUnstructuredList(list client.ObjectList) (*unstructured.UnstructuredList,
 	return result, err
 }
 
-// ToUnstructuredObject returns a copy of object by marshaling through JSON.
+// ToUnstructuredObject returns a copy of object using reflection.
 func ToUnstructuredObject(object client.Object) (*unstructured.Unstructured, error) {
 	content, err := runtime.
 		DefaultUnstructuredConverter.