catalog cache: retry cache population when cache contains an error (#1489)

Signed-off-by: Joe Lanford <joe.lanford@gmail.com>

Author: joelanford

internal/catalogmetadata/cache/cache.go

@@ -60,19 +60,6 @@ func (fsc *filesystemCache) Put(catalogName, resolvedRef string, source io.Reade
 	fsc.mutex.Lock()
 	defer fsc.mutex.Unlock()
 
-	// make sure we only write if this info hasn't been updated
-	// by another thread. The check here, if multiple threads are
-	// updating this, has no way to tell if the current ref is the
-	// newest possible ref. If another thread has already updated
-	// this to be the same value, skip the write logic and return
-	// the cached contents.
-	if cache, err := fsc.get(catalogName, resolvedRef); err == nil && cache != nil {
-		// We only return here if the was no error during
-		// the previous (likely concurrent) cache population attempt.
-		// If there was an error - we want to try and populate the cache again.
-		return cache, nil
-	}
-
 	var cacheFS fs.FS
 	if errToCache == nil {
 		cacheFS, errToCache = fsc.writeFS(catalogName, source)

internal/catalogmetadata/cache/cache_test.go

@@ -96,12 +96,10 @@ func TestFilesystemCachePutAndGet(t *testing.T) {
 	assert.NoError(t, equalFilesystems(actualFSPut, actualFSGet))
 
 	t.Log("Put v1 error into cache")
-	actualFSPut, err = c.Put(catalogName, resolvedRef1, nil, errors.New("fake put error"))
-	// Errors do not override previously successfully populated cache
-	require.NoError(t, err)
-	require.NotNil(t, actualFSPut)
-	assert.NoError(t, equalFilesystems(defaultFS(), actualFSPut))
-	assert.NoError(t, equalFilesystems(actualFSPut, actualFSGet))
+	actualFSPut, err = c.Put(catalogName, resolvedRef1, nil, errors.New("fake v1 put error"))
+	// Errors for an existing resolvedRef should override previously successfully populated cache
+	assert.Equal(t, err, errors.New("fake v1 put error"))
+	assert.Nil(t, actualFSPut)
 
 	t.Log("Put v2 error into cache")
 	actualFSPut, err = c.Put(catalogName, resolvedRef2, nil, errors.New("fake v2 put error"))

internal/controllers/clustercatalog_controller.go

@@ -24,6 +24,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	catalogd "github.com/operator-framework/catalogd/api/v1"
 )
@@ -47,6 +48,12 @@ type ClusterCatalogReconciler struct {
 //+kubebuilder:rbac:groups=olm.operatorframework.io,resources=clustercatalogs,verbs=get;list;watch
 
 func (r *ClusterCatalogReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	l := log.FromContext(ctx).WithName("cluster-catalog")
+	ctx = log.IntoContext(ctx, l)
+
+	l.Info("reconcile starting")
+	defer l.Info("reconcile ending")
+
 	existingCatalog := &catalogd.ClusterCatalog{}
 	err := r.Client.Get(ctx, req.NamespacedName, existingCatalog)
 	if apierrors.IsNotFound(err) {
@@ -70,9 +77,8 @@ func (r *ClusterCatalogReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 
 	catalogFsys, err := r.CatalogCache.Get(existingCatalog.Name, existingCatalog.Status.ResolvedSource.Image.Ref)
 	if err != nil {
-		return ctrl.Result{}, fmt.Errorf("error retrieving cache for catalog %q: %v", existingCatalog.Name, err)
-	}
-	if catalogFsys != nil {
+		l.Info("retrying cache population: found previous error from catalog cache", "cacheErr", err)
+	} else if catalogFsys != nil {
 		// Cache already exists so we do not need to populate it
 		return ctrl.Result{}, nil
 	}

internal/controllers/clustercatalog_controller_test.go

@@ -134,8 +134,8 @@ func TestClusterCatalogReconcilerFinalizers(t *testing.T) {
 					return nil, errors.New("fake error from cache get function")
 				},
 			},
-			wantGetCacheCalled: true,
-			wantErr:            "error retrieving cache for catalog",
+			wantGetCacheCalled:      true,
+			wantPopulateCacheCalled: true,
 		},
 		{
 			name: "catalog does not exist",

internal/controllers/clusterextension_controller.go

@@ -99,7 +99,7 @@ type InstalledBundleGetter interface {
 // The operator controller needs to watch all the bundle objects and reconcile accordingly. Though not ideal, but these permissions are required.
 // This has been taken from rukpak, and an issue was created before to discuss it: https://github.com/operator-framework/rukpak/issues/800.
 func (r *ClusterExtensionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	l := log.FromContext(ctx).WithName("operator-controller")
+	l := log.FromContext(ctx).WithName("cluster-extension")
 	ctx = log.IntoContext(ctx, l)
 
 	l.Info("reconcile starting")