Extract and unify cache bypass logic by creating a CacheProxy struct

Author: serathius

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/storage_factory.go

@@ -79,7 +79,7 @@ func StorageWithCacher() generic.StorageDecorator {
 			})
 		}
 
-		return cacher, destroyFunc, nil
+		return cacherstorage.NewCacheProxy(cacher, s), destroyFunc, nil
 	}
 }
 

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store_test.go

@@ -2459,7 +2459,7 @@ func newTestGenericStoreRegistry(t *testing.T, scheme *runtime.Scheme, hasCacheE
 			}
 		}
 		d := destroyFunc
-		s = cacher
+		s = cacherstorage.NewCacheProxy(cacher, s)
 		destroyFunc = func() {
 			cacher.Stop()
 			d()

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

@@ -339,10 +339,6 @@ type Cacher struct {
 	expiredBookmarkWatchers []*cacheWatcher
 }
 
-func (c *Cacher) RequestWatchProgress(ctx context.Context) error {
-	return c.storage.RequestWatchProgress(ctx)
-}
-
 // NewCacherFromConfig creates a new Cacher responsible for servicing WATCH and LIST requests from
 // its internal cache and updating its cache in the background based on the
 // given configuration.
@@ -495,56 +491,13 @@ func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
 	}
 }
 
-// Versioner implements storage.Interface.
-func (c *Cacher) Versioner() storage.Versioner {
-	return c.storage.Versioner()
-}
-
-// Create implements storage.Interface.
-func (c *Cacher) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
-	return c.storage.Create(ctx, key, obj, out, ttl)
-}
-
-// Delete implements storage.Interface.
-func (c *Cacher) Delete(
-	ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions,
-	validateDeletion storage.ValidateObjectFunc, _ runtime.Object, opts storage.DeleteOptions) error {
-	// Ignore the suggestion and try to pass down the current version of the object
-	// read from cache.
-	if elem, exists, err := c.watchCache.GetByKey(key); err != nil {
-		klog.Errorf("GetByKey returned error: %v", err)
-	} else if exists {
-		// DeepCopy the object since we modify resource version when serializing the
-		// current object.
-		currObj := elem.(*storeElement).Object.DeepCopyObject()
-		return c.storage.Delete(ctx, key, out, preconditions, validateDeletion, currObj, opts)
-	}
-	// If we couldn't get the object, fallback to no-suggestion.
-	return c.storage.Delete(ctx, key, out, preconditions, validateDeletion, nil, opts)
-}
-
 type namespacedName struct {
 	namespace string
 	name      string
 }
 
-// Watch implements storage.Interface.
 func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) {
 	pred := opts.Predicate
-	// if the watch-list feature wasn't set and the resourceVersion is unset
-	// ensure that the rv from which the watch is being served, is the latest
-	// one. "latest" is ensured by serving the watch from
-	// the underlying storage.
-	//
-	// it should never happen due to our validation but let's just be super-safe here
-	// and disable sendingInitialEvents when the feature wasn't enabled
-	if !utilfeature.DefaultFeatureGate.Enabled(features.WatchList) && opts.SendInitialEvents != nil {
-		opts.SendInitialEvents = nil
-	}
-	// TODO: we should eventually get rid of this legacy case
-	if utilfeature.DefaultFeatureGate.Enabled(features.WatchFromStorageWithoutResourceVersion) && opts.SendInitialEvents == nil && opts.ResourceVersion == "" {
-		return c.storage.Watch(ctx, key, opts)
-	}
 	requestedWatchRV, err := c.versioner.ParseResourceVersion(opts.ResourceVersion)
 	if err != nil {
 		return nil, err
@@ -709,58 +662,17 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 	return watcher, nil
 }
 
-// Get implements storage.Interface.
 func (c *Cacher) Get(ctx context.Context, key string, opts storage.GetOptions, objPtr runtime.Object) error {
-	ctx, span := tracing.Start(ctx, "cacher.Get",
-		attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)),
-		attribute.String("key", key),
-		attribute.String("resource-version", opts.ResourceVersion))
-	defer span.End(500 * time.Millisecond)
-	if opts.ResourceVersion == "" {
-		// If resourceVersion is not specified, serve it from underlying
-		// storage (for backward compatibility).
-		span.AddEvent("About to Get from underlying storage")
-		return c.storage.Get(ctx, key, opts, objPtr)
-	}
-
-	if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
-		if !c.ready.check() {
-			// If Cache is not initialized, delegate Get requests to storage
-			// as described in https://kep.k8s.io/4568
-			span.AddEvent("About to Get from underlying storage - cache not initialized")
-			return c.storage.Get(ctx, key, opts, objPtr)
-		}
-	}
-
-	// If resourceVersion is specified, serve it from cache.
-	// It's guaranteed that the returned value is at least that
-	// fresh as the given resourceVersion.
 	getRV, err := c.versioner.ParseResourceVersion(opts.ResourceVersion)
 	if err != nil {
 		return err
 	}
 
-	// Do not create a trace - it's not for free and there are tons
-	// of Get requests. We can add it if it will be really needed.
-
-	if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
-		if getRV == 0 && !c.ready.check() {
-			// If Cacher is not yet initialized and we don't require any specific
-			// minimal resource version, simply forward the request to storage.
-			span.AddEvent("About to Get from underlying storage - cache not initialized and no resourceVersion set")
-			return c.storage.Get(ctx, key, opts, objPtr)
-		}
-		if err := c.ready.wait(ctx); err != nil {
-			return errors.NewServiceUnavailable(err.Error())
-		}
-	}
-
 	objVal, err := conversion.EnforcePtr(objPtr)
 	if err != nil {
 		return err
 	}
 
-	span.AddEvent("About to fetch object from cache")
 	obj, exists, readResourceVersion, err := c.watchCache.WaitUntilFreshAndGet(ctx, getRV, key)
 	if err != nil {
 		return err
@@ -845,32 +757,9 @@ type listResp struct {
 }
 
 // GetList implements storage.Interface
-func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
+func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object, listRV uint64) error {
 	recursive := opts.Recursive
-	resourceVersion := opts.ResourceVersion
 	pred := opts.Predicate
-	if shouldDelegateList(opts) {
-		return c.storage.GetList(ctx, key, opts, listObj)
-	}
-
-	listRV, err := c.versioner.ParseResourceVersion(resourceVersion)
-	if err != nil {
-		return err
-	}
-
-	if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
-		if !c.ready.check() && shouldDelegateListOnNotReadyCache(opts) {
-			// If Cacher is not initialized, delegate List requests to storage
-			// as described in https://kep.k8s.io/4568
-			return c.storage.GetList(ctx, key, opts, listObj)
-		}
-	} else {
-		if listRV == 0 && !c.ready.check() {
-			// If Cacher is not yet initialized and we don't require any specific
-			// minimal resource version, simply forward the request to storage.
-			return c.storage.GetList(ctx, key, opts, listObj)
-		}
-	}
 	// For recursive lists, we need to make sure the key ended with "/" so that we only
 	// get children "directories". e.g. if we have key "/a", "/a/b", "/ab", getting keys
 	// with prefix "/a" will return all three, while with prefix "/a/" will return only
@@ -879,14 +768,6 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
 	if opts.Recursive && !strings.HasSuffix(key, "/") {
 		preparedKey += "/"
 	}
-	requestWatchProgressSupported := etcdfeature.DefaultFeatureSupportChecker.Supports(storage.RequestWatchProgress)
-	consistentRead := resourceVersion == "" && utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache) && requestWatchProgressSupported
-	if consistentRead {
-		listRV, err = storage.GetCurrentResourceVersionFromStorage(ctx, c.storage, c.newListFunc, c.resourcePrefix, c.objectType.String())
-		if err != nil {
-			return err
-		}
-	}
 
 	ctx, span := tracing.Start(ctx, "cacher.GetList",
 		attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)),
@@ -920,24 +801,9 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
 	}
 
 	resp, indexUsed, err := c.listItems(ctx, listRV, preparedKey, pred, recursive)
-	success := "true"
-	fallback := "false"
 	if err != nil {
-		if consistentRead {
-			if storage.IsTooLargeResourceVersion(err) {
-				fallback = "true"
-				err = c.storage.GetList(ctx, key, opts, listObj)
-			}
-			if err != nil {
-				success = "false"
-			}
-			metrics.ConsistentReadTotal.WithLabelValues(c.resourcePrefix, success, fallback).Add(1)
-		}
 		return err
 	}
-	if consistentRead {
-		metrics.ConsistentReadTotal.WithLabelValues(c.resourcePrefix, success, fallback).Add(1)
-	}
 	span.AddEvent("Listed items from cache", attribute.Int("count", len(resp.Items)))
 	// store pointer of eligible objects,
 	// Why not directly put object in the items of listObj?
@@ -987,37 +853,6 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
 	return nil
 }
 
-// GuaranteedUpdate implements storage.Interface.
-func (c *Cacher) GuaranteedUpdate(
-	ctx context.Context, key string, destination runtime.Object, ignoreNotFound bool,
-	preconditions *storage.Preconditions, tryUpdate storage.UpdateFunc, _ runtime.Object) error {
-	// Ignore the suggestion and try to pass down the current version of the object
-	// read from cache.
-	if elem, exists, err := c.watchCache.GetByKey(key); err != nil {
-		klog.Errorf("GetByKey returned error: %v", err)
-	} else if exists {
-		// DeepCopy the object since we modify resource version when serializing the
-		// current object.
-		currObj := elem.(*storeElement).Object.DeepCopyObject()
-		return c.storage.GuaranteedUpdate(ctx, key, destination, ignoreNotFound, preconditions, tryUpdate, currObj)
-	}
-	// If we couldn't get the object, fallback to no-suggestion.
-	return c.storage.GuaranteedUpdate(ctx, key, destination, ignoreNotFound, preconditions, tryUpdate, nil)
-}
-
-// Count implements storage.Interface.
-func (c *Cacher) Count(pathPrefix string) (int64, error) {
-	return c.storage.Count(pathPrefix)
-}
-
-// ReadinessCheck implements storage.Interface.
-func (c *Cacher) ReadinessCheck() error {
-	if !c.ready.check() {
-		return storage.ErrStorageNotReady
-	}
-	return nil
-}
-
 // baseObjectThreadUnsafe omits locking for cachingObject.
 func baseObjectThreadUnsafe(object runtime.Object) runtime.Object {
 	if co, ok := object.(*cachingObject); ok {
@@ -1494,6 +1329,10 @@ func (c *Cacher) setInitialEventsEndBookmarkIfRequested(cacheInterval *watchCach
 	}
 }
 
+func (c *Cacher) Ready() bool {
+	return c.ready.check()
+}
+
 // errWatcher implements watch.Interface to return a single error
 type errWatcher struct {
 	result chan watch.Event

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go

@@ -455,12 +455,12 @@ func withNodeNameAndNamespaceIndex(options *setupOptions) {
 	}
 }
 
-func testSetup(t *testing.T, opts ...setupOption) (context.Context, *Cacher, tearDownFunc) {
+func testSetup(t *testing.T, opts ...setupOption) (context.Context, *CacheProxy, tearDownFunc) {
 	ctx, cacher, _, tearDown := testSetupWithEtcdServer(t, opts...)
 	return ctx, cacher, tearDown
 }
 
-func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context, *Cacher, *etcd3testing.EtcdTestServer, tearDownFunc) {
+func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context, *CacheProxy, *etcd3testing.EtcdTestServer, tearDownFunc) {
 	setupOpts := setupOptions{}
 	opts = append([]setupOption{withDefaults}, opts...)
 	for _, opt := range opts {
@@ -514,31 +514,31 @@ func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context
 		}
 	}
 
-	return ctx, cacher, server, terminate
+	return ctx, NewCacheProxy(cacher, wrappedStorage), server, terminate
 }
 
 func testSetupWithEtcdAndCreateWrapper(t *testing.T, opts ...setupOption) (storage.Interface, tearDownFunc) {
 	_, cacher, _, tearDown := testSetupWithEtcdServer(t, opts...)
 
 	if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
-		if err := cacher.ready.wait(context.TODO()); err != nil {
+		if err := cacher.cacher.ready.wait(context.TODO()); err != nil {
 			t.Fatalf("unexpected error waiting for the cache to be ready")
 		}
 	}
-	return &createWrapper{Cacher: cacher}, tearDown
+	return &createWrapper{CacheProxy: cacher}, tearDown
 }
 
 type createWrapper struct {
-	*Cacher
+	*CacheProxy
 }
 
 func (c *createWrapper) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
-	if err := c.Cacher.Create(ctx, key, obj, out, ttl); err != nil {
+	if err := c.CacheProxy.Create(ctx, key, obj, out, ttl); err != nil {
 		return err
 	}
 	return wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) {
-		currentObj := c.Cacher.newFunc()
-		err := c.Cacher.Get(ctx, key, storage.GetOptions{ResourceVersion: "0"}, currentObj)
+		currentObj := c.CacheProxy.cacher.newFunc()
+		err := c.CacheProxy.Get(ctx, key, storage.GetOptions{ResourceVersion: "0"}, currentObj)
 		if err != nil {
 			if storage.IsNotFound(err) {
 				return false, nil

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_testing_utils_test.go

@@ -78,45 +78,45 @@ func computePodKey(obj *example.Pod) string {
 	return fmt.Sprintf("/pods/%s/%s", obj.Namespace, obj.Name)
 }
 
-func compactStorage(c *Cacher, client *clientv3.Client) storagetesting.Compaction {
+func compactStorage(c *CacheProxy, client *clientv3.Client) storagetesting.Compaction {
 	return func(ctx context.Context, t *testing.T, resourceVersion string) {
 		versioner := storage.APIObjectVersioner{}
 		rv, err := versioner.ParseResourceVersion(resourceVersion)
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		err = c.watchCache.waitUntilFreshAndBlock(context.TODO(), rv)
+		err = c.cacher.watchCache.waitUntilFreshAndBlock(context.TODO(), rv)
 		if err != nil {
 			t.Fatalf("WatchCache didn't caught up to RV: %v", rv)
 		}
-		c.watchCache.RUnlock()
+		c.cacher.watchCache.RUnlock()
 
-		c.watchCache.Lock()
-		defer c.watchCache.Unlock()
-		c.Lock()
-		defer c.Unlock()
+		c.cacher.watchCache.Lock()
+		defer c.cacher.watchCache.Unlock()
+		c.cacher.Lock()
+		defer c.cacher.Unlock()
 
-		if c.watchCache.resourceVersion < rv {
+		if c.cacher.watchCache.resourceVersion < rv {
 			t.Fatalf("Can't compact into a future version: %v", resourceVersion)
 		}
 
-		if len(c.watchers.allWatchers) > 0 || len(c.watchers.valueWatchers) > 0 {
+		if len(c.cacher.watchers.allWatchers) > 0 || len(c.cacher.watchers.valueWatchers) > 0 {
 			// We could consider terminating those watchers, but given
 			// watchcache doesn't really support compaction and we don't
 			// exercise it in tests, we just throw an error here.
 			t.Error("Open watchers are not supported during compaction")
 		}
 
-		for c.watchCache.startIndex < c.watchCache.endIndex {
-			index := c.watchCache.startIndex % c.watchCache.capacity
-			if c.watchCache.cache[index].ResourceVersion > rv {
+		for c.cacher.watchCache.startIndex < c.cacher.watchCache.endIndex {
+			index := c.cacher.watchCache.startIndex % c.cacher.watchCache.capacity
+			if c.cacher.watchCache.cache[index].ResourceVersion > rv {
 				break
 			}
 
-			c.watchCache.startIndex++
+			c.cacher.watchCache.startIndex++
 		}
-		c.watchCache.listResourceVersion = rv
+		c.cacher.watchCache.listResourceVersion = rv
 
 		if _, err = client.KV.Put(ctx, "compact_rev_key", resourceVersion); err != nil {
 			t.Fatalf("Could not update compact_rev_key: %v", err)