resolver: Add support for excluding global catalogs from resolution

The resolver is currently configured to always consider resolving
from global catalogs. In some cases, this may not be
desirable, and the user would like to explicitly ignore global catalogs
for the purposes of resolution. This change enables per-namespace
exclusion from global catalog sources via an annotation on the
operatorgroup.

This is done via a client-side interface that wraps the existing
registry source provider. Also, a queueinformer for OperatorGroups is added
since they are now an input to resolution via the global catalog exclusion annotation.
Namespace resolution will be triggered on changes to an OperatorGroup, in case the
value provided on that annotation by the user changes.

Signed-off-by: Daniel Sover <[email protected]>
Co-authored-by: perdasilva <[email protected]>

Author: exdx

pkg/controller/operators/catalog/og_source_provider.go

@@ -0,0 +1,97 @@
+package catalog
+
+import (
+	"context"
+	"fmt"
+
+	v1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1"
+
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver/cache"
+	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+type OperatorGroupToggleSourceProvider struct {
+	sp       SourceProviderWithInvalidate
+	logger   *logrus.Logger
+	ogLister v1listers.OperatorGroupLister
+}
+
+func NewOperatorGroupToggleSourceProvider(sp SourceProviderWithInvalidate, logger *logrus.Logger,
+	ogLister v1listers.OperatorGroupLister) *OperatorGroupToggleSourceProvider {
+	return &OperatorGroupToggleSourceProvider{
+		sp:       sp,
+		logger:   logger,
+		ogLister: ogLister,
+	}
+}
+
+// SourceProviderWithInvalidate is a client-side interface for the OperatorGroup Toggle
+type SourceProviderWithInvalidate interface {
+	cache.SourceProvider
+	Invalidate(key cache.SourceKey)
+}
+
+const exclusionAnnotation string = "olm.operatorframework.io/exclude-global-namespace-resolution"
+
+func (e *OperatorGroupToggleSourceProvider) Sources(namespaces ...string) map[cache.SourceKey]cache.Source {
+	// Check if annotation is set first
+	resolutionNamespaces, err := e.CheckForExclusion(namespaces...)
+	if err != nil {
+		e.logger.Errorf("error checking namespaces %#v for global resolution exlusion: %s", namespaces, err)
+		// Fail early with a dummy Source that returns an error
+		// Note: this only errors in the case the Lister fails, which should not happen
+		// TODO: Update the Sources interface to return an error
+		m := make(map[cache.SourceKey]cache.Source)
+		key := cache.SourceKey{Name: "og-exclusion-failure", Namespace: namespaces[0]}
+		source := errorSource{err}
+		m[key] = source
+		return m
+	}
+	return e.sp.Sources(resolutionNamespaces...)
+}
+
+func (e *OperatorGroupToggleSourceProvider) Invalidate(key cache.SourceKey) {
+	e.sp.Invalidate(key)
+}
+
+type errorSource struct {
+	error
+}
+
+func (e errorSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) {
+	return nil, e.error
+}
+
+func (e *OperatorGroupToggleSourceProvider) CheckForExclusion(namespaces ...string) ([]string, error) {
+	var defaultResult = namespaces
+	// The first namespace provided is always the current namespace being synced
+	var ownNamespace = namespaces[0]
+	var toggledResult = []string{ownNamespace}
+
+	// Check the OG on the NS provided for the exclusion annotation
+	ogs, err := e.ogLister.OperatorGroups(ownNamespace).List(labels.Everything())
+	if err != nil {
+		return nil, fmt.Errorf("listing operatorgroups in namespace %s: %s", ownNamespace, err)
+	}
+
+	if len(ogs) != 1 {
+		// Problem with the operatorgroup configuration in the namespace, or the operatorgroup has not yet been persisted
+		// Note: a resync will be triggered if/when the operatorgroup becomes available
+		return nil, fmt.Errorf("found %d operatorgroups in namespace %s: expected 1", len(ogs), ownNamespace)
+	}
+
+	var og = ogs[0]
+	if og.Annotations == nil {
+		// No exclusion specified
+		return defaultResult, nil
+	}
+	if val, ok := og.Annotations[exclusionAnnotation]; ok && val == "true" {
+		// Exclusion specified
+		// Ignore the globalNamespace for the purposes of resolution in this namespace
+		e.logger.Printf("excluding global catalogs from resolution in namespace %s", ownNamespace)
+		return toggledResult, nil
+	}
+
+	return defaultResult, nil
+}

pkg/controller/operators/catalog/operator.go

@@ -42,6 +42,7 @@ import (
 	"k8s.io/client-go/util/workqueue"
 
 	"github.com/operator-framework/api/pkg/operators/reference"
+	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/informers/externalversions"
@@ -118,7 +119,7 @@ type Operator struct {
 	bundleUnpackTimeout      time.Duration
 	clientFactory            clients.Factory
 	muInstallPlan            sync.Mutex
-	resolverSourceProvider   *resolver.RegistrySourceProvider
+	resolverSourceProvider   SourceProviderWithInvalidate
 }
 
 type CatalogSourceSyncFunc func(logger *logrus.Entry, in *v1alpha1.CatalogSource) (out *v1alpha1.CatalogSource, continueSync bool, syncError error)
@@ -191,7 +192,8 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		clientFactory:            clients.NewFactory(config),
 	}
 	op.sources = grpc.NewSourceStore(logger, 10*time.Second, 10*time.Minute, op.syncSourceState)
-	op.resolverSourceProvider = resolver.SourceProviderFromRegistryClientProvider(op.sources, logger)
+	op.resolverSourceProvider = NewOperatorGroupToggleSourceProvider(resolver.SourceProviderFromRegistryClientProvider(op.sources, logger),
+		logger, op.lister.OperatorsV1().OperatorGroupLister())
 	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now, ssaClient)
 	res := resolver.NewOperatorStepResolver(lister, crClient, operatorNamespace, op.resolverSourceProvider, logger)
 	op.resolver = resolver.NewInstrumentedResolver(res, metrics.RegisterDependencyResolutionSuccess, metrics.RegisterDependencyResolutionFailure)
@@ -259,7 +261,19 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 
 	operatorGroupInformer := crInformerFactory.Operators().V1().OperatorGroups()
 	op.lister.OperatorsV1().RegisterOperatorGroupLister(metav1.NamespaceAll, operatorGroupInformer.Lister())
-	if err := op.RegisterInformer(operatorGroupInformer.Informer()); err != nil {
+	ogQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "ogs")
+	op.ogQueueSet.Set(metav1.NamespaceAll, ogQueue)
+	operatorGroupQueueInformer, err := queueinformer.NewQueueInformer(
+		ctx,
+		queueinformer.WithLogger(op.logger),
+		queueinformer.WithQueue(ogQueue),
+		queueinformer.WithInformer(operatorGroupInformer.Informer()),
+		queueinformer.WithSyncer(queueinformer.LegacySyncHandler(op.syncOperatorGroups).ToSyncer()),
+	)
+	if err != nil {
+		return nil, err
+	}
+	if err := op.RegisterQueueInformer(operatorGroupQueueInformer); err != nil {
 		return nil, err
 	}
 
@@ -1085,6 +1099,20 @@ func (o *Operator) syncSubscriptions(obj interface{}) error {
 	return nil
 }
 
+// syncOperatorGroups requeues the namespace resolution queue on changes to an operatorgroup
+// This is because the operatorgroup is now an input to resolution via the global catalog exclusion annotation
+func (o *Operator) syncOperatorGroups(obj interface{}) error {
+	og, ok := obj.(*operatorsv1.OperatorGroup)
+	if !ok {
+		o.logger.Debugf("wrong type: %#v", obj)
+		return fmt.Errorf("casting OperatorGroup failed")
+	}
+
+	o.nsResolveQueue.Add(og.GetNamespace())
+
+	return nil
+}
+
 func (o *Operator) nothingToUpdate(logger *logrus.Entry, sub *v1alpha1.Subscription) bool {
 	if sub.Status.InstallPlanRef != nil && sub.Status.State == v1alpha1.SubscriptionStateUpgradePending {
 		logger.Debugf("skipping update: installplan already created")

test/e2e/catalog_exclusion_test.go

@@ -0,0 +1,133 @@
+package e2e
+
+import (
+	"context"
+	"path/filepath"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	"github.com/operator-framework/operator-lifecycle-manager/test/e2e/ctx"
+	"github.com/operator-framework/operator-lifecycle-manager/test/e2e/util"
+	. "github.com/operator-framework/operator-lifecycle-manager/test/e2e/util/gomega"
+	"google.golang.org/grpc/connectivity"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8scontrollerclient "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+const magicCatalogDir = "magiccatalog"
+
+var _ = Describe("Global Catalog Exclusion", func() {
+	var (
+		testNamespace       corev1.Namespace
+		determinedE2eClient *util.DeterminedE2EClient
+		operatorGroup       operatorsv1.OperatorGroup
+		localCatalog        *MagicCatalog
+	)
+
+	BeforeEach(func() {
+		determinedE2eClient = util.NewDeterminedClient(ctx.Ctx().E2EClient())
+
+		By("creating a namespace with an own namespace operator group without annotations")
+		e2eTestNamespace := genName("global-catalog-exclusion-e2e-")
+		operatorGroup = operatorsv1.OperatorGroup{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:   e2eTestNamespace,
+				Name:        genName("og-"),
+				Annotations: nil,
+			},
+			Spec: operatorsv1.OperatorGroupSpec{
+				TargetNamespaces: []string{e2eTestNamespace},
+			},
+		}
+		testNamespace = SetupGeneratedTestNamespaceWithOperatorGroup(e2eTestNamespace, operatorGroup)
+
+		By("creating a broken catalog in the global namespace")
+		globalCatalog := &v1alpha1.CatalogSource{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      genName("bad-global-catalog-"),
+				Namespace: operatorNamespace,
+			},
+			Spec: v1alpha1.CatalogSourceSpec{
+				DisplayName: "Broken Global Catalog Source",
+				SourceType:  v1alpha1.SourceTypeGrpc,
+				Address:     "1.1.1.1:1337", // points to non-existing service
+			},
+		}
+		_ = determinedE2eClient.Create(context.Background(), globalCatalog)
+
+		By("creating a healthy catalog in the test namespace")
+		localCatalogName := genName("good-catsrc-")
+		var err error = nil
+
+		fbcPath := filepath.Join(testdataDir, magicCatalogDir, "fbc_initial.yaml")
+		localCatalog, err = NewMagicCatalogFromFile(determinedE2eClient, testNamespace.GetName(), localCatalogName, fbcPath)
+		Expect(err).To(Succeed())
+
+		// deploy catalog blocks until the catalog has reached a ready state or fails
+		Expect(localCatalog.DeployCatalog(context.Background())).To(Succeed())
+
+		By("checking that the global catalog is broken")
+		// Adding this check here to speed up the test
+		// the global catalog can fail while we wait for the local catalog to get to a ready state
+		EventuallyResource(globalCatalog).Should(HaveGrpcConnectionWithLastConnectionState(connectivity.TransientFailure))
+	})
+
+	AfterEach(func() {
+		TeardownNamespace(testNamespace.GetName())
+	})
+
+	When("a subscription referring to the local catalog is created", func() {
+		var subscription *v1alpha1.Subscription
+
+		BeforeEach(func() {
+			subscription = &v1alpha1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: testNamespace.GetName(),
+					Name:      genName("local-subscription-"),
+				},
+				Spec: &v1alpha1.SubscriptionSpec{
+					CatalogSource:          localCatalog.GetName(),
+					CatalogSourceNamespace: localCatalog.GetNamespace(),
+					Package:                "packageA",
+					Channel:                "stable",
+					InstallPlanApproval:    v1alpha1.ApprovalAutomatic,
+				},
+			}
+
+			By("creating a subscription")
+			_ = determinedE2eClient.Create(context.Background(), subscription)
+		})
+
+		When("the operator group is annotated with olm.operatorframework.io/exclude-global-namespace-resolution=true", func() {
+
+			It("the broken subscription should resolve and have state AtLatest", func() {
+				By("checking that the subscription is not resolving and has a condition with type ResolutionFailed")
+				EventuallyResource(subscription).Should(ContainSubscriptionConditionOfType(v1alpha1.SubscriptionResolutionFailed))
+
+				By("annotating the operator group with olm.operatorframework.io/exclude-global-namespace-resolution=true")
+				Eventually(func() error {
+					annotatedOperatorGroup := operatorGroup.DeepCopy()
+					if err := determinedE2eClient.Get(context.Background(), k8scontrollerclient.ObjectKeyFromObject(annotatedOperatorGroup), annotatedOperatorGroup); err != nil {
+						return err
+					}
+
+					if annotatedOperatorGroup.Annotations == nil {
+						annotatedOperatorGroup.Annotations = map[string]string{}
+					}
+
+					annotatedOperatorGroup.Annotations["olm.operatorframework.io/exclude-global-namespace-resolution"] = "true"
+					if err := determinedE2eClient.Update(context.Background(), annotatedOperatorGroup); err != nil {
+						return err
+					}
+					return nil
+				}).Should(Succeed())
+
+				By("checking that the subscription resolves and has state AtLatest")
+				EventuallyResource(subscription).Should(HaveSubscriptionState(v1alpha1.SubscriptionStateAtLatest))
+			})
+		})
+	})
+})