feat: add Server-Side Apply for object synchronization

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
 	}

cmd/api-syncagent/options.go

@@ -71,13 +71,21 @@ 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 {
 	return &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 {

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)

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,25 +85,27 @@ func Add(
 	prFilter labels.Selector,
 	stateNamespace string,
 	agentName string,
+	enableServerSideApply bool,
 ) error {
 	discoveryClient, err := discovery.NewClient(localManager.GetConfig())
 	if err != nil {
 		return fmt.Errorf("failed to create discovery client: %w", err)
 	}
 
 	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,
 	)

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)

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,