Expose errors generated while retrieving catalog content.

Failing to fetch catalog content should not silently return an empty
cache. Instead, it should fail outright with an error that indicates
which catalog(s) could not be fetched.

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

Author: benluddy

pkg/controller/registry/resolver/cache.go

@@ -8,6 +8,8 @@ import (
 	"sync"
 	"time"
 
+	"k8s.io/apimachinery/pkg/util/errors"
+
 	"github.com/sirupsen/logrus"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
@@ -79,6 +81,19 @@ type NamespacedOperatorCache struct {
 	snapshots  map[registry.CatalogKey]*CatalogSnapshot
 }
 
+func (c *NamespacedOperatorCache) Error() error {
+	var errs []error
+	for key, snapshot := range c.snapshots {
+		snapshot.m.Lock()
+		err := snapshot.err
+		snapshot.m.Unlock()
+		if err != nil {
+			errs = append(errs, fmt.Errorf("error using catalog %s (in namespace %s): %w", key.Name, key.Namespace, err))
+		}
+	}
+	return errors.NewAggregate(errs)
+}
+
 func (c *OperatorCache) Expire(catalog registry.CatalogKey) {
 	c.m.Lock()
 	defer c.m.Unlock()
@@ -183,6 +198,12 @@ func (c *OperatorCache) Namespaced(namespaces ...string) MultiCatalogOperatorFin
 
 func (c *OperatorCache) populate(ctx context.Context, snapshot *CatalogSnapshot, registry client.Interface) {
 	defer snapshot.m.Unlock()
+	defer func() {
+		// Don't cache an errorred snapshot.
+		if snapshot.err != nil {
+			snapshot.expiry = time.Time{}
+		}
+	}()
 
 	c.sem <- struct{}{}
 	defer func() { <-c.sem }()
@@ -195,6 +216,7 @@ func (c *OperatorCache) populate(ctx context.Context, snapshot *CatalogSnapshot,
 	it, err := registry.ListBundles(ctx)
 	if err != nil {
 		snapshot.logger.Errorf("failed to list bundles: %s", err.Error())
+		snapshot.err = err
 		return
 	}
 	c.logger.WithField("catalog", snapshot.key.String()).Debug("updating cache")
@@ -223,6 +245,7 @@ func (c *OperatorCache) populate(ctx context.Context, snapshot *CatalogSnapshot,
 	}
 	if err := it.Error(); err != nil {
 		snapshot.logger.Warnf("error encountered while listing bundles: %s", err.Error())
+		snapshot.err = err
 	}
 	snapshot.operators = operators
 }
@@ -293,6 +316,7 @@ type CatalogSnapshot struct {
 	m         sync.RWMutex
 	pop       context.CancelFunc
 	priority  catalogSourcePriority
+	err       error
 }
 
 func (s *CatalogSnapshot) Cancel() {
@@ -400,6 +424,7 @@ type MultiCatalogOperatorFinder interface {
 	Catalog(registry.CatalogKey) OperatorFinder
 	FindPreferred(*registry.CatalogKey, ...OperatorPredicate) []*Operator
 	WithExistingOperators(*CatalogSnapshot) MultiCatalogOperatorFinder
+	Error() error
 	OperatorFinder
 }
 

pkg/controller/registry/resolver/cache_test.go

@@ -2,6 +2,7 @@ package resolver
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math/rand"
@@ -10,6 +11,7 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -35,6 +37,8 @@ func (s *BundleStreamStub) Recv() (*api.Bundle, error) {
 
 type RegistryClientStub struct {
 	BundleIterator *client.BundleIterator
+
+	ListBundlesError error
 }
 
 func (s *RegistryClientStub) Get() (client.Interface, error) {
@@ -58,7 +62,7 @@ func (s *RegistryClientStub) GetBundleThatProvides(ctx context.Context, group, v
 }
 
 func (s *RegistryClientStub) ListBundles(ctx context.Context) (*client.BundleIterator, error) {
-	return s.BundleIterator, nil
+	return s.BundleIterator, s.ListBundlesError
 }
 
 func (s *RegistryClientStub) GetPackage(ctx context.Context, packageName string) (*api.Package, error) {
@@ -339,3 +343,21 @@ func TestStripPluralRequiredAndProvidedAPIKeys(t *testing.T) {
 	assert.Equal(t, "K.v1.g", result[0].providedAPIs.String())
 	assert.Equal(t, "K2.v2.g2", result[0].requiredAPIs.String())
 }
+
+func TestNamespaceOperatorCacheError(t *testing.T) {
+	rcp := RegistryClientProviderStub{}
+	catsrcLister := operatorlister.NewLister().OperatorsV1alpha1().CatalogSourceLister()
+	key := registry.CatalogKey{Namespace: "dummynamespace", Name: "dummyname"}
+	rcp[key] = &RegistryClientStub{
+		ListBundlesError: errors.New("testing"),
+	}
+
+	logger, _ := test.NewNullLogger()
+	c := NewOperatorCache(rcp, logger, catsrcLister)
+	require.EqualError(t, c.Namespaced("dummynamespace").Error(), "error using catalog dummyname (in namespace dummynamespace): testing")
+	if snapshot, ok := c.snapshots[key]; !ok {
+		t.Fatalf("cache snapshot not found")
+	} else {
+		require.Zero(t, snapshot.expiry)
+	}
+}

pkg/controller/registry/resolver/resolver.go

@@ -103,6 +103,10 @@ func (r *SatResolver) SolveOperators(namespaces []string, csvs []*v1alpha1.Clust
 
 	r.addInvariants(namespacedCache, installables)
 
+	if err := namespacedCache.Error(); err != nil {
+		return nil, err
+	}
+
 	input := make([]solver.Installable, 0)
 	for _, i := range installables {
 		input = append(input, i)

pkg/controller/registry/types.go

@@ -33,7 +33,7 @@ func (k *CatalogKey) Virtual() bool {
 
 func NewVirtualCatalogKey(namespace string) CatalogKey {
 	return CatalogKey{
-		Name: ExistingOperatorKey,
+		Name:      ExistingOperatorKey,
 		Namespace: namespace,
 	}
 }