Clean storage shutdown for allocators

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

@@ -198,6 +198,10 @@ func (r *AllocationBitmap) Restore(rangeSpec string, data []byte) error {
 	return nil
 }
 
+// Destroy cleans up everything on shutdown.
+func (r *AllocationBitmap) Destroy() {
+}
+
 // randomScanStrategy chooses a random address from the provided big.Int, and then
 // scans forward looking for the next available address (it will wrap the range if
 // necessary).

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

@@ -29,6 +29,11 @@ type Interface interface {
 
 	// For testing
 	Free() int
+
+	// Destroy shuts down all internal structures.
+	// Destroy needs to be implemented in thread-safe way and be prepared for being
+	// called more than once.
+	Destroy()
 }
 
 // Snapshottable is an Interface that can be snapshotted and restored. Snapshottable

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

@@ -26,7 +26,6 @@ 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"
@@ -53,6 +52,8 @@ type Etcd struct {
 
 	baseKey  string
 	resource schema.GroupResource
+
+	destroyFn func()
 }
 
 // Etcd implements allocator.Interface and rangeallocation.RangeRegistry
@@ -67,16 +68,13 @@ func NewEtcd(alloc allocator.Snapshottable, baseKey string, config *storagebacke
 		return nil, err
 	}
 
-	// TODO : Remove RegisterStorageCleanup below when PR
+	var once sync.Once
-	// https://github.com/kubernetes/kubernetes/pull/50690
-	// merges as that shuts down storage properly
-	registry.RegisterStorageCleanup(d)
-
 	return &Etcd{
-		alloc:    alloc,
+		alloc:     alloc,
-		storage:  storage,
+		storage:   storage,
-		baseKey:  baseKey,
+		baseKey:   baseKey,
-		resource: config.GroupResource,
+		resource:  config.GroupResource,
+		destroyFn: func() { once.Do(d) },
 	}, nil
 }
 
@@ -220,7 +218,7 @@ func (e *Etcd) CreateOrUpdate(snapshot *api.RangeAllocation) error {
 	return err
 }
 
-// Implements allocator.Interface::Has
+// Has implements allocator.Interface::Has
 func (e *Etcd) Has(item int) bool {
 	e.lock.Lock()
 	defer e.lock.Unlock()
@@ -228,10 +226,15 @@ func (e *Etcd) Has(item int) bool {
 	return e.alloc.Has(item)
 }
 
-// Implements allocator.Interface::Free
+// Free implements allocator.Interface::Free
 func (e *Etcd) Free() int {
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
 	return e.alloc.Free()
 }
+
+// Destroy implement allocator.Interface::Destroy
+func (e *Etcd) Destroy() {
+	e.destroyFn()
+}

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

@@ -39,6 +39,7 @@ type Interface interface {
 	CIDR() net.IPNet
 	IPFamily() api.IPFamily
 	Has(ip net.IP) bool
+	Destroy()
 
 	// DryRun offers a way to try operations without persisting them.
 	DryRun() Interface
@@ -358,6 +359,11 @@ func (r *Range) contains(ip net.IP) (bool, int) {
 	return true, offset
 }
 
+// Destroy shuts down internal allocator.
+func (r *Range) Destroy() {
+	r.alloc.Destroy()
+}
+
 // calculateIPOffset calculates the integer offset of ip from base such that
 // base + offset = ip. It requires ip >= base.
 func calculateIPOffset(base *big.Int, ip net.IP) int {
@@ -427,3 +433,6 @@ func (dry dryRunRange) DryRun() Interface {
 func (dry dryRunRange) Has(ip net.IP) bool {
 	return dry.real.Has(ip)
 }
+
+func (dry dryRunRange) Destroy() {
+}

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

@@ -35,6 +35,7 @@ type Interface interface {
 	Release(int) error
 	ForEach(func(int))
 	Has(int) bool
+	Destroy()
 }
 
 var (
@@ -207,3 +208,8 @@ func (r *PortAllocator) contains(port int) (bool, int) {
 	offset := port - r.portRange.Base
 	return true, offset
 }
+
+// Destroy shuts down internal allocator.
+func (r *PortAllocator) Destroy() {
+	r.alloc.Destroy()
+}

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

@@ -899,6 +899,13 @@ func (al *Allocators) releaseClusterIPs(service *api.Service) (released map[api.
 	return al.releaseIPs(toRelease)
 }
 
+func (al *Allocators) Destroy() {
+	al.serviceNodePorts.Destroy()
+	for _, a := range al.serviceIPAllocatorsByFamily {
+		a.Destroy()
+	}
+}
+
 // This is O(N), but we expect haystack to be small;
 // so small that we expect a linear search to be faster
 func containsNumber(haystack []int, needle int) bool {

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

@@ -157,6 +157,12 @@ func (r *REST) Categories() []string {
 	return []string{"all"}
 }
 
+// Destroy cleans up everything on shutdown.
+func (r *REST) Destroy() {
+	r.Store.Destroy()
+	r.alloc.Destroy()
+}
+
 // StatusREST implements the REST endpoint for changing the status of a service.
 type StatusREST struct {
 	store *genericregistry.Store