mirror of
https://github.com/kubernetes/sample-controller.git
synced 2025-01-19 07:52:52 +08:00
Merge pull request #112200 from pohly/client-go-shared-informer-factory-shutdown
client-go: support waiting for SharedInformerFactory shutdown Kubernetes-commit: 084a412e03816ef3ea57da928ae8fd332d17ab59
This commit is contained in:
Kubernetes Publisher
commit
2180d334f0
8
go.mod
8
go.mod
@ -7,8 +7,8 @@ go 1.19
|
||||
require (
|
||||
k8s.io/api v0.0.0-20220909223647-30ff9916664f
|
||||
k8s.io/apimachinery v0.0.0-20220909223208-6d854d747c21
|
||||
k8s.io/client-go v0.0.0-20220909224245-ab826d2728f3
|
||||
k8s.io/code-generator v0.0.0-20220909222852-c3fdc3ca70f8
|
||||
k8s.io/client-go v0.0.0-20220913183650-18c3338d487f
|
||||
k8s.io/code-generator v0.0.0-20220913182912-2e5cca781290
|
||||
k8s.io/klog/v2 v2.80.1
|
||||
)
|
||||
|
||||
@ -61,6 +61,6 @@ require (
|
||||
replace (
|
||||
k8s.io/api => k8s.io/api v0.0.0-20220909223647-30ff9916664f
|
||||
k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20220909223208-6d854d747c21
|
||||
k8s.io/client-go => k8s.io/client-go v0.0.0-20220909224245-ab826d2728f3
|
||||
k8s.io/code-generator => k8s.io/code-generator v0.0.0-20220909222852-c3fdc3ca70f8
|
||||
k8s.io/client-go => k8s.io/client-go v0.0.0-20220913183650-18c3338d487f
|
||||
k8s.io/code-generator => k8s.io/code-generator v0.0.0-20220913182912-2e5cca781290
|
||||
)
|
||||
|
||||
8
go.sum
8
go.sum
@ -476,10 +476,10 @@ k8s.io/api v0.0.0-20220909223647-30ff9916664f h1:NDgZks7RqnJwWUPDe0tPxMirAowsSG2
|
||||
k8s.io/api v0.0.0-20220909223647-30ff9916664f/go.mod h1:mgirip+ylRYNjZVz7OqYQtrEdhksbpM2LSSH/QNc3wg=
|
||||
k8s.io/apimachinery v0.0.0-20220909223208-6d854d747c21 h1:/RMUsMMVr3xRUWpyOQKagJNRAXt7OfC5R7nJm0BJvz0=
|
||||
k8s.io/apimachinery v0.0.0-20220909223208-6d854d747c21/go.mod h1:uBlVnHT48nY5oV6uG8J4eVKMg56CZnmtHewbsBFKgJA=
|
||||
k8s.io/client-go v0.0.0-20220909224245-ab826d2728f3 h1:PkVMH2O/rFAMo1rTkgR3Z350vHfprbaG295igiMTdWk=
|
||||
k8s.io/client-go v0.0.0-20220909224245-ab826d2728f3/go.mod h1:0uMSNDHUsMhV/kWEx8KQztCZHSWwprrUaWZebaf3QPc=
|
||||
k8s.io/code-generator v0.0.0-20220909222852-c3fdc3ca70f8 h1:8M/ckEnzwRXR2ToMGbcMQIxCiZ9oIVbX/cKd/sIzOCk=
|
||||
k8s.io/code-generator v0.0.0-20220909222852-c3fdc3ca70f8/go.mod h1:73e+BhEs8seYGp/WKXp0MjgcxigA9H393d0o1lO+1JA=
|
||||
k8s.io/client-go v0.0.0-20220913183650-18c3338d487f h1:QdrONbRk42fxstEF/DtMhNhm1yWN1YUr615kKILxoyM=
|
||||
k8s.io/client-go v0.0.0-20220913183650-18c3338d487f/go.mod h1:0uMSNDHUsMhV/kWEx8KQztCZHSWwprrUaWZebaf3QPc=
|
||||
k8s.io/code-generator v0.0.0-20220913182912-2e5cca781290 h1:Trr2hPIkxf6qOwZW8hgT9ljYkwcQAI++8azAY2DjEFg=
|
||||
k8s.io/code-generator v0.0.0-20220913182912-2e5cca781290/go.mod h1:73e+BhEs8seYGp/WKXp0MjgcxigA9H393d0o1lO+1JA=
|
||||
k8s.io/gengo v0.0.0-20220902162205-c0856e24416d h1:U9tB195lKdzwqicbJvyJeOXV7Klv+wNAWENRnXEGi08=
|
||||
k8s.io/gengo v0.0.0-20220902162205-c0856e24416d/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
|
||||
k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
|
||||
|
||||
@ -47,6 +47,11 @@ type sharedInformerFactory struct {
|
||||
// startedInformers is used for tracking which informers have been started.
|
||||
// This allows Start() to be called multiple times safely.
|
||||
startedInformers map[reflect.Type]bool
|
||||
// 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
|
||||
}
|
||||
|
||||
// WithCustomResyncConfig sets a custom resync period for the specified informer types.
|
||||
@ -107,20 +112,39 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
|
||||
return factory
|
||||
}
|
||||
|
||||
// Start initializes all requested informers.
|
||||
func (f *sharedInformerFactory) 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.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
|
||||
go func() {
|
||||
defer f.wg.Done()
|
||||
informer.Run(stopCh)
|
||||
}()
|
||||
f.startedInformers[informerType] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WaitForCacheSync waits for all started informers' cache were synced.
|
||||
func (f *sharedInformerFactory) Shutdown() {
|
||||
f.lock.Lock()
|
||||
f.shuttingDown = true
|
||||
f.lock.Unlock()
|
||||
|
||||
// Will return immediately if there is nothing to wait for.
|
||||
f.wg.Wait()
|
||||
}
|
||||
|
||||
func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
|
||||
informers := func() map[reflect.Type]cache.SharedIndexInformer {
|
||||
f.lock.Lock()
|
||||
@ -167,11 +191,58 @@ func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internal
|
||||
|
||||
// SharedInformerFactory provides shared informers for resources in all known
|
||||
// API group versions.
|
||||
//
|
||||
// It is typically used like this:
|
||||
//
|
||||
// ctx, cancel := context.Background()
|
||||
// defer cancel()
|
||||
// factory := NewSharedInformerFactory(client, resyncPeriod)
|
||||
// defer factory.WaitForStop() // Returns immediately if nothing was started.
|
||||
// genericInformer := factory.ForResource(resource)
|
||||
// typedInformer := factory.SomeAPIGroup().V1().SomeType()
|
||||
// factory.Start(ctx.Done()) // Start processing these informers.
|
||||
// synced := factory.WaitForCacheSync(ctx.Done())
|
||||
// for v, ok := range synced {
|
||||
// if !ok {
|
||||
// fmt.Fprintf(os.Stderr, "caches failed to sync: %v", v)
|
||||
// return
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// // Creating informers can also be created after Start, but then
|
||||
// // Start must be called again:
|
||||
// anotherGenericInformer := factory.ForResource(resource)
|
||||
// factory.Start(ctx.Done())
|
||||
type SharedInformerFactory interface {
|
||||
internalinterfaces.SharedInformerFactory
|
||||
ForResource(resource schema.GroupVersionResource) (GenericInformer, error)
|
||||
|
||||
// Start initializes all requested informers. They are handled in goroutines
|
||||
// which run until the stop channel gets closed.
|
||||
Start(stopCh <-chan struct{})
|
||||
|
||||
// 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()
|
||||
|
||||
// WaitForCacheSync blocks until all started informers' caches were synced
|
||||
// or the stop channel gets closed.
|
||||
WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
|
||||
|
||||
// ForResource gives generic access to a shared informer of the matching type.
|
||||
ForResource(resource schema.GroupVersionResource) (GenericInformer, error)
|
||||
|
||||
// InternalInformerFor returns the SharedIndexInformer for obj using an internal
|
||||
// client.
|
||||
InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer
|
||||
|
||||
Samplecontroller() samplecontroller.Interface
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user