Merge pull request #755 from joelanford/fix-cache-race

🐛 wait for cache start before returning from WaitForCacheSync

pkg/cache/internal/deleg_map.go

@@ -63,11 +63,17 @@ func (m *InformersMap) Start(stop <-chan struct{}) error {
 	return nil
 }
 
-// WaitForCacheSync waits until all the caches have been synced.
+// WaitForCacheSync waits until all the caches have been started and synced.
 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()