diff --git a/pkg/registry/core/service/allocator/bitmap.go b/pkg/registry/core/service/allocator/bitmap.go index 3c12864c114..8cf1377bd2c 100644 --- a/pkg/registry/core/service/allocator/bitmap.go +++ b/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). diff --git a/pkg/registry/core/service/allocator/interfaces.go b/pkg/registry/core/service/allocator/interfaces.go index bd7ebc77b81..41f966858b7 100644 --- a/pkg/registry/core/service/allocator/interfaces.go +++ b/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 diff --git a/pkg/registry/core/service/allocator/storage/storage.go b/pkg/registry/core/service/allocator/storage/storage.go index d965d067458..93cc0b055bd 100644 --- a/pkg/registry/core/service/allocator/storage/storage.go +++ b/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 - // https://github.com/kubernetes/kubernetes/pull/50690 - // merges as that shuts down storage properly - registry.RegisterStorageCleanup(d) - + var once sync.Once return &Etcd{ - alloc: alloc, - storage: storage, - baseKey: baseKey, - resource: config.GroupResource, + alloc: alloc, + storage: storage, + baseKey: baseKey, + 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() +} diff --git a/pkg/registry/core/service/ipallocator/allocator.go b/pkg/registry/core/service/ipallocator/allocator.go index b8d8560dc79..b24a04d7203 100644 --- a/pkg/registry/core/service/ipallocator/allocator.go +++ b/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() { +} diff --git a/pkg/registry/core/service/portallocator/allocator.go b/pkg/registry/core/service/portallocator/allocator.go index d8ba78aed45..4f6cb68caf4 100644 --- a/pkg/registry/core/service/portallocator/allocator.go +++ b/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() +} diff --git a/pkg/registry/core/service/storage/alloc.go b/pkg/registry/core/service/storage/alloc.go index 77f91ebed61..6a354d92a02 100644 --- a/pkg/registry/core/service/storage/alloc.go +++ b/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 { diff --git a/pkg/registry/core/service/storage/storage.go b/pkg/registry/core/service/storage/storage.go index 7565101c4cb..b5330da1df4 100644 --- a/pkg/registry/core/service/storage/storage.go +++ b/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