From 9902ae73872fe46481fe53af0d3af0f6725cdd95 Mon Sep 17 00:00:00 2001 From: Lukasz Szaszkiewicz Date: Tue, 21 Oct 2025 20:56:38 +0200 Subject: [PATCH] apiserver/pkg/storage/cacher/lister_watcher: exposes WatchList semantics --- .../pkg/storage/cacher/cacher_test.go | 9 +++++- .../storage/cacher/cacher_whitebox_test.go | 30 +++++++++++++++++-- .../pkg/storage/cacher/lister_watcher.go | 23 +++++++++----- .../pkg/storage/cacher/lister_watcher_test.go | 15 ++++++++++ 4 files changed, 66 insertions(+), 11 deletions(-) diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go index 31b4a249689..52d079fc855 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_test.go @@ -42,6 +42,7 @@ import ( storagetesting "k8s.io/apiserver/pkg/storage/testing" "k8s.io/apiserver/pkg/storage/value/encrypt/identity" utilfeature "k8s.io/apiserver/pkg/util/feature" + clientfeatures "k8s.io/client-go/features" "k8s.io/client-go/tools/cache" featuregatetesting "k8s.io/component-base/featuregate/testing" "k8s.io/klog/v2" @@ -568,9 +569,15 @@ func testSetupWithEtcdServer(t testing.TB, opts ...setupOption) (context.Context server, etcdStorage := newEtcdTestStorage(t, etcd3testing.PathPrefix()) // Inject one list error to make sure we test the relist case. + listErrors := 1 + if clientfeatures.FeatureGates().Enabled(clientfeatures.WatchListClient) { + // The WatchListClient feature changes the reflector to use WATCH + // instead of LIST, therefore we don't expect any errors + listErrors = 0 + } wrappedStorage := &storagetesting.StorageInjectingListErrors{ Interface: etcdStorage, - Errors: 1, + Errors: listErrors, } config := Config{ diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go index f3505293306..ec31426eb69 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go @@ -31,12 +31,12 @@ import ( "github.com/stretchr/testify/require" "google.golang.org/grpc/metadata" - "k8s.io/apimachinery/pkg/apis/meta/internalversion" - "k8s.io/apimachinery/pkg/apis/meta/internalversion/validation" apiequality "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/apis/meta/internalversion" + "k8s.io/apimachinery/pkg/apis/meta/internalversion/validation" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/fields" "k8s.io/apimachinery/pkg/labels" @@ -55,6 +55,8 @@ import ( etcdfeature "k8s.io/apiserver/pkg/storage/feature" storagetesting "k8s.io/apiserver/pkg/storage/testing" utilfeature "k8s.io/apiserver/pkg/util/feature" + clientfeatures "k8s.io/client-go/features" + clientfeaturestesting "k8s.io/client-go/features/testing" featuregatetesting "k8s.io/component-base/featuregate/testing" k8smetrics "k8s.io/component-base/metrics" "k8s.io/component-base/metrics/testutil" @@ -141,6 +143,10 @@ func (d *dummyStorage) CompactRevision() int64 { return 0 } +func (d *dummyStorage) IsWatchListSemanticsUnSupported() bool { + return true +} + type dummyWatch struct { ch chan watch.Event } @@ -3557,3 +3563,23 @@ func TestRetryAfterForUnreadyCache(t *testing.T) { t.Fatalf("Unexpected retry after: %v", statusError.Status().Details.RetryAfterSeconds) } } + +func TestWatchListSemanticsSimple(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchList, true) + clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.WatchListClient, true) + + // The dummyStore doesn’t support WatchList semantics, + // so we don’t need to prepare a response. + backingStorage := &dummyStorage{} + clock := testingclock.NewFakeClock(time.Now()) + cacher, _, err := newTestCacherWithoutSyncing(backingStorage, clock) + if err != nil { + t.Fatalf("couldn't create cacher: %v", err) + } + defer cacher.Stop() + ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second) + defer cancel() + if err = cacher.ready.wait(ctx); err != nil { + t.Fatalf("error waiting for the cache to be ready, err: %v", err) + } +} diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher.go index d5ac30152b6..552c2d35fcd 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher.go @@ -31,23 +31,26 @@ import ( "k8s.io/apiserver/pkg/storage" utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/watchlist" ) // listerWatcher opaques storage.Interface to expose cache.ListerWatcher. type listerWatcher struct { - storage storage.Interface - resourcePrefix string - newListFunc func() runtime.Object - contextMetadata metadata.MD + storage storage.Interface + resourcePrefix string + newListFunc func() runtime.Object + contextMetadata metadata.MD + unsupportedWatchListSemantics bool } // NewListerWatcher returns a storage.Interface backed ListerWatcher. func NewListerWatcher(storage storage.Interface, resourcePrefix string, newListFunc func() runtime.Object, contextMetadata metadata.MD) cache.ListerWatcher { return &listerWatcher{ - storage: storage, - resourcePrefix: resourcePrefix, - newListFunc: newListFunc, - contextMetadata: contextMetadata, + storage: storage, + resourcePrefix: resourcePrefix, + newListFunc: newListFunc, + contextMetadata: contextMetadata, + unsupportedWatchListSemantics: watchlist.DoesClientNotSupportWatchListSemantics(storage), } } @@ -105,3 +108,7 @@ func (lw *listerWatcher) Watch(options metav1.ListOptions) (watch.Interface, err return lw.storage.Watch(ctx, lw.resourcePrefix, opts) } + +func (lw *listerWatcher) IsWatchListSemanticsUnSupported() bool { + return lw.unsupportedWatchListSemantics +} diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher_test.go index cd4c6b8d543..5e47a3fca3c 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher_test.go @@ -28,10 +28,25 @@ import ( storagetesting "k8s.io/apiserver/pkg/storage/testing" utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/watchlist" featuregatetesting "k8s.io/component-base/featuregate/testing" "k8s.io/utils/ptr" ) +func TestDoesClientSupportWatchListSemanticsForKubeClient(t *testing.T) { + target1 := &dummyStorage{} + if !watchlist.DoesClientNotSupportWatchListSemantics(target1) { + t.Fatalf("dummyStorage should NOT support WatchList semantics") + } + + server, target2 := newEtcdTestStorage(t, "/pods/") + defer server.Terminate(t) + + if watchlist.DoesClientNotSupportWatchListSemantics(target2) { + t.Fatalf("etcd should support WatchList semantics") + } +} + func TestCacherListerWatcher(t *testing.T) { prefix := "/pods/" fn := func() runtime.Object { return &example.PodList{} }