Merge pull request #134941 from p0lyn0mial/upstream-watchlist-unsupported-wl-semantics-cacher-consistency-checker

apiserver/pkg/storage/cacher/lister_watcher: pass RV for request from the watchlist consistency checker Kubernetes-commit: d8f8b017e19af984908153aa1c66eb563bdc5940

Author: k8s-publishing-bot

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)

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)
+}
+})
+}
+}