Merge pull request #30251 from hongchaodeng/r2

Automatic merge from submit-queue

Move new etcd storage (low level storage) into cacher

In an effort for #29888, we are pushing forward this:

What?
- It changes creating etcd storage.Interface impl into creating config
- In creating cacher storage (StorageWithCacher), it passes config created above and new etcd storage inside.

Why?
- We want to expose the information of (etcd) kv client to cacher. Cacher storage uses this information to talk to remote storage.

pkg/registry/service/allocator/etcd/etcd.go

@@ -25,10 +25,12 @@ import (
 	k8serr "k8s.io/kubernetes/pkg/api/errors"
 	storeerr "k8s.io/kubernetes/pkg/api/errors/storage"
 	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/registry/generic"
 	"k8s.io/kubernetes/pkg/registry/rangeallocation"
 	"k8s.io/kubernetes/pkg/registry/service/allocator"
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/storage"
+	"k8s.io/kubernetes/pkg/storage/storagebackend"
 
 	"golang.org/x/net/context"
 )
@@ -58,7 +60,8 @@ 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 unversioned.GroupResource, storage storage.Interface) *Etcd {
+func NewEtcd(alloc allocator.Snapshottable, baseKey string, resource unversioned.GroupResource, config *storagebackend.Config) *Etcd {
+	storage := generic.NewRawStorage(config)
 	return &Etcd{
 		alloc:    alloc,
 		storage:  storage,

pkg/registry/service/allocator/etcd/etcd_test.go

@@ -25,15 +25,16 @@ import (
 	"k8s.io/kubernetes/pkg/registry/service/allocator"
 	"k8s.io/kubernetes/pkg/storage/etcd/etcdtest"
 	etcdtesting "k8s.io/kubernetes/pkg/storage/etcd/testing"
+	"k8s.io/kubernetes/pkg/storage/storagebackend"
 
 	"golang.org/x/net/context"
 )
 
-func newStorage(t *testing.T) (*Etcd, *etcdtesting.EtcdTestServer, allocator.Interface) {
+func newStorage(t *testing.T) (*Etcd, *etcdtesting.EtcdTestServer, allocator.Interface, *storagebackend.Config) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
 	mem := allocator.NewAllocationMap(100, "rangeSpecValue")
 	etcd := NewEtcd(mem, "/ranges/serviceips", api.Resource("serviceipallocations"), etcdStorage)
-	return etcd, server, mem
+	return etcd, server, mem, etcdStorage
 }
 
 func validNewRangeAllocation() *api.RangeAllocation {
@@ -48,7 +49,7 @@ func key() string {
 }
 
 func TestEmpty(t *testing.T) {
-	storage, server, _ := newStorage(t)
+	storage, server, _, _ := newStorage(t)
 	defer server.Terminate(t)
 	if _, err := storage.Allocate(1); !strings.Contains(err.Error(), "cannot allocate resources of type serviceipallocations at this time") {
 		t.Fatal(err)
@@ -56,7 +57,7 @@ func TestEmpty(t *testing.T) {
 }
 
 func TestStore(t *testing.T) {
-	storage, server, backing := newStorage(t)
+	storage, server, backing, config := newStorage(t)
 	defer server.Terminate(t)
 	if err := storage.storage.Create(context.TODO(), key(), validNewRangeAllocation(), nil, 0); err != nil {
 		t.Fatalf("unexpected error: %v", err)
@@ -93,7 +94,7 @@ func TestStore(t *testing.T) {
 	}
 
 	other = allocator.NewAllocationMap(100, "rangeSpecValue")
-	otherStorage := NewEtcd(other, "/ranges/serviceips", api.Resource("serviceipallocations"), storage.storage)
+	otherStorage := NewEtcd(other, "/ranges/serviceips", api.Resource("serviceipallocations"), config)
 	if ok, err := otherStorage.Allocate(2); ok || err != nil {
 		t.Fatal(err)
 	}

pkg/registry/service/etcd/etcd.go

@@ -37,7 +37,7 @@ func NewREST(opts generic.RESTOptions) (*REST, *StatusREST) {
 
 	newListFunc := func() runtime.Object { return &api.ServiceList{} }
 	storageInterface := opts.Decorator(
-		opts.Storage,
+		opts.StorageConfig,
 		cachesize.GetWatchCacheSizeByResource(cachesize.Services),
 		&api.Service{},
 		prefix,

pkg/registry/service/etcd/etcd_test.go

@@ -31,7 +31,7 @@ import (
 
 func newStorage(t *testing.T) (*REST, *StatusREST, *etcdtesting.EtcdTestServer) {
 	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
-	restOptions := generic.RESTOptions{Storage: etcdStorage, Decorator: generic.UndecoratedStorage, DeleteCollectionWorkers: 1}
+	restOptions := generic.RESTOptions{StorageConfig: etcdStorage, Decorator: generic.UndecoratedStorage, DeleteCollectionWorkers: 1}
 	serviceStorage, statusStorage := NewREST(restOptions)
 	return serviceStorage, statusStorage, server
 }

pkg/registry/service/ipallocator/etcd/etcd_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/registry/generic"
 	"k8s.io/kubernetes/pkg/registry/registrytest"
 	"k8s.io/kubernetes/pkg/registry/service/allocator"
 	allocatoretcd "k8s.io/kubernetes/pkg/registry/service/allocator/etcd"
@@ -48,7 +49,7 @@ func newStorage(t *testing.T) (*etcdtesting.EtcdTestServer, ipallocator.Interfac
 		return etcd
 	})
 
-	return server, storage, backing, etcdStorage
+	return server, storage, backing, generic.NewRawStorage(etcdStorage)
 }
 
 func validNewRangeAllocation() *api.RangeAllocation {