Fix TestCRD Flake

The DestroyFunc functions returned by generic.NewRawStorage is never
called when we do a StartTestServer() in the test suite. For a quick
hack for now, added TrackStorageCleanup/RegisterStorageCleanup and
CleanupStorage. Note that unless TrackStorageCleanup is called (which
is called only from the test suite) the other two methods are
no-ops essentially. So no change in behavior at runtime. This vastly
brings down the number of goroutines that are left behind when this
test is executed and should reduce if not eliminate the flakiness
of TestCRD

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()