From bd9e96faf48b73853c60d1a8e6678c7975f18ce5 Mon Sep 17 00:00:00 2001 From: Arnaud FEVRIER Date: Mon, 19 Oct 2026 17:53:35 +0200 Subject: [PATCH 1/2] fix: sync forward status with server-side apply Server-Side Apply removes status from the main-resource payload, so synchronize forward status separately through the status subresource. Signed-off-by: Arnaud FEVRIER --- internal/sync/object_syncer.go | 6 +++++ internal/sync/syncer_ssa_test.go | 41 ++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/internal/sync/object_syncer.go b/internal/sync/object_syncer.go index 2404f91..732b71c 100644 --- a/internal/sync/object_syncer.go +++ b/internal/sync/object_syncer.go @@ -159,6 +159,12 @@ func (s *objectSyncer) Sync(ctx context.Context, log *zap.SugaredLogger, source, return false, nil } + // Status is removed from the SSA payload and must be synchronized through + // the status subresource separately. + if err := s.syncObjectStatusForward(ctx, log, source, newDest); err != nil { + return false, err + } + return s.syncObjectStatus(ctx, log, source, newDest) } diff --git a/internal/sync/syncer_ssa_test.go b/internal/sync/syncer_ssa_test.go index d42a090..075b717 100644 --- a/internal/sync/syncer_ssa_test.go +++ b/internal/sync/syncer_ssa_test.go @@ -29,6 +29,7 @@ import ( "github.com/kcp-dev/logicalcluster/v3" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/tools/record" ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" @@ -180,6 +181,46 @@ func TestObjectSyncerFieldManager(t *testing.T) { } } +func TestObjectSyncerServerSideApplySyncsForwardStatus(t *testing.T) { + const readyPhase = "ready" + + source := makeUnstructuredWithStatus("my-widget", "", map[string]interface{}{"phase": readyPhase}) + dest := makeUnstructuredWithStatus("my-widget", "", map[string]interface{}{"phase": "pending"}) + destClient := buildFakeClientWithStatus(dest) + + syncer := &objectSyncer{ + useServerSideApply: true, + syncStatusForward: true, + subresources: []string{"status"}, + destCreator: func(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) { + return obj.DeepCopy(), nil + }, + eventObjSide: syncSideSource, + } + + ctx := WithEventRecorder(t.Context(), record.NewFakeRecorder(10)) + if _, err := syncer.Sync( + ctx, + zap.NewNop().Sugar(), + syncSide{object: source}, + syncSide{object: dest, client: destClient}, + ); err != nil { + t.Fatalf("Sync returned an error: %v", err) + } + + got := makeUnstructuredWithStatus("my-widget", "", nil) + if err := destClient.Get(ctx, ctrlruntimeclient.ObjectKeyFromObject(dest), got); err != nil { + t.Fatalf("Failed to get destination object: %v", err) + } + phase, _, err := unstructured.NestedString(got.Object, "status", "phase") + if err != nil { + t.Fatalf("Failed to read destination status: %v", err) + } + if phase != readyPhase { + t.Errorf("Expected destination status.phase=ready, got %q", phase) + } +} + // 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 From 247292c0f3e4a62e8f98748984e29d3a822bc544 Mon Sep 17 00:00:00 2001 From: Arnaud FEVRIER Date: Tue, 20 Oct 2026 13:36:25 +0200 Subject: [PATCH 2/2] feat: configure related resource status source Signed-off-by: Arnaud FEVRIER --- .../syncagent.kcp.io_publishedresources.yaml | 30 +- internal/controller/sync/controller.go | 112 +++---- internal/sync/object_syncer.go | 2 +- internal/sync/syncer_related.go | 12 +- .../syncagent/v1alpha1/published_resource.go | 26 +- .../v1alpha1/zz_generated.deepcopy.go | 5 + .../syncagent/v1alpha1/relatedresourcespec.go | 4 +- test/crds/backupswithstatus.yaml | 27 ++ test/e2e/sdk/cel_validations_test.go | 46 ++- test/e2e/sync/related_status_test.go | 301 +++++++++++++++++- 10 files changed, 462 insertions(+), 103 deletions(-) create mode 100644 test/crds/backupswithstatus.yaml diff --git a/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml b/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml index 63c39ac..01901e0 100644 --- a/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml +++ b/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml @@ -807,17 +807,19 @@ spec: type: string syncStatus: description: |- - SyncStatus enables synchronization of the status subresource in the same direction as - the spec (from the origin side to the destination side). When enabled, the agent will - use the status subresource endpoint to update the destination object's status. - This requires the related resource to have a status subresource configured in its CRD. + SyncStatus configures the side from which the status subresource is synchronized. The + agent uses the status subresource endpoint to update the object on the opposite side. + Both representations of the related resource must have a status subresource configured. - - origin: kcp -> status is synced from kcp to the service cluster - - origin: service -> status is synced from the service cluster to kcp + - kcp: status is synced from kcp to the service cluster + - service: status is synced from the service cluster to kcp - For origin: service, Watch must also be configured so that changes to the related - resource's status trigger reconciliation and ensure the informer cache is populated. - type: boolean + Except when both Origin and SyncStatus are kcp, Watch must also be configured so that + status changes trigger reconciliation and the required informer caches are populated. + enum: + - service + - kcp + type: string version: description: |- Version is the API version of the related resource. This can be left blank to automatically @@ -826,9 +828,9 @@ spec: watch: description: |- Watch configures how the agent identifies the owning primary object when a related - resource with origin: kcp changes. When set, the agent sets up a watch on the related - resource type and uses the configured rule to enqueue the correct primary object. - Without this field, changes to origin:kcp related resources do not trigger reconciliation. + resource changes. The agent always watches the origin side. When SyncStatus selects the + opposite side, the agent watches that side as well. Without this field, changes to related + resources do not trigger reconciliation. properties: byOwner: description: |- @@ -901,8 +903,8 @@ spec: rule: '!has(self.group) || (has(self.resource) && has(self.version))' - message: identity hashes can only be used with GVRs rule: '!has(self.identityHash) || (has(self.group) && has(self.version) && has(self.resource))' - - message: watch must be configured when origin is service and syncStatus is true - rule: '!(self.origin == ''service'' && has(self.syncStatus) && self.syncStatus) || has(self.watch)' + - message: watch must be configured unless origin and syncStatus are both kcp + rule: '!has(self.syncStatus) || (self.origin == ''kcp'' && self.syncStatus == ''kcp'') || has(self.watch)' - message: watch must be configured when cleanupPolicy is MatchOrigin and origin is service rule: '!(has(self.cleanupPolicy) && self.cleanupPolicy == ''MatchOrigin'' && self.origin == ''service'') || has(self.watch)' - message: 'cleanup:true conflicts with cleanupPolicy: Orphan' diff --git a/internal/controller/sync/controller.go b/internal/controller/sync/controller.go index 207d148..9b8ff81 100644 --- a/internal/controller/sync/controller.go +++ b/internal/controller/sync/controller.go @@ -36,7 +36,6 @@ import ( corev1 "k8s.io/api/core/v1" apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" - "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/labels" @@ -171,7 +170,7 @@ func Create( return nil, fmt.Errorf("failed to setup local-side watch: %w", err) } - if err := setupRelatedResourceWatches(c, localManager, remoteManager, pubRes, localDummy, remoteDummy, log); err != nil { + if err := setupRelatedResourceWatches(c, localManager, pubRes, localDummy, remoteDummy, log); err != nil { return nil, err } @@ -180,18 +179,18 @@ func Create( return c, nil } -// setupRelatedResourceWatches sets up watches for all related resources that have a Watch -// config, on their respective origin side, so that changes trigger primary reconciliation. +// setupRelatedResourceWatches sets up watches for all related resources that have a Watch config. +// The origin side is always watched. When status is synchronized from the opposite side, that side +// is watched as well so changes in either direction trigger primary reconciliation. func setupRelatedResourceWatches( c mccontroller.Controller, localManager manager.Manager, - remoteManager mcmanager.Manager, pubRes *syncagentv1alpha1.PublishedResource, localDummy, remoteDummy *unstructured.Unstructured, log *zap.SugaredLogger, ) error { - // Deduplication is per-origin to allow the same GVK on both sides. - watchedKcpGVKs := sets.New[schema.GroupVersionKind]() + // Deduplication is per-side to allow the same resource on both sides. + watchedKcpGVRs := sets.New[schema.GroupVersionResource]() watchedServiceGVKs := sets.New[schema.GroupVersionKind]() for _, relRes := range pubRes.Spec.Related { @@ -199,67 +198,70 @@ func setupRelatedResourceWatches( continue } - // this handles the legacy .Kind-based related resources - gvr := projection.RelatedResourceGVR(&relRes) - - // Use the REST mapper of the origin side: related resources may have projected GVKs - // that differ between kcp and the service cluster, so we must resolve using the - // mapper that actually knows about the GVR on that side. - var originRESTMapper meta.RESTMapper - if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginKcp { - originRESTMapper = remoteManager.GetLocalManager().GetRESTMapper() - } else { - originRESTMapper = localManager.GetRESTMapper() - } - - gvk, err := originRESTMapper.KindFor(gvr) - if err != nil { - return fmt.Errorf("failed to determine Kind for related resource %v (origin: %s): %w", gvr, relRes.Origin, err) - } + watchKcp := relRes.Origin == syncagentv1alpha1.RelatedResourceOriginKcp || + (relRes.SyncStatus != nil && *relRes.SyncStatus == syncagentv1alpha1.RelatedResourceOriginKcp) + if watchKcp { + kcpGVR := projection.RelatedResourceKcpGVR(&relRes) + if !watchedKcpGVRs.Has(kcpGVR) { + enqueueForRelated, err := buildKcpRelatedHandler(relRes.Watch, kcpGVR, remoteDummy, log) + if err != nil { + return err + } - relatedDummy := &unstructured.Unstructured{} - relatedDummy.SetGroupVersionKind(gvk) + relatedSource := mcsource.TypedKind(&unstructured.Unstructured{}, enqueueForRelated).WithProjection( + func(cl cluster.Cluster, obj *unstructured.Unstructured) (*unstructured.Unstructured, error) { + gvk, err := cl.GetRESTMapper().KindFor(kcpGVR) + if err != nil { + return nil, fmt.Errorf("failed to determine Kind for kcp-side related resource %v: %w", kcpGVR, err) + } + + projected := obj.DeepCopy() + projected.SetGroupVersionKind(gvk) + return projected, nil + }, + ) + if err := c.MultiClusterWatch(relatedSource); err != nil { + return fmt.Errorf("failed to setup watch for related resource %v on kcp side: %w", kcpGVR, err) + } - if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginKcp { - if watchedKcpGVKs.Has(gvk) { - continue + watchedKcpGVRs.Insert(kcpGVR) + log.Infow("Set up watch for related resource", "gvr", kcpGVR, "side", syncagentv1alpha1.RelatedResourceOriginKcp) } - watchedKcpGVKs.Insert(gvk) + } - enqueueForRelated, err := buildKcpRelatedHandler(relRes.Watch, gvk, remoteDummy, log) + watchService := relRes.Origin == syncagentv1alpha1.RelatedResourceOriginService || + (relRes.SyncStatus != nil && *relRes.SyncStatus == syncagentv1alpha1.RelatedResourceOriginService) + if watchService { + serviceGVR := projection.RelatedResourceServiceGVK(&relRes) + gvk, err := localManager.GetRESTMapper().KindFor(serviceGVR) if err != nil { - return err + return fmt.Errorf("failed to determine Kind for service-side related resource %v: %w", serviceGVR, err) } + if !watchedServiceGVKs.Has(gvk) { + relatedDummy := &unstructured.Unstructured{} + relatedDummy.SetGroupVersionKind(gvk) - if err := c.MultiClusterWatch(mcsource.TypedKind(relatedDummy, enqueueForRelated)); err != nil { - return fmt.Errorf("failed to setup watch for kcp-origin related resource %v: %w", gvk, err) - } - } else { - if watchedServiceGVKs.Has(gvk) { - continue - } - watchedServiceGVKs.Insert(gvk) - - enqueueForRelated, err := buildServiceRelatedHandler(relRes.Watch, gvk, localDummy, localManager, log) - if err != nil { - return err - } + enqueueForRelated, err := buildServiceRelatedHandler(relRes.Watch, gvk, localDummy, localManager, log) + if err != nil { + return err + } + if err := c.Watch(source.TypedKind(localManager.GetCache(), relatedDummy, enqueueForRelated)); err != nil { + return fmt.Errorf("failed to setup watch for related resource %v on service side: %w", gvk, err) + } - if err := c.Watch(source.TypedKind(localManager.GetCache(), relatedDummy, enqueueForRelated)); err != nil { - return fmt.Errorf("failed to setup watch for service-origin related resource %v: %w", gvk, err) + watchedServiceGVKs.Insert(gvk) + log.Infow("Set up watch for related resource", "gvk", gvk, "side", syncagentv1alpha1.RelatedResourceOriginService) } } - - log.Infow("Set up watch for related resource", "gvk", gvk, "origin", relRes.Origin) } return nil } -// buildKcpRelatedHandler constructs the per-cluster event handler for a kcp-origin related resource. +// buildKcpRelatedHandler constructs the per-cluster event handler for a kcp-side related resource. func buildKcpRelatedHandler( watch *syncagentv1alpha1.RelatedResourceWatch, - gvk schema.GroupVersionKind, + gvr schema.GroupVersionResource, remoteDummy *unstructured.Unstructured, log *zap.SugaredLogger, ) (mchandler.TypedEventHandlerFunc[*unstructured.Unstructured, mcreconcile.Request], error) { @@ -287,11 +289,11 @@ func buildKcpRelatedHandler( }, nil default: - return nil, fmt.Errorf("related resource %v (origin: kcp) has Watch set but neither byOwner nor bySelector configured", gvk) + return nil, fmt.Errorf("related resource %v on kcp side has Watch set but neither byOwner nor bySelector configured", gvr) } } -// buildServiceRelatedHandler constructs the event handler for a service-cluster-origin related resource. +// buildServiceRelatedHandler constructs the event handler for a service-side related resource. // It maps the changed related resource back to the remote (kcp) primary via sync metadata on the local primary. func buildServiceRelatedHandler( watch *syncagentv1alpha1.RelatedResourceWatch, @@ -328,7 +330,7 @@ func buildServiceRelatedHandler( case watch.BySelector != nil: selector, err := metav1.LabelSelectorAsSelector(watch.BySelector) if err != nil { - return nil, fmt.Errorf("failed to convert bySelector for service-origin related resource %v: %w", gvk, err) + return nil, fmt.Errorf("failed to convert bySelector for service-side related resource %v: %w", gvk, err) } primaryDummy := localDummy.DeepCopy() return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, _ *unstructured.Unstructured) []mcreconcile.Request { @@ -349,7 +351,7 @@ func buildServiceRelatedHandler( }), nil default: - return nil, fmt.Errorf("related resource %v (origin: service) has Watch set but neither byOwner nor bySelector configured", gvk) + return nil, fmt.Errorf("related resource %v on service side has Watch set but neither byOwner nor bySelector configured", gvk) } } diff --git a/internal/sync/object_syncer.go b/internal/sync/object_syncer.go index 732b71c..500950e 100644 --- a/internal/sync/object_syncer.go +++ b/internal/sync/object_syncer.go @@ -405,7 +405,7 @@ func (s *objectSyncer) syncObjectStatusForward(ctx context.Context, log *zap.Sug if apierrors.IsNotFound(err) { // The /status subresource does not exist on the destination CRD. // Retrying will not help; emit a warning so the user knows what to fix. - s.recordEvent(ctx, source, dest, corev1.EventTypeWarning, "StatusSubresourceMissing", "Cannot sync status: the destination resource does not have a status subresource. Set syncStatus: false or add subresources.status to the CRD.") + s.recordEvent(ctx, source, dest, corev1.EventTypeWarning, "StatusSubresourceMissing", "Cannot sync status: the destination resource does not have a status subresource. Unset syncStatus or add subresources.status to the CRD.") return nil } return fmt.Errorf("failed to update destination object status: %w", err) diff --git a/internal/sync/syncer_related.go b/internal/sync/syncer_related.go index d31e4b4..7c93f8d 100644 --- a/internal/sync/syncer_related.go +++ b/internal/sync/syncer_related.go @@ -154,11 +154,13 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su deletionPolicies[relatedCopyKey(resolved.destination.Namespace, resolved.destination.Name)] = deletionPolicy // When status sync is enabled, include "status" in subresources so it is stripped from - // the spec patch (avoiding a no-op write on resources that have a status subresource). - // The status is then separately written via the status subresource endpoint by syncStatusForward. + // the spec patch and written separately via the status subresource endpoint. relatedSubresources := []string(nil) - if relRes.SyncStatus { + syncStatusForward, syncStatusBack := false, false + if relRes.SyncStatus != nil { relatedSubresources = []string{"status"} + syncStatusForward = *relRes.SyncStatus == relRes.Origin + syncStatusBack = *relRes.SyncStatus != relRes.Origin } syncer := objectSyncer{ @@ -178,8 +180,8 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su return dest, nil }, subresources: relatedSubresources, - syncStatusBack: false, - syncStatusForward: relRes.SyncStatus, + syncStatusBack: syncStatusBack, + syncStatusForward: syncStatusForward, // if the origin is on the remote side, we want to add a finalizer to make // sure we can clean up properly; when forceDelete is enabled, we are in deletion mode and // want to force the deletion, so we ignore the related object's origin (it was taken into diff --git a/sdk/apis/syncagent/v1alpha1/published_resource.go b/sdk/apis/syncagent/v1alpha1/published_resource.go index 0e6fdde..2e77f74 100644 --- a/sdk/apis/syncagent/v1alpha1/published_resource.go +++ b/sdk/apis/syncagent/v1alpha1/published_resource.go @@ -229,7 +229,7 @@ const ( // +kubebuilder:validation:XValidation:rule="!has(self.group) || (has(self.resource) && has(self.version))",message="configuring a group also requires a version and resource" // group is included here because when an identityHash is used, core/v1 cannot possible be targetted // +kubebuilder:validation:XValidation:rule="!has(self.identityHash) || (has(self.group) && has(self.version) && has(self.resource))",message="identity hashes can only be used with GVRs" -// +kubebuilder:validation:XValidation:rule="!(self.origin == 'service' && has(self.syncStatus) && self.syncStatus) || has(self.watch)",message="watch must be configured when origin is service and syncStatus is true" +// +kubebuilder:validation:XValidation:rule="!has(self.syncStatus) || (self.origin == 'kcp' && self.syncStatus == 'kcp') || has(self.watch)",message="watch must be configured unless origin and syncStatus are both kcp" // +kubebuilder:validation:XValidation:rule="!(has(self.cleanupPolicy) && self.cleanupPolicy == 'MatchOrigin' && self.origin == 'service') || has(self.watch)",message="watch must be configured when cleanupPolicy is MatchOrigin and origin is service" // +kubebuilder:validation:XValidation:rule="!(has(self.cleanup) && self.cleanup && has(self.cleanupPolicy) && self.cleanupPolicy == 'Orphan')",message="cleanup:true conflicts with cleanupPolicy: Orphan" type RelatedResourceSpec struct { @@ -308,24 +308,24 @@ type RelatedResourceSpec struct { Mutation *ResourceMutationSpec `json:"mutation,omitempty"` // Watch configures how the agent identifies the owning primary object when a related - // resource with origin: kcp changes. When set, the agent sets up a watch on the related - // resource type and uses the configured rule to enqueue the correct primary object. - // Without this field, changes to origin:kcp related resources do not trigger reconciliation. + // resource changes. The agent always watches the origin side. When SyncStatus selects the + // opposite side, the agent watches that side as well. Without this field, changes to related + // resources do not trigger reconciliation. Watch *RelatedResourceWatch `json:"watch,omitempty"` - // SyncStatus enables synchronization of the status subresource in the same direction as - // the spec (from the origin side to the destination side). When enabled, the agent will - // use the status subresource endpoint to update the destination object's status. - // This requires the related resource to have a status subresource configured in its CRD. + // SyncStatus configures the side from which the status subresource is synchronized. The + // agent uses the status subresource endpoint to update the object on the opposite side. + // Both representations of the related resource must have a status subresource configured. // - // - origin: kcp -> status is synced from kcp to the service cluster - // - origin: service -> status is synced from the service cluster to kcp + // - kcp: status is synced from kcp to the service cluster + // - service: status is synced from the service cluster to kcp // - // For origin: service, Watch must also be configured so that changes to the related - // resource's status trigger reconciliation and ensure the informer cache is populated. + // Except when both Origin and SyncStatus are kcp, Watch must also be configured so that + // status changes trigger reconciliation and the required informer caches are populated. // + // +kubebuilder:validation:Enum=service;kcp // +optional - SyncStatus bool `json:"syncStatus,omitempty"` + SyncStatus *RelatedResourceOrigin `json:"syncStatus,omitempty"` } // EffectiveCleanupPolicy normalizes the deprecated Cleanup bool and the CleanupPolicy field into diff --git a/sdk/apis/syncagent/v1alpha1/zz_generated.deepcopy.go b/sdk/apis/syncagent/v1alpha1/zz_generated.deepcopy.go index 45a1cec..afade2b 100644 --- a/sdk/apis/syncagent/v1alpha1/zz_generated.deepcopy.go +++ b/sdk/apis/syncagent/v1alpha1/zz_generated.deepcopy.go @@ -309,6 +309,11 @@ func (in *RelatedResourceSpec) DeepCopyInto(out *RelatedResourceSpec) { *out = new(RelatedResourceWatch) (*in).DeepCopyInto(*out) } + if in.SyncStatus != nil { + in, out := &in.SyncStatus, &out.SyncStatus + *out = new(RelatedResourceOrigin) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RelatedResourceSpec. diff --git a/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go b/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go index 4196a14..76c5ab2 100644 --- a/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go +++ b/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go @@ -38,7 +38,7 @@ type RelatedResourceSpecApplyConfiguration struct { Object *RelatedResourceObjectApplyConfiguration `json:"object,omitempty"` Mutation *ResourceMutationSpecApplyConfiguration `json:"mutation,omitempty"` Watch *RelatedResourceWatchApplyConfiguration `json:"watch,omitempty"` - SyncStatus *bool `json:"syncStatus,omitempty"` + SyncStatus *syncagentv1alpha1.RelatedResourceOrigin `json:"syncStatus,omitempty"` } // RelatedResourceSpecApplyConfiguration constructs a declarative configuration of the RelatedResourceSpec type for use with @@ -154,7 +154,7 @@ func (b *RelatedResourceSpecApplyConfiguration) WithWatch(value *RelatedResource // WithSyncStatus sets the SyncStatus field in the declarative configuration to the given value // and returns the receiver, so that objects can be built by chaining "With" function invocations. // If called multiple times, the SyncStatus field is set to the value of the last call. -func (b *RelatedResourceSpecApplyConfiguration) WithSyncStatus(value bool) *RelatedResourceSpecApplyConfiguration { +func (b *RelatedResourceSpecApplyConfiguration) WithSyncStatus(value syncagentv1alpha1.RelatedResourceOrigin) *RelatedResourceSpecApplyConfiguration { b.SyncStatus = &value return b } diff --git a/test/crds/backupswithstatus.yaml b/test/crds/backupswithstatus.yaml new file mode 100644 index 0000000..fc7575c --- /dev/null +++ b/test/crds/backupswithstatus.yaml @@ -0,0 +1,27 @@ +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: backupswithstatus.eksempel.no +spec: + group: eksempel.no + scope: Namespaced + names: + plural: backupswithstatus + singular: backupwithstatus + kind: BackupWithStatus + versions: + - name: v1 + served: true + storage: true + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + properties: + spec: + type: object + x-kubernetes-preserve-unknown-fields: true + status: + type: object + x-kubernetes-preserve-unknown-fields: true diff --git a/test/e2e/sdk/cel_validations_test.go b/test/e2e/sdk/cel_validations_test.go index 999f00f..fae78a1 100644 --- a/test/e2e/sdk/cel_validations_test.go +++ b/test/e2e/sdk/cel_validations_test.go @@ -27,6 +27,7 @@ import ( "github.com/kcp-dev/api-syncagent/test/utils" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" ) func TestValidateRelatedResourceSpec(t *testing.T) { @@ -177,36 +178,69 @@ func TestValidateRelatedResourceSpec(t *testing.T) { }, }, { - name: "syncStatus with origin service requires watch", + name: "syncStatus service requires watch", valid: false, spec: syncagentv1alpha1.RelatedResourceSpec{ Origin: syncagentv1alpha1.RelatedResourceOriginService, Resource: "things", Version: "v1", - SyncStatus: true, + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginService), }, }, { - name: "syncStatus with origin service and watch is valid", + name: "syncStatus service with watch is valid", valid: true, spec: syncagentv1alpha1.RelatedResourceSpec{ Origin: syncagentv1alpha1.RelatedResourceOriginService, Resource: "things", Version: "v1", - SyncStatus: true, + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginService), Watch: &syncagentv1alpha1.RelatedResourceWatch{ ByOwner: &syncagentv1alpha1.RelatedResourceWatchByOwner{}, }, }, }, { - name: "syncStatus with origin kcp does not require watch", + name: "syncStatus kcp does not require watch", valid: true, spec: syncagentv1alpha1.RelatedResourceSpec{ Origin: syncagentv1alpha1.RelatedResourceOriginKcp, Resource: "things", Version: "v1", - SyncStatus: true, + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginKcp), + }, + }, + { + name: "syncStatus service with origin kcp requires watch", + valid: false, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginKcp, + Resource: "things", + Version: "v1", + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginService), + }, + }, + { + name: "syncStatus kcp with origin service requires watch", + valid: false, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Resource: "things", + Version: "v1", + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginKcp), + }, + }, + { + name: "syncStatus kcp with origin service and watch is valid", + valid: true, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Resource: "things", + Version: "v1", + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginKcp), + Watch: &syncagentv1alpha1.RelatedResourceWatch{ + ByOwner: &syncagentv1alpha1.RelatedResourceWatchByOwner{}, + }, }, }, { diff --git a/test/e2e/sync/related_status_test.go b/test/e2e/sync/related_status_test.go index 0cf8aa7..216a759 100644 --- a/test/e2e/sync/related_status_test.go +++ b/test/e2e/sync/related_status_test.go @@ -36,6 +36,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/utils/ptr" ctrlruntime "sigs.k8s.io/controller-runtime" ) @@ -51,7 +52,7 @@ func makeCronTabWithStatus(name, namespace string) *unstructured.Unstructured { return obj } -// TestSyncRelatedObjectStatusFromKcp verifies that when syncStatus: true is set on a related +// TestSyncRelatedObjectStatusFromKcp verifies that when syncStatus: kcp is set on a related // resource with origin: kcp, the status is propagated from kcp to the service cluster. func TestSyncRelatedObjectStatusFromKcp(t *testing.T) { const apiExportName = "kcp.example.com" @@ -86,7 +87,7 @@ func TestSyncRelatedObjectStatusFromKcp(t *testing.T) { t.Fatalf("Failed to create CronTabWithStatus PublishedResource: %v", err) } - // Publish CronTabs with a related CronTabWithStatus (origin: kcp, syncStatus: true). + // Publish CronTabs with a related CronTabWithStatus (origin: kcp, syncStatus: kcp). prCrontabs := &syncagentv1alpha1.PublishedResource{ ObjectMeta: metav1.ObjectMeta{Name: "publish-crontabs"}, Spec: syncagentv1alpha1.PublishedResourceSpec{ @@ -110,7 +111,7 @@ func TestSyncRelatedObjectStatusFromKcp(t *testing.T) { Projection: &syncagentv1alpha1.RelatedResourceProjection{ Group: "example.com", }, - SyncStatus: true, + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginKcp), Object: syncagentv1alpha1.RelatedResourceObject{ RelatedResourceObjectSpec: syncagentv1alpha1.RelatedResourceObjectSpec{ Template: &syncagentv1alpha1.TemplateExpression{ @@ -228,7 +229,155 @@ func TestSyncRelatedObjectStatusFromKcp(t *testing.T) { } } -// TestSyncRelatedObjectStatusFromService verifies that when syncStatus: true is set on a related +// TestSyncRelatedObjectStatusBackToKcp verifies that the spec and status of a related resource can +// be synchronized in opposite directions. It updates the service-side status after both copies +// exist to also verify that the service-side related-resource watch triggers reconciliation. +func TestSyncRelatedObjectStatusBackToKcp(t *testing.T) { + const apiExportName = "kcp.example.com" + + ctrlruntime.SetLogger(logr.Discard()) + + ctx := t.Context() + orgKubconfig := utils.CreateOrganization(t, ctx, "related-status-back-to-kcp", apiExportName) + + envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{ + "test/crds/crontab.yaml", + "test/crds/crontabswithstatus.yaml", + }) + + prCronTabsWithStatus := &syncagentv1alpha1.PublishedResource{ + ObjectMeta: metav1.ObjectMeta{Name: "publish-crontabswithstatus"}, + Spec: syncagentv1alpha1.PublishedResourceSpec{ + Resource: syncagentv1alpha1.SourceResourceDescriptor{ + APIGroup: "example.com", + Version: "v1", + Kind: "CronTabWithStatus", + }, + Projection: &syncagentv1alpha1.ResourceProjection{Group: "kcp.example.com"}, + Synchronization: &syncagentv1alpha1.SynchronizationSpec{Enabled: false}, + }, + } + if err := envtestClient.Create(ctx, prCronTabsWithStatus); err != nil { + t.Fatalf("Failed to create CronTabWithStatus PublishedResource: %v", err) + } + + prCrontabs := &syncagentv1alpha1.PublishedResource{ + ObjectMeta: metav1.ObjectMeta{Name: "publish-crontabs"}, + Spec: syncagentv1alpha1.PublishedResourceSpec{ + Resource: syncagentv1alpha1.SourceResourceDescriptor{ + APIGroup: "example.com", + Version: "v1", + Kind: "CronTab", + }, + Naming: &syncagentv1alpha1.ResourceNaming{Name: "{{ .Object.metadata.name }}", Namespace: "synced-{{ .Object.metadata.namespace }}"}, + Projection: &syncagentv1alpha1.ResourceProjection{Group: "kcp.example.com"}, + Related: []syncagentv1alpha1.RelatedResourceSpec{ + { + Identifier: "crontabwithstatus", + Origin: syncagentv1alpha1.RelatedResourceOriginKcp, + Group: "kcp.example.com", + Version: "v1", + Resource: "crontabswithstatus", + Projection: &syncagentv1alpha1.RelatedResourceProjection{Group: "example.com"}, + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginService), + Watch: &syncagentv1alpha1.RelatedResourceWatch{ + BySelector: &metav1.LabelSelector{}, + }, + Object: syncagentv1alpha1.RelatedResourceObject{ + RelatedResourceObjectSpec: syncagentv1alpha1.RelatedResourceObjectSpec{ + Template: &syncagentv1alpha1.TemplateExpression{Template: "my-related"}, + }, + }, + }, + }, + }, + } + if err := envtestClient.Create(ctx, prCrontabs); err != nil { + t.Fatalf("Failed to create CronTab PublishedResource: %v", err) + } + + utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "") + + kcpClusterClient := utils.GetKcpAdminClusterClient(t) + teamClusterPath := logicalcluster.NewPath("root").Join("related-status-back-to-kcp").Join("team-1") + teamClient := kcpClusterClient.Cluster(teamClusterPath) + + utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{ + Group: apiExportName, Version: "v1", Kind: "CronTab", + }) + utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{ + Group: apiExportName, Version: "v1", Kind: "CronTabWithStatus", + }) + + kcpRelated := makeCronTabWithStatus("my-related", "default") + kcpRelated.SetAPIVersion("kcp.example.com/v1") + if err := unstructured.SetNestedField(kcpRelated.Object, "from-kcp", "spec", "cronSpec"); err != nil { + t.Fatalf("Failed to set spec.cronSpec: %v", err) + } + if err := teamClient.Create(ctx, kcpRelated); err != nil { + t.Fatalf("Failed to create related resource in kcp: %v", err) + } + + primary := utils.ToUnstructured(t, &crds.Crontab{ + TypeMeta: metav1.TypeMeta{APIVersion: "kcp.example.com/v1", Kind: "CronTab"}, + ObjectMeta: metav1.ObjectMeta{Name: "my-crontab", Namespace: "default"}, + Spec: crds.CrontabSpec{CronSpec: "* * *", Image: "ubuntu:latest"}, + }) + if err := teamClient.Create(ctx, primary); err != nil { + t.Fatalf("Failed to create primary CronTab: %v", err) + } + + serviceRelated := makeCronTabWithStatus("my-related", "synced-default") + err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 3*time.Minute, false, func(ctx context.Context) (bool, error) { + if err := envtestClient.Get(ctx, types.NamespacedName{Name: "my-related", Namespace: "synced-default"}, serviceRelated); err != nil { + return false, nil + } + cronSpec, _, _ := unstructured.NestedString(serviceRelated.Object, "spec", "cronSpec") + return cronSpec == "from-kcp", nil + }) + if err != nil { + t.Fatalf("Related resource spec was not synchronized to the service cluster: %v", err) + } + + if err := unstructured.SetNestedField(serviceRelated.Object, "from-service", "status", "id"); err != nil { + t.Fatalf("Failed to set status.id: %v", err) + } + if err := envtestClient.Status().Update(ctx, serviceRelated); err != nil { + t.Fatalf("Failed to update service-side status: %v", err) + } + + err = wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 3*time.Minute, false, func(ctx context.Context) (bool, error) { + if err := teamClient.Get(ctx, types.NamespacedName{Name: "my-related", Namespace: "default"}, kcpRelated); err != nil { + return false, nil + } + id, _, _ := unstructured.NestedString(kcpRelated.Object, "status", "id") + return id == "from-service", nil + }) + if err != nil { + t.Fatalf("Related resource status was not synchronized back to kcp: %v", err) + } + + // The origin watch must remain active when status is synchronized in the opposite direction. + if err := unstructured.SetNestedField(kcpRelated.Object, "updated-from-kcp", "spec", "cronSpec"); err != nil { + t.Fatalf("Failed to update spec.cronSpec: %v", err) + } + if err := teamClient.Update(ctx, kcpRelated); err != nil { + t.Fatalf("Failed to update kcp-side spec: %v", err) + } + + err = wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 3*time.Minute, false, func(ctx context.Context) (bool, error) { + if err := envtestClient.Get(ctx, types.NamespacedName{Name: "my-related", Namespace: "synced-default"}, serviceRelated); err != nil { + return false, nil + } + cronSpec, _, _ := unstructured.NestedString(serviceRelated.Object, "spec", "cronSpec") + return cronSpec == "updated-from-kcp", nil + }) + if err != nil { + t.Fatalf("Related resource spec update was not synchronized to the service cluster: %v", err) + } +} + +// TestSyncRelatedObjectStatusFromService verifies that when syncStatus: service is set on a related // resource with origin: service, the status is propagated from the service cluster to kcp. func TestSyncRelatedObjectStatusFromService(t *testing.T) { const apiExportName = "kcp.example.com" @@ -264,7 +413,7 @@ func TestSyncRelatedObjectStatusFromService(t *testing.T) { t.Fatalf("Failed to create CronTabWithStatus PublishedResource: %v", err) } - // Publish CronTabs with a related CronTabWithStatus (origin: service, syncStatus: true). + // Publish CronTabs with a related CronTabWithStatus (origin: service, syncStatus: service). prCrontabs := &syncagentv1alpha1.PublishedResource{ ObjectMeta: metav1.ObjectMeta{Name: "publish-crontabs"}, Spec: syncagentv1alpha1.PublishedResourceSpec{ @@ -288,7 +437,7 @@ func TestSyncRelatedObjectStatusFromService(t *testing.T) { Projection: &syncagentv1alpha1.RelatedResourceProjection{ Group: "kcp.example.com", }, - SyncStatus: true, + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginService), // Watch triggers CronTab reconciliation whenever a service cluster // CronTabWithStatus changes. This also registers an informer so that // syncObjectStatusForward can read the up-to-date status from the cache. @@ -376,6 +525,144 @@ func TestSyncRelatedObjectStatusFromService(t *testing.T) { } } +// TestSyncRelatedObjectStatusFromKcpWithServiceOrigin verifies that spec and status can be +// synchronized in opposite directions when the related resource originates on the service side. +// The projection also changes the resource Kind to ensure the kcp-side watch resolves its GVK +// from the workspace RESTMapper. +func TestSyncRelatedObjectStatusFromKcpWithServiceOrigin(t *testing.T) { + const apiExportName = "kcp.example.com" + + ctrlruntime.SetLogger(logr.Discard()) + + ctx := t.Context() + orgKubconfig := utils.CreateOrganization(t, ctx, "related-status-kcp-service-origin", apiExportName) + + envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{ + "test/crds/crontab.yaml", + "test/crds/crontabswithstatus.yaml", + "test/crds/backupswithstatus.yaml", + }) + + // Publish BackupWithStatus so the related resource can be projected to a different Kind in kcp. + prBackupsWithStatus := &syncagentv1alpha1.PublishedResource{ + ObjectMeta: metav1.ObjectMeta{Name: "publish-backupswithstatus"}, + Spec: syncagentv1alpha1.PublishedResourceSpec{ + Resource: syncagentv1alpha1.SourceResourceDescriptor{ + APIGroup: "eksempel.no", + Version: "v1", + Kind: "BackupWithStatus", + }, + Projection: &syncagentv1alpha1.ResourceProjection{Group: apiExportName}, + Synchronization: &syncagentv1alpha1.SynchronizationSpec{Enabled: false}, + }, + } + if err := envtestClient.Create(ctx, prBackupsWithStatus); err != nil { + t.Fatalf("Failed to create BackupWithStatus PublishedResource: %v", err) + } + + prCrontabs := &syncagentv1alpha1.PublishedResource{ + ObjectMeta: metav1.ObjectMeta{Name: "publish-crontabs"}, + Spec: syncagentv1alpha1.PublishedResourceSpec{ + Resource: syncagentv1alpha1.SourceResourceDescriptor{ + APIGroup: "example.com", + Version: "v1", + Kind: "CronTab", + }, + Naming: &syncagentv1alpha1.ResourceNaming{ + Name: "{{ .Object.metadata.name }}", + Namespace: "synced-{{ .Object.metadata.namespace }}", + }, + Projection: &syncagentv1alpha1.ResourceProjection{Group: apiExportName}, + Related: []syncagentv1alpha1.RelatedResourceSpec{ + { + Identifier: "crontabwithstatus", + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Group: "example.com", + Version: "v1", + Resource: "crontabswithstatus", + Projection: &syncagentv1alpha1.RelatedResourceProjection{ + Group: apiExportName, + Resource: "backupswithstatus", + }, + SyncStatus: ptr.To(syncagentv1alpha1.RelatedResourceOriginKcp), + Watch: &syncagentv1alpha1.RelatedResourceWatch{ + BySelector: &metav1.LabelSelector{}, + }, + Object: syncagentv1alpha1.RelatedResourceObject{ + RelatedResourceObjectSpec: syncagentv1alpha1.RelatedResourceObjectSpec{ + Template: &syncagentv1alpha1.TemplateExpression{Template: "my-related"}, + }, + }, + }, + }, + }, + } + if err := envtestClient.Create(ctx, prCrontabs); err != nil { + t.Fatalf("Failed to create CronTab PublishedResource: %v", err) + } + + utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "") + + kcpClusterClient := utils.GetKcpAdminClusterClient(t) + teamClusterPath := logicalcluster.NewPath("root").Join("related-status-kcp-service-origin").Join("team-1") + teamClient := kcpClusterClient.Cluster(teamClusterPath) + + utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{ + Group: apiExportName, Version: "v1", Kind: "CronTab", + }) + utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{ + Group: apiExportName, Version: "v1", Kind: "BackupWithStatus", + }) + + crontab := utils.ToUnstructured(t, &crds.Crontab{ + TypeMeta: metav1.TypeMeta{APIVersion: apiExportName + "/v1", Kind: "CronTab"}, + ObjectMeta: metav1.ObjectMeta{Name: "my-crontab", Namespace: "default"}, + Spec: crds.CrontabSpec{CronSpec: "* * *", Image: "ubuntu:latest"}, + }) + if err := teamClient.Create(ctx, crontab); err != nil { + t.Fatalf("Failed to create CronTab: %v", err) + } + + ensureNamespace(t, ctx, envtestClient, "synced-default") + serviceRelated := makeCronTabWithStatus("my-related", "synced-default") + if err := envtestClient.Create(ctx, serviceRelated); err != nil { + t.Fatalf("Failed to create CronTabWithStatus in service cluster: %v", err) + } + + kcpRelated := &unstructured.Unstructured{} + kcpRelated.SetAPIVersion(apiExportName + "/v1") + kcpRelated.SetKind("BackupWithStatus") + err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 3*time.Minute, false, func(ctx context.Context) (bool, error) { + return teamClient.Get(ctx, types.NamespacedName{Name: "my-related", Namespace: "default"}, kcpRelated) == nil, nil + }) + if err != nil { + t.Fatalf("Projected BackupWithStatus never appeared in kcp: %v", err) + } + + if err := unstructured.SetNestedField(kcpRelated.Object, "id-kcp-67890", "status", "id"); err != nil { + t.Fatalf("Failed to set status.id: %v", err) + } + if err := teamClient.Status().Update(ctx, kcpRelated); err != nil { + t.Fatalf("Failed to update BackupWithStatus status in kcp: %v", err) + } + + err = wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 3*time.Minute, false, func(ctx context.Context) (bool, error) { + if err := envtestClient.Get( + ctx, + types.NamespacedName{Name: "my-related", Namespace: "synced-default"}, + serviceRelated, + ); err != nil { + return false, nil + } + id, _, _ := unstructured.NestedString(serviceRelated.Object, "status", "id") + return id == "id-kcp-67890", nil + }) + if err != nil { + id, _, _ := unstructured.NestedString(serviceRelated.Object, "status", "id") + t.Fatalf("BackupWithStatus status was not synced to the service cluster (got status.id=%q, want %q)", id, "id-kcp-67890") + } +} + // TestSyncRelatedObjectStatusNotSyncedByDefault verifies that when syncStatus is not set (the // default), status changes on the origin side do not appear on the destination copy. func TestSyncRelatedObjectStatusNotSyncedByDefault(t *testing.T) { @@ -434,7 +721,7 @@ func TestSyncRelatedObjectStatusNotSyncedByDefault(t *testing.T) { Projection: &syncagentv1alpha1.RelatedResourceProjection{ Group: "example.com", }, - // SyncStatus: false (default) + // SyncStatus: nil (default) Object: syncagentv1alpha1.RelatedResourceObject{ RelatedResourceObjectSpec: syncagentv1alpha1.RelatedResourceObjectSpec{ Template: &syncagentv1alpha1.TemplateExpression{