client-go informers: context-aware Start + WaitForCacheSync

Passing a context to StartWithContext enables context-aware reflector
logging. This is the main remaining source of log spam (output to stderr
instead of per-test logger) in controller unit tests.

WaitForCacheSynceWithContext takes advantage of the new cache.WaitFor +
NamedHasSynced functionality to finish "immediately" (= no virtual time
passed) in a synctest bubble. While at it, the return type gets improved so
that a failure is easier to handle.

Kubernetes-commit: 5ff323de791df88880f6e065f5de4b445e5c90ed

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)
 

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 != "" {

tools/cache/event_handler_name_test.go

@@ -76,6 +76,12 @@ func TestNameForHandler(t *testing.T) {
 			handler:  nil,
 			wantName: "<nil>",
 		},
+		"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)

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="<what>").
+// while waiting ("Waiting",  for="<what>"). 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)
 	}

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)
+			}
+		})
+	}
+}