identity: cache: close channel in writing party

pkg/allocator/allocator.go

@@ -88,7 +88,7 @@ const (
 type Allocator struct {
 	// events is a channel which will receive AllocatorEvent as IDs are
 	// added, modified or removed from the allocator
-	events AllocatorEventChan
+	events AllocatorEventSendChan
 
 	// keyType is an instance of the type to be used as allocator key.
 	keyType AllocatorKey
@@ -345,7 +345,7 @@ func WithBackend(backend Backend) AllocatorOption {
 // read while NewAllocator() is being called to ensure that the channel does
 // not block indefinitely while NewAllocator() emits events on it while
 // populating the initial cache.
-func WithEvents(events AllocatorEventChan) AllocatorOption {
+func WithEvents(events AllocatorEventSendChan) AllocatorOption {
 	return func(a *Allocator) { a.events = events }
 }
 
@@ -381,7 +381,7 @@ func WithoutGC() AllocatorOption {
 // GetEvents returns the events channel given to the allocator when
 // constructed.
 // Note: This channel is not owned by the allocator!
-func (a *Allocator) GetEvents() AllocatorEventChan {
+func (a *Allocator) GetEvents() AllocatorEventSendChan {
 	return a.events
 }
 
@@ -867,6 +867,10 @@ func (a *Allocator) startLocalKeySync() {
 // AllocatorEventChan is a channel to receive allocator events on
 type AllocatorEventChan chan AllocatorEvent
 
+// Send- and receive-only versions of the above.
+type AllocatorEventRecvChan = <-chan AllocatorEvent
+type AllocatorEventSendChan = chan<- AllocatorEvent
+
 // AllocatorEvent is an event sent over AllocatorEventChan
 type AllocatorEvent struct {
 	// Typ is the type of event (create / modify / delete)

pkg/identity/cache/allocator.go

@@ -49,6 +49,9 @@ type CachingIdentityAllocator struct {
 
 	identitiesPath string
 
+	// This field exists is to hand out references that are either for sending
+	// and receiving. It should not be used directly without converting it first
+	// to a AllocatorEventSendChan or AllocatorEventRecvChan.
 	events  allocator.AllocatorEventChan
 	watcher identityWatcher
 
@@ -167,7 +170,7 @@ func (m *CachingIdentityAllocator) InitIdentityAllocator(client clientset.Interf
 
 	// Asynchronously set up the global identity allocator since it connects
 	// to the kvstore.
-	go func(owner IdentityAllocatorOwner, events allocator.AllocatorEventChan, minID, maxID idpool.ID) {
+	go func(owner IdentityAllocatorOwner, events allocator.AllocatorEventSendChan, minID, maxID idpool.ID) {
 		m.setupMutex.Lock()
 		defer m.setupMutex.Unlock()
 
@@ -273,7 +276,9 @@ func (m *CachingIdentityAllocator) Close() {
 
 	m.IdentityAllocator.Delete()
 	if m.events != nil {
-		close(m.events)
+		// Have the now only remaining writing party close the events channel,
+		// to ensure we don't panic with 'send on closed channel'.
+		m.localIdentities.close()
 		m.events = nil
 	}
 

pkg/identity/cache/cache.go

@@ -134,7 +134,7 @@ func collectEvent(event allocator.AllocatorEvent, added, deleted IdentityCache)
 }
 
 // watch starts the identity watcher
-func (w *identityWatcher) watch(events allocator.AllocatorEventChan) {
+func (w *identityWatcher) watch(events allocator.AllocatorEventRecvChan) {
 
 	go func() {
 		for {

pkg/identity/cache/local.go

@@ -22,10 +22,10 @@ type localIdentityCache struct {
 	nextNumericIdentity identity.NumericIdentity
 	minID               identity.NumericIdentity
 	maxID               identity.NumericIdentity
-	events              allocator.AllocatorEventChan
+	events              allocator.AllocatorEventSendChan
 }
 
-func newLocalIdentityCache(minID, maxID identity.NumericIdentity, events allocator.AllocatorEventChan) *localIdentityCache {
+func newLocalIdentityCache(minID, maxID identity.NumericIdentity, events allocator.AllocatorEventSendChan) *localIdentityCache {
 	return &localIdentityCache{
 		identitiesByID:      map[identity.NumericIdentity]*identity.Identity{},
 		identitiesByLabels:  map[string]*identity.Identity{},
@@ -191,3 +191,15 @@ func (l *localIdentityCache) GetIdentities() map[identity.NumericIdentity]*ident
 
 	return cache
 }
+
+// close closes the events channel. The local identity cache is the writing
+// party, hence also needs to close the channel.
+func (l *localIdentityCache) close() {
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
+
+	if l.events != nil {
+		close(l.events)
+		l.events = nil
+	}
+}