Merge pull request kubernetes#94364 from wojtek-t/efficient_watch_resumption

Efficient watch resumption

Author: k8s-ci-robot

pkg/controlplane/instance.go

@@ -258,7 +258,7 @@ func (c *Config) createLeaseReconciler() reconcilers.EndpointReconciler {
 	if err != nil {
 		klog.Fatalf("Error determining service IP ranges: %v", err)
 	}
-	leaseStorage, _, err := storagefactory.Create(*config)
+	leaseStorage, _, err := storagefactory.Create(*config, nil)
 	if err != nil {
 		klog.Fatalf("Error creating storage factory: %v", err)
 	}

pkg/registry/core/pod/rest/log_test.go

@@ -30,7 +30,7 @@ import (
 func TestPodLogValidates(t *testing.T) {
 	config, server := registrytest.NewEtcdStorage(t, "")
 	defer server.Terminate(t)
-	s, destroyFunc, err := generic.NewRawStorage(config)
+	s, destroyFunc, err := generic.NewRawStorage(config, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}

pkg/registry/core/service/allocator/storage/storage.go

@@ -62,7 +62,7 @@ var _ rangeallocation.RangeRegistry = &Etcd{}
 // NewEtcd returns an allocator that is backed by Etcd and can manage
 // persisting the snapshot state of allocation after each allocation is made.
 func NewEtcd(alloc allocator.Snapshottable, baseKey string, resource schema.GroupResource, config *storagebackend.Config) (*Etcd, error) {
-	storage, d, err := generic.NewRawStorage(config)
+	storage, d, err := generic.NewRawStorage(config, nil)
 	if err != nil {
 		return nil, err
 	}

pkg/registry/core/service/ipallocator/storage/storage_test.go

@@ -54,7 +54,7 @@ func newStorage(t *testing.T) (*etcd3testing.EtcdTestServer, ipallocator.Interfa
 	if err != nil {
 		t.Fatalf("unexpected error creating etcd: %v", err)
 	}
-	s, d, err := generic.NewRawStorage(etcdStorage)
+	s, d, err := generic.NewRawStorage(etcdStorage, nil)
 	if err != nil {
 		t.Fatalf("Couldn't create storage: %v", err)
 	}

pkg/registry/core/service/portallocator/storage/storage_test.go

@@ -57,7 +57,7 @@ func newStorage(t *testing.T) (*etcd3testing.EtcdTestServer, portallocator.Inter
 	if err != nil {
 		t.Fatalf("unexpected error creating etcd: %v", err)
 	}
-	s, d, err := generic.NewRawStorage(etcdStorage)
+	s, d, err := generic.NewRawStorage(etcdStorage, nil)
 	if err != nil {
 		t.Fatalf("Couldn't create storage: %v", err)
 	}

staging/src/k8s.io/apiserver/pkg/features/kube_features.go

@@ -138,6 +138,12 @@ const (
 	//
 	// Allows sending warning headers in API responses.
 	WarningHeaders featuregate.Feature = "WarningHeaders"
+
+	// owner: @wojtek-t
+	// alpha: v1.20
+	//
+	// Allows for updating watchcache resource version with progress notify events.
+	EfficientWatchResumption featuregate.Feature = "EfficientWatchResumption"
 )
 
 func init() {
@@ -148,18 +154,19 @@ func init() {
 // To add a new feature, define a key for it above and add it here. The features will be
 // available throughout Kubernetes binaries.
 var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{
-	StreamingProxyRedirects: {Default: true, PreRelease: featuregate.Deprecated},
-	ValidateProxyRedirects:  {Default: true, PreRelease: featuregate.Beta},
-	AdvancedAuditing:        {Default: true, PreRelease: featuregate.GA},
-	APIResponseCompression:  {Default: true, PreRelease: featuregate.Beta},
-	APIListChunking:         {Default: true, PreRelease: featuregate.Beta},
-	DryRun:                  {Default: true, PreRelease: featuregate.GA},
-	RemainingItemCount:      {Default: true, PreRelease: featuregate.Beta},
-	ServerSideApply:         {Default: true, PreRelease: featuregate.Beta},
-	StorageVersionHash:      {Default: true, PreRelease: featuregate.Beta},
-	WatchBookmark:           {Default: true, PreRelease: featuregate.GA, LockToDefault: true},
-	APIPriorityAndFairness:  {Default: false, PreRelease: featuregate.Alpha},
-	RemoveSelfLink:          {Default: true, PreRelease: featuregate.Beta},
-	SelectorIndex:           {Default: true, PreRelease: featuregate.Beta},
-	WarningHeaders:          {Default: true, PreRelease: featuregate.Beta},
+	StreamingProxyRedirects:  {Default: true, PreRelease: featuregate.Deprecated},
+	ValidateProxyRedirects:   {Default: true, PreRelease: featuregate.Beta},
+	AdvancedAuditing:         {Default: true, PreRelease: featuregate.GA},
+	APIResponseCompression:   {Default: true, PreRelease: featuregate.Beta},
+	APIListChunking:          {Default: true, PreRelease: featuregate.Beta},
+	DryRun:                   {Default: true, PreRelease: featuregate.GA},
+	RemainingItemCount:       {Default: true, PreRelease: featuregate.Beta},
+	ServerSideApply:          {Default: true, PreRelease: featuregate.Beta},
+	StorageVersionHash:       {Default: true, PreRelease: featuregate.Beta},
+	WatchBookmark:            {Default: true, PreRelease: featuregate.GA, LockToDefault: true},
+	APIPriorityAndFairness:   {Default: false, PreRelease: featuregate.Alpha},
+	RemoveSelfLink:           {Default: true, PreRelease: featuregate.Beta},
+	SelectorIndex:            {Default: true, PreRelease: featuregate.Beta},
+	WarningHeaders:           {Default: true, PreRelease: featuregate.Beta},
+	EfficientWatchResumption: {Default: false, PreRelease: featuregate.Alpha},
 }

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/dryrun_test.go

@@ -38,7 +38,7 @@ import (
 func NewDryRunnableTestStorage(t *testing.T) (DryRunnableStorage, func()) {
 	server, sc := etcd3testing.NewUnsecuredEtcd3TestClientServer(t)
 	sc.Codec = apitesting.TestStorageCodec(codecs, examplev1.SchemeGroupVersion)
-	s, destroy, err := factory.Create(*sc)
+	s, destroy, err := factory.Create(*sc, nil)
 	if err != nil {
 		t.Fatalf("Error creating storage: %v", err)
 	}

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/storage_factory.go

@@ -45,7 +45,7 @@ func StorageWithCacher() generic.StorageDecorator {
 		triggerFuncs storage.IndexerFuncs,
 		indexers *cache.Indexers) (storage.Interface, factory.DestroyFunc, error) {
 
-		s, d, err := generic.NewRawStorage(storageConfig)
+		s, d, err := generic.NewRawStorage(storageConfig, newFunc)
 		if err != nil {
 			return s, d, err
 		}

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store_test.go

@@ -1601,8 +1601,11 @@ func newTestGenericStoreRegistry(t *testing.T, scheme *runtime.Scheme, hasCacheE
 	server, sc := etcd3testing.NewUnsecuredEtcd3TestClientServer(t)
 	strategy := &testRESTStrategy{scheme, names.SimpleNameGenerator, true, false, true}
 
+	newFunc := func() runtime.Object { return &example.Pod{} }
+	newListFunc := func() runtime.Object { return &example.PodList{} }
+
 	sc.Codec = apitesting.TestStorageCodec(codecs, examplev1.SchemeGroupVersion)
-	s, dFunc, err := factory.Create(*sc)
+	s, dFunc, err := factory.Create(*sc, newFunc)
 	if err != nil {
 		t.Fatalf("Error creating storage: %v", err)
 	}
@@ -1617,8 +1620,8 @@ func newTestGenericStoreRegistry(t *testing.T, scheme *runtime.Scheme, hasCacheE
 			ResourcePrefix: podPrefix,
 			KeyFunc:        func(obj runtime.Object) (string, error) { return storage.NoNamespaceKeyFunc(podPrefix, obj) },
 			GetAttrsFunc:   getPodAttrs,
-			NewFunc:        func() runtime.Object { return &example.Pod{} },
-			NewListFunc:    func() runtime.Object { return &example.PodList{} },
+			NewFunc:        newFunc,
+			NewListFunc:    newListFunc,
 			Codec:          sc.Codec,
 		}
 		cacher, err := cacherstorage.NewCacherFromConfig(config)

staging/src/k8s.io/apiserver/pkg/registry/generic/storage_decorator.go

@@ -47,12 +47,12 @@ func UndecoratedStorage(
 	getAttrsFunc storage.AttrFunc,
 	trigger storage.IndexerFuncs,
 	indexers *cache.Indexers) (storage.Interface, factory.DestroyFunc, error) {
-	return NewRawStorage(config)
+	return NewRawStorage(config, newFunc)
 }
 
 // NewRawStorage creates the low level kv storage. This is a work-around for current
 // two layer of same storage interface.
 // TODO: Once cacher is enabled on all registries (event registry is special), we will remove this method.
-func NewRawStorage(config *storagebackend.Config) (storage.Interface, factory.DestroyFunc, error) {
-	return factory.Create(*config)
+func NewRawStorage(config *storagebackend.Config, newFunc func() runtime.Object) (storage.Interface, factory.DestroyFunc, error) {
+	return factory.Create(*config, newFunc)
 }