diff --git a/.gitignore b/.gitignore index 5614a596..ca0333a3 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ .DS_Store /docs/__pycache__ .idea/ +temp/ diff --git a/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml b/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml index 59224836..63c39acc 100644 --- a/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml +++ b/deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml @@ -387,7 +387,26 @@ spec: the destination side (i.e. the original related object will not be touched, regardless of this option). Leaving this disabled, the syncagent will only create copies of the related objects, but never delete them itself. + + Deprecated: use CleanupPolicy instead. When CleanupPolicy is empty, cleanup:true is treated + as OnPrimaryDeletion and cleanup:false as Orphan. type: boolean + cleanupPolicy: + description: |- + CleanupPolicy controls when the syncagent deletes destination copies of this related + resource. The original object on the origin side is never deleted. + + - Orphan (default): copies are never deleted by the agent. + - OnPrimaryDeletion: copies are deleted only when the primary object is deleted. + - MatchOrigin: copies are pruned as soon as their origin object no longer exists, and + all copies are deleted when the primary object is deleted. + + When left empty, the deprecated Cleanup field is used to derive the effective policy. + enum: + - Orphan + - OnPrimaryDeletion + - MatchOrigin + type: string group: description: |- Group is the API group of the related resource. This should be left blank for resources @@ -398,7 +417,13 @@ spec: Identifier is a unique name for this related resource. The name must be unique within one PublishedResource and is the key by which consumers (end users) can identify and consume the related resource. Common names are "connection-details" or "credentials". - The identifier must be an alphanumeric string. + + The identifier is used verbatim as a label value on the synced copies of the related resource, + so it must be a valid label value: a lowercase RFC 1123 label consisting of lowercase + alphanumeric characters or '-', starting and ending with an alphanumeric character, and at + most 63 characters long. + maxLength: 63 + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ type: string identityHash: description: |- @@ -878,6 +903,10 @@ spec: 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 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' + rule: '!(has(self.cleanup) && self.cleanup && has(self.cleanupPolicy) && self.cleanupPolicy == ''Orphan'')' type: array resource: description: |- diff --git a/docs/content/publish-resources/related-resources.md b/docs/content/publish-resources/related-resources.md index 3c0a9751..f41d54ef 100644 --- a/docs/content/publish-resources/related-resources.md +++ b/docs/content/publish-resources/related-resources.md @@ -571,3 +571,60 @@ individually. For each namespace it will again follow the configured source, may template or reference. If again a label selector is used, it will be applied in each namespace and the configured rewrite rule will be evaluated once per found object. In this case, `.Value` is the name of found object. + +## Cleanup + +By default the agent only ever _creates_ copies of related resources; it never deletes them. This +is intentional: copies can be useful as audit trails, may have decoupled lifecycles, or may be owned +by the service provider. The behavior is controlled per related resource via the `cleanupPolicy` +field, which takes one of three values: + +- `Orphan` (default): copies are never deleted by the agent. +- `OnPrimaryDeletion`: copies are deleted only when the primary object is deleted. +- `MatchOrigin`: the destination set is kept equal to the origin set — a copy is pruned as soon as + its origin object no longer exists, and all copies are deleted when the primary object is deleted. + +**The original object on the origin side is never deleted, under any policy.** The agent only ever +deletes destination copies that it created itself (they carry provenance labels identifying the +owning primary object); hand-created objects are never touched. + +`MatchOrigin` prunes copies while the agent reconciles the primary object. For an `origin: service` +related resource this means a `watch` must be configured, otherwise a deleted source object would +only be noticed on the next incidental reconcile of the primary. The CRD enforces this: setting +`cleanupPolicy: MatchOrigin` together with `origin: service` requires a `watch` block. + +{% raw %} +```yaml +apiVersion: syncagent.kcp.io/v1alpha1 +kind: PublishedResource +metadata: + name: publish-crontabs +spec: + resource: + apiGroup: example.com + version: v1 + kind: CronTab + related: + - identifier: credentials + origin: service + kind: Secret + cleanupPolicy: MatchOrigin + object: + selector: + matchLabels: + app: credentials + rewrite: + template: + template: "{{ .Value }}" + watch: + bySelector: + matchLabels: + example.com/managed: "true" +``` +{% endraw %} + +!!! note + The deprecated boolean `cleanup` field still works for backwards compatibility. When + `cleanupPolicy` is not set, `cleanup: true` is treated as `OnPrimaryDeletion` and `cleanup: false` + (or unset) as `Orphan`. New configurations should use `cleanupPolicy` instead; setting + `cleanup: true` together with `cleanupPolicy: Orphan` is rejected as a contradiction. diff --git a/internal/controller/sync/controller.go b/internal/controller/sync/controller.go index 0d40b58d..207d1485 100644 --- a/internal/controller/sync/controller.go +++ b/internal/controller/sync/controller.go @@ -65,6 +65,7 @@ const ( type Reconciler struct { localClient ctrlruntimeclient.Client + localAPIReader ctrlruntimeclient.Reader remoteManager mcmanager.Manager log *zap.SugaredLogger remoteDummy *unstructured.Unstructured @@ -120,6 +121,7 @@ func Create( // setup the reconciler reconciler := &Reconciler{ localClient: localManager.GetClient(), + localAPIReader: localManager.GetAPIReader(), remoteManager: remoteManager, log: log, remoteDummy: remoteDummy, @@ -360,6 +362,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request) return reconcile.Result{}, fmt.Errorf("failed to get cluster: %w", err) } vwClient := cl.GetClient() + vwAPIReader := cl.GetAPIReader() remoteObj := r.remoteDummy.DeepCopy() if err := vwClient.Get(ctx, request.NamespacedName, remoteObj); ctrlruntimeclient.IgnoreNotFound(err) != nil { @@ -410,7 +413,11 @@ func (r *Reconciler) Reconcile(ctx context.Context, request mcreconcile.Request) ctx = sync.WithWorkspacePath(ctx, logicalcluster.NewPath(path)) } - var syncerOpts []sync.ResourceSyncerOption + // pass uncached readers so a MatchOrigin prune can re-confirm an empty origin resolution + // against the API server before deleting all copies of a related resource. + syncerOpts := []sync.ResourceSyncerOption{ + sync.WithLiveReaders(r.localAPIReader, vwAPIReader), + } if r.enableServerSideApply { syncerOpts = append(syncerOpts, sync.WithServerSideApply()) } diff --git a/internal/sync/meta.go b/internal/sync/meta.go index c5bf76f8..c2d9f57e 100644 --- a/internal/sync/meta.go +++ b/internal/sync/meta.go @@ -18,6 +18,7 @@ package sync import ( "context" + "strings" "go.uber.org/zap" @@ -158,3 +159,57 @@ func (k objectKey) Annotations() labels.Set { return s } + +// relatedCopyLabels builds the provenance labels put on a destination copy of a related resource. +// They tie the copy to its owning primary object and the related resource identifier so that all +// copies of a given (primary, identifier) can be enumerated via relatedCopySelector. The owner tuple +// — the primary's logical cluster (workspace), the owning PublishedResource and the primary's +// namespace/name — is hashed into a single related-owner label; because the prune List is +// cluster-wide on the shared destination, all of these dimensions must be part of the identity, and +// folding them into one hash makes that uniqueness inherent in the hash input (and keeps the value a +// valid label regardless of the source lengths or characters). The identifier is used verbatim (the +// API constrains it to a valid label value) and the agent name (when set) is included so that each +// agent only ever prunes its own copies. +func relatedCopyLabels(primary ctrlruntimeclient.Object, clusterName logicalcluster.Name, publishedResourceName, identifier, agentName string) map[string]string { + set := map[string]string{ + relatedOwnerLabel: relatedOwnerHash(clusterName, publishedResourceName, primary.GetNamespace(), primary.GetName()), + relatedIdentifierLabel: identifier, + } + + if agentName != "" { + set[agentNameLabel] = agentName + } + + return set +} + +// relatedOwnerHash hashes the owner tuple (cluster, PublishedResource, primary namespace, primary +// name) into a single label value. The NUL separator keeps the dimensions unambiguous, so e.g. +// cluster "a" + name "bc" cannot collide with cluster "ab" + name "c". A cluster-scoped primary has +// an empty namespace, which folds in cleanly and cannot collide with a namespaced primary (whose +// namespace is never empty). +func relatedOwnerHash(clusterName logicalcluster.Name, publishedResourceName, namespace, name string) string { + return crypto.Hash(strings.Join([]string{string(clusterName), publishedResourceName, namespace, name}, "\x00")) +} + +// relatedCopyAnnotations builds the human-facing provenance annotations (plaintext primary +// namespace/name) for a destination copy of a related resource. +func relatedCopyAnnotations(primary ctrlruntimeclient.Object) map[string]string { + set := map[string]string{ + relatedPrimaryNameAnnotation: primary.GetName(), + } + + if namespace := primary.GetNamespace(); namespace != "" { + set[relatedPrimaryNamespaceAnnotation] = namespace + } + + return set +} + +// relatedCopySelector returns a label selector matching exactly the destination copies created for +// the given primary object and related resource identifier (scoped to the agent when set). Because +// it mirrors relatedCopyLabels, only objects the agent itself labelled are ever selected, so +// hand-created objects are never in scope for pruning. +func relatedCopySelector(primary ctrlruntimeclient.Object, clusterName logicalcluster.Name, publishedResourceName, identifier, agentName string) labels.Selector { + return labels.SelectorFromSet(relatedCopyLabels(primary, clusterName, publishedResourceName, identifier, agentName)) +} diff --git a/internal/sync/meta_test.go b/internal/sync/meta_test.go index 5b3e986b..7d07d327 100644 --- a/internal/sync/meta_test.go +++ b/internal/sync/meta_test.go @@ -17,12 +17,16 @@ limitations under the License. package sync import ( + "strings" "testing" "github.com/kcp-dev/logicalcluster/v3" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + metav1validation "k8s.io/apimachinery/pkg/apis/meta/v1/validation" + "k8s.io/apimachinery/pkg/util/validation/field" + ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" ) func createNewObject(name, namespace string) metav1.Object { @@ -33,6 +37,74 @@ func createNewObject(name, namespace string) metav1.Object { return obj } +// TestRelatedCopyLabelsAreValidLabelSet guards the invariant that lets relatedCopyLabels use the +// related-resource identifier verbatim as a label value: as long as the identifier stays within the +// bounds the CRD enforces (a lowercase RFC 1123 label of at most 63 characters), the resulting +// provenance label set must always be accepted by the API server. Otherwise the apply/create of the +// copy fails and sync silently breaks for that related resource. +func TestRelatedCopyLabelsAreValidLabelSet(t *testing.T) { + // Names and namespaces are hashed, so they can be arbitrarily long/invalid; use overlong ones to + // make sure the hashing keeps the label set valid. + longName := strings.Repeat("a", 300) + + // maxIdentifier is the longest identifier the CRD accepts (63 chars, matching the pattern). + maxIdentifier := "a" + strings.Repeat("b", 61) + "c" + + testcases := []struct { + name string + primary ctrlruntimeclient.Object + clusterName logicalcluster.Name + publishedResName string + identifier string + agentName string + }{ + { + name: "typical namespaced primary", + primary: createNewUnstructured("my-primary", "kube-system"), + clusterName: "root", + publishedResName: "my-published-resource", + identifier: "connection-details", + agentName: "agent-1", + }, + { + // clusterName is a kcp cluster ID (a colon-free hash), used verbatim like the main + // object's remote-object-cluster label; the workspace path (with colons) is never used here. + name: "max-length identifier and overlong name/namespace/published-resource", + primary: createNewUnstructured(longName, longName), + clusterName: "kvdk2spgmbld9mnc", + publishedResName: longName, + identifier: maxIdentifier, + agentName: "", + }, + { + name: "cluster-scoped primary without agent name", + primary: createNewUnstructured(longName, ""), + clusterName: "abc123", + publishedResName: "another-published-resource", + identifier: "credentials", + agentName: "", + }, + } + + for _, testcase := range testcases { + t.Run(testcase.name, func(t *testing.T) { + set := relatedCopyLabels(testcase.primary, testcase.clusterName, testcase.publishedResName, testcase.identifier, testcase.agentName) + + if errs := metav1validation.ValidateLabels(set, field.NewPath("metadata", "labels")); len(errs) > 0 { + t.Fatalf("relatedCopyLabels produced an invalid label set: %v", errs) + } + }) + } +} + +func createNewUnstructured(name, namespace string) ctrlruntimeclient.Object { + obj := &unstructured.Unstructured{} + obj.SetName(name) + obj.SetNamespace(namespace) + + return obj +} + func TestObjectKey(t *testing.T) { testcases := []struct { object metav1.Object diff --git a/internal/sync/object_syncer.go b/internal/sync/object_syncer.go index cabbae03..1dc6abd7 100644 --- a/internal/sync/object_syncer.go +++ b/internal/sync/object_syncer.go @@ -57,6 +57,14 @@ type objectSyncer struct { blockSourceDeletion bool // whether or not to place sync-related metadata on the destination object metadataOnDestination bool + // destLabels are additional labels stamped onto the destination object on every apply. + // Used to attach related-resource provenance (owning primary + identifier) to copies so + // that they can later be enumerated and pruned. These are applied additively and never + // clobber labels the copy already carries. + destLabels map[string]string + // destAnnotations are additional annotations stamped onto the destination object on every + // apply; the human-facing counterpart to destLabels. + destAnnotations map[string]string // optional mutations for both directions of the sync mutator mutation.Mutator // stateStore is capable of remembering the state of a Kubernetes object @@ -173,6 +181,18 @@ func (s *objectSyncer) Sync(ctx context.Context, log *zap.SugaredLogger, source, return false, nil } + // Backfill provenance labels/annotations onto copies that predate them: unlike the create, adopt + // and Server-Side Apply paths, the client-side-merge update path below never stamps them, so a + // copy that already existed when the user opted into a pruning cleanupPolicy would otherwise never + // be enumerated for pruning. Requeue after a restamp so the content sync runs on the next pass. + restamped, err := s.ensureDestMetadataPersisted(ctx, log, dest) + if err != nil { + return false, fmt.Errorf("failed to ensure destination metadata: %w", err) + } + if restamped { + return true, nil + } + requeue, err = s.syncObjectContents(ctx, log, source, dest) if err != nil { return false, fmt.Errorf("failed to synchronize object state: %w", err) @@ -420,6 +440,9 @@ func (s *objectSyncer) applyServerSide(ctx context.Context, log *zap.SugaredLogg s.labelWithAgent(desired) } + // stamp any additional provenance labels/annotations (e.g. related-resource ownership) + s.ensureDestMetadata(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) @@ -503,6 +526,9 @@ func (s *objectSyncer) ensureDestinationObject(ctx context.Context, log *zap.Sug s.labelWithAgent(destObj) } + // stamp any additional provenance labels/annotations (e.g. related-resource ownership) + s.ensureDestMetadata(destObj) + // finally, we can create the destination object objectLog := log.With("dest-object", newObjectKey(destObj, dest.clusterName, logicalcluster.None)) objectLog.Debugw("Creating destination object…") @@ -551,6 +577,7 @@ func (s *objectSyncer) adoptExistingDestinationObject(ctx context.Context, log * ensureAnnotations(existingDestObj, sourceKey.Annotations()) s.labelWithAgent(existingDestObj) + s.ensureDestMetadata(existingDestObj) if err := dest.client.Update(ctx, existingDestObj); err != nil { return fmt.Errorf("failed to upsert current destination object labels: %w", err) @@ -654,3 +681,44 @@ func (s *objectSyncer) labelWithAgent(obj *unstructured.Unstructured) { ensureLabels(obj, map[string]string{agentNameLabel: s.agentName}) } } + +// ensureDestMetadata stamps the syncer's additional destination labels/annotations onto the given +// object. It is additive (it never removes labels the object already carries) and is used to +// attach related-resource provenance to destination copies. +func (s *objectSyncer) ensureDestMetadata(obj *unstructured.Unstructured) { + if len(s.destLabels) > 0 { + ensureLabels(obj, s.destLabels) + } + + if len(s.destAnnotations) > 0 { + ensureAnnotations(obj, s.destAnnotations) + } +} + +// ensureDestMetadataPersisted makes sure the additional destination labels/annotations are present +// on an already-existing destination object, patching it if they are missing. New copies are stamped +// at creation time (ensureDestinationObject) or when adopted, and the Server-Side Apply path restamps +// on every apply, but the client-side-merge update path (syncObjectSpec) does not touch them. Without +// this, a copy that already existed when the user opted into a pruning cleanupPolicy would never +// receive the provenance labels the prune relies on and could therefore never be reclaimed. The +// operation is idempotent: once the labels are present, the diff is empty and no patch is issued. +func (s *objectSyncer) ensureDestMetadataPersisted(ctx context.Context, log *zap.SugaredLogger, dest syncSide) (updated bool, err error) { + if len(s.destLabels) == 0 && len(s.destAnnotations) == 0 { + return false, nil + } + + original := dest.object.DeepCopy() + s.ensureDestMetadata(dest.object) + + if equality.Semantic.DeepEqual(original.GetLabels(), dest.object.GetLabels()) && + equality.Semantic.DeepEqual(original.GetAnnotations(), dest.object.GetAnnotations()) { + return false, nil + } + + log.Debugw("Restamping provenance metadata on existing destination object…", "dest-object", newObjectKey(dest.object, dest.clusterName, logicalcluster.None)) + if err := dest.client.Patch(ctx, dest.object, ctrlruntimeclient.MergeFrom(original)); err != nil { + return false, fmt.Errorf("failed to restamp destination metadata: %w", err) + } + + return true, nil +} diff --git a/internal/sync/object_syncer_test.go b/internal/sync/object_syncer_test.go index 3fdb40d3..947cdbf3 100644 --- a/internal/sync/object_syncer_test.go +++ b/internal/sync/object_syncer_test.go @@ -26,6 +26,7 @@ import ( "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/client-go/tools/record" + ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" ) // fakeStateStore is a minimal ObjectStateStore for unit tests. @@ -221,3 +222,80 @@ func TestSyncObjectContentsForwardStatusRunsEvenOnSpecRequeue(t *testing.T) { t.Errorf("expected dest status.phase=ready after syncObjectContents, got %q — forward status sync did not run on spec requeue", phase) } } + +func TestEnsureDestMetadataPersisted(t *testing.T) { + log := zap.NewNop().Sugar() + + makeCopy := func() *unstructured.Unstructured { + obj := &unstructured.Unstructured{} + obj.SetAPIVersion("test.example.com/v1") + obj.SetKind("Widget") + obj.SetName("my-widget") + obj.SetNamespace("default") + return obj + } + + destLabels := map[string]string{ + relatedOwnerLabel: "testcluster", + relatedIdentifierLabel: "credentials", + } + destAnnotations := map[string]string{ + relatedPrimaryNameAnnotation: "my-primary", + } + + t.Run("backfills provenance metadata onto a pre-existing copy and is idempotent", func(t *testing.T) { + // A copy that predates the feature: it exists on the destination but carries none of the + // provenance labels/annotations the prune relies on. + dest := makeCopy() + destClient := buildFakeClient(dest) + + syncer := &objectSyncer{destLabels: destLabels, destAnnotations: destAnnotations} + destSide := syncSide{object: dest, client: destClient} + + updated, err := syncer.ensureDestMetadataPersisted(t.Context(), log, destSide) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !updated { + t.Fatal("expected updated=true because the copy was missing the provenance metadata") + } + + // The change must be persisted to the API, not just to the in-memory object. + persisted := makeCopy() + if err := destClient.Get(t.Context(), ctrlruntimeclient.ObjectKeyFromObject(dest), persisted); err != nil { + t.Fatalf("failed to get persisted object: %v", err) + } + for k, v := range destLabels { + if persisted.GetLabels()[k] != v { + t.Errorf("expected persisted label %q=%q, got %q", k, v, persisted.GetLabels()[k]) + } + } + if persisted.GetAnnotations()[relatedPrimaryNameAnnotation] != "my-primary" { + t.Errorf("expected persisted annotation, got %q", persisted.GetAnnotations()[relatedPrimaryNameAnnotation]) + } + + // A second call must be a no-op now that the metadata is present. + updated, err = syncer.ensureDestMetadataPersisted(t.Context(), log, syncSide{object: persisted, client: destClient}) + if err != nil { + t.Fatalf("unexpected error on second call: %v", err) + } + if updated { + t.Error("expected updated=false on the second call (idempotent), but got true") + } + }) + + t.Run("no-op when the syncer has no additional destination metadata", func(t *testing.T) { + dest := makeCopy() + destClient := buildFakeClient(dest) + + syncer := &objectSyncer{} + + updated, err := syncer.ensureDestMetadataPersisted(t.Context(), log, syncSide{object: dest, client: destClient}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if updated { + t.Error("expected updated=false when there are no destLabels/destAnnotations") + } + }) +} diff --git a/internal/sync/syncer.go b/internal/sync/syncer.go index 0c51207a..718e49e5 100644 --- a/internal/sync/syncer.go +++ b/internal/sync/syncer.go @@ -46,6 +46,13 @@ type ResourceSyncer struct { localCRD *apiextensionsv1.CustomResourceDefinition subresources []string + // localAPIReader and remoteAPIReader are uncached readers that hit the API server directly. + // They are only used to re-confirm a cache-driven empty related-resource resolution before a + // MatchOrigin prune would delete every copy (see processRelatedResource). They may be nil, in + // which case that destructive full-set prune is skipped rather than run on unverified data. + localAPIReader ctrlruntimeclient.Reader + remoteAPIReader ctrlruntimeclient.Reader + destDummy *unstructured.Unstructured // cached mutators (for those transformers that are expensive to compile, like CEL) @@ -69,6 +76,17 @@ func WithServerSideApply() ResourceSyncerOption { } } +// WithLiveReaders configures uncached readers for the local (service cluster) and remote (kcp) +// sides. They let a MatchOrigin prune re-confirm an empty origin resolution against the API server +// before deleting all copies of a related resource, guarding against a stale/relisting informer +// cache. Either reader may be nil. +func WithLiveReaders(local, remote ctrlruntimeclient.Reader) ResourceSyncerOption { + return func(r *ResourceSyncer) { + r.localAPIReader = local + r.remoteAPIReader = remote + } +} + type MutatorCreatorFunc func(*syncagentv1alpha1.ResourceMutationSpec) (mutation.Mutator, error) func NewResourceSyncer( diff --git a/internal/sync/syncer_related.go b/internal/sync/syncer_related.go index 84a059ee..650da52a 100644 --- a/internal/sync/syncer_related.go +++ b/internal/sync/syncer_related.go @@ -21,6 +21,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "regexp" "slices" "strings" @@ -36,7 +37,10 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -80,17 +84,15 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su eventObjSide = syncSideSource } + // normalize the deprecated cleanup bool and the cleanup policy field into a single policy. + policy := relRes.EffectiveCleanupPolicy() + // find the all objects on the origin side that match the given criteria resolvedObjects, err := resolveRelatedResourceObjects(ctx, origin, dest, relRes) if err != nil { return false, fmt.Errorf("failed to get resolve origin objects: %w", err) } - // no objects were found yet, that's okay - if len(resolvedObjects) == 0 { - return false, nil - } - slices.SortStableFunc(resolvedObjects, func(a, b resolvedObject) int { aKey := ctrlruntimeclient.ObjectKeyFromObject(a.original).String() bKey := ctrlruntimeclient.ObjectKeyFromObject(b.original).String() @@ -106,7 +108,27 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su return false, fmt.Errorf("failed to lookup %v: %w", projectedGVR, err) } - for idx, resolved := range resolvedObjects { + // The primary object (always the kcp side) owns these related copies; stamp its coordinates + // onto every copy so that all copies of this (primary, identifier) can be enumerated later to + // prune stale copies or tear them all down. + primary := remote.object + destLabels := relatedCopyLabels(primary, remote.clusterName, s.pubRes.Name, relRes.Identifier, s.agentName) + destAnnotations := relatedCopyAnnotations(primary) + + // remember which destination copies we (re)synced this pass, so a MatchOrigin prune can delete + // the copies that no longer have a matching origin object. + synced := sets.New[string]() + + // We "forward" the deletion to the related objects only if the primary is already in deletion + // and the related object either originated from the user (so on the service cluster we just + // have a useless copy once the main object has been cleared up) OR the admin explicitly opted + // into a cleanup policy that removes copies on primary deletion. + forceDelete := primaryDeleting && + (policy == syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion || + policy == syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin || + relRes.Origin == syncagentv1alpha1.RelatedResourceOriginKcp) + + for _, resolved := range resolvedObjects { destObject := &unstructured.Unstructured{} destObject.SetAPIVersion(projectedGVK.GroupVersion().String()) destObject.SetKind(projectedGVK.Kind) @@ -127,12 +149,6 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su object: destObject, } - // We "forward" the deletion to the related objects only if the primary is already in deletion - // and the related object either originated from the user (so on the service cluster we just - // have a useless copy once the main object has been cleared up) OR the admin explicitly opted - // into the cleanup procedure to ensure that copies of the related object are removed. - forceDelete := primaryDeleting && (relRes.Cleanup || relRes.Origin == syncagentv1alpha1.RelatedResourceOriginKcp) - // 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. @@ -169,6 +185,9 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su mutator: s.relatedMutators[relRes.Identifier], // we never want to store sync-related metadata inside kcp metadataOnDestination: false, + // stamp owning-primary provenance so that the copies can be enumerated for pruning. + destLabels: destLabels, + destAnnotations: destAnnotations, // events are always created on the kcp side eventObjSide: eventObjSide, // force deletion of related resources when the primary object is being deleted @@ -188,48 +207,286 @@ func (s *ResourceSyncer) processRelatedResource(ctx context.Context, log *zap.Su // too many unnecessary requeues. requeue = requeue || req - // now that the related object was successfully synced, we can remember its details on the - // main object - if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginService { - // TODO: Improve this logic, the added index is just a hack until we find a better solution - // to let the user know about the related object (this annotation is not relevant for the - // syncing logic, it's purely for the end-user). - annotation := fmt.Sprintf("%s%s.%d", relatedObjectAnnotationPrefix, relRes.Identifier, idx) - - value, err := json.Marshal(relatedObjectAnnotation{ - Namespace: resolved.destination.Namespace, - Name: resolved.destination.Name, - APIVersion: resolved.original.GetAPIVersion(), - Kind: resolved.original.GetKind(), - }) + synced.Insert(relatedCopyKey(resolved.destination.Namespace, resolved.destination.Name)) + } + + // Remember the related objects on the primary object for the end-user. This is rebuilt from the + // freshly-resolved set so entries for objects that no longer resolve are dropped; a fully-empty + // resolution is left untouched (see rememberRelatedObjects) to avoid churning the primary. + if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginService { + annRequeue, err := s.rememberRelatedObjects(ctx, log, remote, relRes.Identifier, resolvedObjects) + if err != nil { + return false, err + } + + // we updated the main object, so we requeue immediately because successive patches would + // fail anyway; the prune below then runs on the next reconciliation. + if annRequeue { + return true, nil + } + } + + // Prune / teardown destination copies as configured. Only objects carrying our provenance + // labels are ever considered, and we only ever act on the destination client, so the original + // origin-side objects are never touched. + switch { + case primaryDeleting && (policy == syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion || + policy == syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin): + // On primary teardown, delete ALL labelled copies. This is a superset of the per-object + // deletion performed in the loop above and additionally reclaims copies whose origin object + // had already disappeared mid-life (which the loop can no longer resolve). + selector := relatedCopySelector(primary, remote.clusterName, s.pubRes.Name, relRes.Identifier, s.agentName) + + pruneRequeue, err := s.pruneRelatedCopies(ctx, log, dest, primary, projectedGVK, selector, nil, true) + if err != nil { + return false, fmt.Errorf("failed to tear down related copies: %w", err) + } + + requeue = requeue || pruneRequeue + + case !primaryDeleting && policy == syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin: + // Keep the destination set equal to the origin set: prune every labelled copy whose origin + // object was not resolved this pass. + selector := relatedCopySelector(primary, remote.clusterName, s.pubRes.Name, relRes.Identifier, s.agentName) + + // A fully-empty resolution here would prune *every* copy at once, because the keep set is + // empty. resolvedObjects comes from a cache-backed read, and a relisting or otherwise stale + // informer -- notably a kcp virtual-workspace cache -- can momentarily return nothing even + // though the origin objects still exist. Acting on that would delete and then immediately + // recreate all copies, disrupting consumers (e.g. Secrets mounted by workloads). The + // non-destructive annotation bookkeeping already refuses to act on an empty resolution for + // the same reason (see rememberRelatedObjects); the destructive prune must be at least as + // careful. Before deleting the whole set, re-confirm the emptiness against a live read; if + // the live read disagrees, requeue and let the cache converge. A genuinely empty origin is + // still pruned, so the single-object mid-life case the feature promises keeps working. + // + // Partial under-resolution (a non-empty but incomplete set) is intentionally not guarded: + // it prunes at most a subset and self-heals on the next pass, matching the cache semantics + // the rest of the sync already relies on. + if len(resolvedObjects) == 0 { + confirmedEmpty, checked, err := s.confirmOriginEmpty(ctx, origin, dest, relRes) if err != nil { - return false, fmt.Errorf("failed to encode related object annotation: %w", err) + return false, fmt.Errorf("failed to confirm empty origin before pruning related copies: %w", err) + } + + switch { + case !checked: + // No live reader configured to confirm against (e.g. in unit tests). Do not risk + // deleting the whole set on an unverified empty resolution; teardown still reclaims + // the copies if the primary is ever deleted. + log.Debug("Skipping full related-copy prune on empty resolution: no live reader configured to confirm it.") + return requeue, nil + + case !confirmedEmpty: + log.Warn("Skipping related-copy prune: origin resolved to no objects from cache, but a live read still sees origin objects; requeueing to let the cache converge.") + return true, nil } + } - annotations := remote.object.GetAnnotations() - existing := annotations[annotation] + pruneRequeue, err := s.pruneRelatedCopies(ctx, log, dest, primary, projectedGVK, selector, synced, false) + if err != nil { + return false, fmt.Errorf("failed to prune related copies: %w", err) + } - if existing != string(value) { - oldState := remote.object.DeepCopy() + requeue = requeue || pruneRequeue + } - annotations[annotation] = string(value) - remote.object.SetAnnotations(annotations) + return requeue, nil +} - log.Debug("Remembering related object in main object…") - if err := remote.client.Patch(ctx, remote.object, ctrlruntimeclient.MergeFrom(oldState)); err != nil { - return false, fmt.Errorf("failed to update related data in remote object: %w", err) - } +// relatedCopyKey builds the namespace/name key used to track and match destination copies. +func relatedCopyKey(namespace, name string) string { + return namespace + "/" + name +} - // requeue (since this updated the main object, we do actually want to - // requeue immediately because successive patches would fail anyway) - return true, nil +// rememberRelatedObjects writes the human-facing provenance annotations onto the primary object, +// one per related copy (indexed). It rebuilds the full set for the given identifier from the +// currently resolved objects, so annotations for objects that are no longer resolved are removed and +// do not accumulate. It reports requeue=true when it patched the primary object. +func (s *ResourceSyncer) rememberRelatedObjects(ctx context.Context, log *zap.SugaredLogger, remote syncSide, identifier string, resolvedObjects []resolvedObject) (requeue bool, err error) { + // When nothing resolves there is nothing new to remember, and we deliberately do not treat this + // as "clear all annotations for this identifier". These annotations are purely informational and + // the prune is the authoritative cleanup for the copies themselves; wiping them on every empty + // pass would otherwise churn the primary object with patches and requeues for all cleanup + // policies, not just the pruning ones (this used to be avoided by an early return before the + // resolved set was allowed to be empty so the prune could run). + if len(resolvedObjects) == 0 { + return false, nil + } + + // TODO: Improve this logic, the added index is just a hack until we find a better solution to + // let the user know about the related object (this annotation is not relevant for the syncing + // logic, it's purely for the end-user). + prefix := fmt.Sprintf("%s%s.", relatedObjectAnnotationPrefix, identifier) + + desired := map[string]string{} + for idx, resolved := range resolvedObjects { + value, err := json.Marshal(relatedObjectAnnotation{ + Namespace: resolved.destination.Namespace, + Name: resolved.destination.Name, + APIVersion: resolved.original.GetAPIVersion(), + Kind: resolved.original.GetKind(), + }) + if err != nil { + return false, fmt.Errorf("failed to encode related object annotation: %w", err) + } + + desired[fmt.Sprintf("%s%d", prefix, idx)] = string(value) + } + + annotations := remote.object.GetAnnotations() + if annotations == nil { + annotations = map[string]string{} + } + + // determine whether the existing annotations for this identifier already match the desired set. + changed := false + for key := range annotations { + if strings.HasPrefix(key, prefix) { + if _, ok := desired[key]; !ok { + changed = true + break + } + } + } + if !changed { + for key, value := range desired { + if annotations[key] != value { + changed = true + break } } } + if !changed { + return false, nil + } + + oldState := remote.object.DeepCopy() + + // drop all existing entries for this identifier, then add the freshly computed ones. + for key := range annotations { + if strings.HasPrefix(key, prefix) { + delete(annotations, key) + } + } + maps.Copy(annotations, desired) + remote.object.SetAnnotations(annotations) + + log.Debug("Remembering related objects in main object…") + if err := remote.client.Patch(ctx, remote.object, ctrlruntimeclient.MergeFrom(oldState)); err != nil { + return false, fmt.Errorf("failed to update related data in remote object: %w", err) + } + + return true, nil +} + +// pruneRelatedCopies lists the destination copies matching the given (primary + identifier + agent) +// selector and deletes those that are no longer wanted. When deleteAll is true, every matching copy +// is deleted (primary teardown); otherwise only copies whose key is not in the keep set are deleted +// (mid-life prune). It only ever operates on the destination client, so origin objects are never +// touched, and it only ever sees objects that carry our provenance labels, so hand-created objects +// are never in scope. +func (s *ResourceSyncer) pruneRelatedCopies(ctx context.Context, log *zap.SugaredLogger, dest syncSide, primary *unstructured.Unstructured, projectedGVK schema.GroupVersionKind, selector labels.Selector, keep sets.Set[string], deleteAll bool) (requeue bool, err error) { + list := &unstructured.UnstructuredList{} + list.SetAPIVersion(projectedGVK.GroupVersion().String()) + list.SetKind(projectedGVK.Kind + "List") + + // List cluster-wide (across all namespaces). A related resource can map its copies into a + // namespace other than the primary's (via spec.object.namespace rewrites), so scoping the List + // to a single namespace would miss — and therefore never prune — copies that landed elsewhere. + // The label selector already scopes the result to this primary + identifier + agent, so only + // copies this agent created for this primary can match. + listOpts := []ctrlruntimeclient.ListOption{ + ctrlruntimeclient.MatchingLabelsSelector{Selector: selector}, + } + + if err := dest.client.List(ctx, list, listOpts...); err != nil { + return false, fmt.Errorf("failed to list related copies: %w", err) + } + + recorder := recorderFromContext(ctx) + + for i := range list.Items { + item := &list.Items[i] + + if !deleteAll && keep.Has(relatedCopyKey(item.GetNamespace(), item.GetName())) { + continue + } + + // already being deleted; come back once it is gone. + if item.GetDeletionTimestamp() != nil { + requeue = true + continue + } + + log.Debugw("Pruning related object copy…", "namespace", item.GetNamespace(), "name", item.GetName()) + if err := dest.client.Delete(ctx, item); err != nil { + if apierrors.IsNotFound(err) { + continue + } + + return false, fmt.Errorf("failed to delete related copy: %w", err) + } + + if recorder != nil { + recorder.Eventf(primary, corev1.EventTypeNormal, "ObjectCleanup", "Deleted orphaned copy %s/%s of a related resource.", item.GetNamespace(), item.GetName()) + } + + requeue = true + } + return requeue, nil } +// liveReadClient serves reads from an uncached reader (hitting the API server directly) while +// borrowing the RESTMapper, scheme and everything else from an underlying cached client. It lets a +// destructive prune re-run the origin resolution against the API server instead of trusting a +// possibly-stale informer cache, without duplicating the resolution logic. +type liveReadClient struct { + ctrlruntimeclient.Client + reader ctrlruntimeclient.Reader +} + +func (c *liveReadClient) Get(ctx context.Context, key ctrlruntimeclient.ObjectKey, obj ctrlruntimeclient.Object, opts ...ctrlruntimeclient.GetOption) error { + return c.reader.Get(ctx, key, obj, opts...) +} + +func (c *liveReadClient) List(ctx context.Context, list ctrlruntimeclient.ObjectList, opts ...ctrlruntimeclient.ListOption) error { + return c.reader.List(ctx, list, opts...) +} + +// confirmOriginEmpty re-runs the origin resolution against a live (uncached) reader to double-check +// a cache-driven empty resolution before a MatchOrigin prune deletes all copies of a related +// resource. It returns checked=false when no live reader is configured for the origin side, in +// which case the caller must not treat the emptiness as authoritative. When a reader is configured, +// confirmedEmpty reports whether the live resolution also yields no objects. +func (s *ResourceSyncer) confirmOriginEmpty(ctx context.Context, origin, dest syncSide, relRes syncagentv1alpha1.RelatedResourceSpec) (confirmedEmpty bool, checked bool, err error) { + var reader ctrlruntimeclient.Reader + if relRes.Origin == syncagentv1alpha1.RelatedResourceOriginService { + reader = s.localAPIReader + } else { + reader = s.remoteAPIReader + } + + if reader == nil { + return false, false, nil + } + + liveOrigin := syncSide{ + clusterName: origin.clusterName, + client: &liveReadClient{Client: origin.client, reader: reader}, + object: origin.object, + } + + resolved, err := resolveRelatedResourceObjects(ctx, liveOrigin, dest, relRes) + if err != nil { + return false, true, err + } + + return len(resolved) == 0, true, nil +} + // resolvedObject is the result of following the configuration of a related resources. It contains // the original object (on the origin side of the related resource) and the target name to be used // on the destination side of the sync. diff --git a/internal/sync/syncer_related_test.go b/internal/sync/syncer_related_test.go index 92dcd248..ac042855 100644 --- a/internal/sync/syncer_related_test.go +++ b/internal/sync/syncer_related_test.go @@ -19,13 +19,430 @@ package sync import ( "testing" + "go.uber.org/zap" + 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" + corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/sets" + ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" ) +func TestEffectiveCleanupPolicy(t *testing.T) { + testcases := []struct { + name string + cleanup bool + policy syncagentv1alpha1.RelatedResourceCleanupPolicy + expected syncagentv1alpha1.RelatedResourceCleanupPolicy + }{ + { + name: "no cleanup, no policy defaults to Orphan", + expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan, + }, + { + name: "legacy cleanup:true maps to OnPrimaryDeletion", + cleanup: true, + expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion, + }, + { + name: "explicit Orphan wins over cleanup:false", + policy: syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan, + expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan, + }, + { + name: "explicit policy wins over legacy cleanup:true", + cleanup: true, + policy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + expected: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + }, + { + name: "explicit OnPrimaryDeletion without cleanup", + policy: syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion, + expected: syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion, + }, + { + name: "explicit MatchOrigin without cleanup", + policy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + expected: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + }, + } + + for _, tc := range testcases { + t.Run(tc.name, func(t *testing.T) { + spec := &syncagentv1alpha1.RelatedResourceSpec{ + Cleanup: tc.cleanup, + CleanupPolicy: tc.policy, + } + + if got := spec.EffectiveCleanupPolicy(); got != tc.expected { + t.Errorf("expected %q, got %q", tc.expected, got) + } + }) + } +} + +func TestRelatedCopyLabelsSelectorRoundTrip(t *testing.T) { + primary := &unstructured.Unstructured{} + primary.SetName("my-primary") + primary.SetNamespace("some-namespace") + + const ( + clusterName = logicalcluster.Name("cluster-a") + publishedRes = "published-resource-a" + identifier = "credentials" + agentName = "agent-1" + ) + + labelSet := relatedCopyLabels(primary, clusterName, publishedRes, identifier, agentName) + + // the labels produced for a copy must match the selector used to find them again. + selector := relatedCopySelector(primary, clusterName, publishedRes, identifier, agentName) + if !selector.Matches(labels.Set(labelSet)) { + t.Errorf("selector %q does not match its own labels %v", selector, labelSet) + } + + // a selector for a different identifier must not match. + otherIdentifier := relatedCopySelector(primary, clusterName, publishedRes, "other", agentName) + if otherIdentifier.Matches(labels.Set(labelSet)) { + t.Errorf("selector for a different identifier unexpectedly matched labels %v", labelSet) + } + + // a selector for a different agent must not match. + otherAgent := relatedCopySelector(primary, clusterName, publishedRes, identifier, "agent-2") + if otherAgent.Matches(labels.Set(labelSet)) { + t.Errorf("selector for a different agent unexpectedly matched labels %v", labelSet) + } + + // a selector for a different primary object must not match. + otherPrimary := &unstructured.Unstructured{} + otherPrimary.SetName("other-primary") + otherPrimary.SetNamespace("some-namespace") + if relatedCopySelector(otherPrimary, clusterName, publishedRes, identifier, agentName).Matches(labels.Set(labelSet)) { + t.Errorf("selector for a different primary unexpectedly matched labels %v", labelSet) + } + + // a selector for a primary in a different logical cluster (workspace) must not match; the + // destination is shared across workspaces, so two primaries with identical name+namespace in + // different clusters must not prune each other's copies. + if relatedCopySelector(primary, "cluster-b", publishedRes, identifier, agentName).Matches(labels.Set(labelSet)) { + t.Errorf("selector for a different cluster unexpectedly matched labels %v", labelSet) + } + + // a selector for a different owning PublishedResource must not match; two PublishedResources that + // project to the same Kind, reuse an identifier and have primaries sharing name+namespace must + // not prune each other's copies. + if relatedCopySelector(primary, clusterName, "published-resource-b", identifier, agentName).Matches(labels.Set(labelSet)) { + t.Errorf("selector for a different published resource unexpectedly matched labels %v", labelSet) + } + + // a cluster-scoped primary (no namespace) still produces a valid, round-tripping label set; its + // empty namespace folds into the owner hash and cannot collide with a namespaced primary. + clusterPrimary := &unstructured.Unstructured{} + clusterPrimary.SetName("cluster-primary") + clusterLabels := relatedCopyLabels(clusterPrimary, clusterName, publishedRes, identifier, agentName) + if _, ok := clusterLabels[relatedOwnerLabel]; !ok { + t.Errorf("expected a related-owner label, got %v", clusterLabels) + } + if !relatedCopySelector(clusterPrimary, clusterName, publishedRes, identifier, agentName).Matches(labels.Set(clusterLabels)) { + t.Errorf("cluster-scoped selector does not match its own labels %v", clusterLabels) + } + + // the provenance is exactly the three labels (owner + identifier + agent); the four separate + // cluster/PublishedResource/name/namespace labels were collapsed into the single owner hash. + if len(labelSet) != 3 { + t.Errorf("expected exactly 3 provenance labels (owner, identifier, agent), got %d: %v", len(labelSet), labelSet) + } +} + +func TestRememberRelatedObjectsSkipsEmptyResolution(t *testing.T) { + const identifier = "credentials" + + relatedAnnotation := relatedObjectAnnotationPrefix + identifier + ".0" + + // A primary that already carries a related-object annotation for this identifier (from a previous + // pass) plus an unrelated annotation that must never be touched. + primary := newUnstructured(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-primary", + Namespace: "default", + Annotations: map[string]string{ + relatedAnnotation: `{"name":"secret-copy","namespace":"default","apiVersion":"v1","kind":"Secret"}`, + "example.com/keep": "yes", + }, + }, + }) + + client := buildFakeClient(primary) + remote := syncSide{object: primary.DeepCopy(), client: client} + + s := &ResourceSyncer{} + + // With an empty resolved set, the informational annotations must be left untouched: no patch, no + // requeue. Wiping them here would churn the primary object for every cleanup policy, not just the + // pruning ones (the prune below is the authoritative cleanup for the copies themselves). + requeue, err := s.rememberRelatedObjects(t.Context(), zap.NewNop().Sugar(), remote, identifier, nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if requeue { + t.Error("expected requeue=false for an empty resolution, but got true") + } + + persisted := &unstructured.Unstructured{} + persisted.SetGroupVersionKind(primary.GroupVersionKind()) + if err := client.Get(t.Context(), ctrlruntimeclient.ObjectKeyFromObject(primary), persisted); err != nil { + t.Fatalf("failed to get persisted object: %v", err) + } + + if got := persisted.GetAnnotations()[relatedAnnotation]; got == "" { + t.Error("expected the pre-existing related-object annotation to be preserved, but it was wiped") + } + if got := persisted.GetAnnotations()["example.com/keep"]; got != "yes" { + t.Errorf("unrelated annotation must be preserved, got %q", got) + } +} + +func TestPruneRelatedCopies(t *testing.T) { + log := zap.NewNop().Sugar() + + const ( + clusterName = logicalcluster.Name("cluster-a") + publishedRes = "pr-a" + identifier = "credentials" + agentName = "agent-1" + ) + + primary := &unstructured.Unstructured{} + primary.SetAPIVersion("v1") + primary.SetKind("ConfigMap") + primary.SetName("my-primary") + primary.SetNamespace("default") + + secretGVK := schema.GroupVersionKind{Version: "v1", Kind: "Secret"} + selector := relatedCopySelector(primary, clusterName, publishedRes, identifier, agentName) + + // makeCopy builds a Secret carrying the provenance labels for the given identifier (so it is only + // in scope for that identifier's selector), optionally with a finalizer. + makeCopy := func(name, namespace, forIdentifier string, withFinalizer bool) *unstructured.Unstructured { + meta := metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: relatedCopyLabels(primary, clusterName, publishedRes, forIdentifier, agentName), + } + if withFinalizer { + meta.Finalizers = []string{"example.com/test"} + } + return newUnstructured(&corev1.Secret{ObjectMeta: meta}) + } + + // listSecretNames returns the names of all Secrets currently present, for assertions. + listSecretNames := func(t *testing.T, client ctrlruntimeclient.Client) sets.Set[string] { + t.Helper() + list := &corev1.SecretList{} + if err := client.List(t.Context(), list); err != nil { + t.Fatalf("failed to list secrets: %v", err) + } + names := sets.New[string]() + for _, item := range list.Items { + names.Insert(item.Name) + } + return names + } + + // "foreign" is a copy for a different related resource identifier on the same primary; the + // credentials selector must never match it, so it is never pruned. + foreign := makeCopy("foreign-copy", "default", "other-identifier", false) + + t.Run("mid-life prune deletes copies outside the keep set and leaves the rest", func(t *testing.T) { + keepMe := makeCopy("keep-me", "default", identifier, false) + pruneMe := makeCopy("prune-me", "other-ns", identifier, false) // also proves cross-namespace pruning + client := buildFakeClient(keepMe, pruneMe, foreign) + + keep := sets.New(relatedCopyKey("default", "keep-me")) + requeue, err := (&ResourceSyncer{}).pruneRelatedCopies(t.Context(), log, syncSide{client: client}, primary, secretGVK, selector, keep, false) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !requeue { + t.Error("expected requeue=true after deleting a copy") + } + + remaining := listSecretNames(t, client) + if !remaining.Has("keep-me") { + t.Error("kept copy was unexpectedly deleted") + } + if remaining.Has("prune-me") { + t.Error("stale copy outside the keep set was not pruned") + } + if !remaining.Has("foreign-copy") { + t.Error("copy for a different identifier must not be pruned") + } + }) + + t.Run("teardown deletes every matching copy but not foreign copies", func(t *testing.T) { + a := makeCopy("copy-a", "default", identifier, false) + b := makeCopy("copy-b", "default", identifier, false) + client := buildFakeClient(a, b, foreign) + + requeue, err := (&ResourceSyncer{}).pruneRelatedCopies(t.Context(), log, syncSide{client: client}, primary, secretGVK, selector, nil, true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !requeue { + t.Error("expected requeue=true after deleting copies") + } + + remaining := listSecretNames(t, client) + if remaining.Has("copy-a") || remaining.Has("copy-b") { + t.Errorf("expected all matching copies to be deleted, still present: %v", remaining) + } + if !remaining.Has("foreign-copy") { + t.Error("copy for a different identifier must not be pruned") + } + }) + + t.Run("empty match set is a no-op", func(t *testing.T) { + client := buildFakeClient(foreign) + + requeue, err := (&ResourceSyncer{}).pruneRelatedCopies(t.Context(), log, syncSide{client: client}, primary, secretGVK, selector, nil, true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if requeue { + t.Error("expected requeue=false when nothing matches the selector") + } + if !listSecretNames(t, client).Has("foreign-copy") { + t.Error("non-matching copy must be left untouched") + } + }) + + t.Run("copy already being deleted requeues instead of erroring", func(t *testing.T) { + deleting := makeCopy("deleting-copy", "default", identifier, true) + client := buildFakeClient(deleting) + + // Delete it once: because it has a finalizer, the fake client only sets a deletion timestamp + // and retains the object, mimicking a copy that is mid-deletion. + if err := client.Delete(t.Context(), deleting); err != nil { + t.Fatalf("failed to start deletion: %v", err) + } + + requeue, err := (&ResourceSyncer{}).pruneRelatedCopies(t.Context(), log, syncSide{client: client}, primary, secretGVK, selector, nil, true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !requeue { + t.Error("expected requeue=true so the reconcile comes back once the copy is gone") + } + if !listSecretNames(t, client).Has("deleting-copy") { + t.Error("a copy that is already being deleted must not be treated as an error or vanish early") + } + }) +} + +// TestConfirmOriginEmpty guards the safety check that prevents a MatchOrigin prune from deleting +// every related copy on a merely-transiently-empty resolution: the primary's cache-backed origin +// read can momentarily return nothing (relisting/stale informer) even though the origin objects +// still exist. confirmOriginEmpty re-runs the resolution against a live reader to distinguish a +// genuine empty origin (prune) from a stale one (skip + requeue). +func TestConfirmOriginEmpty(t *testing.T) { + // A related Secret found via a label selector in a fixed namespace. + relRes := syncagentv1alpha1.RelatedResourceSpec{ + Identifier: "credentials", + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Kind: "Secret", + Object: syncagentv1alpha1.RelatedResourceObject{ + RelatedResourceObjectSpec: syncagentv1alpha1.RelatedResourceObjectSpec{ + Selector: &syncagentv1alpha1.RelatedResourceObjectSelector{ + LabelSelector: metav1.LabelSelector{ + MatchLabels: map[string]string{"app": "credentials"}, + }, + Rewrite: syncagentv1alpha1.RelatedResourceSelectorRewrite{ + Template: &syncagentv1alpha1.TemplateExpression{Template: "{{ .Value }}"}, + }, + }, + }, + // static namespace so resolution never depends on the primary's namespace. + Namespace: &syncagentv1alpha1.RelatedResourceObjectSpec{ + Template: &syncagentv1alpha1.TemplateExpression{Template: "dummy-namespace"}, + }, + }, + } + + // origin: service, so origin = local (service cluster), dest = remote (kcp). + originPrimary := &unstructured.Unstructured{} + originPrimary.SetName("my-primary") + destPrimary := &unstructured.Unstructured{} + destPrimary.SetName("my-primary") + + credSecret := newUnstructured(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "dummy-namespace", + Name: "cred-1", + Labels: map[string]string{"app": "credentials"}, + }, + }) + + // The origin-side cached client is EMPTY: this simulates the reconcile seeing a stale/relisting + // cache that returns no objects even though the origin Secret still exists. + origin := syncSide{client: buildFakeClient(), object: originPrimary} + dest := syncSide{client: buildFakeClient(), object: destPrimary} + + t.Run("no live reader configured is reported as not-checked", func(t *testing.T) { + s := &ResourceSyncer{} + + empty, checked, err := s.confirmOriginEmpty(t.Context(), origin, dest, relRes) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if checked { + t.Error("expected checked=false when no live reader is configured") + } + if empty { + t.Error("expected empty=false (unverified) when no live reader is configured") + } + }) + + t.Run("live read still sees origin objects: not confirmed empty", func(t *testing.T) { + // The live reader sees the Secret the stale cache missed, so the destructive prune must be + // held back. + s := &ResourceSyncer{localAPIReader: buildFakeClient(credSecret)} + + empty, checked, err := s.confirmOriginEmpty(t.Context(), origin, dest, relRes) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !checked { + t.Fatal("expected checked=true when a live reader is configured") + } + if empty { + t.Error("expected confirmedEmpty=false because the live read still sees the origin Secret") + } + }) + + t.Run("live read also empty: confirmed empty", func(t *testing.T) { + // Both the cache and the live read agree there is nothing, so the prune may proceed. + s := &ResourceSyncer{localAPIReader: buildFakeClient()} + + empty, checked, err := s.confirmOriginEmpty(t.Context(), origin, dest, relRes) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !checked { + t.Fatal("expected checked=true when a live reader is configured") + } + if !empty { + t.Error("expected confirmedEmpty=true because the live read also finds no origin objects") + } + }) +} + func TestResolveRelatedResourceObjects(t *testing.T) { // in kcp primaryObject := newUnstructured(&dummyv1alpha1.Thing{ diff --git a/internal/sync/types.go b/internal/sync/types.go index e048ec58..f51de4e8 100644 --- a/internal/sync/types.go +++ b/internal/sync/types.go @@ -53,6 +53,26 @@ const ( // full annotation name, the annotation value is a JSON string containing GVK and // metadata of the related object. relatedObjectAnnotationPrefix = "related-resources.syncagent.kcp.io/" + + // The following label/annotations are put on the destination copies of related resources to + // link them back to their owning primary object and the related resource identifier. They allow + // the agent to List all copies belonging to a specific primary + identifier so that it can prune + // copies whose origin object no longer exists (cleanupPolicy: MatchOrigin) or delete all copies + // on primary teardown. + // + // The owner tuple — the primary's logical cluster (workspace), the owning PublishedResource and + // the primary's namespace/name — is hashed into a single related-owner label. The destination is + // shared across all kcp workspaces and all PublishedResources, so all of these dimensions must be + // part of the identity; folding them into one hash makes that uniqueness inherent in the hash + // input and keeps the value a valid label regardless of the source lengths or characters. The + // identifier and agent name are already valid label values and are kept verbatim (and queryable); + // the plaintext primary name/namespace are kept as annotations for humans. + + relatedOwnerLabel = "syncagent.kcp.io/related-owner" + relatedIdentifierLabel = "syncagent.kcp.io/related-identifier" + + relatedPrimaryNamespaceAnnotation = "syncagent.kcp.io/related-primary-namespace" + relatedPrimaryNameAnnotation = "syncagent.kcp.io/related-primary-name" ) func OwnedBy(obj ctrlruntimeclient.Object, agentName string) bool { diff --git a/sdk/apis/syncagent/v1alpha1/published_resource.go b/sdk/apis/syncagent/v1alpha1/published_resource.go index 77737290..0e6fdde9 100644 --- a/sdk/apis/syncagent/v1alpha1/published_resource.go +++ b/sdk/apis/syncagent/v1alpha1/published_resource.go @@ -201,6 +201,25 @@ const ( RelatedResourceOriginKcp RelatedResourceOrigin = "kcp" ) +// RelatedResourceCleanupPolicy controls when the syncagent deletes the destination copies of a +// related resource. The original object on the origin side is never deleted, regardless of the +// chosen policy. +// +// +kubebuilder:validation:Enum=Orphan;OnPrimaryDeletion;MatchOrigin +type RelatedResourceCleanupPolicy string + +const ( + // RelatedResourceCleanupPolicyOrphan never deletes copies (default; == legacy cleanup:false). + RelatedResourceCleanupPolicyOrphan RelatedResourceCleanupPolicy = "Orphan" + // RelatedResourceCleanupPolicyOnPrimaryDeletion deletes copies only when the primary object + // is deleted (== legacy cleanup:true). + RelatedResourceCleanupPolicyOnPrimaryDeletion RelatedResourceCleanupPolicy = "OnPrimaryDeletion" + // RelatedResourceCleanupPolicyMatchOrigin keeps the destination set equal to the origin set: + // a copy is pruned as soon as its origin object is gone, and all copies are deleted when the + // primary object is deleted. + RelatedResourceCleanupPolicyMatchOrigin RelatedResourceCleanupPolicy = "MatchOrigin" +) + // RelatedResourceSpec describes a single related resource, which might point to // any number of actual Kubernetes objects. // @@ -211,11 +230,19 @@ const ( // 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.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 { // Identifier is a unique name for this related resource. The name must be unique within one // PublishedResource and is the key by which consumers (end users) can identify and consume the // related resource. Common names are "connection-details" or "credentials". - // The identifier must be an alphanumeric string. + // + // The identifier is used verbatim as a label value on the synced copies of the related resource, + // so it must be a valid label value: a lowercase RFC 1123 label consisting of lowercase + // alphanumeric characters or '-', starting and ending with an alphanumeric character, and at + // most 63 characters long. + // +kubebuilder:validation:MaxLength=63 + // +kubebuilder:validation:Pattern=`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$` Identifier string `json:"identifier"` // +kubebuilder:validation:Enum=service;kcp @@ -248,8 +275,24 @@ type RelatedResourceSpec struct { // the destination side (i.e. the original related object will not be touched, regardless of this // option). Leaving this disabled, the syncagent will only create copies of the related objects, // but never delete them itself. + // + // Deprecated: use CleanupPolicy instead. When CleanupPolicy is empty, cleanup:true is treated + // as OnPrimaryDeletion and cleanup:false as Orphan. Cleanup bool `json:"cleanup,omitempty"` + // CleanupPolicy controls when the syncagent deletes destination copies of this related + // resource. The original object on the origin side is never deleted. + // + // - Orphan (default): copies are never deleted by the agent. + // - OnPrimaryDeletion: copies are deleted only when the primary object is deleted. + // - MatchOrigin: copies are pruned as soon as their origin object no longer exists, and + // all copies are deleted when the primary object is deleted. + // + // When left empty, the deprecated Cleanup field is used to derive the effective policy. + // + // +optional + CleanupPolicy RelatedResourceCleanupPolicy `json:"cleanupPolicy,omitempty"` + // Projection is used to change the GVK of a related resource on the opposite side of // its origin. // All fields in the projection are optional. If a field is set, it will overwrite @@ -285,6 +328,21 @@ type RelatedResourceSpec struct { SyncStatus bool `json:"syncStatus,omitempty"` } +// EffectiveCleanupPolicy normalizes the deprecated Cleanup bool and the CleanupPolicy field into +// a single effective policy. CleanupPolicy takes precedence; when it is empty, cleanup:true maps +// to OnPrimaryDeletion and cleanup:false to Orphan. +func (r *RelatedResourceSpec) EffectiveCleanupPolicy() RelatedResourceCleanupPolicy { + if r.CleanupPolicy != "" { + return r.CleanupPolicy + } + + if r.Cleanup { + return RelatedResourceCleanupPolicyOnPrimaryDeletion + } + + return RelatedResourceCleanupPolicyOrphan +} + // RelatedResourceWatch configures how the watch handler maps a changed related resource // back to its owning primary object. // Exactly one of ByOwner or BySelector must be set. diff --git a/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go b/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go index 5517ac2b..4196a14b 100644 --- a/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go +++ b/sdk/applyconfiguration/syncagent/v1alpha1/relatedresourcespec.go @@ -25,19 +25,20 @@ import ( // RelatedResourceSpecApplyConfiguration represents a declarative configuration of the RelatedResourceSpec type for use // with apply. type RelatedResourceSpecApplyConfiguration struct { - Identifier *string `json:"identifier,omitempty"` - Origin *syncagentv1alpha1.RelatedResourceOrigin `json:"origin,omitempty"` - Group *string `json:"group,omitempty"` - Version *string `json:"version,omitempty"` - Resource *string `json:"resource,omitempty"` - Kind *string `json:"kind,omitempty"` - IdentityHash *string `json:"identityHash,omitempty"` - Cleanup *bool `json:"cleanup,omitempty"` - Projection *RelatedResourceProjectionApplyConfiguration `json:"projection,omitempty"` - Object *RelatedResourceObjectApplyConfiguration `json:"object,omitempty"` - Mutation *ResourceMutationSpecApplyConfiguration `json:"mutation,omitempty"` - Watch *RelatedResourceWatchApplyConfiguration `json:"watch,omitempty"` - SyncStatus *bool `json:"syncStatus,omitempty"` + Identifier *string `json:"identifier,omitempty"` + Origin *syncagentv1alpha1.RelatedResourceOrigin `json:"origin,omitempty"` + Group *string `json:"group,omitempty"` + Version *string `json:"version,omitempty"` + Resource *string `json:"resource,omitempty"` + Kind *string `json:"kind,omitempty"` + IdentityHash *string `json:"identityHash,omitempty"` + Cleanup *bool `json:"cleanup,omitempty"` + CleanupPolicy *syncagentv1alpha1.RelatedResourceCleanupPolicy `json:"cleanupPolicy,omitempty"` + Projection *RelatedResourceProjectionApplyConfiguration `json:"projection,omitempty"` + Object *RelatedResourceObjectApplyConfiguration `json:"object,omitempty"` + Mutation *ResourceMutationSpecApplyConfiguration `json:"mutation,omitempty"` + Watch *RelatedResourceWatchApplyConfiguration `json:"watch,omitempty"` + SyncStatus *bool `json:"syncStatus,omitempty"` } // RelatedResourceSpecApplyConfiguration constructs a declarative configuration of the RelatedResourceSpec type for use with @@ -110,6 +111,14 @@ func (b *RelatedResourceSpecApplyConfiguration) WithCleanup(value bool) *Related return b } +// WithCleanupPolicy sets the CleanupPolicy 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 CleanupPolicy field is set to the value of the last call. +func (b *RelatedResourceSpecApplyConfiguration) WithCleanupPolicy(value syncagentv1alpha1.RelatedResourceCleanupPolicy) *RelatedResourceSpecApplyConfiguration { + b.CleanupPolicy = &value + return b +} + // WithProjection sets the Projection 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 Projection field is set to the value of the last call. diff --git a/test/e2e/sdk/cel_validations_test.go b/test/e2e/sdk/cel_validations_test.go index 96da0369..999f00f9 100644 --- a/test/e2e/sdk/cel_validations_test.go +++ b/test/e2e/sdk/cel_validations_test.go @@ -209,6 +209,61 @@ func TestValidateRelatedResourceSpec(t *testing.T) { SyncStatus: true, }, }, + { + name: "cleanupPolicy MatchOrigin with origin service requires watch", + valid: false, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Resource: "things", + Version: "v1", + CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + }, + }, + { + name: "cleanupPolicy MatchOrigin with origin service and watch is valid", + valid: true, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Resource: "things", + Version: "v1", + CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + Watch: &syncagentv1alpha1.RelatedResourceWatch{ + ByOwner: &syncagentv1alpha1.RelatedResourceWatchByOwner{}, + }, + }, + }, + { + name: "cleanupPolicy MatchOrigin with origin kcp does not require watch", + valid: true, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginKcp, + Resource: "things", + Version: "v1", + CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + }, + }, + { + name: "cleanup true conflicts with cleanupPolicy Orphan", + valid: false, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Resource: "things", + Version: "v1", + Cleanup: true, + CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyOrphan, + }, + }, + { + name: "cleanup true with cleanupPolicy OnPrimaryDeletion is valid", + valid: true, + spec: syncagentv1alpha1.RelatedResourceSpec{ + Origin: syncagentv1alpha1.RelatedResourceOriginService, + Resource: "things", + Version: "v1", + Cleanup: true, + CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyOnPrimaryDeletion, + }, + }, } alphaNum := regexp.MustCompile(`[^a-z0-9]`) @@ -218,12 +273,22 @@ func TestValidateRelatedResourceSpec(t *testing.T) { crdName := strings.ToLower(tt.name) crdName = alphaNum.ReplaceAllLiteralString(crdName, "-") + spec := tt.spec + + // Identifier is required and validated as a label value. None of these cases exercise + // identifier validation itself, so give them a valid identifier unless they set their + // own; otherwise the identifier pattern would reject the spec before the rule actually + // under test is reached. + if spec.Identifier == "" { + spec.Identifier = "test-identifier" + } + pubRes := &syncagentv1alpha1.PublishedResource{ ObjectMeta: metav1.ObjectMeta{ Name: "test-" + crdName, }, Spec: syncagentv1alpha1.PublishedResourceSpec{ - Related: []syncagentv1alpha1.RelatedResourceSpec{tt.spec}, + Related: []syncagentv1alpha1.RelatedResourceSpec{spec}, }, } diff --git a/test/e2e/sync/cleanup_test.go b/test/e2e/sync/cleanup_test.go index a1891b10..faca8175 100644 --- a/test/e2e/sync/cleanup_test.go +++ b/test/e2e/sync/cleanup_test.go @@ -414,3 +414,399 @@ spec: t.Errorf("Failed to get Secret on service cluster: %v", err) } } + +// matchOriginPublishedResource builds a PublishedResource that publishes CronTabs with a related +// Secret (found via a label selector so that a whole set of objects can match) using +// cleanupPolicy: MatchOrigin. The related object name is preserved on both sides. +func matchOriginPublishedResource(origin syncagentv1alpha1.RelatedResourceOrigin, watchLabels map[string]string) *syncagentv1alpha1.PublishedResource { + return &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: "credentials", + Origin: origin, + Kind: "Secret", + CleanupPolicy: syncagentv1alpha1.RelatedResourceCleanupPolicyMatchOrigin, + Object: syncagentv1alpha1.RelatedResourceObject{ + RelatedResourceObjectSpec: syncagentv1alpha1.RelatedResourceObjectSpec{ + Selector: &syncagentv1alpha1.RelatedResourceObjectSelector{ + LabelSelector: metav1.LabelSelector{ + MatchLabels: map[string]string{ + "app": "credentials", + }, + }, + Rewrite: syncagentv1alpha1.RelatedResourceSelectorRewrite{ + Template: &syncagentv1alpha1.TemplateExpression{ + // keep the same name on both sides + Template: "{{ .Value }}", + }, + }, + }, + }, + }, + Watch: &syncagentv1alpha1.RelatedResourceWatch{ + BySelector: &metav1.LabelSelector{ + MatchLabels: watchLabels, + }, + }, + }, + }, + }, + } +} + +func credentialSecret(name, namespace string) *corev1.Secret { + return &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{ + "app": "credentials", + }, + }, + Data: map[string][]byte{ + "password": []byte("hunter2"), + }, + Type: corev1.SecretTypeOpaque, + } +} + +func waitForSecret(t *testing.T, ctx context.Context, client ctrlruntimeclient.Client, key types.NamespacedName) { + t.Helper() + + err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) { + return client.Get(ctx, key, &corev1.Secret{}) == nil, nil + }) + if err != nil { + t.Fatalf("Secret %v did not appear: %v", key, err) + } +} + +func waitForSecretGone(t *testing.T, ctx context.Context, client ctrlruntimeclient.Client, key types.NamespacedName) { + t.Helper() + + err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) { + return apierrors.IsNotFound(client.Get(ctx, key, &corev1.Secret{})), nil + }) + if err != nil { + t.Fatalf("Secret %v was not pruned: %v", key, err) + } +} + +func requireSecretExists(t *testing.T, ctx context.Context, client ctrlruntimeclient.Client, key types.NamespacedName, msg string) { + t.Helper() + + if err := client.Get(ctx, key, &corev1.Secret{}); err != nil { + t.Errorf("%s: %v", msg, err) + } +} + +// TestRelatedResourceMatchOriginServiceOrigin verifies that with cleanupPolicy: MatchOrigin and +// origin: service, deleting one source object mid-life prunes exactly its kcp copy while the other +// copy and the surviving source object remain, and that primary teardown removes all kcp copies +// without touching the service-cluster sources. +func TestRelatedResourceMatchOriginServiceOrigin(t *testing.T) { + const ( + apiExportName = "kcp.example.com" + orgWorkspace = "match-origin-service" + ) + + ctx := t.Context() + ctrlruntime.SetLogger(logr.Discard()) + + orgKubconfig := utils.CreateOrganization(t, ctx, orgWorkspace, apiExportName) + + envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{ + "test/crds/crontab.yaml", + }) + + watchLabels := map[string]string{"syncagent-e2e": "find-me"} + + t.Log("Publishing CRDs…") + if err := envtestClient.Create(ctx, matchOriginPublishedResource(syncagentv1alpha1.RelatedResourceOriginService, watchLabels)); err != nil { + t.Fatalf("Failed to create PublishedResource: %v", err) + } + + utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "") + + kcpClusterClient := utils.GetKcpAdminClusterClient(t) + teamClusterPath := logicalcluster.NewPath("root").Join(orgWorkspace).Join("team-1") + teamClient := kcpClusterClient.Cluster(teamClusterPath) + + utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{ + Group: "kcp.example.com", + Version: "v1", + Kind: "CronTab", + }) + + t.Log("Creating CronTab in kcp…") + crontab := utils.YAMLToUnstructured(t, ` +apiVersion: kcp.example.com/v1 +kind: CronTab +metadata: + namespace: default + name: my-crontab +spec: + cronSpec: '* * *' + image: ubuntu:latest +`) + crontab.SetLabels(watchLabels) + + if err := teamClient.Create(ctx, crontab); err != nil { + t.Fatalf("Failed to create CronTab in kcp: %v", err) + } + + // wait for the CronTab to sync down (this also creates the synced-default namespace) + t.Log("Waiting for CronTab to be synced to service cluster…") + serviceCrontab := &unstructured.Unstructured{} + serviceCrontab.SetAPIVersion("example.com/v1") + serviceCrontab.SetKind("CronTab") + err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) { + return envtestClient.Get(ctx, types.NamespacedName{Namespace: "synced-default", Name: "my-crontab"}, serviceCrontab) == nil, nil + }) + if err != nil { + t.Fatalf("CronTab was not synced to service cluster: %v", err) + } + + // create two related Secrets on the service cluster + t.Log("Creating credential Secrets on service cluster…") + for _, name := range []string{"cred-1", "cred-2"} { + if err := envtestClient.Create(ctx, credentialSecret(name, "synced-default")); err != nil { + t.Fatalf("Failed to create Secret %s: %v", name, err) + } + } + + // wait for both to be synced up to kcp + t.Log("Waiting for both Secrets to be synced to kcp…") + waitForSecret(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"}) + waitForSecret(t, ctx, teamClient, types.NamespacedName{Name: "cred-2", Namespace: "default"}) + + // delete one source object while the primary still lives + t.Log("Deleting cred-1 on the service cluster…") + if err := envtestClient.Delete(ctx, credentialSecret("cred-1", "synced-default")); err != nil { + t.Fatalf("Failed to delete source Secret cred-1: %v", err) + } + + // its kcp copy should be pruned… + t.Log("Verifying that the cred-1 copy in kcp is pruned…") + waitForSecretGone(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"}) + + // …while the other copy and the surviving source object remain untouched. + requireSecretExists(t, ctx, teamClient, types.NamespacedName{Name: "cred-2", Namespace: "default"}, "cred-2 copy in kcp should remain") + requireSecretExists(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"}, "cred-2 source on service cluster should remain") + + // now tear down the primary + t.Log("Deleting CronTab in kcp…") + if err := teamClient.Delete(ctx, crontab); err != nil { + t.Fatalf("Failed to delete CronTab: %v", err) + } + + // all kcp copies should be gone + t.Log("Verifying that all related copies in kcp are gone…") + waitForSecretGone(t, ctx, teamClient, types.NamespacedName{Name: "cred-2", Namespace: "default"}) + + // the service-cluster source must never be touched + requireSecretExists(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"}, "cred-2 source on service cluster should remain after teardown") +} + +// TestRelatedResourceMatchOriginTeardownReclaimsOrphan reproduces the teardown-orphan bug: a source +// object is deleted mid-life (orphaning its copy), then the primary is deleted. Even the orphaned +// copy must be reclaimed on teardown. +func TestRelatedResourceMatchOriginTeardownReclaimsOrphan(t *testing.T) { + const ( + apiExportName = "kcp.example.com" + orgWorkspace = "match-origin-orphan" + ) + + ctx := t.Context() + ctrlruntime.SetLogger(logr.Discard()) + + orgKubconfig := utils.CreateOrganization(t, ctx, orgWorkspace, apiExportName) + + envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{ + "test/crds/crontab.yaml", + }) + + watchLabels := map[string]string{"syncagent-e2e": "find-me"} + + t.Log("Publishing CRDs…") + if err := envtestClient.Create(ctx, matchOriginPublishedResource(syncagentv1alpha1.RelatedResourceOriginService, watchLabels)); err != nil { + t.Fatalf("Failed to create PublishedResource: %v", err) + } + + utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "") + + kcpClusterClient := utils.GetKcpAdminClusterClient(t) + teamClusterPath := logicalcluster.NewPath("root").Join(orgWorkspace).Join("team-1") + teamClient := kcpClusterClient.Cluster(teamClusterPath) + + utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{ + Group: "kcp.example.com", + Version: "v1", + Kind: "CronTab", + }) + + t.Log("Creating CronTab in kcp…") + crontab := utils.YAMLToUnstructured(t, ` +apiVersion: kcp.example.com/v1 +kind: CronTab +metadata: + namespace: default + name: my-crontab +spec: + cronSpec: '* * *' + image: ubuntu:latest +`) + crontab.SetLabels(watchLabels) + + if err := teamClient.Create(ctx, crontab); err != nil { + t.Fatalf("Failed to create CronTab in kcp: %v", err) + } + + t.Log("Waiting for CronTab to be synced to service cluster…") + serviceCrontab := &unstructured.Unstructured{} + serviceCrontab.SetAPIVersion("example.com/v1") + serviceCrontab.SetKind("CronTab") + err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) { + return envtestClient.Get(ctx, types.NamespacedName{Namespace: "synced-default", Name: "my-crontab"}, serviceCrontab) == nil, nil + }) + if err != nil { + t.Fatalf("CronTab was not synced to service cluster: %v", err) + } + + t.Log("Creating credential Secret on service cluster…") + if err := envtestClient.Create(ctx, credentialSecret("cred-1", "synced-default")); err != nil { + t.Fatalf("Failed to create Secret: %v", err) + } + + t.Log("Waiting for Secret to be synced to kcp…") + waitForSecret(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"}) + + // Delete the source object, but immediately tear down the primary as well. We do NOT wait for + // the mid-life prune to run first, so that the copy may still be orphaned when teardown begins; + // the List-based teardown must reclaim it regardless. + t.Log("Deleting source Secret and then the CronTab…") + if err := envtestClient.Delete(ctx, credentialSecret("cred-1", "synced-default")); err != nil { + t.Fatalf("Failed to delete source Secret: %v", err) + } + if err := teamClient.Delete(ctx, crontab); err != nil { + t.Fatalf("Failed to delete CronTab: %v", err) + } + + t.Log("Verifying that no orphaned copy survives in kcp…") + waitForSecretGone(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"}) +} + +// TestRelatedResourceMatchOriginKcpOrigin verifies the selector-shrink case for origin: kcp: when a +// source object is relabelled so it no longer matches the selector (but is not deleted), its +// service-side copy is pruned. This is a case the finalizer path alone misses. +func TestRelatedResourceMatchOriginKcpOrigin(t *testing.T) { + const ( + apiExportName = "kcp.example.com" + orgWorkspace = "match-origin-kcp" + ) + + ctx := t.Context() + ctrlruntime.SetLogger(logr.Discard()) + + orgKubconfig := utils.CreateOrganization(t, ctx, orgWorkspace, apiExportName) + + envtestKubeconfig, envtestClient, _ := utils.RunEnvtest(t, []string{ + "test/crds/crontab.yaml", + }) + + watchLabels := map[string]string{"syncagent-e2e": "find-me"} + + t.Log("Publishing CRDs…") + if err := envtestClient.Create(ctx, matchOriginPublishedResource(syncagentv1alpha1.RelatedResourceOriginKcp, watchLabels)); err != nil { + t.Fatalf("Failed to create PublishedResource: %v", err) + } + + utils.RunAgent(ctx, t, "bob", orgKubconfig, envtestKubeconfig, apiExportName, "") + + kcpClusterClient := utils.GetKcpAdminClusterClient(t) + teamClusterPath := logicalcluster.NewPath("root").Join(orgWorkspace).Join("team-1") + teamClient := kcpClusterClient.Cluster(teamClusterPath) + + utils.WaitForBoundAPI(t, ctx, teamClient, schema.GroupVersionKind{ + Group: "kcp.example.com", + Version: "v1", + Kind: "CronTab", + }) + + t.Log("Creating CronTab in kcp…") + crontab := utils.YAMLToUnstructured(t, ` +apiVersion: kcp.example.com/v1 +kind: CronTab +metadata: + namespace: default + name: my-crontab +spec: + cronSpec: '* * *' + image: ubuntu:latest +`) + crontab.SetLabels(watchLabels) + + if err := teamClient.Create(ctx, crontab); err != nil { + t.Fatalf("Failed to create CronTab in kcp: %v", err) + } + + // wait for the CronTab to be synced down so the synced-default namespace exists + t.Log("Waiting for CronTab to be synced to service cluster…") + serviceCrontab := &unstructured.Unstructured{} + serviceCrontab.SetAPIVersion("example.com/v1") + serviceCrontab.SetKind("CronTab") + err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 30*time.Second, false, func(ctx context.Context) (bool, error) { + return envtestClient.Get(ctx, types.NamespacedName{Namespace: "synced-default", Name: "my-crontab"}, serviceCrontab) == nil, nil + }) + if err != nil { + t.Fatalf("CronTab was not synced to service cluster: %v", err) + } + + // create two related Secrets in kcp (origin: kcp) + t.Log("Creating credential Secrets in kcp…") + for _, name := range []string{"cred-1", "cred-2"} { + if err := teamClient.Create(ctx, credentialSecret(name, "default")); err != nil { + t.Fatalf("Failed to create Secret %s: %v", name, err) + } + } + + // both should be synced down to the service cluster + t.Log("Waiting for both Secrets to be synced to the service cluster…") + waitForSecret(t, ctx, envtestClient, types.NamespacedName{Name: "cred-1", Namespace: "synced-default"}) + waitForSecret(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"}) + + // relabel cred-1 in kcp so it no longer matches the selector (but is NOT deleted) + t.Log("Relabelling cred-1 in kcp so it no longer matches…") + kcpSecret := &corev1.Secret{} + if err := teamClient.Get(ctx, types.NamespacedName{Name: "cred-1", Namespace: "default"}, kcpSecret); err != nil { + t.Fatalf("Failed to get cred-1 in kcp: %v", err) + } + kcpSecret.Labels = map[string]string{"app": "something-else"} + if err := teamClient.Update(ctx, kcpSecret); err != nil { + t.Fatalf("Failed to relabel cred-1 in kcp: %v", err) + } + + // its service-side copy should be pruned… + t.Log("Verifying that the cred-1 copy on the service cluster is pruned…") + waitForSecretGone(t, ctx, envtestClient, types.NamespacedName{Name: "cred-1", Namespace: "synced-default"}) + + // …while the still-matching copy remains and the relabelled source object is untouched. + requireSecretExists(t, ctx, envtestClient, types.NamespacedName{Name: "cred-2", Namespace: "synced-default"}, "cred-2 copy on service cluster should remain") + requireSecretExists(t, ctx, teamClient, types.NamespacedName{Name: "cred-1", Namespace: "default"}, "relabelled cred-1 source in kcp should remain") +} diff --git a/test/e2e/sync/related_test.go b/test/e2e/sync/related_test.go index 7a0f0d9d..ccc6696a 100644 --- a/test/e2e/sync/related_test.go +++ b/test/e2e/sync/related_test.go @@ -1754,10 +1754,14 @@ func compareSecrets(t *testing.T, actual, expected corev1.Secret) error { } func compareUnstructured(actual, expected *unstructured.Unstructured) error { - // ensure the secret in kcp does not have any sync-related metadata + // ensure the copy does not have any sync-related metadata: the internal kcp claim labels, the + // cluster annotation, and the agent's own provenance metadata (the agent-name label plus the + // related-resource ownership labels/annotations the agent stamps on every copy so it can + // enumerate them for pruning). None of these are part of the object's meaningful content. labels := actual.GetLabels() maps.DeleteFunc(labels, func(k, v string) bool { - return strings.HasPrefix(k, "claimed.internal.apis.kcp.io/") + return strings.HasPrefix(k, "claimed.internal.apis.kcp.io/") || + strings.HasPrefix(k, "syncagent.kcp.io/") }) if len(labels) == 0 { labels = nil @@ -1766,6 +1770,9 @@ func compareUnstructured(actual, expected *unstructured.Unstructured) error { annotations := actual.GetAnnotations() delete(annotations, "kcp.io/cluster") + maps.DeleteFunc(annotations, func(k, v string) bool { + return strings.HasPrefix(k, "syncagent.kcp.io/") + }) if len(annotations) == 0 { annotations = nil }