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 b3b215ba3ac..d5ac30152b6 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 @@ -18,6 +18,7 @@ package cacher import ( "context" + "fmt" "google.golang.org/grpc/metadata" @@ -26,7 +27,9 @@ import ( "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/features" "k8s.io/apiserver/pkg/storage" + utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/tools/cache" ) @@ -88,5 +91,17 @@ func (lw *listerWatcher) Watch(options metav1.ListOptions) (watch.Interface, err if lw.contextMetadata != nil { ctx = metadata.NewOutgoingContext(ctx, lw.contextMetadata) } + + // we need the below check because the listWatcher bypasses the REST layer, + // so the options are not validated. Without this, we might end up in a situation + // where streaming is requested, but the FeatureGate is disabled, + // and the bookmark will not be sent + // + // in such a case, client-go is going to fall back to a standard LIST on any error + // returned for watch-list requests + if isListWatchRequest(opts) && !utilfeature.DefaultFeatureGate.Enabled(features.WatchList) { + return nil, fmt.Errorf("sendInitialEvents is forbidden for watch unless the WatchList feature gate is enabled") + } + return lw.storage.Watch(ctx, lw.resourcePrefix, opts) } 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 18a918506ea..247fd85cc4d 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 @@ -177,3 +177,29 @@ func TestCacherListerWatcherListWatch(t *testing.T) { storagetesting.TestCheckResultsInStrictOrder(t, w, expectedWatchEvents) storagetesting.TestCheckNoMoreResultsWithIgnoreFunc(t, w, nil) } + +func TestCacherListerWatcherWhenListWatchDisabled(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchList, false) + + prefix := "pods" + fn := func() runtime.Object { return &example.PodList{} } + server, store := newEtcdTestStorage(t, prefix) + defer server.Terminate(t) + + lw := NewListerWatcher(store, prefix, fn, nil) + target := cache.ToListerWatcherWithContext(lw) + watchListOptions := metav1.ListOptions{ + Watch: true, + AllowWatchBookmarks: true, + SendInitialEvents: ptr.To(true), + } + _, err := target.WatchWithContext(context.TODO(), watchListOptions) + if err == nil { + t.Fatalf("Expected error, but got none") + } + + expectedErrMsg := "sendInitialEvents is forbidden for watch unless the WatchList feature gate is enabled" + if err.Error() != expectedErrMsg { + t.Fatalf("Expected error %q, but got %q", expectedErrMsg, err.Error()) + } +}