diff --git a/go.mod b/go.mod index ddfb3a73b..a65d0d5d9 100644 --- a/go.mod +++ b/go.mod @@ -44,9 +44,9 @@ require ( gopkg.in/evanphx/json-patch.v4 v4.12.0 gopkg.in/natefinch/lumberjack.v2 v2.2.1 gopkg.in/square/go-jose.v2 v2.6.0 - k8s.io/api v0.0.0-20240531003526-c114cd746b5a + k8s.io/api v0.0.0-20240605203554-c0840f2e39d3 k8s.io/apimachinery v0.0.0-20240603234208-703232ea6da4 - k8s.io/client-go v0.0.0-20240604003933-7c6e307a725f + k8s.io/client-go v0.0.0-20240607044045-1c075dc88cc6 k8s.io/component-base v0.0.0-20240531004836-3486e8d18caf k8s.io/klog/v2 v2.120.1 k8s.io/kms v0.0.0-20240528085127-26a6cff8e4bf diff --git a/go.sum b/go.sum index 9e568fbb7..0ce11f156 100644 --- a/go.sum +++ b/go.sum @@ -380,12 +380,12 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= -k8s.io/api v0.0.0-20240531003526-c114cd746b5a h1:P8nQ3iz4FxeKN26Y8fz9qoEkUZy/DkQPtBmkVyriAS0= -k8s.io/api v0.0.0-20240531003526-c114cd746b5a/go.mod h1:2VfykmUr8OqDStfcJWPvSW182MtoxAMeWsJHXwxqzXo= +k8s.io/api v0.0.0-20240605203554-c0840f2e39d3 h1:xyY+alozd0hMtW6l/nPRvuKw++ci+kcETN3ncQ/9XkY= +k8s.io/api v0.0.0-20240605203554-c0840f2e39d3/go.mod h1:A81Hv2cdUeI+qLTFOIAFAg/t5jV5W88m9ygoJiL4vxI= k8s.io/apimachinery v0.0.0-20240603234208-703232ea6da4 h1:On2XjWF6loaEJ/GLZH+ZyovYHWfOZUgl9qUdC8Mn05o= k8s.io/apimachinery v0.0.0-20240603234208-703232ea6da4/go.mod h1:ClkKrTMwhmMjgsEHpX2w3F+YKj0ctDOaAxqL7clxG0U= -k8s.io/client-go v0.0.0-20240604003933-7c6e307a725f h1:7A1fNtVKUFjDecsZm2r/Ec3DtdVhuylFo6Hst/cD+HI= -k8s.io/client-go v0.0.0-20240604003933-7c6e307a725f/go.mod h1:5farkvv+2p9AXn5JduImtwwykMh2JcZQO9mlYXp2qEA= +k8s.io/client-go v0.0.0-20240607044045-1c075dc88cc6 h1:57zN8hshN7m8UlDCxzYIdSGilnij3Jy0hDr5SC8ETU8= +k8s.io/client-go v0.0.0-20240607044045-1c075dc88cc6/go.mod h1:XRxDCxqxB6eB3UE/8p7agi/qt7e7WP0nfmmTKP/R4f4= k8s.io/component-base v0.0.0-20240531004836-3486e8d18caf h1:+fSEnCvnEYfGmGtp8/KAmuCG2H/DG91PNcbneTJTRe8= k8s.io/component-base v0.0.0-20240531004836-3486e8d18caf/go.mod h1:+RdT910JB0Ke8zGWTWKLba1mVLmboEEJzaj5m6AMT+o= k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw= diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go index c73063a2e..c25296590 100644 --- a/pkg/features/kube_features.go +++ b/pkg/features/kube_features.go @@ -173,6 +173,13 @@ const ( // to a chunking list request. RemainingItemCount featuregate.Feature = "RemainingItemCount" + // owner: @wojtek-t + // beta: v1.31 + // + // Enables resilient watchcache initialization to avoid controlplane + // overload. + ResilientWatchCacheInitialization featuregate.Feature = "ResilientWatchCacheInitialization" + // owner: @serathius // beta: v1.30 // @@ -353,6 +360,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS RemainingItemCount: {Default: true, PreRelease: featuregate.GA, LockToDefault: true}, // remove in 1.32 + ResilientWatchCacheInitialization: {Default: true, PreRelease: featuregate.Beta}, + RetryGenerateName: {Default: true, PreRelease: featuregate.Beta}, SeparateCacheWatchRPC: {Default: true, PreRelease: featuregate.Beta}, diff --git a/pkg/storage/cacher/cacher.go b/pkg/storage/cacher/cacher.go index c9a6d51ed..a9230e112 100644 --- a/pkg/storage/cacher/cacher.go +++ b/pkg/storage/cacher/cacher.go @@ -532,9 +532,18 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions return nil, err } - readyGeneration, err := c.ready.waitAndReadGeneration(ctx) - if err != nil { - return nil, errors.NewServiceUnavailable(err.Error()) + var readyGeneration int + if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + var ok bool + readyGeneration, ok = c.ready.checkAndReadGeneration() + if !ok { + return nil, errors.NewTooManyRequests("storage is (re)initializing", 1) + } + } else { + readyGeneration, err = c.ready.waitAndReadGeneration(ctx) + if err != nil { + return nil, errors.NewServiceUnavailable(err.Error()) + } } // determine the namespace and name scope of the watch, first from the request, secondarily from the field selector @@ -676,6 +685,14 @@ func (c *Cacher) Get(ctx context.Context, key string, opts storage.GetOptions, o 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 + 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. @@ -684,16 +701,18 @@ func (c *Cacher) Get(ctx context.Context, key string, opts storage.GetOptions, o return err } - 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. - return c.storage.Get(ctx, key, opts, objPtr) - } - // 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 err := c.ready.wait(ctx); err != nil { - return errors.NewServiceUnavailable(err.Error()) + + 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. + 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) @@ -743,6 +762,14 @@ func shouldDelegateList(opts storage.ListOptions) bool { return consistentReadFromStorage || hasContinuation || hasLimit || unsupportedMatch } +func shouldDelegateListOnNotReadyCache(opts storage.ListOptions) bool { + pred := opts.Predicate + noLabelSelector := pred.Label == nil || pred.Label.Empty() + noFieldSelector := pred.Field == nil || pred.Field.Empty() + hasLimit := pred.Limit > 0 + return noLabelSelector && noFieldSelector && hasLimit +} + func (c *Cacher) listItems(ctx context.Context, listRV uint64, key string, pred storage.SelectionPredicate, recursive bool) ([]interface{}, uint64, string, error) { if !recursive { obj, exists, readResourceVersion, err := c.watchCache.WaitUntilFreshAndGet(ctx, listRV, key) @@ -770,10 +797,19 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio if err != nil { return err } - 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) + + 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) + } } requestWatchProgressSupported := etcdfeature.DefaultFeatureSupportChecker.Supports(storage.RequestWatchProgress) if resourceVersion == "" && utilfeature.DefaultFeatureGate.Enabled(features.ConsistentListFromCache) && requestWatchProgressSupported { @@ -788,8 +824,16 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio attribute.Stringer("type", c.groupResource)) defer span.End(500 * time.Millisecond) - if err := c.ready.wait(ctx); err != nil { - return errors.NewServiceUnavailable(err.Error()) + if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if !c.ready.check() { + // If Cacher is not initialized, reject List requests + // as described in https://kep.k8s.io/4568 + return errors.NewTooManyRequests("storage is (re)initializing", 1) + } + } else { + if err := c.ready.wait(ctx); err != nil { + return errors.NewServiceUnavailable(err.Error()) + } } span.AddEvent("Ready") diff --git a/pkg/storage/cacher/cacher_test.go b/pkg/storage/cacher/cacher_test.go index 1b39a8d87..cdd318a9d 100644 --- a/pkg/storage/cacher/cacher_test.go +++ b/pkg/storage/cacher/cacher_test.go @@ -464,14 +464,25 @@ func testSetupWithEtcdServer(t *testing.T, opts ...setupOption) (context.Context t.Fatalf("Failed to inject list errors: %v", err) } + if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + // The tests assume that Get/GetList/Watch calls shouldn't fail. + // However, 429 error can now be returned if watchcache is under initialization. + // To avoid rewriting all tests, we wait for watcache to initialize. + if err := cacher.ready.wait(ctx); err != nil { + t.Fatal(err) + } + } + return ctx, cacher, server, terminate } func testSetupWithEtcdAndCreateWrapper(t *testing.T, opts ...setupOption) (storage.Interface, tearDownFunc) { _, cacher, _, tearDown := testSetupWithEtcdServer(t, opts...) - if err := cacher.ready.wait(context.TODO()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.TODO()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } return &createWrapper{Cacher: cacher}, tearDown } diff --git a/pkg/storage/cacher/cacher_whitebox_test.go b/pkg/storage/cacher/cacher_whitebox_test.go index cc34ced7f..0407fafbe 100644 --- a/pkg/storage/cacher/cacher_whitebox_test.go +++ b/pkg/storage/cacher/cacher_whitebox_test.go @@ -53,7 +53,7 @@ import ( "k8s.io/utils/pointer" ) -func newTestCacher(s storage.Interface) (*Cacher, storage.Versioner, error) { +func newTestCacherWithoutSyncing(s storage.Interface) (*Cacher, storage.Versioner, error) { prefix := "pods" config := Config{ Storage: s, @@ -79,9 +79,27 @@ func newTestCacher(s storage.Interface) (*Cacher, storage.Versioner, error) { Clock: clock.RealClock{}, } cacher, err := NewCacherFromConfig(config) + return cacher, storage.APIObjectVersioner{}, err } +func newTestCacher(s storage.Interface) (*Cacher, storage.Versioner, error) { + cacher, versioner, err := newTestCacherWithoutSyncing(s) + if err != nil { + return nil, versioner, err + } + + if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + // The tests assume that Get/GetList/Watch calls shouldn't fail. + // However, 429 error can now be returned if watchcache is under initialization. + // To avoid rewriting all tests, we wait for watcache to initialize. + if err := cacher.ready.wait(context.Background()); err != nil { + return nil, storage.APIObjectVersioner{}, err + } + } + return cacher, versioner, nil +} + type dummyStorage struct { sync.RWMutex err error @@ -222,10 +240,12 @@ func testGetListCacheBypass(t *testing.T, options storage.ListOptions, expectByp result := &example.PodList{} - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } + // Inject error to underlying layer and check if cacher is not bypassed. backingStorage.getListFn = func(_ context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error { currentResourceVersion := "42" @@ -267,9 +287,10 @@ func TestGetListNonRecursiveCacheBypass(t *testing.T) { } result := &example.PodList{} - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } // Inject error to underlying layer and check if cacher is not bypassed. @@ -301,9 +322,10 @@ func TestGetCacheBypass(t *testing.T) { result := &example.Pod{} - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } // Inject error to underlying layer and check if cacher is not bypassed. @@ -333,9 +355,10 @@ func TestWatchCacheBypass(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } _, err = cacher.Watch(context.TODO(), "pod/ns", storage.ListOptions{ @@ -375,6 +398,43 @@ func TestWatchCacheBypass(t *testing.T) { } } +func TestTooManyRequestsNotReturned(t *testing.T) { + // Ensure that with ResilientWatchCacheInitialization feature disabled, we don't return 429 + // errors when watchcache is not initialized. + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ResilientWatchCacheInitialization, false) + + dummyErr := fmt.Errorf("dummy") + backingStorage := &dummyStorage{err: dummyErr} + cacher, _, err := newTestCacherWithoutSyncing(backingStorage) + if err != nil { + t.Fatalf("Couldn't create cacher: %v", err) + } + defer cacher.Stop() + + opts := storage.ListOptions{ + ResourceVersion: "0", + Predicate: storage.Everything, + } + + // Cancel the request so that it doesn't hang forever. + listCtx, listCancel := context.WithTimeout(context.Background(), 250*time.Millisecond) + defer listCancel() + + result := &example.PodList{} + err = cacher.GetList(listCtx, "/pods/ns", opts, result) + if err != nil && apierrors.IsTooManyRequests(err) { + t.Errorf("Unexpected 429 error without ResilientWatchCacheInitialization feature for List") + } + + watchCtx, watchCancel := context.WithTimeout(context.Background(), 250*time.Millisecond) + defer watchCancel() + + _, err = cacher.Watch(watchCtx, "/pods/ns", opts) + if err != nil && apierrors.IsTooManyRequests(err) { + t.Errorf("Unexpected 429 error without ResilientWatchCacheInitialization feature for Watch") + } +} + func TestEmptyWatchEventCache(t *testing.T) { server, etcdStorage := newEtcdTestStorage(t, etcd3testing.PathPrefix()) defer server.Terminate(t) @@ -471,7 +531,7 @@ func TestWatchNotHangingOnStartupFailure(t *testing.T) { // constantly failing lists to the underlying storage. dummyErr := fmt.Errorf("dummy") backingStorage := &dummyStorage{err: dummyErr} - cacher, _, err := newTestCacher(backingStorage) + cacher, _, err := newTestCacherWithoutSyncing(backingStorage) if err != nil { t.Fatalf("Couldn't create cacher: %v", err) } @@ -489,8 +549,14 @@ func TestWatchNotHangingOnStartupFailure(t *testing.T) { // Ensure that it terminates when its context is cancelled // (e.g. the request is terminated for whatever reason). _, err = cacher.Watch(ctx, "pods/ns", storage.ListOptions{ResourceVersion: "0"}) - if err == nil || err.Error() != apierrors.NewServiceUnavailable(context.Canceled.Error()).Error() { - t.Errorf("Unexpected error: %#v", err) + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err == nil || err.Error() != apierrors.NewServiceUnavailable(context.Canceled.Error()).Error() { + t.Errorf("Unexpected error: %#v", err) + } + } else { + if err == nil || err.Error() != apierrors.NewTooManyRequests("storage is (re)initializing", 1).Error() { + t.Errorf("Unexpected error: %#v", err) + } } } @@ -502,9 +568,10 @@ func TestWatcherNotGoingBackInTime(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } // Ensure there is some budget for slowing down processing. @@ -588,9 +655,10 @@ func TestCacherDontAcceptRequestsStopped(t *testing.T) { t.Fatalf("Couldn't create cacher: %v", err) } - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } w, err := cacher.Watch(context.Background(), "pods/ns", storage.ListOptions{ResourceVersion: "0", Predicate: storage.Everything}) @@ -623,17 +691,32 @@ func TestCacherDontAcceptRequestsStopped(t *testing.T) { IgnoreNotFound: true, ResourceVersion: "1", }, result) - if err == nil { - t.Fatalf("Success to create Get: %v", err) + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err == nil { + t.Fatalf("Success to create Get: %v", err) + } + } else { + if err != nil { + t.Fatalf("Failed to get object: %v:", err) + } } listResult := &example.PodList{} err = cacher.GetList(context.TODO(), "pods/ns", storage.ListOptions{ ResourceVersion: "1", Recursive: true, + Predicate: storage.SelectionPredicate{ + Limit: 500, + }, }, listResult) - if err == nil { - t.Fatalf("Success to create GetList: %v", err) + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err == nil { + t.Fatalf("Success to create GetList: %v", err) + } + } else { + if err != nil { + t.Fatalf("Failed to list objects: %v", err) + } } select { @@ -762,10 +845,12 @@ func TestCacherNoLeakWithMultipleWatchers(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } + pred := storage.Everything pred.AllowWatchBookmarks = true @@ -841,9 +926,10 @@ func testCacherSendBookmarkEvents(t *testing.T, allowWatchBookmarks, expectedBoo } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } pred := storage.Everything pred.AllowWatchBookmarks = allowWatchBookmarks @@ -941,9 +1027,10 @@ func TestCacherSendsMultipleWatchBookmarks(t *testing.T) { // resolution how frequency we recompute. cacher.bookmarkWatchers.bookmarkFrequency = time.Second - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } pred := storage.Everything pred.AllowWatchBookmarks = true @@ -1011,9 +1098,10 @@ func TestDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } // Ensure there is some budget for slowing down processing. @@ -1089,9 +1177,10 @@ func TestBookmarksOnResourceVersionUpdates(t *testing.T) { // Ensure that bookmarks are sent more frequently than every 1m. cacher.bookmarkWatchers = newTimeBucketWatchers(clock.RealClock{}, 2*time.Second) - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } makePod := func(i int) *examplev1.Pod { @@ -1167,9 +1256,10 @@ func TestStartingResourceVersion(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } // Ensure there is some budget for slowing down processing. @@ -1247,9 +1337,10 @@ func TestDispatchEventWillNotBeBlockedByTimedOutWatcher(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } // Ensure there is some budget for slowing down processing. @@ -1389,9 +1480,10 @@ func TestCachingDeleteEvents(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } fooPredicate := storage.SelectionPredicate{ @@ -1471,9 +1563,10 @@ func testCachingObjects(t *testing.T, watchersCount int) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } dispatchedEvents := []*watchCacheEvent{} @@ -1567,10 +1660,12 @@ func TestCacheIntervalInvalidationStopsWatch(t *testing.T) { } defer cacher.Stop() - // Wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } + // Ensure there is enough budget for slow processing since // the entire watch cache is going to be served through the // interval and events won't be popped from the cacheWatcher's @@ -1754,8 +1849,11 @@ func TestWaitUntilWatchCacheFreshAndForceAllEvents(t *testing.T) { t.Fatalf("Couldn't create cacher: %v", err) } defer cacher.Stop() - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } w, err := cacher.Watch(context.Background(), "pods/ns", scenario.opts) @@ -1911,9 +2009,10 @@ func TestWatchListIsSynchronisedWhenNoEventsFromStoreReceived(t *testing.T) { require.NoError(t, err, "failed to create cacher") defer cacher.Stop() - // wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } pred := storage.Everything @@ -1942,9 +2041,10 @@ func TestForgetWatcher(t *testing.T) { require.NoError(t, err) defer cacher.Stop() - // wait until cacher is initialized. - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } assertCacherInternalState := func(expectedWatchersCounter, expectedValueWatchersCounter int) { @@ -2334,8 +2434,11 @@ func TestGetBookmarkAfterResourceVersionLockedFunc(t *testing.T) { require.NoError(t, err, "couldn't create cacher") defer cacher.Stop() - if err := cacher.ready.wait(context.Background()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.Background()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } cacher.watchCache.UpdateResourceVersion(fmt.Sprintf("%d", scenario.watchCacheResourceVersion)) @@ -2395,8 +2498,11 @@ func TestWatchStreamSeparation(t *testing.T) { featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.SeparateCacheWatchRPC, tc.separateCacheWatchRPC) _, cacher, _, terminate := testSetupWithEtcdServer(t) t.Cleanup(terminate) - if err := cacher.ready.wait(context.TODO()); err != nil { - t.Fatalf("unexpected error waiting for the cache to be ready") + + if !utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) { + if err := cacher.ready.wait(context.TODO()); err != nil { + t.Fatalf("unexpected error waiting for the cache to be ready") + } } getCacherRV := func() uint64 {