Merge pull request #134770 from p0lyn0mial/upstream-watchlist-unsupported-wl-cacher

apiserver/pkg/storage/cacher/lister_watcher: exposes WatchList semantics

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{

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

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

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{} }