From 34708549961d095d4226f981664c8c39ce6500c9 Mon Sep 17 00:00:00 2001 From: Hannes Blut Date: Tue, 16 Jun 2026 16:49:11 +0200 Subject: [PATCH] feat: add flag --enable-server-side-apply on-behalf-of: @eon-se opensource@eon.com Signed-off-by: Hannes Blut --- cmd/api-syncagent/main.go | 2 +- cmd/api-syncagent/options.go | 9 + internal/controller/sync/controller.go | 44 ++- internal/controller/syncmanager/controller.go | 46 +-- internal/sync/object_syncer.go | 114 ++++++ internal/sync/syncer.go | 24 +- internal/sync/syncer_related.go | 3 + internal/sync/syncer_ssa_test.go | 337 ++++++++++++++++++ 8 files changed, 537 insertions(+), 42 deletions(-) create mode 100644 internal/sync/syncer_ssa_test.go diff --git a/cmd/api-syncagent/main.go b/cmd/api-syncagent/main.go index 218f7a90..0f79d6cc 100644 --- a/cmd/api-syncagent/main.go +++ b/cmd/api-syncagent/main.go @@ -196,7 +196,7 @@ func run(ctx context.Context, log *zap.SugaredLogger, opts *Options) error { // before it can start any sync controllers. That discovery logic is encapsulated in the ResourceProber. prober := discovery.NewResourceProber(endpoint.EndpointSlice.Config, endpointSliceCluster.GetClient(), endpoint.EndpointSlice.Name) - return syncmanager.Add(ctx, mgr, prober, dmcm, log, opts.PublishedResourceSelector, opts.Namespace, opts.AgentName) + return syncmanager.Add(ctx, mgr, prober, dmcm, log, opts.PublishedResourceSelector, opts.Namespace, opts.AgentName, opts.EnableServerSideApply) }); err != nil { return err } diff --git a/cmd/api-syncagent/options.go b/cmd/api-syncagent/options.go index 0ad8aba1..89976d27 100644 --- a/cmd/api-syncagent/options.go +++ b/cmd/api-syncagent/options.go @@ -71,6 +71,13 @@ type Options struct { HealthAddr string DisabledControllers []string + + // EnableServerSideApply switches the sync controller to use Kubernetes + // Server-Side Apply when writing synchronized objects to the local + // cluster. This preserves fields owned by other field managers (e.g. + // Crossplane writing spec.resourceRef onto a claim) and hopefully avoids + // accidentally creating duplicate composite resources. + EnableServerSideApply bool } func NewOptions() *Options { @@ -78,6 +85,7 @@ func NewOptions() *Options { LogOptions: log.NewDefaultOptions(), PublishedResourceSelector: labels.Everything(), MetricsAddr: "127.0.0.1:8085", + EnableServerSideApply: false, } } @@ -95,6 +103,7 @@ func (o *Options) AddFlags(flags *pflag.FlagSet) { flags.StringVar(&o.MetricsAddr, "metrics-address", o.MetricsAddr, "host and port to serve Prometheus metrics via /metrics (HTTP)") flags.StringVar(&o.HealthAddr, "health-address", o.HealthAddr, "host and port to serve probes via /readyz and /healthz (HTTP)") flags.StringSliceVar(&o.DisabledControllers, "disabled-controllers", o.DisabledControllers, fmt.Sprintf("comma-separated list of controllers (out of %v) to disable (can be given multiple times)", sets.List(availableControllers))) + flags.BoolVar(&o.EnableServerSideApply, "enable-server-side-apply", o.EnableServerSideApply, "use Kubernetes Server-Side Apply when writing synchronized objects to the local cluster (recommended; preserves fields owned by other controllers such as Crossplane's claim binder)") } func (o *Options) Validate() error { diff --git a/internal/controller/sync/controller.go b/internal/controller/sync/controller.go index a3bfc638..ad5f8884 100644 --- a/internal/controller/sync/controller.go +++ b/internal/controller/sync/controller.go @@ -64,14 +64,15 @@ const ( ) type Reconciler struct { - localClient ctrlruntimeclient.Client - remoteManager mcmanager.Manager - log *zap.SugaredLogger - remoteDummy *unstructured.Unstructured - pubRes *syncagentv1alpha1.PublishedResource - localCRD *apiextensionsv1.CustomResourceDefinition - stateNamespace string - agentName string + localClient ctrlruntimeclient.Client + remoteManager mcmanager.Manager + log *zap.SugaredLogger + remoteDummy *unstructured.Unstructured + pubRes *syncagentv1alpha1.PublishedResource + localCRD *apiextensionsv1.CustomResourceDefinition + stateNamespace string + agentName string + enableServerSideApply bool } // Create creates a new controller and importantly does *not* add it to the manager, @@ -84,6 +85,7 @@ func Create( discoveryClient *discovery.Client, stateNamespace string, agentName string, + enableServerSideApply bool, log *zap.SugaredLogger, numWorkers int, ) (mccontroller.Controller, error) { @@ -117,14 +119,15 @@ func Create( // setup the reconciler reconciler := &Reconciler{ - localClient: localManager.GetClient(), - remoteManager: remoteManager, - log: log, - remoteDummy: remoteDummy, - pubRes: pubRes, - stateNamespace: stateNamespace, - agentName: agentName, - localCRD: localCRD, + localClient: localManager.GetClient(), + remoteManager: remoteManager, + log: log, + remoteDummy: remoteDummy, + pubRes: pubRes, + stateNamespace: stateNamespace, + agentName: agentName, + localCRD: localCRD, + enableServerSideApply: enableServerSideApply, } ctrlOptions := mccontroller.Options{ @@ -389,7 +392,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request) return reconcile.Result{}, nil } - recorder := cl.GetEventRecorderFor(ControllerName) //nolint:staticcheck // https://github.com/kcp-dev/api-syncagent/issues/157 + recorder := cl.GetEventRecorderFor(ControllerName) ctx = sync.WithClusterName(ctx, logicalcluster.Name(request.ClusterName)) ctx = sync.WithEventRecorder(ctx, recorder) @@ -407,8 +410,13 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request) ctx = sync.WithWorkspacePath(ctx, logicalcluster.NewPath(path)) } + var syncerOpts []sync.ResourceSyncerOption + if r.enableServerSideApply { + syncerOpts = append(syncerOpts, sync.WithServerSideApply()) + } + // sync main object - syncer, err := sync.NewResourceSyncer(log, r.localClient, vwClient, r.pubRes, r.localCRD, mutation.NewMutator, r.stateNamespace, r.agentName) + syncer, err := sync.NewResourceSyncer(log, r.localClient, vwClient, r.pubRes, r.localCRD, mutation.NewMutator, r.stateNamespace, r.agentName, syncerOpts...) if err != nil { recorder.Event(remoteObj, corev1.EventTypeWarning, "ReconcilingError", "Failed to process object: a provider-side issue has occurred.") return reconcile.Result{}, fmt.Errorf("failed to create syncer: %w", err) diff --git a/internal/controller/syncmanager/controller.go b/internal/controller/syncmanager/controller.go index 840ce325..99fcc2d2 100644 --- a/internal/controller/syncmanager/controller.go +++ b/internal/controller/syncmanager/controller.go @@ -57,15 +57,16 @@ type Reconciler struct { // also triggered. ctx context.Context - localManager manager.Manager - dmcm *kcp.DynamicMultiClusterManager - log *zap.SugaredLogger - recorder record.EventRecorder - discoveryClient *discovery.Client - resourceProber *discovery.ResourceProber - prFilter labels.Selector - stateNamespace string - agentName string + localManager manager.Manager + dmcm *kcp.DynamicMultiClusterManager + log *zap.SugaredLogger + recorder record.EventRecorder + discoveryClient *discovery.Client + resourceProber *discovery.ResourceProber + prFilter labels.Selector + stateNamespace string + agentName string + enableServerSideApply bool syncCancelsLock sync.RWMutex // A map of sync controllers, one for each PublishedResource, using their @@ -84,6 +85,7 @@ func Add( prFilter labels.Selector, stateNamespace string, agentName string, + enableServerSideApply bool, ) error { discoveryClient, err := discovery.NewClient(localManager.GetConfig()) if err != nil { @@ -91,18 +93,19 @@ func Add( } reconciler := &Reconciler{ - ctx: ctx, - localManager: localManager, - dmcm: dmcm, - log: log, - recorder: localManager.GetEventRecorderFor(ControllerName), //nolint:staticcheck // https://github.com/kcp-dev/api-syncagent/issues/157 - discoveryClient: discoveryClient, - prFilter: prFilter, - stateNamespace: stateNamespace, - agentName: agentName, - resourceProber: resourceProber, - syncCancelsLock: sync.RWMutex{}, - syncCancels: map[string]context.CancelCauseFunc{}, + ctx: ctx, + localManager: localManager, + dmcm: dmcm, + log: log, + recorder: localManager.GetEventRecorderFor(ControllerName), + discoveryClient: discoveryClient, + prFilter: prFilter, + stateNamespace: stateNamespace, + agentName: agentName, + resourceProber: resourceProber, + enableServerSideApply: enableServerSideApply, + syncCancelsLock: sync.RWMutex{}, + syncCancels: map[string]context.CancelCauseFunc{}, } bldr := builder. @@ -188,6 +191,7 @@ func (r *Reconciler) ensureSyncController(ctx context.Context, log *zap.SugaredL r.discoveryClient, r.stateNamespace, r.agentName, + r.enableServerSideApply, r.log, numSyncWorkers, ) diff --git a/internal/sync/object_syncer.go b/internal/sync/object_syncer.go index a362e635..cabbae03 100644 --- a/internal/sync/object_syncer.go +++ b/internal/sync/object_syncer.go @@ -69,6 +69,13 @@ type objectSyncer struct { // being deleted; used to clean up related resources when the primary object // is being deleted. forceDelete bool + // useServerSideApply switches the syncer from client-side merge patches + // (backed by a last-known-state secret) to Kubernetes Server-Side Apply + // using a stable field manager. SSA preserves fields owned by other + // controllers (e.g. Crossplane's spec.resourceRef on a claim) and removes + // the full update fallback that could otherwise clobber such + // fields. + useServerSideApply bool } type syncSideType int @@ -126,6 +133,24 @@ func (s *objectSyncer) Sync(ctx context.Context, log *zap.SugaredLogger, source, return false, fmt.Errorf("failed to apply mutations: %w", err) } + // Server-Side Apply path: a single Apply patch handles create-or-update + // in one round-trip and the API server tracks our managed fields. Fields + // the syncagent does not include in the payload (e.g. spec.resourceRef + // written by Crossplane on a downstream claim) are preserved by SSA. + if s.useServerSideApply { + newDest, err := s.applyServerSide(ctx, log, source, dest) + if err != nil { + return false, fmt.Errorf("failed to apply destination object: %w", err) + } + + // Do not back-sync status onto an object that is being deleted. + if newDest.object == nil || newDest.object.GetDeletionTimestamp() != nil { + return false, nil + } + + return s.syncObjectStatus(ctx, log, source, newDest) + } + // if no destination object exists yet, attempt to create it; // note that the object _might_ exist, but we were not able to find it because of broken labels if dest.object == nil { @@ -361,6 +386,95 @@ func (s *objectSyncer) syncObjectStatusForward(ctx context.Context, log *zap.Sug return nil } +// applyServerSide is the Server-Side Apply equivalent of the +// ensureDestinationObject + syncObjectSpec pair. It builds the desired +// destination object from the (mutated) source, strips runtime metadata, +// drops subresources (status is reconciled separately) and applies the +// resulting object as a single SSA patch with a stable field manager. +// +// SSA preserves every field that is owned by another field manager. +func (s *objectSyncer) applyServerSide(ctx context.Context, log *zap.SugaredLogger, source, dest syncSide) (syncSide, error) { + // build the desired destination object (GVK projection + renaming). + desired, err := s.destCreator(source.object) + if err != nil { + return dest, fmt.Errorf("failed to create destination object: %w", err) + } + + // the API server cannot create the namespace for us via SSA + if err := s.ensureNamespace(ctx, log, dest.client, desired.GetNamespace()); err != nil { + return dest, fmt.Errorf("failed to ensure destination namespace: %w", err) + } + + // drop runtime metadata that must not be part of an apply payload + // (uid, resourceVersion, managedFields, ownerRefs, …). stripMetadata + // also runs the unsyncable label/annotation filter for us. + if err := stripMetadata(desired); err != nil { + return dest, fmt.Errorf("failed to strip metadata from destination object: %w", err) + } + + // remember the connection between the source and destination object + sourceObjKey := newObjectKey(source.object, source.clusterName, source.workspacePath) + if s.metadataOnDestination { + ensureLabels(desired, sourceObjKey.Labels()) + ensureAnnotations(desired, sourceObjKey.Annotations()) + s.labelWithAgent(desired) + } + + // do not claim ownership of subresource content via the main resource + // apply; status is reconciled separately and "scale" is never written. + s.removeSubresources(desired) + + // do not apply onto an object that is currently being deleted; SSA would + // otherwise revive removed labels/annotations and confuse the deletion + // flow on the next reconcile. + if dest.object != nil && dest.object.GetDeletionTimestamp() != nil { + log.Debugw("Destination object is in deletion, skipping SSA", "dest-object", newObjectKey(dest.object, dest.clusterName, logicalcluster.None)) + return dest, nil + } + + created := dest.object == nil + fieldManager := s.fieldManager() + + objectLog := log.With( + "dest-object", newObjectKey(desired, dest.clusterName, logicalcluster.None), + "field-manager", fieldManager, + ) + objectLog.Debugw("Applying destination object (server-side apply)…") + + // ForceOwnership lets us take over fields that may have been written by + // the legacy client-side-apply field manager during a previous version + // of the syncagent, and to recover from rare conflicts with other + // controllers that briefly write the same path. + if err := dest.client.Apply(ctx, ctrlruntimeclient.ApplyConfigurationFromUnstructured(desired), + ctrlruntimeclient.FieldOwner(fieldManager), + ctrlruntimeclient.ForceOwnership, + ); err != nil { + return dest, fmt.Errorf("failed to server-side apply destination object: %w", err) + } + + if created { + s.recordEvent(ctx, source, syncSide{object: desired}, corev1.EventTypeNormal, "ObjectPlaced", "Object has been placed.") + } + + // hand the server's view of the object (now including any fields owned + // by other managers, e.g. spec.resourceRef from Crossplane) back to the + // caller so that status back-syncing can read it without an extra GET. + dest.object = desired + return dest, nil +} + +// fieldManager returns the SSA field manager string used by this syncer. +// It is stable across restarts and unique per agent name, so that multi- +// agent setups (the same API synced from multiple kcp's into one backend +// cluster) each get their own ownership track. +func (s *objectSyncer) fieldManager() string { + const base = "api-syncagent" + if s.agentName == "" { + return base + } + return base + "/" + s.agentName +} + func (s *objectSyncer) ensureDestinationObject(ctx context.Context, log *zap.SugaredLogger, source, dest syncSide) error { // create a copy of the source with GVK projected and renaming rules applied destObj, err := s.destCreator(source.object) diff --git a/internal/sync/syncer.go b/internal/sync/syncer.go index 7108fca1..0c51207a 100644 --- a/internal/sync/syncer.go +++ b/internal/sync/syncer.go @@ -54,10 +54,21 @@ type ResourceSyncer struct { agentName string + useServerSideApply bool + // newObjectStateStore is used for testing purposes newObjectStateStore newObjectStateStoreFunc } +// ResourceSyncerOption configures a ResourceSyncer at construction time. +type ResourceSyncerOption func(*ResourceSyncer) + +func WithServerSideApply() ResourceSyncerOption { + return func(r *ResourceSyncer) { + r.useServerSideApply = true + } +} + type MutatorCreatorFunc func(*syncagentv1alpha1.ResourceMutationSpec) (mutation.Mutator, error) func NewResourceSyncer( @@ -69,6 +80,7 @@ func NewResourceSyncer( mutatorCreator MutatorCreatorFunc, stateNamespace string, agentName string, + opts ...ResourceSyncerOption, ) (*ResourceSyncer, error) { // create a dummy that represents the type used on the local service cluster localGVK, err := projection.PublishedResourceSourceGVK(localCRD, pubRes) @@ -116,7 +128,7 @@ func NewResourceSyncer( relatedMutators[rr.Identifier] = mutator } - return &ResourceSyncer{ + r := &ResourceSyncer{ log: log.With("local-gvk", localGVK, "remote-gvk", remoteGVK), localClient: localClient, remoteClient: remoteClient, @@ -128,7 +140,13 @@ func NewResourceSyncer( relatedMutators: relatedMutators, agentName: agentName, newObjectStateStore: newKubernetesStateStoreCreator(stateNamespace), - }, nil + } + + for _, opt := range opts { + opt(r) + } + + return r, nil } // Process is the primary entrypoint for object synchronization. This function will create/update @@ -196,6 +214,8 @@ func (s *ResourceSyncer) Process(ctx context.Context, remoteObj *unstructured.Un // together and can be found. metadataOnDestination: true, eventObjSide: syncSideSource, + // propagate the SSA mode to the per-object syncer + useServerSideApply: s.useServerSideApply, } // When the primary object is being deleted, clean up related resources FIRST, diff --git a/internal/sync/syncer_related.go b/internal/sync/syncer_related.go index f8281ab6..84a059ee 100644 --- a/internal/sync/syncer_related.go +++ b/internal/sync/syncer_related.go @@ -173,6 +173,9 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su eventObjSide: eventObjSide, // force deletion of related resources when the primary object is being deleted forceDelete: forceDelete, + // propagate the SSA mode chosen for the primary syncer to keep + // behavior consistent across the whole resource graph + useServerSideApply: s.useServerSideApply, } req, err := syncer.Sync(ctx, log, sourceSide, destSide) diff --git a/internal/sync/syncer_ssa_test.go b/internal/sync/syncer_ssa_test.go new file mode 100644 index 00000000..abb57403 --- /dev/null +++ b/internal/sync/syncer_ssa_test.go @@ -0,0 +1,337 @@ +/* +Copyright 2026 The KCP Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package sync + +import ( + "context" + "testing" + + "go.uber.org/zap" + + "github.com/kcp-dev/api-syncagent/internal/mutation" + dummyv1alpha1 "github.com/kcp-dev/api-syncagent/internal/sync/apis/dummy/v1alpha1" + syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1" + + "github.com/kcp-dev/logicalcluster/v3" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/tools/record" + ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +// TestSyncerServerSideApplyCreate verifies that the SSA code path successfully +// creates a brand-new local object from a remote one. This is a smoke test for +// the WithServerSideApply() option; the actual field-manager-aware merge +// behaviour is provided by the kube-apiserver in real clusters (and is the +// reason this option fixes Crossplane's duplicate-composite-resource issue). +func TestSyncerServerSideApplyCreate(t *testing.T) { + const stateNamespace = "kcp-system" + clusterName := logicalcluster.Name("testcluster") + + pubRes := &syncagentv1alpha1.PublishedResource{ + Spec: syncagentv1alpha1.PublishedResourceSpec{ + Resource: syncagentv1alpha1.SourceResourceDescriptor{ + APIGroup: dummyv1alpha1.GroupName, + Version: dummyv1alpha1.GroupVersion, + Kind: "Thing", + }, + Projection: &syncagentv1alpha1.ResourceProjection{ + Group: "remote.example.corp", + Kind: "RemoteThing", + }, + Naming: &syncagentv1alpha1.ResourceNaming{ + Name: "$remoteClusterName-$remoteName", + }, + }, + } + + remoteObject := newUnstructured(&dummyv1alpha1.Thing{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-test-thing", + }, + Spec: dummyv1alpha1.ThingSpec{ + Username: "Colonel Mustard", + }, + }, withGroupKind("remote.example.corp", "RemoteThing")) + + localClient := buildFakeClientWithStatus() + remoteClient := buildFakeClientWithStatus(remoteObject) + + syncer, err := NewResourceSyncer( + zap.NewNop().Sugar(), + localClient, + remoteClient, + pubRes, + loadCRD("things"), + func(*syncagentv1alpha1.ResourceMutationSpec) (mutation.Mutator, error) { return nil, nil }, + stateNamespace, + "textor-the-doctor", + WithServerSideApply(), + ) + if err != nil { + t.Fatalf("Failed to create syncer: %v", err) + } + + ctx := t.Context() + ctx = WithClusterName(ctx, clusterName) + ctx = WithEventRecorder(ctx, record.NewFakeRecorder(99)) + + // We loop a couple of times because the finalizer-on-source step still + // happens once before SSA can place the destination object. + for i := 0; i < 5; i++ { + requeue, err := syncer.Process(ctx, remoteObject) + if err != nil { + t.Fatalf("Process failed on iteration %d: %v", i, err) + } + if !requeue { + break + } + if err := remoteClient.Get(ctx, ctrlruntimeclient.ObjectKeyFromObject(remoteObject), remoteObject); err != nil { + t.Fatalf("Failed to re-fetch remote object on iteration %d: %v", i, err) + } + } + + // The local object should now exist and carry the agent label + remote-name annotation. + localKey := ctrlruntimeclient.ObjectKey{Name: "testcluster-my-test-thing"} + got := newUnstructured(&dummyv1alpha1.Thing{}) + if err := localClient.Get(ctx, localKey, got); err != nil { + t.Fatalf("Local object was not created: %v", err) + } + + if got.GetLabels()[agentNameLabel] != "textor-the-doctor" { + t.Errorf("Expected agent label %q on local object, got labels %v", "textor-the-doctor", got.GetLabels()) + } + if got.GetAnnotations()[remoteObjectNameAnnotation] != "my-test-thing" { + t.Errorf("Expected remote-name annotation, got annotations %v", got.GetAnnotations()) + } + + username, found, err := nestedString(got.Object, "spec", "username") + if err != nil || !found || username != "Colonel Mustard" { + t.Errorf("Expected spec.username=Colonel Mustard on local object, got %q (found=%v, err=%v)", username, found, err) + } +} + +// nestedString is a tiny helper to avoid pulling unstructured.NestedString into +// every test that needs to assert on a single field. +func nestedString(obj map[string]any, fields ...string) (string, bool, error) { + current := any(obj) + for _, f := range fields { + m, ok := current.(map[string]any) + if !ok { + return "", false, nil + } + v, ok := m[f] + if !ok { + return "", false, nil + } + current = v + } + s, ok := current.(string) + return s, ok, nil +} + +// TestObjectSyncerFieldManager verifies the field-manager string used for +// Server-Side Apply. It is intentionally stable across restarts so that +// SSA-tracked field ownership is preserved between reconciles. Multi-agent +// setups (the same API synced into multiple kcp's from one service cluster) +// must each get a distinct field manager, otherwise they would overwrite each +// other's fields. +func TestObjectSyncerFieldManager(t *testing.T) { + testcases := []struct { + name string + agentName string + expected string + }{ + { + name: "no agent name uses the bare base", + agentName: "", + expected: "api-syncagent", + }, + { + name: "agent name is appended after a slash", + agentName: "textor-the-doctor", + expected: "api-syncagent/textor-the-doctor", + }, + } + + for _, tc := range testcases { + t.Run(tc.name, func(t *testing.T) { + s := &objectSyncer{agentName: tc.agentName} + if got := s.fieldManager(); got != tc.expected { + t.Errorf("fieldManager() = %q, want %q", got, tc.expected) + } + }) + } +} + +// TestSyncerServerSideApplyFieldOwner verifies that the syncer issues a +// Server-Side Apply call with the expected FieldOwner and ForceOwnership +// options and that the resulting object carries managedFields owned by the +// syncer's stable field manager. This guards against regressions where +// either the on-wire SSA configuration drifts, or where future refactors +// silently fall back to a non-SSA write path (which is what historically +// caused Crossplane's duplicate-composite-resource issue). +func TestSyncerServerSideApplyFieldOwner(t *testing.T) { + const ( + stateNamespace = "kcp-system" + agentName = "textor-the-doctor" + wantFM = "api-syncagent/textor-the-doctor" + ) + clusterName := logicalcluster.Name("testcluster") + + pubRes := &syncagentv1alpha1.PublishedResource{ + Spec: syncagentv1alpha1.PublishedResourceSpec{ + Resource: syncagentv1alpha1.SourceResourceDescriptor{ + APIGroup: dummyv1alpha1.GroupName, + Version: dummyv1alpha1.GroupVersion, + Kind: "Thing", + }, + Projection: &syncagentv1alpha1.ResourceProjection{ + Group: "remote.example.corp", + Kind: "RemoteThing", + }, + Naming: &syncagentv1alpha1.ResourceNaming{ + Name: "$remoteClusterName-$remoteName", + }, + }, + } + + remoteObject := newUnstructured(&dummyv1alpha1.Thing{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-test-thing", + }, + Spec: dummyv1alpha1.ThingSpec{ + Username: "Colonel Mustard", + }, + }, withGroupKind("remote.example.corp", "RemoteThing")) + + // Intercept Apply calls on the local (destination) client so we can + // snapshot the ApplyOptions that the syncer used. + type capture struct { + fieldOwner string + force bool + } + var captures []capture + + // Build the destination fake client with managedFields tracking enabled. + // The default builder strips managedFields on read; with this option the + // fake client preserves them, so we can assert on the entries written by + // the syncer's Apply call. + underlying := newFakeClientBuilder().WithReturnManagedFields().Build() + localClient := interceptor.NewClient(underlying, interceptor.Funcs{ + Apply: func(ctx context.Context, c ctrlruntimeclient.WithWatch, obj runtime.ApplyConfiguration, opts ...ctrlruntimeclient.ApplyOption) error { + ao := &ctrlruntimeclient.ApplyOptions{} + ao.ApplyOptions(opts) + + cp := capture{ + fieldOwner: ao.FieldManager, + } + if ao.Force != nil { + cp.force = *ao.Force + } + captures = append(captures, cp) + + return c.Apply(ctx, obj, opts...) + }, + }) + remoteClient := buildFakeClientWithStatus(remoteObject) + + syncer, err := NewResourceSyncer( + zap.NewNop().Sugar(), + localClient, + remoteClient, + pubRes, + loadCRD("things"), + func(*syncagentv1alpha1.ResourceMutationSpec) (mutation.Mutator, error) { return nil, nil }, + stateNamespace, + agentName, + WithServerSideApply(), + ) + if err != nil { + t.Fatalf("Failed to create syncer: %v", err) + } + + ctx := t.Context() + ctx = WithClusterName(ctx, clusterName) + ctx = WithEventRecorder(ctx, record.NewFakeRecorder(99)) + + for i := 0; i < 5; i++ { + requeue, err := syncer.Process(ctx, remoteObject) + if err != nil { + t.Fatalf("Process failed on iteration %d: %v", i, err) + } + if !requeue { + break + } + if err := remoteClient.Get(ctx, ctrlruntimeclient.ObjectKeyFromObject(remoteObject), remoteObject); err != nil { + t.Fatalf("Failed to re-fetch remote object on iteration %d: %v", i, err) + } + } + + // 1. Assert on the on-wire SSA configuration: every Apply call must use the + // syncer's stable field manager and must set ForceOwnership=true. + if len(captures) == 0 { + t.Fatalf("Expected at least one Apply call on the destination client; none were captured") + } + for i, c := range captures { + if c.fieldOwner != wantFM { + t.Errorf("Apply call %d used FieldOwner %q, want %q", i, c.fieldOwner, wantFM) + } + if !c.force { + t.Errorf("Apply call %d did not set ForceOwnership=true (captured force=%v)", i, c.force) + } + } + + // 2. Assert on the resulting object: managedFields must contain an entry + // owned by the syncer with Operation=Apply. This is what actually + // causes the kube-apiserver to preserve fields owned by other + // controllers (e.g. Crossplane's spec.resourceRef writes). + localKey := ctrlruntimeclient.ObjectKey{Name: "testcluster-my-test-thing"} + got := newUnstructured(&dummyv1alpha1.Thing{}) + if err := localClient.Get(ctx, localKey, got); err != nil { + t.Fatalf("Local object was not created: %v", err) + } + + mfs := got.GetManagedFields() + if len(mfs) == 0 { + t.Fatalf("Expected managedFields on the destination object, got none") + } + + var syncerEntry *metav1.ManagedFieldsEntry + for i := range mfs { + if mfs[i].Manager == wantFM { + syncerEntry = &mfs[i] + break + } + } + if syncerEntry == nil { + var managers []string + for _, e := range mfs { + managers = append(managers, e.Manager) + } + t.Fatalf("Expected a managedFields entry owned by %q, found managers: %v", wantFM, managers) + } + + if syncerEntry.Operation != metav1.ManagedFieldsOperationApply { + t.Errorf("Expected managedFields entry operation %q, got %q", metav1.ManagedFieldsOperationApply, syncerEntry.Operation) + } + if syncerEntry.FieldsV1 == nil || len(syncerEntry.FieldsV1.Raw) == 0 { + t.Errorf("Expected managedFields entry %q to declare owned fields (FieldsV1 was empty)", wantFM) + } +}