(bugfix): rewrite the contentmanager implementation

to fix bugs associated with insufficient permissions resulting
in halting reconciliation of all ClusterExtension and informer
sync errors not being reported via the ClusterExtension
status conditions.

Signed-off-by: everettraven <[email protected]>

Author: everettraven

Diff for: api/v1alpha1/clusterextension_types.go

@@ -116,6 +116,7 @@ const (
 	// TODO(user): add more Types, here and into init()
 	TypeInstalled = "Installed"
 	TypeResolved  = "Resolved"
+	TypeManaged   = "Managed"
 
 	// TypeDeprecated is a rollup condition that is present when
 	// any of the deprecated conditions are present.
@@ -139,6 +140,8 @@ const (
 	ReasonUnpackFailed  = "UnpackFailed"
 
 	ReasonErrorGettingReleaseState = "ErrorGettingReleaseState"
+
+	ReasonManagementFailed = "ManagementFailed"
 )
 
 func init() {
@@ -151,6 +154,7 @@ func init() {
 		TypeChannelDeprecated,
 		TypeBundleDeprecated,
 		TypeUnpacked,
+		TypeManaged,
 	)
 	// TODO(user): add Reasons from above
 	conditionsets.ConditionReasons = append(conditionsets.ConditionReasons,
@@ -164,6 +168,7 @@ func init() {
 		ReasonUnpackSuccess,
 		ReasonUnpackFailed,
 		ReasonErrorGettingReleaseState,
+		ReasonManagementFailed,
 	)
 }
 

Diff for: cmd/manager/main.go

@@ -254,14 +254,22 @@ func main() {
 		Preflights:         preflights,
 	}
 
+	cm := contentmanager.NewManager(clientRestConfigMapper, mgr.GetConfig(), mgr.GetRESTMapper())
+	contentManagerFinalizerKey := fmt.Sprintf("%s/contentmanager-cleanup", domain)
+	clusterExtensionFinalizers.Register(contentManagerFinalizerKey, finalizerFunc(func(ctx context.Context, obj client.Object) (crfinalizer.Result, error) {
+		ext := obj.(*ocv1alpha1.ClusterExtension)
+		cm.Delete(ext)
+		return crfinalizer.Result{}, nil
+	}))
+
 	if err = (&controllers.ClusterExtensionReconciler{
 		Client:                cl,
 		Resolver:              resolver,
 		Unpacker:              unpacker,
 		Applier:               applier,
 		InstalledBundleGetter: &controllers.DefaultInstalledBundleGetter{ActionClientGetter: acg},
 		Finalizers:            clusterExtensionFinalizers,
-		Watcher:               contentmanager.New(clientRestConfigMapper, mgr.GetConfig(), mgr.GetRESTMapper()),
+		Manager:               cm,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ClusterExtension")
 		os.Exit(1)

Diff for: config/samples/olm_v1alpha1_clusterextension.yaml

@@ -36,10 +36,10 @@ rules:
 # Manage ArgoCD CRDs
 - apiGroups: [apiextensions.k8s.io]
   resources: [customresourcedefinitions]
-  verbs: [create]
+  verbs: [create, list, watch]
 - apiGroups: [apiextensions.k8s.io]
   resources: [customresourcedefinitions]
-  verbs: [get, list, watch, update, patch, delete]
+  verbs: [get, update, patch, delete]
   resourceNames:
   - appprojects.argoproj.io
   - argocds.argoproj.io

Diff for: internal/applier/helm.go

@@ -1,18 +1,22 @@
 package applier
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"io/fs"
 	"strings"
 
 	"helm.sh/helm/v3/pkg/action"
 	"helm.sh/helm/v3/pkg/chart"
 	"helm.sh/helm/v3/pkg/chartutil"
+	"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/apis/meta/v1/unstructured"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	helmclient "github.com/operator-framework/helm-operator-plugins/pkg/client"
@@ -21,6 +25,7 @@ import (
 	"github.com/operator-framework/operator-controller/internal/rukpak/convert"
 	"github.com/operator-framework/operator-controller/internal/rukpak/preflights/crdupgradesafety"
 	"github.com/operator-framework/operator-controller/internal/rukpak/util"
+	apimachyaml "k8s.io/apimachinery/pkg/util/yaml"
 )
 
 const (
@@ -63,7 +68,11 @@ func (h *Helm) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1alpha1.Clust
 		return nil, "", err
 	}
 
-	rel, desiredRel, state, err := h.getReleaseState(ac, ext, chrt, values, labels)
+	post := &postrenderer{
+		labels: labels,
+	}
+
+	rel, desiredRel, state, err := h.getReleaseState(ac, ext, chrt, values, post)
 	if err != nil {
 		return nil, "", err
 	}
@@ -96,7 +105,7 @@ func (h *Helm) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1alpha1.Clust
 			install.CreateNamespace = false
 			install.Labels = labels
 			return nil
-		})
+		}, helmclient.AppendInstallPostRenderer(post))
 		if err != nil {
 			return nil, state, err
 		}
@@ -105,7 +114,7 @@ func (h *Helm) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1alpha1.Clust
 			upgrade.MaxHistory = maxHelmReleaseHistory
 			upgrade.Labels = labels
 			return nil
-		})
+		}, helmclient.AppendUpgradePostRenderer(post))
 		if err != nil {
 			return nil, state, err
 		}
@@ -125,7 +134,7 @@ func (h *Helm) Apply(ctx context.Context, contentFS fs.FS, ext *ocv1alpha1.Clust
 	return relObjects, state, nil
 }
 
-func (h *Helm) getReleaseState(cl helmclient.ActionInterface, ext *ocv1alpha1.ClusterExtension, chrt *chart.Chart, values chartutil.Values, labels map[string]string) (*release.Release, *release.Release, string, error) {
+func (h *Helm) getReleaseState(cl helmclient.ActionInterface, ext *ocv1alpha1.ClusterExtension, chrt *chart.Chart, values chartutil.Values, post postrender.PostRenderer) (*release.Release, *release.Release, string, error) {
 	currentRelease, err := cl.Get(ext.GetName())
 	if err != nil && !errors.Is(err, driver.ErrReleaseNotFound) {
 		return nil, nil, StateError, err
@@ -138,9 +147,8 @@ func (h *Helm) getReleaseState(cl helmclient.ActionInterface, ext *ocv1alpha1.Cl
 		desiredRelease, err := cl.Install(ext.GetName(), ext.Spec.InstallNamespace, chrt, values, func(i *action.Install) error {
 			i.DryRun = true
 			i.DryRunOption = "server"
-			i.Labels = labels
 			return nil
-		})
+		}, helmclient.AppendInstallPostRenderer(post))
 		if err != nil {
 			return nil, nil, StateError, err
 		}
@@ -150,9 +158,8 @@ func (h *Helm) getReleaseState(cl helmclient.ActionInterface, ext *ocv1alpha1.Cl
 		upgrade.MaxHistory = maxHelmReleaseHistory
 		upgrade.DryRun = true
 		upgrade.DryRunOption = "server"
-		upgrade.Labels = labels
 		return nil
-	})
+	}, helmclient.AppendUpgradePostRenderer(post))
 	if err != nil {
 		return currentRelease, nil, StateError, err
 	}
@@ -164,3 +171,33 @@ func (h *Helm) getReleaseState(cl helmclient.ActionInterface, ext *ocv1alpha1.Cl
 	}
 	return currentRelease, desiredRelease, relState, nil
 }
+
+type postrenderer struct {
+	labels  map[string]string
+	cascade postrender.PostRenderer
+}
+
+func (p *postrenderer) Run(renderedManifests *bytes.Buffer) (*bytes.Buffer, error) {
+	var buf bytes.Buffer
+	dec := apimachyaml.NewYAMLOrJSONDecoder(renderedManifests, 1024)
+	for {
+		obj := unstructured.Unstructured{}
+		err := dec.Decode(&obj)
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			return nil, err
+		}
+		obj.SetLabels(util.MergeMaps(obj.GetLabels(), p.labels))
+		b, err := obj.MarshalJSON()
+		if err != nil {
+			return nil, err
+		}
+		buf.Write(b)
+	}
+	if p.cascade != nil {
+		return p.cascade.Run(&buf)
+	}
+	return &buf, nil
+}

Diff for: internal/contentmanager/cache.go

@@ -0,0 +1,115 @@
+package contentmanager
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+// Cache is a storage mechanism for keeping track of
+// managed content sources. It also exposes methods
+// for retrieving the stored references of the managed
+// content
+type Cache interface {
+	// Get attempts to fetch the object with the provided namespace/name and GroupVersionKind
+	Get(client.ObjectKey, schema.GroupVersionKind) (runtime.Object, error)
+	// List attempts to list all the objects with the provided GroupVersionKind and label selector
+	List(schema.GroupVersionKind, labels.Selector) ([]runtime.Object, error)
+	// Close stops all sources belonging to the cache
+	Close()
+	// AddSource adds a new ReaderCloserSyncincSource to the cache for the provided GroupVersionKind
+	AddSource(schema.GroupVersionKind, ReaderCloserSyncingSource)
+	// RemoveSource stops and removes a source from cache for the provided GroupVersionKind
+	RemoveSource(schema.GroupVersionKind)
+	// Sources returns the current mapping of GroupVersionKind to ReaderCloserSyncingSource
+	Sources() map[schema.GroupVersionKind]ReaderCloserSyncingSource
+}
+
+// ReaderCloserSyncingSource is a wrapper of the controller-runtime
+// source.SyncingSource that includes methods for:
+//   - Getting a specific resource synced by the source
+//   - Listing resources synced by the source
+//   - Closing the source, stopping it's interaction with the Kubernetes API server and reaction to events
+type ReaderCloserSyncingSource interface {
+	source.SyncingSource
+	// Get attempts to fetch the object with the provided namespace/name
+	Get(client.ObjectKey) (runtime.Object, error)
+	// List attempts to list all the objects with the provided label selector
+	List(labels.Selector) ([]runtime.Object, error)
+	// Close stops the source
+	Close()
+}
+
+type cache struct {
+	sources map[schema.GroupVersionKind]ReaderCloserSyncingSource
+	mu      sync.Mutex
+}
+
+func NewCache() Cache {
+	return &cache{
+		sources: make(map[schema.GroupVersionKind]ReaderCloserSyncingSource),
+	}
+}
+
+var _ Cache = (*cache)(nil)
+
+func (c *cache) Close() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	for _, source := range c.sources {
+		source.Close()
+	}
+}
+
+func (c *cache) Get(key client.ObjectKey, gvk schema.GroupVersionKind) (runtime.Object, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	source, ok := c.sources[gvk]
+	if !ok {
+		return nil, fmt.Errorf("no source for GVK %q", gvk)
+	}
+
+	return source.Get(key)
+}
+
+func (c *cache) List(gvk schema.GroupVersionKind, selector labels.Selector) ([]runtime.Object, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	source, ok := c.sources[gvk]
+	if !ok {
+		return nil, fmt.Errorf("no source for GVK %q", gvk)
+	}
+
+	return source.List(selector)
+}
+
+func (c *cache) AddSource(gvk schema.GroupVersionKind, source ReaderCloserSyncingSource) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, ok := c.sources[gvk]; !ok {
+		c.sources[gvk] = source
+	}
+}
+
+func (c *cache) RemoveSource(gvk schema.GroupVersionKind) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if source, ok := c.sources[gvk]; ok {
+		source.Close()
+	}
+	delete(c.sources, gvk)
+}
+
+func (c *cache) Sources() map[schema.GroupVersionKind]ReaderCloserSyncingSource {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.sources
+}