Merge pull request kubernetes#130475 from serathius/watchcache-consistency

Implement consistency checking

Author: k8s-ci-robot

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

@@ -71,15 +71,17 @@ func StorageWithCacher() generic.StorageDecorator {
 		if err != nil {
 			return nil, func() {}, err
 		}
+		delegator := cacherstorage.NewCacheDelegator(cacher, s)
 		var once sync.Once
 		destroyFunc := func() {
 			once.Do(func() {
+				delegator.Stop()
 				cacher.Stop()
 				d()
 			})
 		}
 
-		return cacherstorage.NewCacheDelegator(cacher, s), destroyFunc, nil
+		return delegator, destroyFunc, nil
 	}
 }
 

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

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

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

@@ -729,7 +729,7 @@ type listResp struct {
 }
 
 // GetList implements storage.Interface
-func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object, listRV uint64) error {
+func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
 	// 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
@@ -738,6 +738,10 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
 	if opts.Recursive && !strings.HasSuffix(key, "/") {
 		preparedKey += "/"
 	}
+	listRV, err := c.versioner.ParseResourceVersion(opts.ResourceVersion)
+	if err != nil {
+		return err
+	}
 
 	ctx, span := tracing.Start(ctx, "cacher.GetList",
 		attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)),

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

@@ -484,10 +484,6 @@ func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context
 		t.Fatalf("Failed to initialize cacher: %v", err)
 	}
 	ctx := context.Background()
-	terminate := func() {
-		cacher.Stop()
-		server.Terminate(t)
-	}
 
 	// Since some tests depend on the fact that GetList shouldn't fail,
 	// we wait until the error from the underlying storage is consumed.
@@ -503,8 +499,14 @@ func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context
 			t.Fatal(err)
 		}
 	}
+	delegator := NewCacheDelegator(cacher, wrappedStorage)
+	terminate := func() {
+		delegator.Stop()
+		cacher.Stop()
+		server.Terminate(t)
+	}
 
-	return ctx, NewCacheDelegator(cacher, wrappedStorage), server, terminate
+	return ctx, delegator, server, terminate
 }
 
 func testSetupWithEtcdAndCreateWrapper(t *testing.T, opts ...setupOption) (storage.Interface, tearDownFunc) {

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

@@ -207,6 +207,15 @@ func (d *dummyStorage) GetCurrentResourceVersion(ctx context.Context) (uint64, e
 	return 100, nil
 }
 
+type dummyCacher struct {
+	dummyStorage
+	ready bool
+}
+
+func (d *dummyCacher) Ready() bool {
+	return d.ready
+}
+
 func TestGetListCacheBypass(t *testing.T) {
 	type opts struct {
 		ResourceVersion      string
@@ -326,6 +335,7 @@ func testGetListCacheBypass(t *testing.T, options storage.ListOptions, expectByp
 	}
 	defer cacher.Stop()
 	delegator := NewCacheDelegator(cacher, backingStorage)
+	defer delegator.Stop()
 	result := &example.PodList{}
 	if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
 		if err := cacher.ready.wait(context.Background()); err != nil {
@@ -450,6 +460,7 @@ apiserver_watch_cache_consistent_read_total{fallback="true", resource="pods", su
 			}
 			defer cacher.Stop()
 			delegator := NewCacheDelegator(cacher, backingStorage)
+			defer delegator.Stop()
 			if err := cacher.ready.wait(context.Background()); err != nil {
 				t.Fatalf("unexpected error waiting for the cache to be ready")
 			}
@@ -533,6 +544,7 @@ func TestGetListNonRecursiveCacheBypass(t *testing.T) {
 	}
 	defer cacher.Stop()
 	delegator := NewCacheDelegator(cacher, backingStorage)
+	defer delegator.Stop()
 
 	pred := storage.SelectionPredicate{
 		Limit: 500,
@@ -572,6 +584,7 @@ func TestGetCacheBypass(t *testing.T) {
 	}
 	defer cacher.Stop()
 	delegator := NewCacheDelegator(cacher, backingStorage)
+	defer delegator.Stop()
 
 	result := &example.Pod{}
 
@@ -608,6 +621,7 @@ func TestWatchCacheBypass(t *testing.T) {
 	}
 	defer cacher.Stop()
 	delegator := NewCacheDelegator(cacher, backingStorage)
+	defer delegator.Stop()
 
 	if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
 		if err := cacher.ready.wait(context.Background()); err != nil {
@@ -645,6 +659,7 @@ func TestTooManyRequestsNotReturned(t *testing.T) {
 	}
 	defer cacher.Stop()
 	delegator := NewCacheDelegator(cacher, backingStorage)
+	defer delegator.Stop()
 
 	opts := storage.ListOptions{
 		ResourceVersion: "0",
@@ -890,6 +905,7 @@ func TestCacherDontAcceptRequestsStopped(t *testing.T) {
 		t.Fatalf("Couldn't create cacher: %v", err)
 	}
 	delegator := NewCacheDelegator(cacher, backingStorage)
+	defer delegator.Stop()
 
 	if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
 		if err := cacher.ready.wait(context.Background()); err != nil {
@@ -2326,6 +2342,7 @@ func BenchmarkCacher_GetList(b *testing.B) {
 				}
 				defer cacher.Stop()
 				delegator := NewCacheDelegator(cacher, store)
+				defer delegator.Stop()
 
 				// prepare result and pred
 				parsedField, err := fields.ParseSelector("spec.nodeName=node-0")
@@ -3207,6 +3224,7 @@ func TestRetryAfterForUnreadyCache(t *testing.T) {
 	}
 	result := &example.PodList{}
 	delegator := NewCacheDelegator(cacher, backingStorage)
+	defer delegator.Stop()
 	err = delegator.GetList(context.TODO(), "/pods/ns", opts, result)
 
 	if !apierrors.IsTooManyRequests(err) {