NO-ISSUE: Synchronize From Upstream Repositories #610 ( With commit to fix the tests after Marketplace Removal )

cmd/operator-controller/main.go

@@ -629,7 +629,7 @@ func (c *boxcutterReconcilerConfigurator) Configure(ceReconciler *controllers.Cl
 		controllers.HandleFinalizers(c.finalizers),
 		controllers.MigrateStorage(storageMigrator),
 		controllers.RetrieveRevisionStates(revisionStatesGetter),
-		controllers.ResolveBundle(c.resolver),
+		controllers.ResolveBundle(c.resolver, c.mgr.GetClient()),
 		controllers.UnpackBundle(c.imagePuller, c.imageCache),
 		controllers.ApplyBundleWithBoxcutter(appl.Apply),
 	}
@@ -748,7 +748,7 @@ func (c *helmReconcilerConfigurator) Configure(ceReconciler *controllers.Cluster
 	ceReconciler.ReconcileSteps = []controllers.ReconcileStepFunc{
 		controllers.HandleFinalizers(c.finalizers),
 		controllers.RetrieveRevisionStates(revisionStatesGetter),
-		controllers.ResolveBundle(c.resolver),
+		controllers.ResolveBundle(c.resolver, c.mgr.GetClient()),
 		controllers.UnpackBundle(c.imagePuller, c.imageCache),
 		controllers.ApplyBundle(appl),
 	}

commitchecker.yaml

@@ -1,4 +1,4 @@
-expectedMergeBase: 8167ff8fd3ab3880459f9d1e5626c6e3073e3428
+expectedMergeBase: 6e4f192699f5c039fa2b92b01372a150274447bd
 upstreamBranch: main
 upstreamOrg: operator-framework
 upstreamRepo: operator-controller

internal/operator-controller/applier/boxcutter.go

@@ -312,21 +312,38 @@ func (bc *Boxcutter) createOrUpdate(ctx context.Context, user user.Info, rev *oc
 	return bc.Client.Patch(ctx, rev, client.Apply, client.FieldOwner(bc.FieldOwner), client.ForceOwnership)
 }
 
-func (bc *Boxcutter) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.ClusterExtension, objectLabels, revisionAnnotations map[string]string) error {
-	// Generate desired revision
-	desiredRevision, err := bc.RevisionGenerator.GenerateRevision(ctx, contentFS, ext, objectLabels, revisionAnnotations)
+func (bc *Boxcutter) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.ClusterExtension, objectLabels, revisionAnnotations map[string]string) (bool, string, error) {
+	// List existing revisions first to validate cluster connectivity before checking contentFS.
+	// This ensures we fail fast on API errors rather than attempting fallback behavior when
+	// cluster access is unavailable (since the ClusterExtensionRevision controller also requires
+	// API access to maintain resources). The revision list is also needed to determine if fallback
+	// is possible when contentFS is nil (at least one revision must exist).
+	existingRevisions, err := bc.getExistingRevisions(ctx, ext.GetName())
 	if err != nil {
-		return err
+		return false, "", err
 	}
 
-	if err := controllerutil.SetControllerReference(ext, desiredRevision, bc.Scheme); err != nil {
-		return fmt.Errorf("set ownerref: %w", err)
+	// If contentFS is nil, we're maintaining the current state without catalog access.
+	// In this case, we should use the existing installed revision without generating a new one.
+	if contentFS == nil {
+		if len(existingRevisions) == 0 {
+			return false, "", fmt.Errorf("catalog content unavailable and no revision installed")
+		}
+		// Returning true here signals that the rollout has succeeded using the current revision.
+		// This assumes the ClusterExtensionRevision controller is running and will continue to
+		// reconcile, apply, and maintain the resources defined in that revision via Server-Side Apply,
+		// ensuring the workload keeps running even when catalog access is unavailable.
+		return true, "", nil
 	}
 
-	// List all existing revisions
-	existingRevisions, err := bc.getExistingRevisions(ctx, ext.GetName())
+	// Generate desired revision
+	desiredRevision, err := bc.RevisionGenerator.GenerateRevision(ctx, contentFS, ext, objectLabels, revisionAnnotations)
 	if err != nil {
-		return err
+		return false, "", err
+	}
+
+	if err := controllerutil.SetControllerReference(ext, desiredRevision, bc.Scheme); err != nil {
+		return false, "", fmt.Errorf("set ownerref: %w", err)
 	}
 
 	currentRevision := &ocv1.ClusterExtensionRevision{}
@@ -348,7 +365,7 @@ func (bc *Boxcutter) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.Clust
 			// inplace patch was successful, no changes in phases
 			state = StateUnchanged
 		default:
-			return fmt.Errorf("patching %s Revision: %w", desiredRevision.Name, err)
+			return false, "", fmt.Errorf("patching %s Revision: %w", desiredRevision.Name, err)
 		}
 	}
 
@@ -362,7 +379,7 @@ func (bc *Boxcutter) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.Clust
 		case StateNeedsInstall:
 			err := preflight.Install(ctx, plainObjs)
 			if err != nil {
-				return err
+				return false, "", err
 			}
 		// TODO: jlanford's IDE says that "StateNeedsUpgrade" condition is always true, but
 		//   it isn't immediately obvious why that is. Perhaps len(existingRevisions) is
@@ -371,7 +388,7 @@ func (bc *Boxcutter) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.Clust
 		case StateNeedsUpgrade:
 			err := preflight.Upgrade(ctx, plainObjs)
 			if err != nil {
-				return err
+				return false, "", err
 			}
 		}
 	}
@@ -385,15 +402,15 @@ func (bc *Boxcutter) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.Clust
 		desiredRevision.Spec.Revision = revisionNumber
 
 		if err = bc.garbageCollectOldRevisions(ctx, prevRevisions); err != nil {
-			return fmt.Errorf("garbage collecting old revisions: %w", err)
+			return false, "", fmt.Errorf("garbage collecting old revisions: %w", err)
 		}
 
 		if err := bc.createOrUpdate(ctx, getUserInfo(ext), desiredRevision); err != nil {
-			return fmt.Errorf("creating new Revision: %w", err)
+			return false, "", fmt.Errorf("creating new Revision: %w", err)
 		}
 	}
 
-	return nil
+	return true, "", nil
 }
 
 // runPreAuthorizationChecks runs PreAuthorization checks if the PreAuthorizer is set. An error will be returned if

internal/operator-controller/applier/boxcutter_test.go

@@ -991,14 +991,18 @@ func TestBoxcutter_Apply(t *testing.T) {
 					labels.PackageNameKey:   "test-package",
 				}
 			}
-			err := boxcutter.Apply(t.Context(), testFS, ext, nil, revisionAnnotations)
+			completed, status, err := boxcutter.Apply(t.Context(), testFS, ext, nil, revisionAnnotations)
 
 			// Assert
 			if tc.expectedErr != "" {
 				require.Error(t, err)
 				assert.Contains(t, err.Error(), tc.expectedErr)
+				assert.False(t, completed)
+				assert.Empty(t, status)
 			} else {
 				require.NoError(t, err)
+				assert.True(t, completed)
+				assert.Empty(t, status)
 			}
 
 			if tc.validate != nil {
@@ -1190,10 +1194,12 @@ func Test_PreAuthorizer_Integration(t *testing.T) {
 				RevisionGenerator: dummyGenerator,
 				PreAuthorizer:     tc.preAuthorizer(t),
 			}
-			err := boxcutter.Apply(t.Context(), dummyBundleFs, ext, nil, revisionAnnotations)
+			completed, status, err := boxcutter.Apply(t.Context(), dummyBundleFs, ext, nil, revisionAnnotations)
 			if tc.validate != nil {
 				tc.validate(t, err)
 			}
+			_ = completed
+			_ = status
 		})
 	}
 }

internal/operator-controller/applier/helm.go

@@ -84,6 +84,16 @@ func (h *Helm) runPreAuthorizationChecks(ctx context.Context, ext *ocv1.ClusterE
 }
 
 func (h *Helm) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.ClusterExtension, objectLabels map[string]string, storageLabels map[string]string) (bool, string, error) {
+	// If contentFS is nil, we're maintaining the current state without catalog access.
+	// In this case, reconcile the existing Helm release if it exists.
+	if contentFS == nil {
+		ac, err := h.ActionClientGetter.ActionClientFor(ctx, ext)
+		if err != nil {
+			return false, "", err
+		}
+		return h.reconcileExistingRelease(ctx, ac, ext)
+	}
+
 	chrt, err := h.buildHelmChart(contentFS, ext)
 	if err != nil {
 		return false, "", err
@@ -178,6 +188,62 @@ func (h *Helm) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1.ClusterExte
 	return true, "", nil
 }
 
+// reconcileExistingRelease reconciles an existing Helm release without catalog access.
+// This is used when the catalog is unavailable but we need to maintain the current installation.
+// It reconciles the release to actively maintain resources, and sets up watchers for monitoring/observability.
+func (h *Helm) reconcileExistingRelease(ctx context.Context, ac helmclient.ActionInterface, ext *ocv1.ClusterExtension) (bool, string, error) {
+	rel, err := ac.Get(ext.GetName())
+	if errors.Is(err, driver.ErrReleaseNotFound) {
+		return false, "", fmt.Errorf("catalog content unavailable and no release installed")
+	}
+	if err != nil {
+		return false, "", fmt.Errorf("failed to get current release: %w", err)
+	}
+
+	// Reconcile the existing release to ensure resources are maintained
+	if err := ac.Reconcile(rel); err != nil {
+		// Reconcile failed - resources NOT maintained
+		// Return false (rollout failed) with error
+		return false, "", err
+	}
+
+	// At this point: Reconcile succeeded - resources ARE maintained (applied to cluster via Server-Side Apply)
+	// The operations below are for setting up watches to detect drift (i.e., if someone manually modifies the
+	// resources). If watch setup fails, the resources are still successfully maintained, but we won't detect
+	// and auto-correct manual modifications. We return true (rollout succeeded) and log watch errors.
+	logger := klog.FromContext(ctx)
+
+	relObjects, err := util.ManifestObjects(strings.NewReader(rel.Manifest), fmt.Sprintf("%s-release-manifest", rel.Name))
+	if err != nil {
+		logger.Error(err, "failed to parse manifest objects, cannot set up drift detection watches (resources are applied but drift detection disabled)")
+		return true, "", nil
+	}
+
+	logger.V(1).Info("setting up drift detection watches on managed objects")
+
+	// Defensive nil checks to prevent panics if Manager or Watcher not properly initialized
+	if h.Manager == nil {
+		logger.Error(fmt.Errorf("manager is nil"), "Manager not initialized, cannot set up drift detection watches (resources are applied but drift detection disabled)")
+		return true, "", nil
+	}
+	cache, err := h.Manager.Get(ctx, ext)
+	if err != nil {
+		logger.Error(err, "failed to get managed content cache, cannot set up drift detection watches (resources are applied but drift detection disabled)")
+		return true, "", nil
+	}
+
+	if h.Watcher == nil {
+		logger.Error(fmt.Errorf("watcher is nil"), "Watcher not initialized, cannot set up drift detection watches (resources are applied but drift detection disabled)")
+		return true, "", nil
+	}
+	if err := cache.Watch(ctx, h.Watcher, relObjects...); err != nil {
+		logger.Error(err, "failed to set up drift detection watches (resources are applied but drift detection disabled)")
+		return true, "", nil
+	}
+
+	return true, "", nil
+}
+
 func (h *Helm) buildHelmChart(bundleFS fs.FS, ext *ocv1.ClusterExtension) (*chart.Chart, error) {
 	if h.HelmChartProvider == nil {
 		return nil, errors.New("HelmChartProvider is nil")

internal/operator-controller/controllers/boxcutter_reconcile_steps.go

@@ -94,7 +94,7 @@ func MigrateStorage(m StorageMigrator) ReconcileStepFunc {
 	}
 }
 
-func ApplyBundleWithBoxcutter(apply func(ctx context.Context, contentFS fs.FS, ext *ocv1.ClusterExtension, objectLabels, revisionAnnotations map[string]string) error) ReconcileStepFunc {
+func ApplyBundleWithBoxcutter(apply func(ctx context.Context, contentFS fs.FS, ext *ocv1.ClusterExtension, objectLabels, revisionAnnotations map[string]string) (bool, string, error)) ReconcileStepFunc {
 	return func(ctx context.Context, state *reconcileState, ext *ocv1.ClusterExtension) (*ctrl.Result, error) {
 		l := log.FromContext(ctx)
 		revisionAnnotations := map[string]string{
@@ -109,7 +109,8 @@ func ApplyBundleWithBoxcutter(apply func(ctx context.Context, contentFS fs.FS, e
 		}
 
 		l.Info("applying bundle contents")
-		if err := apply(ctx, state.imageFS, ext, objLbls, revisionAnnotations); err != nil {
+		_, _, err := apply(ctx, state.imageFS, ext, objLbls, revisionAnnotations)
+		if err != nil {
 			// If there was an error applying the resolved bundle,
 			// report the error via the Progressing condition.
 			setStatusProgressing(ext, wrapErrorWithResolutionInfo(state.resolvedRevisionMetadata.BundleMetadata, err))

internal/operator-controller/controllers/boxcutter_reconcile_steps_apply_test.go

@@ -133,8 +133,8 @@ func TestApplyBundleWithBoxcutter(t *testing.T) {
 				imageFS: fstest.MapFS{},
 			}
 
-			stepFunc := ApplyBundleWithBoxcutter(func(_ context.Context, _ fs.FS, _ *ocv1.ClusterExtension, _, _ map[string]string) error {
-				return nil
+			stepFunc := ApplyBundleWithBoxcutter(func(_ context.Context, _ fs.FS, _ *ocv1.ClusterExtension, _, _ map[string]string) (bool, string, error) {
+				return true, "", nil
 			})
 			result, err := stepFunc(ctx, state, ext)
 			require.NoError(t, err)

internal/operator-controller/controllers/clusterextension_admission_test.go

@@ -13,7 +13,9 @@ import (
 )
 
 func TestClusterExtensionSourceConfig(t *testing.T) {
-	sourceTypeEmptyError := "Invalid value: null"
+	// NOTE: Kubernetes validation error format for JSON null values varies across K8s versions.
+	// We check for the common part "Invalid value:" which appears in all versions.
+	sourceTypeEmptyError := "Invalid value:"
 	sourceTypeMismatchError := "spec.source.sourceType: Unsupported value"
 	sourceConfigInvalidError := "spec.source: Invalid value"
 	// unionField represents the required Catalog or (future) Bundle field required by SourceConfig

internal/operator-controller/controllers/clusterextension_controller.go

@@ -168,6 +168,8 @@ func (r *ClusterExtensionReconciler) Reconcile(ctx context.Context, req ctrl.Req
 
 // ensureAllConditionsWithReason checks that all defined condition types exist in the given ClusterExtension,
 // and assigns a specified reason and custom message to any missing condition.
+//
+//nolint:unparam // reason parameter is designed to be flexible, even if current callers use the same value
 func ensureAllConditionsWithReason(ext *ocv1.ClusterExtension, reason v1alpha1.ConditionReason, message string) {
 	for _, condType := range conditionsets.ConditionTypes {
 		cond := apimeta.FindStatusCondition(ext.Status.Conditions, condType)