A mish-mash of improvements and fixes

- remove unnecessary flag
- optimize catalog watch handler
- fix finalizers (also stop using rukpak-based finalizers and keys)
- Add some reminder TODO comments about more improvements (need to
  convert to issues)

Signed-off-by: Joe Lanford <[email protected]>

Author: joelanford

cmd/manager/main.go

@@ -17,7 +17,7 @@ limitations under the License.
 package main
 
 import (
-	"crypto/x509"
+	"context"
 	"flag"
 	"fmt"
 	"net/url"
@@ -44,7 +44,7 @@ import (
 	"github.com/operator-framework/rukpak/pkg/source"
 	"github.com/operator-framework/rukpak/pkg/storage"
 
-	"github.com/operator-framework/operator-controller/api/v1alpha1"
+	ocv1alpha1 "github.com/operator-framework/operator-controller/api/v1alpha1"
 	"github.com/operator-framework/operator-controller/internal/catalogmetadata/cache"
 	catalogclient "github.com/operator-framework/operator-controller/internal/catalogmetadata/client"
 	"github.com/operator-framework/operator-controller/internal/controllers"
@@ -74,14 +74,13 @@ func podNamespace() string {
 
 func main() {
 	var (
-		metricsAddr                 string
-		enableLeaderElection        bool
-		probeAddr                   string
-		cachePath                   string
-		operatorControllerVersion   bool
-		systemNamespace             string
-		provisionerStorageDirectory string
-		caCert                      string
+		metricsAddr               string
+		enableLeaderElection      bool
+		probeAddr                 string
+		cachePath                 string
+		operatorControllerVersion bool
+		systemNamespace           string
+		caCert                    string
 	)
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
@@ -92,7 +91,6 @@ func main() {
 	flag.StringVar(&cachePath, "cache-path", "/var/cache", "The local directory path used for filesystem based caching")
 	flag.BoolVar(&operatorControllerVersion, "version", false, "Prints operator-controller version information")
 	flag.StringVar(&systemNamespace, "system-namespace", "", "Configures the namespace that gets used to deploy system resources.")
-	flag.StringVar(&provisionerStorageDirectory, "provisioner-storage-dir", storage.DefaultBundleCacheDir, "The directory that is used to store bundle contents.")
 	opts := zap.Options{
 		Development: true,
 	}
@@ -114,7 +112,7 @@ func main() {
 		systemNamespace = podNamespace()
 	}
 
-	dependentRequirement, err := k8slabels.NewRequirement(labels.OwnerKindKey, selection.In, []string{v1alpha1.ClusterExtensionKind})
+	dependentRequirement, err := k8slabels.NewRequirement(labels.OwnerKindKey, selection.In, []string{ocv1alpha1.ClusterExtensionKind})
 	if err != nil {
 		setupLog.Error(err, "unable to create dependent label selector for cache")
 		os.Exit(1)
@@ -130,7 +128,7 @@ func main() {
 		LeaderElectionID:       "9c4404e7.operatorframework.io",
 		Cache: crcache.Options{
 			ByObject: map[client.Object]crcache.ByObject{
-				&v1alpha1.ClusterExtension{}: {},
+				&ocv1alpha1.ClusterExtension{}: {},
 			},
 			DefaultNamespaces: map[string]crcache.Config{
 				systemNamespace:       {},
@@ -162,9 +160,14 @@ func main() {
 	cl := mgr.GetClient()
 	catalogClient := catalogclient.New(cl, cache.NewFilesystemCache(cachePath, httpClient))
 
-	cfgGetter, err := helmclient.NewActionConfigGetter(mgr.GetConfig(), mgr.GetRESTMapper(), helmclient.StorageNamespaceMapper(func(o client.Object) (string, error) {
-		return systemNamespace, nil
-	}))
+	installNamespaceMapper := helmclient.ObjectToStringMapper(func(obj client.Object) (string, error) {
+		ext := obj.(*ocv1alpha1.ClusterExtension)
+		return ext.Spec.InstallNamespace, nil
+	})
+	cfgGetter, err := helmclient.NewActionConfigGetter(mgr.GetConfig(), mgr.GetRESTMapper(),
+		helmclient.StorageNamespaceMapper(installNamespaceMapper),
+		helmclient.ClientNamespaceMapper(installNamespaceMapper),
+	)
 	if err != nil {
 		setupLog.Error(err, "unable to config for creating helm client")
 		os.Exit(1)
@@ -176,37 +179,47 @@ func main() {
 		os.Exit(1)
 	}
 
-	bundleFinalizers := crfinalizer.NewFinalizers()
-	unpacker, err := source.NewDefaultUnpacker(mgr, systemNamespace, filepath.Join(cachePath, "unpack"), (*x509.CertPool)(nil))
-	if err != nil {
-		setupLog.Error(err, "unable to create unpacker")
-		os.Exit(1)
+	clusterExtensionFinalizers := crfinalizer.NewFinalizers()
+	unpacker := &source.ImageRegistry{
+		BaseCachePath: filepath.Join(cachePath, "unpack"),
+		// TODO: This needs to be derived per extension via ext.Spec.InstallNamespace
+		AuthNamespace: systemNamespace,
 	}
 
-	if err := bundleFinalizers.Register(finalizer.CleanupUnpackCacheKey, &finalizer.CleanupUnpackCache{Unpacker: unpacker}); err != nil {
-		setupLog.Error(err, "unable to register finalizer", "finalizerKey", finalizer.CleanupUnpackCacheKey)
+	domain := ocv1alpha1.GroupVersion.Group
+	cleanupUnpackCacheKey := fmt.Sprintf("%s/cleanup-unpack-cache", domain)
+	deleteCachedBundleKey := fmt.Sprintf("%s/delete-cached-bundle", domain)
+	if err := clusterExtensionFinalizers.Register(cleanupUnpackCacheKey, finalizerFunc(func(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
+		ext := obj.(*ocv1alpha1.ClusterExtension)
+		return crfinalizer.Result{}, os.RemoveAll(filepath.Join(unpacker.BaseCachePath, ext.GetName()))
+	})); err != nil {
+		setupLog.Error(err, "unable to register finalizer", "finalizerKey", cleanupUnpackCacheKey)
 		os.Exit(1)
 	}
 
 	localStorage := &storage.LocalDirectory{
-		RootDirectory: provisionerStorageDirectory,
+		RootDirectory: filepath.Join(cachePath, "bundles"),
 		URL:           url.URL{},
 	}
-
-	if err := bundleFinalizers.Register(finalizer.DeleteCachedBundleKey, &finalizer.DeleteCachedBundle{Storage: localStorage}); err != nil {
-		setupLog.Error(err, "unable to register finalizer", "finalizerKey", finalizer.DeleteCachedBundleKey)
+	if err := clusterExtensionFinalizers.Register(deleteCachedBundleKey, finalizerFunc(func(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
+		ext := obj.(*ocv1alpha1.ClusterExtension)
+		return crfinalizer.Result{}, localStorage.Delete(ctx, ext)
+	})); err != nil {
+		setupLog.Error(err, "unable to register finalizer", "finalizerKey", deleteCachedBundleKey)
 		os.Exit(1)
 	}
 
+	if err := clusterExtensionFinalizers.Register(
+
 	if err = (&controllers.ClusterExtensionReconciler{
 		Client:                cl,
-		ReleaseNamespace:      systemNamespace,
 		BundleProvider:        catalogClient,
 		ActionClientGetter:    acg,
 		Unpacker:              unpacker,
 		Storage:               localStorage,
 		InstalledBundleGetter: &controllers.DefaultInstalledBundleGetter{ActionClientGetter: acg},
 		Handler:               registryv1handler.HandlerFunc(registry.HandleBundleDeployment),
+		Finalizers:            clusterExtensionFinalizers,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ClusterExtension")
 		os.Exit(1)
@@ -229,3 +242,9 @@ func main() {
 		os.Exit(1)
 	}
 }
+
+type finalizerFunc func(ctx context.Context, obj client.Object) (crfinalizer.Result, error)
+
+func (f finalizerFunc) Finalize(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
+	return f(ctx, obj)
+}

config/base/rbac/role.yaml

@@ -27,25 +27,15 @@ rules:
 - apiGroups:
   - ""
   resources:
-  - configmaps
-  verbs:
-  - list
-  - watch
-- apiGroups:
-  - ""
-  resources:
-  - pods
+  - secrets
   verbs:
   - create
   - delete
+  - get
   - list
+  - patch
+  - update
   - watch
-- apiGroups:
-  - ""
-  resources:
-  - pods/log
-  verbs:
-  - get
 - apiGroups:
   - olm.operatorframework.io
   resources:

internal/controllers/clusterextension_controller.go

@@ -36,7 +36,6 @@ import (
 	"helm.sh/helm/v3/pkg/postrender"
 	"helm.sh/helm/v3/pkg/release"
 	"helm.sh/helm/v3/pkg/storage/driver"
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -48,9 +47,9 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	crcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
+	crfinalizer "sigs.k8s.io/controller-runtime/pkg/finalizer"
 	crhandler "sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -80,7 +79,6 @@ import (
 // ClusterExtensionReconciler reconciles a ClusterExtension object
 type ClusterExtensionReconciler struct {
 	client.Client
-	ReleaseNamespace      string
 	BundleProvider        BundleProvider
 	Unpacker              rukpaksource.Unpacker
 	ActionClientGetter    helmclient.ActionClientGetter
@@ -91,6 +89,7 @@ type ClusterExtensionReconciler struct {
 	controller            crcontroller.Controller
 	cache                 cache.Cache
 	InstalledBundleGetter InstalledBundleGetter
+	Finalizers            crfinalizer.Finalizers
 }
 
 type InstalledBundleGetter interface {
@@ -104,9 +103,7 @@ const (
 //+kubebuilder:rbac:groups=olm.operatorframework.io,resources=clusterextensions,verbs=get;list;watch
 //+kubebuilder:rbac:groups=olm.operatorframework.io,resources=clusterextensions/status,verbs=update;patch
 //+kubebuilder:rbac:groups=olm.operatorframework.io,resources=clusterextensions/finalizers,verbs=update
-//+kubebuilder:rbac:groups=core,resources=pods,verbs=list;watch;create;delete
-//+kubebuilder:rbac:groups=core,resources=configmaps,verbs=list;watch
-//+kubebuilder:rbac:groups=core,resources=pods/log,verbs=get
+//+kubebuilder:rbac:groups=core,resources=secrets,verbs=create;update;patch;delete;get;list;watch
 //+kubebuilder:rbac:groups=*,resources=*,verbs=*
 
 //+kubebuilder:rbac:groups=catalogd.operatorframework.io,resources=clustercatalogs,verbs=list;watch
@@ -196,6 +193,25 @@ func checkForUnexpectedFieldChange(a, b ocv1alpha1.ClusterExtension) bool {
 */
 //nolint:unparam
 func (r *ClusterExtensionReconciler) reconcile(ctx context.Context, ext *ocv1alpha1.ClusterExtension) (ctrl.Result, error) {
+	finalizeResult, err := r.Finalizers.Finalize(ctx, ext)
+	if err != nil {
+		// TODO: For now, this error handling follows the pattern of other error handling.
+		//  Namely: zero just about everything out, throw our hands up, and return an error.
+		//  This is not ideal, and we should consider a more nuanced approach that resolves
+		//  as much status as possible before returning, or at least keeps previous state if
+		//  it is properly labeled with its observed generation.
+		ext.Status.ResolvedBundle = nil
+		ext.Status.InstalledBundle = nil
+		setResolvedStatusConditionFailed(ext, err.Error())
+		ensureAllConditionsWithReason(ext, ocv1alpha1.ReasonResolutionFailed, err.Error())
+		return ctrl.Result{}, err
+	}
+	if finalizeResult.Updated || finalizeResult.StatusUpdated {
+		// On create: make sure the finalizer is applied before we do anything
+		// On delete: make sure we do nothing after the finalizer is removed
+		return ctrl.Result{}, nil
+	}
+
 	// run resolution
 	bundle, err := r.resolve(ctx, *ext)
 	if err != nil {
@@ -300,7 +316,7 @@ func (r *ClusterExtensionReconciler) reconcile(ctx context.Context, ext *ocv1alp
 
 	switch state {
 	case stateNeedsInstall:
-		rel, err = ac.Install(ext.GetName(), r.ReleaseNamespace, chrt, values, func(install *action.Install) error {
+		rel, err = ac.Install(ext.GetName(), ext.Spec.InstallNamespace, chrt, values, func(install *action.Install) error {
 			install.CreateNamespace = false
 			install.Labels = map[string]string{labels.BundleNameKey: bundle.Name, labels.PackageNameKey: bundle.Package, labels.BundleVersionKey: bundleVersion.String()}
 			return nil
@@ -310,7 +326,7 @@ func (r *ClusterExtensionReconciler) reconcile(ctx context.Context, ext *ocv1alp
 			return ctrl.Result{}, err
 		}
 	case stateNeedsUpgrade:
-		rel, err = ac.Upgrade(ext.GetName(), r.ReleaseNamespace, chrt, values, helmclient.AppendUpgradePostRenderer(post))
+		rel, err = ac.Upgrade(ext.GetName(), ext.Spec.InstallNamespace, chrt, values, helmclient.AppendUpgradePostRenderer(post))
 		if err != nil {
 			setInstalledStatusConditionFailed(ext, fmt.Sprintf("%s:%v", ocv1alpha1.ReasonUpgradeFailed, err))
 			return ctrl.Result{}, err
@@ -574,7 +590,6 @@ func (r *ClusterExtensionReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				return true
 			},
 		}).
-		Watches(&corev1.Pod{}, mapOwneeToOwnerHandler(mgr.GetClient(), mgr.GetLogger(), &ocv1alpha1.ClusterExtension{})).
 		Build(r)
 
 	if err != nil {
@@ -587,47 +602,12 @@ func (r *ClusterExtensionReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return nil
 }
 
-func mapOwneeToOwnerHandler(cl client.Client, log logr.Logger, owner client.Object) crhandler.EventHandler {
-	return crhandler.EnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []reconcile.Request {
-		ownerGVK, err := apiutil.GVKForObject(owner, cl.Scheme())
-		if err != nil {
-			log.Error(err, "map ownee to owner: lookup GVK for owner")
-			return nil
-		}
-
-		type ownerInfo struct {
-			key types.NamespacedName
-			gvk schema.GroupVersionKind
-		}
-		var oi *ownerInfo
-
-		for _, ref := range obj.GetOwnerReferences() {
-			gv, err := schema.ParseGroupVersion(ref.APIVersion)
-			if err != nil {
-				log.Error(err, fmt.Sprintf("map ownee to owner: parse ownee's owner reference group version %q", ref.APIVersion))
-				return nil
-			}
-			refGVK := gv.WithKind(ref.Kind)
-			if refGVK == ownerGVK && ref.Controller != nil && *ref.Controller {
-				oi = &ownerInfo{
-					key: types.NamespacedName{Name: ref.Name},
-					gvk: ownerGVK,
-				}
-				break
-			}
-		}
-		if oi == nil {
-			return nil
-		}
-		return []reconcile.Request{{NamespacedName: oi.key}}
-	})
-}
-
 // Generate reconcile requests for all cluster extensions affected by a catalog change
 func clusterExtensionRequestsForCatalog(c client.Reader, logger logr.Logger) crhandler.MapFunc {
 	return func(ctx context.Context, _ client.Object) []reconcile.Request {
 		// no way of associating an extension to a catalog so create reconcile requests for everything
-		clusterExtensions := ocv1alpha1.ClusterExtensionList{}
+		clusterExtensions := metav1.PartialObjectMetadataList{}
+		clusterExtensions.SetGroupVersionKind(ocv1alpha1.GroupVersion.WithKind("ClusterExtensionList"))
 		err := c.List(ctx, &clusterExtensions)
 		if err != nil {
 			logger.Error(err, "unable to enqueue cluster extensions for catalog reconcile")
@@ -655,16 +635,16 @@ const (
 	stateError        releaseState = "Error"
 )
 
-func (r *ClusterExtensionReconciler) getReleaseState(cl helmclient.ActionInterface, obj metav1.Object, chrt *chart.Chart, values chartutil.Values, post *postrenderer) (*release.Release, releaseState, error) {
-	currentRelease, err := cl.Get(obj.GetName())
+func (r *ClusterExtensionReconciler) getReleaseState(cl helmclient.ActionInterface, ext *ocv1alpha1.ClusterExtension, chrt *chart.Chart, values chartutil.Values, post *postrenderer) (*release.Release, releaseState, error) {
+	currentRelease, err := cl.Get(ext.GetName())
 	if err != nil && !errors.Is(err, driver.ErrReleaseNotFound) {
 		return nil, stateError, err
 	}
 	if errors.Is(err, driver.ErrReleaseNotFound) {
 		return nil, stateNeedsInstall, nil
 	}
 
-	desiredRelease, err := cl.Upgrade(obj.GetName(), r.ReleaseNamespace, chrt, values, func(upgrade *action.Upgrade) error {
+	desiredRelease, err := cl.Upgrade(ext.GetName(), ext.Spec.InstallNamespace, chrt, values, func(upgrade *action.Upgrade) error {
 		upgrade.DryRun = true
 		return nil
 	}, helmclient.AppendUpgradePostRenderer(post))

test/e2e/cluster_extension_install_test.go

@@ -146,6 +146,10 @@ func TestClusterExtensionInstallReResolvesWhenNewCatalog(t *testing.T) {
 	t.Log("By creating the ClusterExtension resource")
 	require.NoError(t, c.Create(context.Background(), clusterExtension))
 
+	// TODO: this isn't a good precondition because a missing package results in
+	//  exponential backoff retries. So we can't be sure that the re-reconcile is a result of
+	//  the catalog becoming available because it could also be a retry of the initial failed
+	//  resolution.
 	t.Log("By failing to find ClusterExtension during resolution")
 	require.EventuallyWithT(t, func(ct *assert.CollectT) {
 		assert.NoError(ct, c.Get(context.Background(), types.NamespacedName{Name: clusterExtension.Name}, clusterExtension))
@@ -257,13 +261,13 @@ func TestClusterExtensionForceInstallNonSuccessorVersion(t *testing.T) {
 		assert.Equal(ct, &ocv1alpha1.BundleMetadata{Name: "prometheus-operator.1.0.0", Version: "1.0.0"}, clusterExtension.Status.ResolvedBundle)
 	}, pollDuration, pollInterval)
 
-	t.Log("It does not allow to upgrade the ClusterExtension to a non-successor version")
+	t.Log("It allows to upgrade the ClusterExtension to a non-successor version")
 	t.Log("By updating the ClusterExtension resource to a non-successor version")
 	// 1.2.0 does not replace/skip/skipRange 1.0.0.
 	clusterExtension.Spec.Version = "1.2.0"
 	clusterExtension.Spec.UpgradeConstraintPolicy = ocv1alpha1.UpgradeConstraintPolicyIgnore
 	require.NoError(t, c.Update(context.Background(), clusterExtension))
-	t.Log("By eventually reporting an unsatisfiable resolution")
+	t.Log("By eventually reporting a satisfiable resolution")
 	require.EventuallyWithT(t, func(ct *assert.CollectT) {
 		assert.NoError(ct, c.Get(context.Background(), types.NamespacedName{Name: clusterExtension.Name}, clusterExtension))
 		cond := apimeta.FindStatusCondition(clusterExtension.Status.Conditions, ocv1alpha1.TypeResolved)