Make CatalogSource the source of truth for available catalogs.

Internally, the catalog operator has always maintained a set of
registry clients for each CatalogSource. Although this set is
reconciled toward containing a client per CatalogSource object, there
is some latency before changes made to CatalogSources are reflected in
the client set, and differences between CatalogSources and client set
membership are a potential source of error.

Instead, the catalog operator should list CatalogSources from its
informer cache to determine which catalogs are reachable from a given
namespace.

Signed-off-by: Ben Luddy <[email protected]>

Author: benluddy

pkg/controller/operators/catalog/operator.go

@@ -191,7 +191,7 @@ 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 = resolver.SourceProviderFromRegistryClientProvider(op.sources, lister.OperatorsV1alpha1().CatalogSourceLister(), logger)
 	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)

pkg/controller/registry/grpc/source.go

@@ -256,40 +256,16 @@ func (s *SourceStore) Remove(key registry.CatalogKey) error {
 	return source.Conn.Close()
 }
 
-func (s *SourceStore) AsClients(namespaces ...string) map[registry.CatalogKey]registry.ClientInterface {
-	refs := map[registry.CatalogKey]registry.ClientInterface{}
-	s.sourcesLock.RLock()
-	defer s.sourcesLock.RUnlock()
-	for key, source := range s.sources {
-		if source.LastConnect.IsZero() {
-			continue
-		}
-		for _, namespace := range namespaces {
-			if key.Namespace == namespace {
-				refs[key] = registry.NewClientFromConn(source.Conn)
-			}
-		}
-	}
-
-	// TODO : remove unhealthy
-	return refs
-}
-
 func (s *SourceStore) ClientsForNamespaces(namespaces ...string) map[registry.CatalogKey]client.Interface {
 	refs := map[registry.CatalogKey]client.Interface{}
 	s.sourcesLock.RLock()
 	defer s.sourcesLock.RUnlock()
 	for key, source := range s.sources {
-		if source.LastConnect.IsZero() {
-			continue
-		}
 		for _, namespace := range namespaces {
 			if key.Namespace == namespace {
 				refs[key] = client.NewClientFromConn(source.Conn)
 			}
 		}
 	}
-
-	// TODO : remove unhealthy
 	return refs
 }

pkg/controller/registry/resolver/source_registry.go

@@ -8,12 +8,15 @@ import (
 	"time"
 
 	"github.com/blang/semver/v4"
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	v1alpha1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver/cache"
 	"github.com/operator-framework/operator-registry/pkg/api"
 	"github.com/operator-framework/operator-registry/pkg/client"
 	opregistry "github.com/operator-framework/operator-registry/pkg/registry"
 	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/labels"
 )
 
 // todo: move to pkg/controller/operators/catalog
@@ -66,22 +69,41 @@ func (i *sourceInvalidator) GetValidChannel(key cache.SourceKey) <-chan struct{}
 
 type RegistrySourceProvider struct {
 	rcp         RegistryClientProvider
+	catsrcs     v1alpha1listers.CatalogSourceLister
 	logger      logrus.StdLogger
 	invalidator *sourceInvalidator
 }
 
-func SourceProviderFromRegistryClientProvider(rcp RegistryClientProvider, logger logrus.StdLogger) *RegistrySourceProvider {
+func SourceProviderFromRegistryClientProvider(rcp RegistryClientProvider, catsrcs v1alpha1listers.CatalogSourceLister, logger logrus.StdLogger) *RegistrySourceProvider {
 	return &RegistrySourceProvider{
-		rcp:    rcp,
-		logger: logger,
+		rcp:     rcp,
+		catsrcs: catsrcs,
+		logger:  logger,
 		invalidator: &sourceInvalidator{
 			validChans: make(map[cache.SourceKey]chan struct{}),
 			ttl:        5 * time.Minute,
 		},
 	}
 }
 
+type errorSource struct {
+	error
+}
+
+func (s errorSource) Snapshot(_ context.Context) (*cache.Snapshot, error) {
+	return nil, s.error
+}
+
 func (a *RegistrySourceProvider) Sources(namespaces ...string) map[cache.SourceKey]cache.Source {
+	var catalogs []*v1alpha1.CatalogSource
+	for _, namespace := range namespaces {
+		cats, err := a.catsrcs.CatalogSources(namespace).List(labels.Everything())
+		if err != nil {
+			// todo: this must be properly exposed
+			panic(err)
+		}
+		catalogs = append(catalogs, cats...)
+	}
 	result := make(map[cache.SourceKey]cache.Source)
 	for key, client := range a.rcp.ClientsForNamespaces(namespaces...) {
 		result[cache.SourceKey(key)] = &registrySource{
@@ -91,6 +113,14 @@ func (a *RegistrySourceProvider) Sources(namespaces ...string) map[cache.SourceK
 			invalidator: a.invalidator,
 		}
 	}
+	for _, cat := range catalogs {
+		key := cache.SourceKey{Name: cat.Name, Namespace: cat.Namespace}
+		if _, ok := result[key]; !ok {
+			result[key] = errorSource{
+				error: fmt.Errorf("registry server not reachable for catalogsource %s/%s", cat.Namespace, cat.Name),
+			}
+		}
+	}
 	return result
 }
 

pkg/controller/registry/resolver/step_resolver.go

@@ -57,7 +57,6 @@ func NewOperatorStepResolver(lister operatorlister.OperatorLister, client versio
 	cacheSourceProvider := &mergedSourceProvider{
 		sps: []cache.SourceProvider{
 			sourceProvider,
-			//SourceProviderFromRegistryClientProvider(provider, log),
 			&csvSourceProvider{
 				csvLister: lister.OperatorsV1alpha1().ClusterServiceVersionLister(),
 				subLister: lister.OperatorsV1alpha1().SubscriptionLister(),

test/e2e/subscription_e2e_test.go

@@ -19,6 +19,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
+	"k8s.io/apimachinery/pkg/api/errors"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -27,6 +28,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/operator-framework/api/pkg/lib/version"
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	operatorsv1alpha1 "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/controller/install"
@@ -62,6 +64,82 @@ var _ = Describe("Subscription", func() {
 		TeardownNamespace(generatedNamespace.GetName())
 	})
 
+	When("a registry server is not started for a grpc-type CatalogSource", func() {
+		var (
+			catsrc *v1alpha1.CatalogSource
+		)
+
+		BeforeEach(func() {
+			catsrc = &v1alpha1.CatalogSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace:    generatedNamespace.GetName(),
+					GenerateName: "without-registry-server-",
+				},
+				Spec: v1alpha1.CatalogSourceSpec{
+					SourceType: v1alpha1.SourceTypeGrpc,
+					Image:      "@", // bad image ref, pod creation will fail
+				},
+			}
+			Expect(ctx.Ctx().Client().Create(context.Background(), catsrc)).To(Succeed())
+		})
+
+		AfterEach(func() {
+			Eventually(func() error {
+				if catsrc == nil {
+					return nil
+				}
+				return ctx.Ctx().Client().Delete(context.Background(), catsrc)
+			}).Should(Or(
+				Succeed(),
+				WithTransform(errors.IsNotFound, BeTrue()),
+			))
+		})
+
+		It("should indicate ErrorPreventedResolution on a dependent Subscription status", func() {
+			sub := v1alpha1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace:    generatedNamespace.GetName(),
+					GenerateName: "test-subscription",
+				},
+				Spec: &v1alpha1.SubscriptionSpec{
+					Package: "whatever",
+				},
+			}
+			Expect(ctx.Ctx().Client().Create(context.Background(), &sub)).To(Succeed())
+
+			getCondition := func() (v1alpha1.SubscriptionCondition, error) {
+				if err := ctx.Ctx().Client().Get(context.Background(), client.ObjectKeyFromObject(&sub), &sub); err != nil {
+					return v1alpha1.SubscriptionCondition{}, err
+				}
+				cond := sub.Status.GetCondition(v1alpha1.SubscriptionResolutionFailed)
+				ctx.Ctx().Logf("%#v\n", cond)
+				return v1alpha1.SubscriptionCondition{
+					Type:   cond.Type,
+					Reason: cond.Reason,
+					Status: cond.Status,
+				}, nil
+			}
+
+			// this doesn't seem very robust. basically, this subscription condition should arrive directly at True/ErrorPreventedResolution without passing through any other non-Unknown states
+
+			Consistently(getCondition).Should(And(
+				Not(Equal(v1alpha1.SubscriptionCondition{
+					Type:   v1alpha1.SubscriptionResolutionFailed,
+					Reason: "ConstraintsNotSatisfiable",
+					Status: corev1.ConditionTrue,
+				})),
+			))
+
+			Eventually(getCondition).Should(
+				Equal(v1alpha1.SubscriptionCondition{
+					Type:   v1alpha1.SubscriptionResolutionFailed,
+					Reason: "ErrorPreventedResolution",
+					Status: corev1.ConditionTrue,
+				}),
+			)
+		})
+	})
+
 	When("an entry in the middle of a channel does not provide a required GVK", func() {
 		var (
 			teardown func()