diff --git a/cmd/object-controller/main_test.go b/cmd/object-controller/main_test.go index 6fea75f96b..cdd86235a7 100644 --- a/cmd/object-controller/main_test.go +++ b/cmd/object-controller/main_test.go @@ -19,6 +19,7 @@ import ( "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" ocv1 "github.com/operator-framework/operator-controller/api/v1" "github.com/operator-framework/operator-controller/internal/object-controller/scheme" @@ -60,7 +61,7 @@ func TestVersionWithoutCluster(t *testing.T) { require.NotEmpty(t, output.String()) } -// Exercise the actual manager and revision engine with only the ClusterObjectSet CRD installed. +// Exercise the actual manager and revision engine without ClusterExtension or ClusterCatalog CRDs. func TestStandaloneController(t *testing.T) { ctrl.SetLogger(testr.New(t)) testEnv := test.NewEnv() @@ -93,7 +94,21 @@ func TestStandaloneController(t *testing.T) { defer syncCancel() require.True(t, mgr.GetCache().WaitForCacheSync(syncCtx), "manager cache did not synchronize") - for _, name := range []string{"inline", "secret-ref"} { + waitForStableRevision := func(t *testing.T, cos *ocv1.ClusterObjectSet) { + t.Helper() + lastVersion, unchangedSince := cos.ResourceVersion, time.Now() + require.Eventually(t, func() bool { + if err := cl.Get(ctx, client.ObjectKeyFromObject(cos), cos); err != nil { + return false + } + if cos.ResourceVersion != lastVersion { + lastVersion, unchangedSince = cos.ResourceVersion, time.Now() + } + return time.Since(unchangedSince) >= time.Second + }, 5*time.Second, 100*time.Millisecond) + } + + for _, name := range []string{"inline", "secret-ref", "ce-owned-secret-ref", "ownerless-secret-ref", "mutable-secret-ref"} { t.Run(name, func(t *testing.T) { ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{GenerateName: "standalone-"}} require.NoError(t, cl.Create(ctx, ns)) @@ -103,12 +118,13 @@ func TestStandaloneController(t *testing.T) { "data": map[string]any{"hello": "world"}, }} obj := ocv1.ClusterObjectSetObject{Object: manifest} - if name == "secret-ref" { + var secret *corev1.Secret + if name != "inline" { data, err := json.Marshal(manifest.Object) require.NoError(t, err) - secret := &corev1.Secret{ + secret = &corev1.Secret{ ObjectMeta: metav1.ObjectMeta{Name: "content", Namespace: ns.Name}, - Immutable: ptr.To(true), Data: map[string][]byte{"object": data}, + Immutable: ptr.To(name != "mutable-secret-ref"), Data: map[string][]byte{"object": data}, } require.NoError(t, cl.Create(ctx, secret)) obj = ocv1.ClusterObjectSetObject{Ref: ocv1.ObjectSourceRef{Name: secret.Name, Namespace: secret.Namespace, Key: "object"}} @@ -122,7 +138,57 @@ func TestStandaloneController(t *testing.T) { Phases: []ocv1.ClusterObjectSetPhase{{Name: "deploy", Objects: []ocv1.ClusterObjectSetObject{obj}}}, }, } - require.NoError(t, cl.Create(ctx, cos)) + var deployment *ocv1.ClusterObjectDeployment + if name == "ce-owned-secret-ref" { + // Follow the production CE -> COD -> COS path: the deployment + // creates the revision, while its source remains owned by CE. + deployment = &ocv1.ClusterObjectDeployment{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: ocv1.ClusterObjectDeploymentSpec{ + Template: ocv1.ClusterObjectDeploymentTemplate{ + Spec: ocv1.ClusterObjectDeploymentTemplateSpec{ + Phases: cos.Spec.Phases, CollisionProtection: cos.Spec.CollisionProtection, + }, + }, + }, + } + require.NoError(t, cl.Create(ctx, deployment)) + } else { + require.NoError(t, cl.Create(ctx, cos)) + } + if secret != nil && name != "ownerless-secret-ref" { + if name == "ce-owned-secret-ref" || name == "mutable-secret-ref" { + // Packed sources are owned by ClusterExtension, which is not + // installed in this standalone manager's API server. + secret.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: ocv1.GroupVersion.String(), Kind: "ClusterExtension", + Name: "test-extension", UID: "test-extension", Controller: ptr.To(true), + }} + } else { + require.NoError(t, controllerutil.SetControllerReference(cos, secret, scheme.Scheme)) + } + require.NoError(t, cl.Update(ctx, secret)) + } + rolloutTimeout := time.Minute + if name == "mutable-secret-ref" { + require.EventuallyWithT(t, func(collect *assert.CollectT) { + if !assert.NoError(collect, cl.Get(ctx, client.ObjectKeyFromObject(cos), cos)) { + return + } + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + if assert.NotNil(collect, condition) { + assert.Equal(collect, ocv1.ClusterObjectSetReasonPreconditionsNotMet, condition.Reason) + assert.Contains(collect, condition.Message, "must be immutable but are not") + } + }, 5*time.Second, 100*time.Millisecond) + + // Let status-triggered reconciliations settle before changing only + // the Secret. Recovery must precede the 10-second polling retry. + waitForStableRevision(t, cos) + secret.Immutable = ptr.To(true) + require.NoError(t, cl.Update(ctx, secret)) + rolloutTimeout = 5 * time.Second + } require.EventuallyWithT(t, func(collect *assert.CollectT) { if !assert.NoError(collect, cl.Get(ctx, client.ObjectKeyFromObject(cos), cos)) { return @@ -133,12 +199,59 @@ func TestStandaloneController(t *testing.T) { assert.Equal(collect, metav1.ConditionTrue, ready.Status) assert.Equal(collect, ocv1.ClusterObjectSetReasonAllObjectsReady, ready.Reason) } - }, time.Minute, 100*time.Millisecond) + }, rolloutTimeout, 100*time.Millisecond) cm := &corev1.ConfigMap{} require.NoError(t, cl.Get(ctx, client.ObjectKey{Name: name, Namespace: ns.Name}, cm)) require.Equal(t, "world", cm.Data["hello"]) require.NotNil(t, metav1.GetControllerOf(cm)) require.Equal(t, cos.UID, metav1.GetControllerOf(cm).UID) + if deployment != nil { + require.NotNil(t, metav1.GetControllerOf(cos)) + require.Equal(t, deployment.UID, metav1.GetControllerOf(cos).UID) + } + + if secret != nil { + // Let initial status-triggered reconciliations finish before changing + // only the source. A completed COS does not poll. + waitForStableRevision(t, cos) + + // Replacing a source Secret must trigger content verification, + // and restoring it must unblock + // reconciliation without changing the COS or its managed objects. + original := secret.DeepCopy() + require.NoError(t, cl.Delete(ctx, secret)) + secret.ResourceVersion = "" + secret.UID = "" + changed := manifest.DeepCopy() + changed.Object["data"] = map[string]any{"hello": "changed"} + secret.Data["object"], err = json.Marshal(changed.Object) + require.NoError(t, err) + require.NoError(t, cl.Create(ctx, secret)) + require.EventuallyWithT(t, func(collect *assert.CollectT) { + if !assert.NoError(collect, cl.Get(ctx, client.ObjectKeyFromObject(cos), cos)) { + return + } + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + if assert.NotNil(collect, condition) { + assert.Equal(collect, ocv1.ClusterObjectSetReasonPreconditionsNotMet, condition.Reason) + assert.Contains(collect, condition.Message, "resolved content of 1 phase(s) has changed") + } + }, 30*time.Second, 100*time.Millisecond) + + require.NoError(t, cl.Delete(ctx, secret)) + original.ResourceVersion = "" + original.UID = "" + require.NoError(t, cl.Create(ctx, original)) + require.EventuallyWithT(t, func(collect *assert.CollectT) { + if !assert.NoError(collect, cl.Get(ctx, client.ObjectKeyFromObject(cos), cos)) { + return + } + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + if assert.NotNil(collect, condition) { + assert.Equal(collect, ocv1.ClusterObjectSetReasonAllObjectsReady, condition.Reason) + } + }, 30*time.Second, 100*time.Millisecond) + } // Observe managed-object changes without updating the ClusterObjectSet. originalUID := cm.UID @@ -157,10 +270,85 @@ func TestStandaloneController(t *testing.T) { // The controller releases its finalizer independently of ClusterExtension. // The owner reference above lets Kubernetes garbage-collect the ConfigMap; // envtest does not run that garbage collector. + if deployment != nil { + require.NoError(t, cl.Delete(ctx, deployment)) + } require.NoError(t, cl.Delete(ctx, cos)) require.Eventually(t, func() bool { return apierrors.IsNotFound(cl.Get(ctx, client.ObjectKeyFromObject(cos), cos)) }, time.Minute, 100*time.Millisecond) }) } + + t.Run("shared-secret-ref", func(t *testing.T) { + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{GenerateName: "standalone-shared-"}} + require.NoError(t, cl.Create(ctx, ns)) + manifest := func(name, value string) []byte { + data, err := json.Marshal(&corev1.ConfigMap{ + TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"}, + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns.Name}, + Data: map[string]string{"hello": value}, + }) + require.NoError(t, err) + return data + } + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "content", Namespace: ns.Name}, + Immutable: ptr.To(true), + Data: map[string][]byte{ + "first": manifest("shared-first", "world"), + "second": manifest("shared-second", "world"), + }, + } + require.NoError(t, cl.Create(ctx, secret)) + revisions := make([]*ocv1.ClusterObjectSet, 0, 2) + for _, key := range []string{"first", "second"} { + cos := &ocv1.ClusterObjectSet{ + ObjectMeta: metav1.ObjectMeta{Name: "shared-" + key + "-1"}, + Spec: ocv1.ClusterObjectSetSpec{ + Group: "shared-" + key, Revision: 1, + LifecycleState: ocv1.ClusterObjectSetLifecycleStateActive, + CollisionProtection: ocv1.CollisionProtectionPrevent, + Phases: []ocv1.ClusterObjectSetPhase{{Name: "deploy", Objects: []ocv1.ClusterObjectSetObject{{ + Ref: ocv1.ObjectSourceRef{Namespace: secret.Namespace, Name: secret.Name, Key: key}, + }}}}, + }, + } + require.NoError(t, cl.Create(ctx, cos)) + revisions = append(revisions, cos) + } + waitForReason := func(reason string, timeout time.Duration) { + t.Helper() + require.EventuallyWithT(t, func(collect *assert.CollectT) { + for _, cos := range revisions { + if !assert.NoError(collect, cl.Get(ctx, client.ObjectKeyFromObject(cos), cos)) { + continue + } + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + if assert.NotNil(collect, condition) { + assert.Equal(collect, reason, condition.Reason, "revision %s", cos.Name) + } + } + }, timeout, 100*time.Millisecond) + } + waitForReason(ocv1.ClusterObjectSetReasonAllObjectsReady, time.Minute) + for _, cos := range revisions { + waitForStableRevision(t, cos) + } + + changed := secret.DeepCopy() + changed.ResourceVersion = "" + changed.UID = "" + changed.Data["first"] = manifest("shared-first", "changed") + changed.Data["second"] = manifest("shared-second", "changed") + require.NoError(t, cl.Delete(ctx, secret)) + require.NoError(t, cl.Create(ctx, changed)) + waitForReason(ocv1.ClusterObjectSetReasonPreconditionsNotMet, 5*time.Second) + + require.NoError(t, cl.Delete(ctx, changed)) + secret.ResourceVersion = "" + secret.UID = "" + require.NoError(t, cl.Create(ctx, secret)) + waitForReason(ocv1.ClusterObjectSetReasonAllObjectsReady, 5*time.Second) + }) } diff --git a/internal/object-controller/controllers/clusterobjectset_controller.go b/internal/object-controller/controllers/clusterobjectset_controller.go index 31308a5cab..7ef3726078 100644 --- a/internal/object-controller/controllers/clusterobjectset_controller.go +++ b/internal/object-controller/controllers/clusterobjectset_controller.go @@ -3,21 +3,17 @@ package controllers import ( - "bytes" - "compress/gzip" "context" "crypto/sha256" "encoding/json" "errors" "fmt" - "io" "slices" "strings" "time" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/equality" - apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -49,11 +45,10 @@ const ( clusterObjectSetTeardownFinalizer = "olm.operatorframework.io/teardown" // validationRetryInterval is how often the reconciler re-checks a revision that is - // Ready=False/PreconditionsNotMet. These failures stem from inputs that are not watched - // — a mutable or changed referenced Secret, an invalid referenced manifest, or a - // preflight dry-run rejected until RBAC or a dependency CRD propagates — and the ref - // spec is immutable, so a periodic retry lets the revision recover (once the Secret is - // replaced or the cluster state catches up) without a controller restart. + // Ready=False/PreconditionsNotMet. These failures can clear without a spec change, + // such as when a referenced Secret is replaced or RBAC and dependency CRDs propagate. + // Secret events normally trigger recovery immediately; the timer also covers + // dependencies that are not watched. validationRetryInterval = 10 * time.Second ) @@ -152,28 +147,28 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl remaining, hasDeadline := durationUntilDeadline(c.Clock, cos) isDeadlineExceeded := hasDeadline && remaining <= 0 - if err := c.verifyReferencedSecretsImmutable(ctx, cos); err != nil { + secretReader := newReferencedSecretReader(c.Client) + if err := secretReader.verifyImmutable(ctx, cos); err != nil { var mutErr *mutableSecretError if errors.As(err, &mutErr) { - // A confirmed mutable Secret requires user action (replacing the immutable Secret), - // which is not watched, so keep retrying so the revision can recover. + // A confirmed mutable Secret requires user action. Keep the periodic retry + // alongside the source watch so the revision can recover. // PreconditionsNotMet takes precedence over ProgressDeadlineExceeded: it is more actionable. - l.Error(err, "referenced Secret is mutable, marking revision not ready") + l.V(1).Info("referenced Secret is mutable, marking revision not ready", "error", err.Error()) markAsNotReady(cos, ocv1.ClusterObjectSetReasonPreconditionsNotMet, err.Error()) return ctrl.Result{RequeueAfter: validationRetryInterval}, nil } // A Secret read error is transient; leave the Ready condition unchanged and // return the error for rate-limited backoff. - return ctrl.Result{}, fmt.Errorf("verifying referenced Secrets: %v", err) + return ctrl.Result{}, fmt.Errorf("verifying referenced Secrets: %w", err) } - phases, currentPhases, opts, err := c.buildBoxcutterPhases(ctx, cos) + phases, currentPhases, opts, err := c.buildBoxcutterPhases(ctx, cos, secretReader) if err != nil { var contentErr *referencedContentError if errors.As(err, &contentErr) { - // Invalid referenced Secret content (missing key or malformed manifest). The Secret - // is not watched and the ref is immutable, so keep retrying so the revision can - // recover once the Secret is replaced with valid content. + // Invalid referenced Secret content can recover when the source is replaced. + // Keep the periodic retry alongside the source watch. l.Error(err, "invalid referenced content, marking revision not ready") markAsNotReady(cos, ocv1.ClusterObjectSetReasonPreconditionsNotMet, fmt.Sprintf("Referenced content is invalid: %v. Fix the Secret content.", err)) return ctrl.Result{RequeueAfter: validationRetryInterval}, nil @@ -196,15 +191,15 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl return ctrl.Result{}, fmt.Errorf("converting to boxcutter revision: %v", err) } - if len(cos.Status.ObservedPhases) == 0 { - cos.Status.ObservedPhases = currentPhases - } else if err := verifyObservedPhases(cos.Status.ObservedPhases, currentPhases); err != nil { - // The resolved content of a referenced Secret changed since first reconciliation. - // Secrets are not watched, so keep retrying so the revision can recover once the - // original content is restored. - l.Error(err, "resolved phases content changed, marking revision not ready") - markAsNotReady(cos, ocv1.ClusterObjectSetReasonPreconditionsNotMet, fmt.Sprintf("Referenced Secret content no longer matches the original rollout: %v. Restore the original content.", err)) - return ctrl.Result{RequeueAfter: validationRetryInterval}, nil + if len(cos.Status.ObservedPhases) > 0 { + if err := verifyObservedPhases(cos.Status.ObservedPhases, currentPhases); err != nil { + // The resolved content of a referenced Secret changed since first reconciliation. + // Keep retrying alongside the source watch so the revision can recover once the + // original content is restored. + l.Error(err, "resolved phases content changed, marking revision not ready") + markAsNotReady(cos, ocv1.ClusterObjectSetReasonPreconditionsNotMet, fmt.Sprintf("Referenced Secret content no longer matches the original rollout: %v. Restore the original content.", err)) + return ctrl.Result{RequeueAfter: validationRetryInterval}, nil + } } revisionEngine, err := c.RevisionEngineFactory.CreateRevisionEngine(ctx, cos) @@ -230,6 +225,9 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl if err := c.ensureFinalizer(ctx, cos, clusterObjectSetTeardownFinalizer); err != nil { return ctrl.Result{}, fmt.Errorf("error ensuring teardown finalizer: %v", err) } + if len(cos.Status.ObservedPhases) == 0 { + cos.Status.ObservedPhases = currentPhases + } if err := c.establishWatch(ctx, cos, revision); err != nil { return ctrl.Result{}, fmt.Errorf("establish watch: %v", err) @@ -467,9 +465,28 @@ func (c *ClusterObjectSetReconciler) SetupWithManager(mgr ctrl.Manager) error { predicate.ResourceVersionChangedPredicate{}, ), ). + // Source Secrets may be shared or owned by another controller. + WatchesMetadata(&corev1.Secret{}, + handler.EnqueueRequestsFromMapFunc(c.mapSecretToClusterObjectSets), + builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}), + ). Complete(c) } +func (c *ClusterObjectSetReconciler) mapSecretToClusterObjectSets(ctx context.Context, secret client.Object) []ctrl.Request { + key := client.ObjectKeyFromObject(secret) + revisions := &ocv1.ClusterObjectSetList{} + if err := c.Client.List(ctx, revisions, client.MatchingFields{clusterObjectSetSourceRefIndex: key.String()}); err != nil { + log.FromContext(ctx).Error(err, "listing ClusterObjectSets referencing Secret", "secret", key) + return nil + } + requests := make([]ctrl.Request, 0, len(revisions.Items)) + for i := range revisions.Items { + requests = append(requests, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(&revisions.Items[i])}) + } + return requests +} + func (c *ClusterObjectSetReconciler) establishWatch(ctx context.Context, cos *ocv1.ClusterObjectSet, revision boxcutter.RevisionBuilder) error { gvks := sets.New[schema.GroupVersionKind]() for _, phase := range revision.GetPhases() { @@ -578,14 +595,7 @@ func (c *ClusterObjectSetReconciler) listOtherActiveRevisions( // without a periodic retry: a spec edit re-triggers reconciliation. type configValidationError struct{ error } -// referencedContentError marks a validation failure caused by the content of a referenced -// Secret (a missing key or a malformed manifest). Referenced Secrets are not watched and the -// phase/ref spec is immutable, so the failure can only clear when the Secret is replaced. -// The reconciler reports it as Ready=False/PreconditionsNotMet but keeps retrying periodically -// so the revision can recover without a controller restart. -type referencedContentError struct{ error } - -func (c *ClusterObjectSetReconciler) buildBoxcutterPhases(ctx context.Context, cos *ocv1.ClusterObjectSet) ([]boxcutter.Phase, []ocv1.ObservedPhase, []boxcutter.RevisionReconcileOption, error) { +func (c *ClusterObjectSetReconciler) buildBoxcutterPhases(ctx context.Context, cos *ocv1.ClusterObjectSet, secretReader *referencedSecretReader) ([]boxcutter.Phase, []ocv1.ObservedPhase, []boxcutter.RevisionReconcileOption, error) { siblings, err := c.listSiblingRevisions(ctx, cos) if err != nil { return nil, nil, nil, fmt.Errorf("listing sibling revisions: %w", err) @@ -618,7 +628,7 @@ func (c *ClusterObjectSetReconciler) buildBoxcutterPhases(ctx context.Context, c case specObj.Object.Object != nil: obj = specObj.Object.DeepCopy() case specObj.Ref.Name != "": - resolved, err := c.resolveObjectRef(ctx, specObj.Ref) + resolved, err := secretReader.resolveObjectRef(ctx, specObj.Ref) if err != nil { return nil, nil, nil, fmt.Errorf("resolving ref in phase %q: %w", specPhase.Name, err) } @@ -659,47 +669,6 @@ func (c *ClusterObjectSetReconciler) buildBoxcutterPhases(ctx context.Context, c return phases, observedPhases, opts, nil } -// resolveObjectRef fetches the referenced Secret, reads the value at the specified key, -// auto-detects gzip compression, and deserializes into an unstructured.Unstructured. -func (c *ClusterObjectSetReconciler) resolveObjectRef(ctx context.Context, ref ocv1.ObjectSourceRef) (*unstructured.Unstructured, error) { - secret := &corev1.Secret{} - key := client.ObjectKey{Name: ref.Name, Namespace: ref.Namespace} - if err := c.Client.Get(ctx, key, secret); err != nil { - return nil, fmt.Errorf("getting Secret %s/%s: %w", ref.Namespace, ref.Name, err) - } - - data, ok := secret.Data[ref.Key] - if !ok { - return nil, &referencedContentError{fmt.Errorf("key %q not found in Secret %s/%s", ref.Key, ref.Namespace, ref.Name)} - } - - // Auto-detect gzip compression (magic bytes 0x1f 0x8b) - if len(data) >= 2 && data[0] == 0x1f && data[1] == 0x8b { - reader, err := gzip.NewReader(bytes.NewReader(data)) - if err != nil { - return nil, &referencedContentError{fmt.Errorf("creating gzip reader for key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)} - } - defer reader.Close() - const maxDecompressedSize = 10 * 1024 * 1024 // 10 MiB - limited := io.LimitReader(reader, maxDecompressedSize+1) - decompressed, err := io.ReadAll(limited) - if err != nil { - return nil, &referencedContentError{fmt.Errorf("decompressing key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)} - } - if len(decompressed) > maxDecompressedSize { - return nil, &referencedContentError{fmt.Errorf("decompressed data for key %q in Secret %s/%s exceeds maximum size (%d bytes)", ref.Key, ref.Namespace, ref.Name, maxDecompressedSize)} - } - data = decompressed - } - - obj := &unstructured.Unstructured{} - if err := json.Unmarshal(data, &obj.Object); err != nil { - return nil, &referencedContentError{fmt.Errorf("unmarshaling object from key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)} - } - - return obj, nil -} - // EffectiveCollisionProtection resolves the collision protection value using // the inheritance hierarchy: object > phase > spec > default ("Prevent"). func EffectiveCollisionProtection(cp ...ocv1.CollisionProtection) ocv1.CollisionProtection { @@ -874,76 +843,3 @@ func verifyObservedPhases(stored, current []ocv1.ObservedPhase) error { } return nil } - -// mutableSecretError is returned by verifyReferencedSecretsImmutable when one or -// more referenced Secrets are confirmed to be mutable. It distinguishes a confirmed -// precondition failure (reported as PreconditionsNotMet and retried on a timer) from -// a transient Secret read error (surfaced as a plain error for backoff requeue). -type mutableSecretError struct { - msg string -} - -func (e *mutableSecretError) Error() string { return e.msg } - -// verifyReferencedSecretsImmutable checks that all referenced Secrets -// have Immutable set to true. A confirmed mutable Secret is returned as a -// *mutableSecretError listing every misconfigured Secret; a failure to read a -// Secret is returned as a plain (retryable) error. -func (c *ClusterObjectSetReconciler) verifyReferencedSecretsImmutable(ctx context.Context, cos *ocv1.ClusterObjectSet) error { - type secretRef struct { - name string - namespace string - } - seen := make(map[secretRef]struct{}) - var refs []secretRef - - for _, phase := range cos.Spec.Phases { - for _, obj := range phase.Objects { - if obj.Ref.Name == "" { - continue - } - sr := secretRef{name: obj.Ref.Name, namespace: obj.Ref.Namespace} - if _, ok := seen[sr]; !ok { - seen[sr] = struct{}{} - refs = append(refs, sr) - } - } - } - - var mutableSecrets []string - var readErr error - for _, ref := range refs { - secret := &corev1.Secret{} - key := client.ObjectKey{Name: ref.name, Namespace: ref.namespace} - if err := c.Client.Get(ctx, key, secret); err != nil { - if apierrors.IsNotFound(err) { - // Secret not yet available — skip verification. - // resolveObjectRef will handle the not-found with a retryable error. - continue - } - // Remember the first read error but keep checking the remaining refs: a - // confirmed mutable Secret is a more actionable failure and takes precedence - // over a transient read error. - if readErr == nil { - readErr = fmt.Errorf("getting Secret %s/%s: %w", ref.namespace, ref.name, err) - } - continue - } - - if secret.Immutable == nil || !*secret.Immutable { - mutableSecrets = append(mutableSecrets, fmt.Sprintf("%s/%s", ref.namespace, ref.name)) - } - } - - if len(mutableSecrets) > 0 { - return &mutableSecretError{msg: fmt.Sprintf( - "Referenced Secret(s) must be immutable but are not: %s. Recreate them with immutable set to true.", - strings.Join(mutableSecrets, ", "))} - } - - if readErr != nil { - return readErr - } - - return nil -} diff --git a/internal/object-controller/controllers/clusterobjectset_controller_internal_test.go b/internal/object-controller/controllers/clusterobjectset_controller_internal_test.go index 61e2d097c7..d700ba4757 100644 --- a/internal/object-controller/controllers/clusterobjectset_controller_internal_test.go +++ b/internal/object-controller/controllers/clusterobjectset_controller_internal_test.go @@ -17,6 +17,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "k8s.io/utils/ptr" + ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/client/interceptor" @@ -460,8 +461,6 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { WithObjects(immutableSecret.DeepCopy()). Build() - reconciler := &ClusterObjectSetReconciler{Client: testClient} - cos := &ocv1.ClusterObjectSet{ Spec: ocv1.ClusterObjectSetSpec{ Phases: []ocv1.ClusterObjectSetPhase{{ @@ -477,7 +476,7 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { }, } - err := reconciler.verifyReferencedSecretsImmutable(t.Context(), cos) + err := newReferencedSecretReader(testClient).verifyImmutable(t.Context(), cos) require.NoError(t, err) }) @@ -490,8 +489,6 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { WithObjects(mutableSecret). Build() - reconciler := &ClusterObjectSetReconciler{Client: testClient} - cos := &ocv1.ClusterObjectSet{ Spec: ocv1.ClusterObjectSetSpec{ Phases: []ocv1.ClusterObjectSetPhase{{ @@ -507,7 +504,7 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { }, } - err := reconciler.verifyReferencedSecretsImmutable(t.Context(), cos) + err := newReferencedSecretReader(testClient).verifyImmutable(t.Context(), cos) require.Error(t, err) assert.Contains(t, err.Error(), "must be immutable but are not") }) @@ -517,8 +514,6 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { WithScheme(testScheme). Build() - reconciler := &ClusterObjectSetReconciler{Client: testClient} - cos := &ocv1.ClusterObjectSet{ Spec: ocv1.ClusterObjectSetSpec{ Phases: []ocv1.ClusterObjectSetPhase{{ @@ -530,7 +525,7 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { }, } - err := reconciler.verifyReferencedSecretsImmutable(t.Context(), cos) + err := newReferencedSecretReader(testClient).verifyImmutable(t.Context(), cos) require.NoError(t, err) }) @@ -550,8 +545,6 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { }). Build() - reconciler := &ClusterObjectSetReconciler{Client: testClient} - cos := &ocv1.ClusterObjectSet{ Spec: ocv1.ClusterObjectSetSpec{ Phases: []ocv1.ClusterObjectSetPhase{{ @@ -564,8 +557,73 @@ func TestVerifyReferencedSecretsImmutable(t *testing.T) { }, } - err := reconciler.verifyReferencedSecretsImmutable(t.Context(), cos) + err := newReferencedSecretReader(testClient).verifyImmutable(t.Context(), cos) require.NoError(t, err) assert.Equal(t, int32(1), secretGetCount.Load(), "secret should be fetched only once despite multiple references") }) } + +func TestMapSecretToClusterObjectSets(t *testing.T) { + testScheme := runtime.NewScheme() + require.NoError(t, ocv1.AddToScheme(testScheme)) + newRevision := func(name string, refs ...ocv1.ObjectSourceRef) *ocv1.ClusterObjectSet { + cos := &ocv1.ClusterObjectSet{ObjectMeta: metav1.ObjectMeta{Name: name}} + phase := ocv1.ClusterObjectSetPhase{Name: "deploy"} + for _, ref := range refs { + phase.Objects = append(phase.Objects, ocv1.ClusterObjectSetObject{Ref: ref}) + } + cos.Spec.Phases = []ocv1.ClusterObjectSetPhase{phase} + return cos + } + ref := ocv1.ObjectSourceRef{Namespace: "storage-a", Name: "content", Key: "first"} + otherKey := ref + otherKey.Key = "second" + otherNamespace := ref + otherNamespace.Namespace = "storage-b" + otherSecret := ref + otherSecret.Name = "other" + testClient := testutil.WithIndexes(t, fake.NewClientBuilder().WithScheme(testScheme), SetupIndexes). + WithObjects( + newRevision("first", ref, ref), + newRevision("second", otherKey), + newRevision("other-namespace", otherNamespace), + newRevision("other-secret", otherSecret), + newRevision("inline"), + ).Build() + reconciler := &ClusterObjectSetReconciler{Client: testClient} + for _, tc := range []struct { + name string + key client.ObjectKey + want []ctrl.Request + }{ + { + name: "all referencing revisions, once each and regardless of data key", + key: client.ObjectKey{Namespace: "storage-a", Name: "content"}, + want: []ctrl.Request{ + {NamespacedName: client.ObjectKey{Name: "first"}}, + {NamespacedName: client.ObjectKey{Name: "second"}}, + }, + }, + { + name: "same Secret name in another namespace", + key: client.ObjectKey{Namespace: "storage-b", Name: "content"}, + want: []ctrl.Request{{NamespacedName: client.ObjectKey{Name: "other-namespace"}}}, + }, + { + name: "another Secret in the same namespace", + key: client.ObjectKey{Namespace: "storage-a", Name: "other"}, + want: []ctrl.Request{{NamespacedName: client.ObjectKey{Name: "other-secret"}}}, + }, + { + name: "unreferenced Secret", + key: client.ObjectKey{Namespace: "storage-a", Name: "unreferenced"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + metadata := &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{ + Namespace: tc.key.Namespace, Name: tc.key.Name, + }} + assert.ElementsMatch(t, tc.want, reconciler.mapSecretToClusterObjectSets(t.Context(), metadata)) + }) + } +} diff --git a/internal/object-controller/controllers/clusterobjectset_controller_test.go b/internal/object-controller/controllers/clusterobjectset_controller_test.go index 51f3c8502d..2bc68faa4d 100644 --- a/internal/object-controller/controllers/clusterobjectset_controller_test.go +++ b/internal/object-controller/controllers/clusterobjectset_controller_test.go @@ -938,7 +938,7 @@ func Test_ClusterObjectSetReconciler_Reconcile_VerifyReferencedSecrets(t *testin ext := newTestClusterExtension() rev1 := newTestClusterObjectSet(t, clusterObjectSetName, ext, testScheme) - // Reference two Secrets so verifyReferencedSecretsImmutable reads both. + // Reference two Secrets so immutable verification reads both. rev1.Spec.Phases = []ocv1.ClusterObjectSetPhase{{ Name: "everything", Objects: []ocv1.ClusterObjectSetObject{ diff --git a/internal/object-controller/controllers/indexes.go b/internal/object-controller/controllers/indexes.go index 45c97a7161..8605071460 100644 --- a/internal/object-controller/controllers/indexes.go +++ b/internal/object-controller/controllers/indexes.go @@ -4,7 +4,9 @@ package controllers import ( "context" + "fmt" + "k8s.io/apimachinery/pkg/util/sets" "sigs.k8s.io/controller-runtime/pkg/client" ocv1 "github.com/operator-framework/operator-controller/api/v1" @@ -15,16 +17,38 @@ import ( // in a group. const clusterObjectSetGroupIndex = ".spec.group" +// clusterObjectSetSourceRefIndex maps a source Secret's namespace/name to all +// revisions that reference it, regardless of the Secret's owner references. +const clusterObjectSetSourceRefIndex = ".spec.phases.objects.ref" + // SetupIndexes registers the cache indexes used to look up ClusterObjectSet -// revisions by group. Both the ClusterObjectSet and ClusterObjectDeployment -// controllers rely on the spec.group index, so it is registered once per manager -// rather than per controller. +// revisions by group and referenced Secret. The group index is shared with the +// ClusterObjectDeployment controller, so indexes are registered once per manager. func SetupIndexes(ctx context.Context, indexer client.FieldIndexer) error { - return indexer.IndexField(ctx, &ocv1.ClusterObjectSet{}, clusterObjectSetGroupIndex, func(obj client.Object) []string { + if err := indexer.IndexField(ctx, &ocv1.ClusterObjectSet{}, clusterObjectSetGroupIndex, func(obj client.Object) []string { cos, ok := obj.(*ocv1.ClusterObjectSet) if !ok || cos.Spec.Group == "" { return nil } return []string{cos.Spec.Group} + }); err != nil { + return fmt.Errorf("indexing ClusterObjectSet groups: %w", err) + } + return indexer.IndexField(ctx, &ocv1.ClusterObjectSet{}, clusterObjectSetSourceRefIndex, func(obj client.Object) []string { + cos, ok := obj.(*ocv1.ClusterObjectSet) + if !ok { + return nil + } + refs := sets.New[string]() + for _, phase := range cos.Spec.Phases { + for _, obj := range phase.Objects { + if obj.Ref.Name == "" { + continue + } + key := client.ObjectKey{Namespace: obj.Ref.Namespace, Name: obj.Ref.Name} + refs.Insert(key.String()) + } + } + return sets.List(refs) }) } diff --git a/internal/object-controller/controllers/referenced_secrets.go b/internal/object-controller/controllers/referenced_secrets.go new file mode 100644 index 0000000000..753eeaea18 --- /dev/null +++ b/internal/object-controller/controllers/referenced_secrets.go @@ -0,0 +1,138 @@ +//go:build !standard + +package controllers + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "fmt" + "io" + "strings" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/util/sets" + "sigs.k8s.io/controller-runtime/pkg/client" + + ocv1 "github.com/operator-framework/operator-controller/api/v1" +) + +// referencedSecretReader shares a Secret snapshot between verification and object +// decoding within one reconciliation. A new reader must be used on the next +// reconciliation so that deleted and recreated Secrets are read again. +type referencedSecretReader struct { + reader client.Reader + secrets map[client.ObjectKey]*corev1.Secret +} + +func newReferencedSecretReader(reader client.Reader) *referencedSecretReader { + return &referencedSecretReader{reader: reader, secrets: make(map[client.ObjectKey]*corev1.Secret)} +} + +func (r *referencedSecretReader) get(ctx context.Context, key client.ObjectKey) (*corev1.Secret, error) { + if secret, ok := r.secrets[key]; ok { + return secret, nil + } + secret := &corev1.Secret{} + if err := r.reader.Get(ctx, key, secret); err != nil { + return nil, err + } + r.secrets[key] = secret + return secret, nil +} + +// mutableSecretError distinguishes a confirmed mutable Secret from a transient +// read error, so reconciliation can report an actionable precondition failure. +type mutableSecretError struct { + msg string +} + +func (e *mutableSecretError) Error() string { return e.msg } + +// verifyImmutable checks every distinct referenced Secret. Confirmed mutable +// Secrets take precedence over read errors; decoding must use this same reader. +func (r *referencedSecretReader) verifyImmutable(ctx context.Context, cos *ocv1.ClusterObjectSet) error { + seen := sets.New[client.ObjectKey]() + var mutableSecrets []string + var readErr error + + for _, phase := range cos.Spec.Phases { + for _, obj := range phase.Objects { + if obj.Ref.Name == "" { + continue + } + key := client.ObjectKey{Name: obj.Ref.Name, Namespace: obj.Ref.Namespace} + if seen.Has(key) { + continue + } + seen.Insert(key) + secret, err := r.get(ctx, key) + if err != nil { + // Keep checking so a confirmed mutable Secret takes precedence over a read error. + if readErr == nil { + readErr = fmt.Errorf("getting Secret %s/%s: %w", key.Namespace, key.Name, err) + } + continue + } + + if secret.Immutable == nil || !*secret.Immutable { + mutableSecrets = append(mutableSecrets, fmt.Sprintf("%s/%s", key.Namespace, key.Name)) + } + } + } + + if len(mutableSecrets) > 0 { + return &mutableSecretError{msg: fmt.Sprintf( + "Referenced Secret(s) must be immutable but are not: %s. Recreate them with immutable set to true.", + strings.Join(mutableSecrets, ", "))} + } + + return readErr +} + +// referencedContentError marks invalid Secret content, such as a missing key or +// malformed manifest. Replacing the source can clear it without changing the ref. +type referencedContentError struct{ error } + +// resolveObjectRef decodes the verified Secret snapshot, auto-detecting gzip +// compression, into an unstructured Kubernetes object. +func (r *referencedSecretReader) resolveObjectRef(ctx context.Context, ref ocv1.ObjectSourceRef) (*unstructured.Unstructured, error) { + key := client.ObjectKey{Name: ref.Name, Namespace: ref.Namespace} + secret, err := r.get(ctx, key) + if err != nil { + return nil, fmt.Errorf("getting Secret %s/%s: %w", ref.Namespace, ref.Name, err) + } + + data, ok := secret.Data[ref.Key] + if !ok { + return nil, &referencedContentError{fmt.Errorf("key %q not found in Secret %s/%s", ref.Key, ref.Namespace, ref.Name)} + } + + // Auto-detect gzip compression (magic bytes 0x1f 0x8b). + if len(data) >= 2 && data[0] == 0x1f && data[1] == 0x8b { + reader, err := gzip.NewReader(bytes.NewReader(data)) + if err != nil { + return nil, &referencedContentError{fmt.Errorf("creating gzip reader for key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)} + } + defer reader.Close() + const maxDecompressedSize = 10 * 1024 * 1024 // 10 MiB + limited := io.LimitReader(reader, maxDecompressedSize+1) + decompressed, err := io.ReadAll(limited) + if err != nil { + return nil, &referencedContentError{fmt.Errorf("decompressing key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)} + } + if len(decompressed) > maxDecompressedSize { + return nil, &referencedContentError{fmt.Errorf("decompressed data for key %q in Secret %s/%s exceeds maximum size (%d bytes)", ref.Key, ref.Namespace, ref.Name, maxDecompressedSize)} + } + data = decompressed + } + + obj := &unstructured.Unstructured{} + if err := json.Unmarshal(data, &obj.Object); err != nil { + return nil, &referencedContentError{fmt.Errorf("unmarshaling object from key %q in Secret %s/%s: %w", ref.Key, ref.Namespace, ref.Name, err)} + } + + return obj, nil +} diff --git a/internal/object-controller/controllers/referenced_secrets_test.go b/internal/object-controller/controllers/referenced_secrets_test.go new file mode 100644 index 0000000000..d2ee584285 --- /dev/null +++ b/internal/object-controller/controllers/referenced_secrets_test.go @@ -0,0 +1,281 @@ +package controllers_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + clocktesting "k8s.io/utils/clock/testing" + "k8s.io/utils/ptr" + "pkg.package-operator.run/boxcutter/machinery" + machinerytypes "pkg.package-operator.run/boxcutter/machinery/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + ocv1 "github.com/operator-framework/operator-controller/api/v1" + "github.com/operator-framework/operator-controller/internal/object-controller/controllers" + "github.com/operator-framework/operator-controller/internal/testutil" +) + +func TestReferencedSecretsReadOncePerReconcile(t *testing.T) { + ctx := t.Context() + mockCtrl := gomock.NewController(t) + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "content", Namespace: "source-a"}, + Immutable: ptr.To(true), Data: map[string][]byte{}, + } + cos := newRefTestCOS("packed", ocv1.ObjectSourceRef{}) + cos.Spec.Phases = nil + for phaseIndex := range 2 { + phase := ocv1.ClusterObjectSetPhase{Name: fmt.Sprintf("phase-%d", phaseIndex)} + for objectIndex := range 25 { + name := fmt.Sprintf("cm-%d-%d", phaseIndex, objectIndex) + secret.Data[name] = []byte(fmt.Sprintf(`{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":%q,"namespace":"target"}}`, name)) + phase.Objects = append(phase.Objects, ocv1.ClusterObjectSetObject{Ref: ocv1.ObjectSourceRef{Name: secret.Name, Namespace: secret.Namespace, Key: name}}) + } + cos.Spec.Phases = append(cos.Spec.Phases, phase) + } + otherSecret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: secret.Name, Namespace: "source-b"}, + Immutable: ptr.To(true), + Data: map[string][]byte{"other": []byte(`{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"other","namespace":"target"}}`)}, + } + cos.Spec.Phases[1].Objects = append(cos.Spec.Phases[1].Objects, ocv1.ClusterObjectSetObject{Ref: ocv1.ObjectSourceRef{Name: otherSecret.Name, Namespace: otherSecret.Namespace, Key: "other"}}) + reads := map[client.ObjectKey]int{} + cl := testutil.WithIndexes(t, fake.NewClientBuilder().WithScheme(newSchemeWithCoreV1(t)), controllers.SetupIndexes). + WithObjects(secret, otherSecret, cos). + WithStatusSubresource(&ocv1.ClusterObjectSet{}). + WithInterceptorFuncs(interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*corev1.Secret); ok { + reads[key]++ + } + return cl.Get(ctx, key, obj, opts...) + }, + }).Build() + engine := newMockRevisionEngineWithReconcile(mockCtrl, + func(context.Context, machinerytypes.Revision, ...machinerytypes.RevisionReconcileOption) (machinery.RevisionResult, error) { + return newMockRevisionResult(mockCtrl, revisionResultConfig{isComplete: true}), nil + }, nil) + reconciler := &controllers.ClusterObjectSetReconciler{ + Client: cl, TrackingCache: newMockTrackingCache(mockCtrl, cl, nil), + RevisionEngineFactory: newMockRevisionEngineFactoryWithEngine(mockCtrl, engine, nil), + Clock: clocktesting.NewFakeClock(metav1.Now().Time), + } + reconcile := func(wantReads int, wantReason string) { + t.Helper() + _, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(cos)}) + require.NoError(t, err) + require.Equal(t, wantReads, reads[client.ObjectKeyFromObject(secret)]) + require.Equal(t, wantReads, reads[client.ObjectKeyFromObject(otherSecret)]) + require.NoError(t, cl.Get(ctx, client.ObjectKeyFromObject(cos), cos)) + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + require.NotNil(t, condition) + require.Equal(t, wantReason, condition.Reason) + } + reconcile(1, ocv1.ClusterObjectSetReasonAllObjectsReady) + + // Replace immediately after the first successful reconciliation: the original + // digest must already be recorded, without relying on another reconciliation. + require.NoError(t, cl.Delete(ctx, secret)) + changed := secret.DeepCopy() + changed.ResourceVersion = "" + changed.Data["cm-0-0"] = []byte(`{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"cm-0-0","namespace":"target"},"data":{"value":"changed"}}`) + require.NoError(t, cl.Create(ctx, changed)) + reconcile(2, ocv1.ClusterObjectSetReasonPreconditionsNotMet) + require.Contains(t, meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady).Message, "resolved content of 1 phase(s) has changed") + + require.NoError(t, cl.Delete(ctx, changed)) + secret.ResourceVersion = "" + require.NoError(t, cl.Create(ctx, secret)) + reconcile(3, ocv1.ClusterObjectSetReasonAllObjectsReady) + reconcile(4, ocv1.ClusterObjectSetReasonAllObjectsReady) +} + +func TestReferencedSecretReadFailureRetries(t *testing.T) { + ctx := t.Context() + mockCtrl := gomock.NewController(t) + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "content", Namespace: "source"}, + Immutable: ptr.To(true), + Data: map[string][]byte{"object": []byte(`{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"cm","namespace":"target"}}`)}, + } + cos := newRefTestCOS("retry", ocv1.ObjectSourceRef{Name: secret.Name, Namespace: secret.Namespace, Key: "object"}) + failReads := true + cl := testutil.WithIndexes(t, fake.NewClientBuilder().WithScheme(newSchemeWithCoreV1(t)), controllers.SetupIndexes). + WithObjects(secret, cos). + WithStatusSubresource(&ocv1.ClusterObjectSet{}). + WithInterceptorFuncs(interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*corev1.Secret); ok && failReads { + return apierrors.NewServiceUnavailable("temporary API outage") + } + return cl.Get(ctx, key, obj, opts...) + }, + }).Build() + engine := newMockRevisionEngineWithReconcile(mockCtrl, + func(context.Context, machinerytypes.Revision, ...machinerytypes.RevisionReconcileOption) (machinery.RevisionResult, error) { + return newMockRevisionResult(mockCtrl, revisionResultConfig{inTransition: true}), nil + }, nil) + reconciler := &controllers.ClusterObjectSetReconciler{ + Client: cl, TrackingCache: newMockTrackingCache(mockCtrl, cl, nil), + RevisionEngineFactory: newMockRevisionEngineFactoryWithEngine(mockCtrl, engine, nil), + Clock: clocktesting.NewFakeClock(metav1.Now().Time), + } + req := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(cos)} + for range 2 { + _, err := reconciler.Reconcile(ctx, req) + require.ErrorContains(t, err, "temporary API outage") + require.NoError(t, cl.Get(ctx, req.NamespacedName, cos)) + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + require.NotNil(t, condition) + require.Equal(t, metav1.ConditionUnknown, condition.Status) + require.Equal(t, ocv1.ClusterObjectSetReasonPending, condition.Reason) + } + failReads = false + _, err := reconciler.Reconcile(ctx, req) + require.NoError(t, err) + require.NoError(t, cl.Get(ctx, req.NamespacedName, cos)) + require.Equal(t, ocv1.ClusterObjectSetReasonObjectsNotReady, meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady).Reason) +} + +func TestReferencedSecretSnapshotSurvivesReplacementBeforeDecoding(t *testing.T) { + mockCtrl := gomock.NewController(t) + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "content", Namespace: "source"}, + Immutable: ptr.To(true), + Data: map[string][]byte{"object": []byte(`{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"verified","namespace":"target"}}`)}, + } + cos := newRefTestCOS("snapshot", ocv1.ObjectSourceRef{ + Name: secret.Name, Namespace: secret.Namespace, Key: "object", + }) + reads := 0 + cl := testutil.WithIndexes(t, fake.NewClientBuilder().WithScheme(newSchemeWithCoreV1(t)), controllers.SetupIndexes). + WithObjects(secret, cos). + WithStatusSubresource(&ocv1.ClusterObjectSet{}). + WithInterceptorFuncs(interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if err := cl.Get(ctx, key, obj, opts...); err != nil { + return err + } + if _, ok := obj.(*corev1.Secret); ok { + reads++ + if reads == 1 { + // Replace the source after returning its verified content, + // before decoding gets a chance to read it again. + replacement := secret.DeepCopy() + replacement.ResourceVersion = "" + replacement.Immutable = ptr.To(false) + replacement.Data["object"] = []byte(`{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"unverified","namespace":"target"}}`) + require.NoError(t, cl.Delete(ctx, secret)) + require.NoError(t, cl.Create(ctx, replacement)) + } + } + return nil + }, + }).Build() + engine := newMockRevisionEngineWithReconcile(mockCtrl, + func(_ context.Context, revision machinerytypes.Revision, _ ...machinerytypes.RevisionReconcileOption) (machinery.RevisionResult, error) { + require.Equal(t, "verified", revision.GetPhases()[0].GetObjects()[0].GetName(), + "the engine must receive the exact content verified as immutable") + return newMockRevisionResult(mockCtrl, revisionResultConfig{isComplete: true}), nil + }, nil) + reconciler := &controllers.ClusterObjectSetReconciler{ + Client: cl, TrackingCache: newMockTrackingCache(mockCtrl, cl, nil), + RevisionEngineFactory: newMockRevisionEngineFactoryWithEngine(mockCtrl, engine, nil), + } + req := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(cos)} + _, err := reconciler.Reconcile(t.Context(), req) + require.NoError(t, err) + require.Equal(t, 1, reads) + + // A fresh reconciliation must observe the replacement, rather than reuse the + // previous snapshot or apply its mutable payload. + _, err = reconciler.Reconcile(t.Context(), req) + require.NoError(t, err) + require.Equal(t, 2, reads) + require.NoError(t, cl.Get(t.Context(), req.NamespacedName, cos)) + require.Equal(t, ocv1.ClusterObjectSetReasonPreconditionsNotMet, + meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady).Reason) +} + +func TestMutableReferencedSecretRequeues(t *testing.T) { + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "content", Namespace: "source"}, + Immutable: ptr.To(false), + } + cos := newRefTestCOS("wait-for-immutable-secret", ocv1.ObjectSourceRef{ + Name: secret.Name, Namespace: secret.Namespace, Key: "object", + }) + cl := fake.NewClientBuilder().WithScheme(newSchemeWithCoreV1(t)). + WithStatusSubresource(&ocv1.ClusterObjectSet{}). + WithObjects(secret, cos).Build() + reconciler := &controllers.ClusterObjectSetReconciler{Client: cl} + + result, err := reconciler.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(cos)}) + require.NoError(t, err) + require.Equal(t, ctrl.Result{RequeueAfter: 10 * time.Second}, result) + require.NoError(t, cl.Get(t.Context(), client.ObjectKeyFromObject(cos), cos)) + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + require.NotNil(t, condition) + require.Equal(t, metav1.ConditionFalse, condition.Status) + require.Equal(t, ocv1.ClusterObjectSetReasonPreconditionsNotMet, condition.Reason) + require.Contains(t, condition.Message, "source/content") +} + +func TestMissingReferencedSecretRetriesBeforeDecoding(t *testing.T) { + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "content", Namespace: "source"}, + Immutable: ptr.To(false), + Data: map[string][]byte{"object": []byte(`{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"cm","namespace":"target"}}`)}, + } + cos := newRefTestCOS("missing-secret", ocv1.ObjectSourceRef{ + Name: secret.Name, Namespace: secret.Namespace, Key: "object", + }) + reads := 0 + cl := fake.NewClientBuilder().WithScheme(newSchemeWithCoreV1(t)).WithObjects(secret, cos). + WithStatusSubresource(&ocv1.ClusterObjectSet{}). + WithInterceptorFuncs(interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*corev1.Secret); ok { + reads++ + if reads == 1 { + // Model a Secret created just after verification's first read. + return apierrors.NewNotFound(schema.GroupResource{Resource: "secrets"}, key.Name) + } + } + return cl.Get(ctx, key, obj, opts...) + }, + }).Build() + reconciler := &controllers.ClusterObjectSetReconciler{Client: cl} + req := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(cos)} + + _, err := reconciler.Reconcile(t.Context(), req) + require.True(t, apierrors.IsNotFound(err), "expected a retryable NotFound error, got %v", err) + require.Equal(t, 1, reads, "must not reread and decode a Secret that was not verified") + require.NoError(t, cl.Get(t.Context(), req.NamespacedName, cos)) + condition := meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + require.NotNil(t, condition) + require.Equal(t, metav1.ConditionUnknown, condition.Status) + require.Equal(t, ocv1.ClusterObjectSetReasonPending, condition.Reason) + + result, err := reconciler.Reconcile(t.Context(), req) + require.NoError(t, err) + require.Equal(t, ctrl.Result{RequeueAfter: 10 * time.Second}, result) + require.Equal(t, 2, reads) + require.NoError(t, cl.Get(t.Context(), req.NamespacedName, cos)) + condition = meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeReady) + require.NotNil(t, condition) + require.Equal(t, ocv1.ClusterObjectSetReasonPreconditionsNotMet, condition.Reason) + require.Contains(t, condition.Message, "source/content") +} diff --git a/internal/object-controller/controllers/resolve_ref_test.go b/internal/object-controller/controllers/resolve_ref_test.go index ddaf2cc218..e7d9df6823 100644 --- a/internal/object-controller/controllers/resolve_ref_test.go +++ b/internal/object-controller/controllers/resolve_ref_test.go @@ -11,6 +11,7 @@ import ( "github.com/stretchr/testify/require" "go.uber.org/mock/gomock" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" apimachineryruntime "k8s.io/apimachinery/pkg/runtime" @@ -180,7 +181,8 @@ func TestResolveObjectRef_SecretNotFound(t *testing.T) { NamespacedName: types.NamespacedName{Name: cos.Name}, }) require.Error(t, err) - assert.Contains(t, err.Error(), "resolving ref") + assert.True(t, apierrors.IsNotFound(err)) + assert.Contains(t, err.Error(), "getting Secret olmv1-system/nonexistent-secret") } func TestResolveObjectRef_KeyNotFound(t *testing.T) {