diff --git a/dynamic/dynamicinformer/informer.go b/dynamic/dynamicinformer/informer.go index 9785613d..62d01339 100644 --- a/dynamic/dynamicinformer/informer.go +++ b/dynamic/dynamicinformer/informer.go @@ -61,6 +61,12 @@ type dynamicSharedInformerFactory struct { // This allows Start() to be called multiple times safely. startedInformers map[schema.GroupVersionResource]bool tweakListOptions TweakListOptionsFunc + + // wg tracks how many goroutines were started. + wg sync.WaitGroup + // shuttingDown is true when Shutdown has been called. It may still be running + // because it needs to wait for goroutines. + shuttingDown bool } var _ DynamicSharedInformerFactory = &dynamicSharedInformerFactory{} @@ -86,9 +92,21 @@ func (f *dynamicSharedInformerFactory) Start(stopCh <-chan struct{}) { f.lock.Lock() defer f.lock.Unlock() + if f.shuttingDown { + return + } + for informerType, informer := range f.informers { if !f.startedInformers[informerType] { - go informer.Informer().Run(stopCh) + f.wg.Add(1) + // We need a new variable in each loop iteration, + // otherwise the goroutine would use the loop variable + // and that keeps changing. + informer := informer.Informer() + go func() { + defer f.wg.Done() + informer.Run(stopCh) + }() f.startedInformers[informerType] = true } } @@ -116,6 +134,15 @@ func (f *dynamicSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) return res } +func (f *dynamicSharedInformerFactory) Shutdown() { + // Will return immediately if there is nothing to wait for. + defer f.wg.Wait() + + f.lock.Lock() + defer f.lock.Unlock() + f.shuttingDown = true +} + // NewFilteredDynamicInformer constructs a new informer for a dynamic type. func NewFilteredDynamicInformer(client dynamic.Interface, gvr schema.GroupVersionResource, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions TweakListOptionsFunc) informers.GenericInformer { return &dynamicInformer{ diff --git a/dynamic/dynamicinformer/interface.go b/dynamic/dynamicinformer/interface.go index 083977c3..0419ef4f 100644 --- a/dynamic/dynamicinformer/interface.go +++ b/dynamic/dynamicinformer/interface.go @@ -24,9 +24,28 @@ import ( // DynamicSharedInformerFactory provides access to a shared informer and lister for dynamic client type DynamicSharedInformerFactory interface { + // Start initializes all requested informers. They are handled in goroutines + // which run until the stop channel gets closed. Start(stopCh <-chan struct{}) + + // ForResource gives generic access to a shared informer of the matching type. ForResource(gvr schema.GroupVersionResource) informers.GenericInformer + + // WaitForCacheSync blocks until all started informers' caches were synced + // or the stop channel gets closed. WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool + + // Shutdown marks a factory as shutting down. At that point no new + // informers can be started anymore and Start will return without + // doing anything. + // + // In addition, Shutdown blocks until all goroutines have terminated. For that + // to happen, the close channel(s) that they were started with must be closed, + // either before Shutdown gets called or while it is waiting. + // + // Shutdown may be called multiple times, even concurrently. All such calls will + // block until all goroutines have terminated. + Shutdown() } // TweakListOptionsFunc defines the signature of a helper function diff --git a/metadata/metadatainformer/informer.go b/metadata/metadatainformer/informer.go index e4ebd61f..c211a4b7 100644 --- a/metadata/metadatainformer/informer.go +++ b/metadata/metadatainformer/informer.go @@ -60,6 +60,11 @@ type metadataSharedInformerFactory struct { // This allows Start() to be called multiple times safely. startedInformers map[schema.GroupVersionResource]bool tweakListOptions TweakListOptionsFunc + // wg tracks how many goroutines were started. + wg sync.WaitGroup + // shuttingDown is true when Shutdown has been called. It may still be running + // because it needs to wait for goroutines. + shuttingDown bool } var _ SharedInformerFactory = &metadataSharedInformerFactory{} @@ -85,9 +90,21 @@ func (f *metadataSharedInformerFactory) Start(stopCh <-chan struct{}) { f.lock.Lock() defer f.lock.Unlock() + if f.shuttingDown { + return + } + for informerType, informer := range f.informers { if !f.startedInformers[informerType] { - go informer.Informer().Run(stopCh) + f.wg.Add(1) + // We need a new variable in each loop iteration, + // otherwise the goroutine would use the loop variable + // and that keeps changing. + informer := informer.Informer() + go func() { + defer f.wg.Done() + informer.Run(stopCh) + }() f.startedInformers[informerType] = true } } @@ -115,6 +132,15 @@ func (f *metadataSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) return res } +func (f *metadataSharedInformerFactory) Shutdown() { + // Will return immediately if there is nothing to wait for. + defer f.wg.Wait() + + f.lock.Lock() + defer f.lock.Unlock() + f.shuttingDown = true +} + // NewFilteredMetadataInformer constructs a new informer for a metadata type. func NewFilteredMetadataInformer(client metadata.Interface, gvr schema.GroupVersionResource, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions TweakListOptionsFunc) informers.GenericInformer { return &metadataInformer{ diff --git a/metadata/metadatainformer/interface.go b/metadata/metadatainformer/interface.go index 732e565c..9f61706c 100644 --- a/metadata/metadatainformer/interface.go +++ b/metadata/metadatainformer/interface.go @@ -24,9 +24,28 @@ import ( // SharedInformerFactory provides access to a shared informer and lister for dynamic client type SharedInformerFactory interface { + // Start initializes all requested informers. They are handled in goroutines + // which run until the stop channel gets closed. Start(stopCh <-chan struct{}) + + // ForResource gives generic access to a shared informer of the matching type. ForResource(gvr schema.GroupVersionResource) informers.GenericInformer + + // WaitForCacheSync blocks until all started informers' caches were synced + // or the stop channel gets closed. WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool + + // Shutdown marks a factory as shutting down. At that point no new + // informers can be started anymore and Start will return without + // doing anything. + // + // In addition, Shutdown blocks until all goroutines have terminated. For that + // to happen, the close channel(s) that they were started with must be closed, + // either before Shutdown gets called or while it is waiting. + // + // Shutdown may be called multiple times, even concurrently. All such calls will + // block until all goroutines have terminated. + Shutdown() } // TweakListOptionsFunc defines the signature of a helper function