Skip to content

OCPBUGS-44813: UPSTREAM: <drop>: catalog cache: retry cache population when cache contains an error#197

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
openshift-merge-bot merged 1 commit into from
Nov 21, 2024
Merged

OCPBUGS-44813: UPSTREAM: <drop>: catalog cache: retry cache population when cache contains an error #197

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 0 additions & 13 deletions internal/catalogmetadata/cache/cache.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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
}

Comment on lines -63 to -75

@joelanfordjoelanfordNov 21, 2024

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Multiple threads can't be simultaneously Put-ing because we hold a write lock for the duration of the Put call. This is leftover from a time when we performed this read with a read lock and then proceeded to a write lock.

We also want to do the expected thing and always update the cache with the latest Put, which ensures eventual consistency when combined with the updates made to the reconciler.

var cacheFS fs.FS
if errToCache == nil {
cacheFS, errToCache = fsc.writeFS(catalogName, source)
Expand Down
10 changes: 4 additions & 6 deletions internal/catalogmetadata/cache/cache_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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)
Comment on lines +99 to +102

@joelanfordjoelanfordNov 21, 2024

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With the changes in internal/catalogmetadata/cache/cache.go, we now expect Put to always update the cache, even for the same resolvedRef.


t.Log("Put v2 error into cache")
actualFSPut, err = c.Put(catalogName, resolvedRef2, nil, errors.New("fake v2 put error"))
Expand Down
12 changes: 9 additions & 3 deletions internal/controllers/clustercatalog_controller.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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"
)
Expand All@@ -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) {
Expand All@@ -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
}
Expand Down
4 changes: 2 additions & 2 deletions internal/controllers/clustercatalog_controller_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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",
Expand Down
2 changes: 1 addition & 1 deletion internal/controllers/clusterextension_controller.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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")
Expand Down