Merge pull request #54704 from dims/try-clean-up-stores

cmd/kube-apiserver/app/testing/BUILD

@@ -40,6 +40,7 @@ go_library(
         "//cmd/kube-apiserver/app/options:go_default_library",
         "//pkg/api/legacyscheme:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/registry/generic/registry:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/etcd/testing:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/client-go/rest:go_default_library",

cmd/kube-apiserver/app/testing/testserver.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/registry/generic/registry"
 	etcdtesting "k8s.io/apiserver/pkg/storage/etcd/testing"
 	"k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
@@ -46,8 +47,15 @@ type TearDownFunc func()
 func StartTestServer(t *testing.T) (result *restclient.Config, tearDownForCaller TearDownFunc, err error) {
 	var tmpDir string
 	var etcdServer *etcdtesting.EtcdTestServer
+
+	// TODO : Remove TrackStorageCleanup below when PR
+	// https://github.com/kubernetes/kubernetes/pull/50690
+	// merges as that shuts down storage properly
+	registry.TrackStorageCleanup()
+
 	stopCh := make(chan struct{})
 	tearDown := func() {
+		registry.CleanupStorage()
 		close(stopCh)
 		if etcdServer != nil {
 			etcdServer.Terminate(t)

pkg/registry/core/service/allocator/storage/BUILD

@@ -34,6 +34,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/registry/generic:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/registry/generic/registry:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/errors:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",

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

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/registry/generic"
+	"k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/apiserver/pkg/storage"
 	storeerr "k8s.io/apiserver/pkg/storage/errors"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
@@ -61,7 +62,13 @@ 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 {
-	storage, _ := generic.NewRawStorage(config)
+	storage, d := generic.NewRawStorage(config)
+
+	// TODO : Remove RegisterStorageCleanup below when PR
+	// https://github.com/kubernetes/kubernetes/pull/50690
+	// merges as that shuts down storage properly
+	registry.RegisterStorageCleanup(d)
+
 	return &Etcd{
 		alloc:    alloc,
 		storage:  storage,

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

@@ -65,6 +65,42 @@ func StorageWithCacher(capacity int) generic.StorageDecorator {
 			d()
 		}
 
+		// TODO : Remove RegisterStorageCleanup below when PR
+		// https://github.com/kubernetes/kubernetes/pull/50690
+		// merges as that shuts down storage properly
+		RegisterStorageCleanup(destroyFunc)
+
 		return cacher, destroyFunc
 	}
 }
+
+// TODO : Remove all the code below when PR
+// https://github.com/kubernetes/kubernetes/pull/50690
+// merges as that shuts down storage properly
+// HACK ALERT : Track the destroy methods to call them
+// from the test harness. TrackStorageCleanup will be called
+// only from the test harness, so Register/Cleanup will be
+// no-op at runtime.
+
+var cleanup []func() = nil
+
+func TrackStorageCleanup() {
+	cleanup = make([]func(), 0)
+}
+
+func RegisterStorageCleanup(fn func()) {
+	if cleanup == nil {
+		return
+	}
+	cleanup = append(cleanup, fn)
+}
+
+func CleanupStorage() {
+	if cleanup == nil {
+		return
+	}
+	for _, d := range cleanup {
+		d()
+	}
+	cleanup = nil
+}

staging/src/k8s.io/apiserver/pkg/storage/cacher.go

@@ -649,6 +649,12 @@ func (c *Cacher) isStopped() bool {
 }
 
 func (c *Cacher) Stop() {
+	// TODO : Do not check for isStopped (and return) when PR
+	// https://github.com/kubernetes/kubernetes/pull/50690
+	// merges as that shuts down storage properly
+	if c.isStopped() {
+		return
+	}
 	c.stopLock.Lock()
 	c.stopped = true
 	c.stopLock.Unlock()