diff --git a/informers/factory.go b/informers/factory.go index fe7b1242e..c7c735fed 100644 --- a/informers/factory.go +++ b/informers/factory.go @@ -19,6 +19,7 @@ limitations under the License. package informers import ( + context "context" reflect "reflect" sync "sync" time "time" @@ -26,6 +27,7 @@ import ( v1 "k8s.io/apimachinery/pkg/apis/meta/v1" runtime "k8s.io/apimachinery/pkg/runtime" schema "k8s.io/apimachinery/pkg/runtime/schema" + wait "k8s.io/apimachinery/pkg/util/wait" admissionregistration "k8s.io/client-go/informers/admissionregistration" apiserverinternal "k8s.io/client-go/informers/apiserverinternal" apps "k8s.io/client-go/informers/apps" @@ -158,6 +160,10 @@ func NewSharedInformerFactoryWithOptions(client kubernetes.Interface, defaultRes } func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) { + f.StartWithContext(wait.ContextForChannel(stopCh)) +} + +func (f *sharedInformerFactory) StartWithContext(ctx context.Context) { f.lock.Lock() defer f.lock.Unlock() @@ -167,15 +173,9 @@ func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) { for informerType, informer := range f.informers { if !f.startedInformers[informerType] { - 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.wg.Go(func() { + informer.RunWithContext(ctx) + }) f.startedInformers[informerType] = true } } @@ -192,6 +192,11 @@ func (f *sharedInformerFactory) Shutdown() { } func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool { + result := f.WaitForCacheSyncWithContext(wait.ContextForChannel(stopCh)) + return result.Synced +} + +func (f *sharedInformerFactory) WaitForCacheSyncWithContext(ctx context.Context) cache.SyncResult { informers := func() map[reflect.Type]cache.SharedIndexInformer { f.lock.Lock() defer f.lock.Unlock() @@ -205,10 +210,31 @@ func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[ref return informers }() - res := map[reflect.Type]bool{} - for informType, informer := range informers { - res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced) + // Wait for informers to sync, without polling. + cacheSyncs := make([]cache.DoneChecker, 0, len(informers)) + for _, informer := range informers { + cacheSyncs = append(cacheSyncs, informer.HasSyncedChecker()) } + cache.WaitFor(ctx, "" /* no logging */, cacheSyncs...) + + res := cache.SyncResult{ + Synced: make(map[reflect.Type]bool, len(informers)), + } + failed := false + for informType, informer := range informers { + hasSynced := informer.HasSynced() + if !hasSynced { + failed = true + } + res.Synced[informType] = hasSynced + } + if failed { + // context.Cause is more informative than ctx.Err(). + // This must be non-nil, otherwise WaitFor wouldn't have stopped + // prematurely. + res.Err = context.Cause(ctx) + } + return res } @@ -247,27 +273,46 @@ func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internal // 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 -// } +// handle, err := typeInformer.Informer().AddEventHandler(...) +// if err != nil { +// return fmt.Errorf("register event handler: %v", err) +// } +// defer typeInformer.Informer().RemoveEventHandler(handle) // Avoids leaking goroutines. +// factory.StartWithContext(ctx) // Start processing these informers. +// synced := factory.WaitForCacheSyncWithContext(ctx) +// if err := synced.AsError(); err != nil { +// return err +// } +// for v := range synced { +// // Only if desired log some information similar to this. +// fmt.Fprintf(os.Stdout, "cache synced: %s", v) +// } +// +// // Also make sure that all of the initial cache events have been delivered. +// if !WaitFor(ctx, "event handler sync", handle.HasSyncedChecker()) { +// // Must have failed because of context. +// return fmt.Errorf("sync event handler: %w", context.Cause(ctx)) // } // // // Creating informers can also be created after Start, but then // // Start must be called again: // anotherGenericInformer := factory.ForResource(resource) -// factory.Start(ctx.Done()) +// factory.StartWithContext(ctx) type SharedInformerFactory interface { internalinterfaces.SharedInformerFactory // Start initializes all requested informers. They are handled in goroutines // which run until the stop channel gets closed. // Warning: Start does not block. When run in a go-routine, it will race with a later WaitForCacheSync. + // + // Contextual logging: StartWithContext should be used instead of Start in code which supports contextual logging. Start(stopCh <-chan struct{}) + // StartWithContext initializes all requested informers. They are handled in goroutines + // which run until the context gets canceled. + // Warning: StartWithContext does not block. When run in a go-routine, it will race with a later WaitForCacheSync. + StartWithContext(ctx context.Context) + // Shutdown marks a factory as shutting down. At that point no new // informers can be started anymore and Start will return without // doing anything. @@ -282,8 +327,14 @@ type SharedInformerFactory interface { // WaitForCacheSync blocks until all started informers' caches were synced // or the stop channel gets closed. + // + // Contextual logging: WaitForCacheSync should be used instead of WaitForCacheSync in code which supports contextual logging. It also returns a more useful result. WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool + // WaitForCacheSyncWithContext blocks until all started informers' caches were synced + // or the context gets canceled. + WaitForCacheSyncWithContext(ctx context.Context) cache.SyncResult + // ForResource gives generic access to a shared informer of the matching type. ForResource(resource schema.GroupVersionResource) (GenericInformer, error) diff --git a/tools/cache/event_handler_name.go b/tools/cache/event_handler_name.go index 9489f9b47..d466fc6e2 100644 --- a/tools/cache/event_handler_name.go +++ b/tools/cache/event_handler_name.go @@ -50,7 +50,9 @@ func nameForHandler(handler ResourceEventHandler) (name string) { value = value.Elem() } if value.Type().Kind() == reflect.Pointer { - value = value.Elem() + if !value.IsNil() { + value = value.Elem() + } } name := value.Type().PkgPath() if name != "" { diff --git a/tools/cache/event_handler_name_test.go b/tools/cache/event_handler_name_test.go index d17f66a08..9cb1434f9 100644 --- a/tools/cache/event_handler_name_test.go +++ b/tools/cache/event_handler_name_test.go @@ -76,6 +76,12 @@ func TestNameForHandler(t *testing.T) { handler: nil, wantName: "", }, + "stored-nil": { + // This is a bit odd, but one unit test actually registered + // such an event handler and it somehow worked. + handler: (*mockHandler)(nil), + wantName: "*cache.mockHandler", + }, } { t.Run(name, func(t *testing.T) { gotName := nameForHandler(tc.handler) diff --git a/tools/cache/shared_informer.go b/tools/cache/shared_informer.go index 503cd2e2d..0f145b72a 100644 --- a/tools/cache/shared_informer.go +++ b/tools/cache/shared_informer.go @@ -20,7 +20,9 @@ import ( "context" "errors" "fmt" + "reflect" "slices" + "strings" "sync" "sync/atomic" "time" @@ -433,7 +435,10 @@ func WaitForCacheSync(stopCh <-chan struct{}, cacheSyncs ...InformerSynced) bool // before all activities are completed. // // If a non-nil "what" is provided, then progress information is logged -// while waiting ("Waiting", for=""). +// while waiting ("Waiting", for=""). Verbosity is V(0). This can +// be made less verbose by the caller with: +// +// WaitFor(klog.NewContext(ctx, klog.FromContext(ctx).V(2)), ...) // // In contrast to other WaitForCacheSync alternatives, this one here doesn't // need polling, which makes it react immediately. When used in a synctest unit @@ -529,6 +534,43 @@ func IsDone(checker DoneChecker) bool { } } +// SyncResult is the result of a shared informer factory's WaitForCacheSyncWithContext. +// Under the hood such factories use [WaitFor] to wait for all instantiated informers, +// then provide this summary of what was synced. +// +// Note that the informers may have synced already before all event handlers registered with +// those informers have synced. Code which wants to be sure that all of its state is up-to-date +// should do its own WaitFor with the informer's HasSyncedChecker() *and* the +// registration handle's HasSyncChecker() results. +type SyncResult struct { + // Err is nil if all informer caches were synced, otherwise it is + // the reason why waiting for cache syncing stopped (= context.Cause(ctx)). + Err error + + // Synced maps each registered informer in a SharedInformerFactory to + // true if it has synced, false otherwise. + Synced map[reflect.Type]bool +} + +// AsError turns a SyncResult into an error if not all caches were synced, +// otherwise it returns nil. The error wraps context.Cause(ctx) and +// includes information about the informers which were not synced. +func (c SyncResult) AsError() error { + if c.Err == nil { + return nil + } + + unsynced := make([]string, 0, len(c.Synced)) + for t, synced := range c.Synced { + if !synced { + unsynced = append(unsynced, t.String()) + } + } + slices.Sort(unsynced) + + return fmt.Errorf("failed to sync all caches: %s: %w", strings.Join(unsynced, ", "), c.Err) +} + // `*sharedIndexInformer` implements SharedIndexInformer and has three // main components. One is an indexed local cache, `indexer Indexer`. // The second main component is a Controller that pulls @@ -888,6 +930,10 @@ func (s *sharedIndexInformer) AddEventHandlerWithOptions(handler ResourceEventHa // thread adding them and the counter is temporarily zero). listener.add(addNotification{newObj: item, isInInitialList: true}) } + + // Initial list is added, now we can allow the listener to detect that "upstream has synced". + s.processor.wg.Start(listener.watchSynced) + return handle, nil } @@ -1008,7 +1054,8 @@ func (p *sharedProcessor) addListener(listener *processorListener) ResourceEvent p.listeners[listener] = true if p.listenersStarted { - p.wg.Start(listener.watchSynced) + // Not starting listener.watchSynced! + // The caller must first add the initial list, then start it. p.wg.Start(listener.run) p.wg.Start(listener.pop) } diff --git a/tools/cache/wait_test.go b/tools/cache/wait_test.go index b53683689..f03b48dd8 100644 --- a/tools/cache/wait_test.go +++ b/tools/cache/wait_test.go @@ -22,6 +22,7 @@ import ( "errors" "fmt" "os" + "reflect" "runtime" "sync" "testing" @@ -198,3 +199,48 @@ func (m *mockChecker) Done() <-chan struct{} { } return m.done } + +func TestSyncResult(t *testing.T) { + for name, tc := range map[string]struct { + result SyncResult + expectAsError string + }{ + "empty": {}, + "one": { + result: SyncResult{ + Err: errors.New("my custom cancellation reason"), + Synced: map[reflect.Type]bool{ + reflect.TypeFor[int](): true, + reflect.TypeFor[string](): false, + }, + }, + expectAsError: "failed to sync all caches: string: my custom cancellation reason", + }, + "many": { + result: SyncResult{ + Err: errors.New("my custom cancellation reason"), + Synced: map[reflect.Type]bool{ + reflect.TypeFor[int](): false, + reflect.TypeFor[string](): false, + }, + }, + expectAsError: "failed to sync all caches: int, string: my custom cancellation reason", + }, + } { + + t.Run(name, func(t *testing.T) { + actual := tc.result.AsError() + switch { + case tc.expectAsError == "" && actual != nil: + t.Fatalf("expected no error, got %v", actual) + case tc.expectAsError != "" && actual == nil: + t.Fatalf("expected %q, got no error", actual) + case tc.expectAsError != "" && actual != nil && actual.Error() != tc.expectAsError: + t.Fatalf("expected %q, got %q", tc.expectAsError, actual.Error()) + } + if tc.result.Err != nil && !errors.Is(actual, tc.result.Err) { + t.Errorf("actual error %+v should wrap %v but doesn't", actual, tc.result.Err) + } + }) + } +}