github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-18 13:57:38 +00:00
Code Issues Projects Releases Wiki Activity

client-go informers: context-aware Start + WaitForCacheSync

Browse Source 
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.
This commit is contained in:
Patrick Ohly
2025-11-21 16:23:44 +01:00
parent fdcbb6cba9
commit 5ff323de79
17 changed files with 992 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
staging/src/k8s.io/apiextensions-apiserver/examples/client-go/pkg/client/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
import (
context "context"
reflect "reflect"
sync "sync"
time "time"
@@ -29,6 +30,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"
cache "k8s.io/client-go/tools/cache"
)
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)

91
staging/src/k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
import (
context "context"
reflect "reflect"
sync "sync"
time "time"
@@ -29,6 +30,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"
cache "k8s.io/client-go/tools/cache"
)
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client clientset.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)

91
staging/src/k8s.io/client-go/informers/factory.go
View File

@@ -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)

4
staging/src/k8s.io/client-go/tools/cache/event_handler_name.go vendored
View File

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

6
staging/src/k8s.io/client-go/tools/cache/event_handler_name_test.go vendored
View File

@@ -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)

51
staging/src/k8s.io/client-go/tools/cache/shared_informer.go vendored
View File

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

46
staging/src/k8s.io/client-go/tools/cache/wait_test.go vendored
View File

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

145
staging/src/k8s.io/code-generator/cmd/informer-gen/generators/factory.go
View File

@@ -74,9 +74,15 @@ func (g *factoryGenerator) GenerateType(c *generator.Context, t *types.Type, w i
gvNewFuncs[groupPkgName] = c.Universe.Function(types.Name{Package: path.Join(g.outputPackage, groupPkgName), Name: "New"})
}
m := map[string]interface{}{
"cacheDoneChecker": c.Universe.Type(cacheDoneChecker),
"cacheInformerName": c.Universe.Type(cacheInformerName),
"cacheSharedIndexInformer": c.Universe.Type(cacheSharedIndexInformer),
"cacheSyncResult": c.Universe.Type(cacheSyncResult),
"cacheTransformFunc": c.Universe.Type(cacheTransformFunc),
"cacheWaitFor": c.Universe.Function(cacheWaitForFunc),
"contextContext": c.Universe.Type(contextContext),
"contextCause": c.Universe.Function(contextCauseFunc),
"fmtErrorf": c.Universe.Function(fmtErrorfFunc),
"groupVersions": g.groupVersions,
"gvInterfaces": gvInterfaces,
"gvNewFuncs": gvNewFuncs,
@@ -88,10 +94,12 @@ func (g *factoryGenerator) GenerateType(c *generator.Context, t *types.Type, w i
"reflectType": c.Universe.Type(reflectType),
"runtimeObject": c.Universe.Type(runtimeObject),
"schemaGroupVersionResource": c.Universe.Type(schemaGroupVersionResource),
"stringsBuilder": c.Universe.Type(stringsBuilder),
"syncMutex": c.Universe.Type(syncMutex),
"timeDuration": c.Universe.Type(timeDuration),
"namespaceAll": c.Universe.Type(metav1NamespaceAll),
"object": c.Universe.Type(metav1Object),
"waitContextForChannel": c.Universe.Function(waitContextForChannelFunc),
}
sw.Do(sharedInformerFactoryStruct, m)
@@ -208,6 +216,10 @@ func NewSharedInformerFactoryWithOptions(client {{.clientSetInterface|raw}}, def
}
func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
f.StartWithContext({{.waitContextForChannel|raw}}(stopCh))
}
func (f *sharedInformerFactory) StartWithContext(ctx {{.contextContext|raw}}) {
f.lock.Lock()
defer f.lock.Unlock()
@@ -217,15 +229,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
}
}
@@ -243,24 +249,50 @@ func (f *sharedInformerFactory) Shutdown() {
}
func (f *sharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool {
informers := func()map[reflect.Type]cache.SharedIndexInformer{
f.lock.Lock()
defer f.lock.Unlock()
result := f.WaitForCacheSyncWithContext(wait.ContextForChannel(stopCh))
return result.Synced
}
informers := map[reflect.Type]cache.SharedIndexInformer{}
for informerType, informer := range f.informers {
if f.startedInformers[informerType] {
informers[informerType] = informer
}
}
return informers
}()
func (f *sharedInformerFactory) WaitForCacheSyncWithContext(ctx context.Context) {{.cacheSyncResult|raw}} {
informers := func() map[{{.reflectType|raw}}]{{.cacheSharedIndexInformer|raw}} {
f.lock.Lock()
defer f.lock.Unlock()
res := map[reflect.Type]bool{}
for informType, informer := range informers {
res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced)
}
return res
informers := map[{{.reflectType|raw}}]{{.cacheSharedIndexInformer|raw}}{}
for informerType, informer := range f.informers {
if f.startedInformers[informerType] {
informers[informerType] = informer
}
}
return informers
}()
// Wait for informers to sync, without polling.
cacheSyncs := make([]{{.cacheDoneChecker|raw}}, 0, len(informers))
for _, informer := range informers {
cacheSyncs = append(cacheSyncs, informer.HasSyncedChecker())
}
{{.cacheWaitFor|raw}}(ctx, "" /* no logging */, cacheSyncs...)
res := {{.cacheSyncResult|raw}} {
Synced: make(map[{{.reflectType|raw}}]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 = {{.contextCause|raw}}(ctx)
}
return res
}
// InformerFor returns the SharedIndexInformer for obj using an internal
@@ -294,33 +326,52 @@ var sharedInformerFactoryInterface = `
//
// It is typically used like this:
//
// ctx, cancel := context.WithCancel(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
// }
// }
// ctx, cancel := context.WithCancel(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()
// 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)
// }
//
// // Creating informers can also be created after Start, but then
// // Start must be called again:
// anotherGenericInformer := factory.ForResource(resource)
// factory.Start(ctx.Done())
// // 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.StartWithContext(ctx)
type SharedInformerFactory interface {
{{.informerFactoryInterface|raw}}
// 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.
@@ -335,7 +386,13 @@ type SharedInformerFactory interface {
// WaitForCacheSync blocks until all started informers' caches were synced
// or the stop channel gets closed.
WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
//
// 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[{{.reflectType|raw}}]bool
// WaitForCacheSyncWithContext blocks until all started informers' caches were synced
// or the context gets canceled.
WaitForCacheSyncWithContext(ctx {{.contextContext|raw}}) {{.cacheSyncResult|raw}}
// ForResource gives generic access to a shared informer of the matching type.
ForResource(resource {{.schemaGroupVersionResource|raw}}) (GenericInformer, error)

6
staging/src/k8s.io/code-generator/cmd/informer-gen/generators/types.go
View File

@@ -20,6 +20,7 @@ import "k8s.io/gengo/v2/types"
var (
apiScheme = types.Name{Package: "k8s.io/kubernetes/pkg/api/legacyscheme", Name: "Scheme"}
cacheDoneChecker = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "DoneChecker"}
cacheGenericLister = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "GenericLister"}
cacheIndexers = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "Indexers"}
cacheInformerName = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "InformerName"}
@@ -31,9 +32,12 @@ var (
cacheNewSharedIndexInformerWithOptions = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "NewSharedIndexInformerWithOptions"}
cacheSharedIndexInformer = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "SharedIndexInformer"}
cacheSharedIndexInformerOptions = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "SharedIndexInformerOptions"}
cacheSyncResult = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "SyncResult"}
cacheTransformFunc = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "TransformFunc"}
cacheToListWatcherWithWatchListSemanticsFunc = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "ToListWatcherWithWatchListSemantics"}
cacheWaitForFunc = types.Name{Package: "k8s.io/client-go/tools/cache", Name: "WaitFor"}
contextBackgroundFunc = types.Name{Package: "context", Name: "Background"}
contextCauseFunc = types.Name{Package: "context", Name: "Cause"}
contextContext = types.Name{Package: "context", Name: "Context"}
fmtErrorfFunc = types.Name{Package: "fmt", Name: "Errorf"}
listOptions = types.Name{Package: "k8s.io/kubernetes/pkg/apis/core", Name: "ListOptions"}
@@ -41,10 +45,12 @@ var (
runtimeObject = types.Name{Package: "k8s.io/apimachinery/pkg/runtime", Name: "Object"}
schemaGroupResource = types.Name{Package: "k8s.io/apimachinery/pkg/runtime/schema", Name: "GroupResource"}
schemaGroupVersionResource = types.Name{Package: "k8s.io/apimachinery/pkg/runtime/schema", Name: "GroupVersionResource"}
stringsBuilder = types.Name{Package: "strings", Name: "Builder"}
syncMutex = types.Name{Package: "sync", Name: "Mutex"}
timeDuration = types.Name{Package: "time", Name: "Duration"}
v1ListOptions = types.Name{Package: "k8s.io/apimachinery/pkg/apis/meta/v1", Name: "ListOptions"}
metav1NamespaceAll = types.Name{Package: "k8s.io/apimachinery/pkg/apis/meta/v1", Name: "NamespaceAll"}
metav1Object = types.Name{Package: "k8s.io/apimachinery/pkg/apis/meta/v1", Name: "Object"}
waitContextForChannelFunc = types.Name{Package: "k8s.io/apimachinery/pkg/util/wait", Name: "ContextForChannel"}
watchInterface = types.Name{Package: "k8s.io/apimachinery/pkg/watch", Name: "Interface"}
)

91
staging/src/k8s.io/code-generator/examples/HyphenGroup/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
versioned "k8s.io/code-generator/examples/HyphenGroup/clientset/versioned"
example "k8s.io/code-generator/examples/HyphenGroup/informers/externalversions/example"
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)

91
staging/src/k8s.io/code-generator/examples/MixedCase/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
versioned "k8s.io/code-generator/examples/MixedCase/clientset/versioned"
example "k8s.io/code-generator/examples/MixedCase/informers/externalversions/example"
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)

91
staging/src/k8s.io/code-generator/examples/apiserver/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
versioned "k8s.io/code-generator/examples/apiserver/clientset/versioned"
core "k8s.io/code-generator/examples/apiserver/informers/externalversions/core"
@@ -142,6 +144,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -151,15 +157,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
}
}
@@ -176,6 +176,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()
@@ -189,10 +194,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
}
@@ -231,27 +257,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.
@@ -266,8 +311,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)

91
staging/src/k8s.io/code-generator/examples/crd/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
versioned "k8s.io/code-generator/examples/crd/clientset/versioned"
conflicting "k8s.io/code-generator/examples/crd/informers/externalversions/conflicting"
@@ -142,6 +144,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -151,15 +157,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
}
}
@@ -176,6 +176,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()
@@ -189,10 +194,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
}
@@ -231,27 +257,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.
@@ -266,8 +311,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)

91
staging/src/k8s.io/code-generator/examples/single/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
versioned "k8s.io/code-generator/examples/single/clientset/versioned"
api "k8s.io/code-generator/examples/single/informers/externalversions/api"
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)

91
staging/src/k8s.io/kube-aggregator/pkg/client/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
clientset "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset"
apiregistration "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration"
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client clientset.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)

91
staging/src/k8s.io/sample-apiserver/pkg/generated/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
versioned "k8s.io/sample-apiserver/pkg/generated/clientset/versioned"
internalinterfaces "k8s.io/sample-apiserver/pkg/generated/informers/externalversions/internalinterfaces"
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)

91
staging/src/k8s.io/sample-controller/pkg/generated/informers/externalversions/factory.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package externalversions
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"
cache "k8s.io/client-go/tools/cache"
versioned "k8s.io/sample-controller/pkg/generated/clientset/versioned"
internalinterfaces "k8s.io/sample-controller/pkg/generated/informers/externalversions/internalinterfaces"
@@ -139,6 +141,10 @@ func NewSharedInformerFactoryWithOptions(client versioned.Interface, defaultResy
}
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()
@@ -148,15 +154,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
}
}
@@ -173,6 +173,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()
@@ -186,10 +191,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
}
@@ -228,27 +254,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.
@@ -263,8 +308,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)