From 1ca34bd0d2db099298232d6ff1a217da2321677a Mon Sep 17 00:00:00 2001 From: Lukasz Szaszkiewicz Date: Wed, 29 Oct 2025 09:03:48 +0100 Subject: [PATCH] apiserver/pkg/storage/cacher/lister_watcher: pass RV for request from the watchlist consistency checker --- .../pkg/storage/cacher/lister_watcher.go | 45 ++++++++--- .../pkg/storage/cacher/lister_watcher_test.go | 77 +++++++++++++++++++ 2 files changed, 112 insertions(+), 10 deletions(-) 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 552c2d35fcd..e037ed4ab8d 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,26 +31,30 @@ import ( "k8s.io/apiserver/pkg/storage" utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/consistencydetector" "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 - unsupportedWatchListSemantics bool + storage storage.Interface + resourcePrefix string + newListFunc func() runtime.Object + contextMetadata metadata.MD + + unsupportedWatchListSemantics bool + watchListConsistencyCheckEnabled 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, - unsupportedWatchListSemantics: watchlist.DoesClientNotSupportWatchListSemantics(storage), + storage: storage, + resourcePrefix: resourcePrefix, + newListFunc: newListFunc, + contextMetadata: contextMetadata, + unsupportedWatchListSemantics: watchlist.DoesClientNotSupportWatchListSemantics(storage), + watchListConsistencyCheckEnabled: consistencydetector.IsDataConsistencyDetectionForWatchListEnabled(), } } @@ -69,6 +73,27 @@ func (lw *listerWatcher) List(options metav1.ListOptions) (runtime.Object, error Predicate: pred, Recursive: true, } + + // The ConsistencyChecker built into reflectors for the WatchList feature is responsible + // for verifying that the data received from the server (potentially from the watch cache) + // is consistent with the data stored in etcd. + // + // To perform this verification, the checker uses the ResourceVersion obtained from the initial request + // and sets the ResourceVersionMatch so that it retrieves exactly the same data directly from etcd. + // This allows comparing both data sources and confirming their consistency. + // + // The code below checks whether the incoming request originates from the ConsistencyChecker. + // If so, it allows explicitly setting the ResourceVersion. + // + // As of Oct 2025, reflector on its own is not setting RVM=Exact. + // However, even if that changes in the meantime, we would have to propagate that + // down to storage to ensure the correct semantics of the request. + watchListEnabled := utilfeature.DefaultFeatureGate.Enabled(features.WatchList) + supportedRVM := options.ResourceVersionMatch == metav1.ResourceVersionMatchExact + if watchListEnabled && lw.watchListConsistencyCheckEnabled && supportedRVM { + storageOpts.ResourceVersion = options.ResourceVersion + } + ctx := context.Background() if lw.contextMetadata != nil { ctx = metadata.NewOutgoingContext(ctx, lw.contextMetadata) 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 5e47a3fca3c..ec4317a2547 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 @@ -25,6 +25,7 @@ import ( "k8s.io/apimachinery/pkg/watch" "k8s.io/apiserver/pkg/apis/example" "k8s.io/apiserver/pkg/features" + "k8s.io/apiserver/pkg/storage" storagetesting "k8s.io/apiserver/pkg/storage/testing" utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/tools/cache" @@ -218,3 +219,79 @@ func TestCacherListerWatcherWhenListWatchDisabled(t *testing.T) { t.Fatalf("Expected error %q, but got %q", expectedErrMsg, err.Error()) } } + +func TestListerWatcherListResourceVersionPropagation(t *testing.T) { + scenarios := []struct { + name string + options metav1.ListOptions + watchListEnabled bool + watchListConsistencyCheckEnabled bool + expectedStorageResourceVer string + }{ + { + name: "WatchList FG disabled - RV not propagated", + options: metav1.ListOptions{ + ResourceVersion: "123", + ResourceVersionMatch: metav1.ResourceVersionMatchExact, + }, + watchListEnabled: false, + watchListConsistencyCheckEnabled: true, + expectedStorageResourceVer: "", + }, + { + name: "WatchList consistency check disabled - RV not propagated", + options: metav1.ListOptions{ + ResourceVersion: "123", + ResourceVersionMatch: metav1.ResourceVersionMatchExact, + }, + watchListEnabled: true, + watchListConsistencyCheckEnabled: false, + expectedStorageResourceVer: "", + }, + { + name: "Unsupported RVM - RV not propagated", + options: metav1.ListOptions{ + ResourceVersion: "123", + ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan, + }, + watchListEnabled: true, + watchListConsistencyCheckEnabled: true, + expectedStorageResourceVer: "", + }, + { + name: "all conditions satisfied - RV propagated", + options: metav1.ListOptions{ + ResourceVersion: "123", + ResourceVersionMatch: metav1.ResourceVersionMatchExact, + }, + watchListEnabled: true, + watchListConsistencyCheckEnabled: true, + expectedStorageResourceVer: "123", + }, + } + + for _, scenario := range scenarios { + t.Run(scenario.name, func(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchList, scenario.watchListEnabled) + + backingStorage := &dummyStorage{} + var capturedOpts storage.ListOptions + backingStorage.getListFn = func(ctx context.Context, resPrefix string, opts storage.ListOptions, listObj runtime.Object) error { + capturedOpts = opts + return nil + } + + targetInterface := NewListerWatcher(backingStorage, "/pods/", func() runtime.Object { return &example.PodList{} }, nil) + target := targetInterface.(*listerWatcher) + target.watchListConsistencyCheckEnabled = scenario.watchListConsistencyCheckEnabled + + if _, err := target.List(scenario.options); err != nil { + t.Fatalf("List returned error: %v", err) + } + + if capturedOpts.ResourceVersion != scenario.expectedStorageResourceVer { + t.Fatalf("expected storage ResourceVersion %q, got %q", scenario.expectedStorageResourceVer, capturedOpts.ResourceVersion) + } + }) + } +}