wait for cache start before returning from WaitForCacheSync

There is currently an unintuitive race condition between starting
a cache, waiting for it to sync, and using it the first time.

The crux of the problem is that WaitForCacheSync can return true
BEFORE the Start function has had a chance to set started to true.

This means that Get and List can return ErrCacheNotStarted even
after WaitForCacheSync has returned true.

This commit adds a channel and a wait function that is called from
WaitForCacheSync to ensure that started has been set to true before
returning.

Author: joelanford

pkg/cache/internal/deleg_map.go

@@ -68,6 +68,12 @@ func (m *InformersMap) WaitForCacheSync(stop <-chan struct{}) bool {
 	syncedFuncs := append([]cache.InformerSynced(nil), m.structured.HasSyncedFuncs()...)
 	syncedFuncs = append(syncedFuncs, m.unstructured.HasSyncedFuncs()...)
 
+	if !m.structured.waitForStarted(stop) {
+		return false
+	}
+	if !m.unstructured.waitForStarted(stop) {
+		return false
+	}
 	return cache.WaitForCacheSync(stop, syncedFuncs...)
 }
 

pkg/cache/internal/informers_map.go

@@ -54,6 +54,7 @@ func newSpecificInformersMap(config *rest.Config,
 		codecs:            serializer.NewCodecFactory(scheme),
 		paramCodec:        runtime.NewParameterCodec(scheme),
 		resync:            resync,
+		startWait:         make(chan struct{}),
 		createListWatcher: createListWatcher,
 		namespace:         namespace,
 	}
@@ -104,6 +105,10 @@ type specificInformersMap struct {
 	// start is true if the informers have been started
 	started bool
 
+	// startWait is a channel that is closed after the
+	// informer has been started.
+	startWait chan struct{}
+
 	// createClient knows how to create a client and a list object,
 	// and allows for abstracting over the particulars of structured vs
 	// unstructured objects.
@@ -131,10 +136,20 @@ func (ip *specificInformersMap) Start(stop <-chan struct{}) {
 
 		// Set started to true so we immediately start any informers added later.
 		ip.started = true
+		close(ip.startWait)
 	}()
 	<-stop
 }
 
+func (ip *specificInformersMap) waitForStarted(stop <-chan struct{}) bool {
+	select {
+	case <-ip.startWait:
+		return true
+	case <-stop:
+		return false
+	}
+}
+
 // HasSyncedFuncs returns all the HasSynced functions for the informers in this map.
 func (ip *specificInformersMap) HasSyncedFuncs() []cache.InformerSynced {
 	ip.mu.RLock()