diff --git a/staging/src/k8s.io/client-go/tools/cache/controller.go b/staging/src/k8s.io/client-go/tools/cache/controller.go index 1913fc309fe..fa78a4ca7b1 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller.go @@ -599,6 +599,16 @@ func processDeltas( if err := processReplacedAllInfo(handler, info, clientState, isInInitialList, keyFunc); err != nil { return err } + case SyncAll: + _, ok := obj.(SyncAllInfo) + if !ok { + return fmt.Errorf("SyncAll did not contain SyncAllInfo: %T", obj) + } + objs := clientState.List() + for _, obj := range objs { + handler.OnUpdate(obj, obj) + } + return nil case Sync, Replaced, Added, Updated: if old, exists, err := clientState.Get(obj); err == nil && exists { if err := clientState.Update(obj); err != nil { @@ -790,12 +800,18 @@ func newInformer(clientState Store, options InformerOptions, keyFunc KeyFunc) Co func newQueueFIFO(clientState Store, transform TransformFunc) Queue { if clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.InOrderInformers) { - return NewRealFIFOWithOptions(RealFIFOOptions{ - KeyFunction: MetaNamespaceKeyFunc, - KnownObjects: clientState, - Transformer: transform, - AtomicEvents: clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.AtomicFIFO), - }) + options := RealFIFOOptions{ + KeyFunction: MetaNamespaceKeyFunc, + Transformer: transform, + } + // If atomic events are enabled, unset clientState in the case of atomic events as we cannot pass a + // store to an atomic fifo. + if clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.AtomicFIFO) { + options.AtomicEvents = true + } else { + options.KnownObjects = clientState + } + return NewRealFIFOWithOptions(options) } else { return NewDeltaFIFOWithOptions(DeltaFIFOOptions{ KnownObjects: clientState, diff --git a/staging/src/k8s.io/client-go/tools/cache/controller_test.go b/staging/src/k8s.io/client-go/tools/cache/controller_test.go index 6bd51218270..c4ee71f477e 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller_test.go @@ -912,10 +912,13 @@ func TestReplaceEvents(t *testing.T) { source.Shutdown() }) store := NewStore(DeletionHandlingMetaNamespaceKeyFunc) - fifo := NewRealFIFOWithOptions(RealFIFOOptions{ - KnownObjects: store, + fifoOptions := RealFIFOOptions{ AtomicEvents: atomic, - }) + } + if !atomic { + fifoOptions.KnownObjects = store + } + fifo := NewRealFIFOWithOptions(fifoOptions) recorder := newEventRecorder(store) cfg := &Config{ @@ -1044,11 +1047,15 @@ func TestResetWatch(t *testing.T) { t.Cleanup(func() { source.Shutdown() }) + store := NewStore(DeletionHandlingMetaNamespaceKeyFunc) - fifo := NewRealFIFOWithOptions(RealFIFOOptions{ - KnownObjects: store, + fifoOptions := RealFIFOOptions{ AtomicEvents: atomic, - }) + } + if !atomic { + fifoOptions.KnownObjects = store + } + fifo := NewRealFIFOWithOptions(fifoOptions) recorder := newEventRecorder(store) cfg := &Config{ diff --git a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go index c0f7416bd45..059c78bd366 100644 --- a/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go @@ -187,6 +187,9 @@ const ( ReplacedAll DeltaType = "ReplacedAll" // Sync is for synthetic events during a periodic resync. Sync DeltaType = "Sync" + // SyncAll indicates all known objects should be reprocessed. + // This event contains an object of type SyncAllInfo. + SyncAll DeltaType = "SyncAll" ) // Delta is a member of Deltas (a list of Delta objects) which diff --git a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go index 699f87c4b20..76a68e17f1e 100644 --- a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go +++ b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go @@ -37,7 +37,8 @@ type RealFIFOOptions struct { // KnownObjects is expected to return a list of keys that the consumer of // this queue "knows about". It is used to decide which items are missing // when Replace() is called; 'Deleted' deltas are produced for the missing items. - // KnownObjects is required. + // KnownObjects is required if AtomicEvents is false since it is used to + // query the state of the internal store for Replace and Resync handling. KnownObjects KeyListerGetter // If set, will be called for objects before enqueueing them. Please @@ -46,7 +47,8 @@ type RealFIFOOptions struct { // AtomicEvents is used to specify whether the RealFIFO will emit events // atomically or not. If it is set, a single event will be emitted - // atomically for Replace operations. + // atomically for Replace and Resync operations. + // If AtomicEvents is true, KnownObjects must be nil. AtomicEvents bool } @@ -78,7 +80,8 @@ type RealFIFO struct { keyFunc KeyFunc // knownObjects list keys that are "known" --- affecting Delete(), - // Replace(), and Resync() + // Replace(), and Resync(). + // It is nil if emitAtomicEvents is true. knownObjects KeyListerGetter // Indication the queue is closed. @@ -92,9 +95,12 @@ type RealFIFO struct { // batchSize determines the maximum number of objects we can combine into a batch. batchSize int - // emitAtomicEvents defines whether a replace should only send out one - // batch of objects on a replace event call rather than a number of delete - // and replace events. + // emitAtomicEvents defines whether events like Replace and Resync should be emitted + // atomically rather than as a series of events. This means that any call to the FIFO + // will emit a single event. + // If it is set: + // * a single ReplacedAll event will be emitted instead of multiple Replace events + // * a single SyncAll event will be emitted instead of multiple Sync events emitAtomicEvents bool } @@ -107,6 +113,10 @@ type ReplacedAllInfo struct { Objects []interface{} } +// SyncAllInfo is the object associated with a Delta of type=SyncAll +// It is used to trigger a resync of the entire queue. +type SyncAllInfo struct{} + var ( _ = Queue(&RealFIFO{}) // RealFIFO is a Queue _ = TransformingStore(&RealFIFO{}) // RealFIFO implements TransformingStore to allow memory optimizations @@ -221,6 +231,16 @@ func (f *RealFIFO) addReplaceToItemsLocked(objs []interface{}, resourceVersion s return nil } +func (f *RealFIFO) addResyncToItemsLocked() error { + f.items = append(f.items, Delta{ + Type: SyncAll, + Object: SyncAllInfo{}, + }) + f.cond.Broadcast() + + return nil +} + // Add inserts an item, and puts it in the queue. The item is only enqueued // if it doesn't already exist in the set. func (f *RealFIFO) Add(obj interface{}) error { @@ -545,6 +565,10 @@ func (f *RealFIFO) Resync() error { f.lock.Lock() defer f.lock.Unlock() + if f.emitAtomicEvents { + return f.addResyncToItemsLocked() + } + if f.knownObjects == nil { return nil } @@ -593,6 +617,8 @@ func (f *RealFIFO) Transformer() TransformFunc { // NewRealFIFO returns a Store which can be used to queue up items to // process. +// +// Deprecated: Use NewRealFIFOWithOptions instead. func NewRealFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter, transformer TransformFunc) *RealFIFO { return NewRealFIFOWithOptions(RealFIFOOptions{ KeyFunction: keyFunc, @@ -608,8 +634,16 @@ func NewRealFIFOWithOptions(opts RealFIFOOptions) *RealFIFO { opts.KeyFunction = MetaNamespaceKeyFunc } - if opts.KnownObjects == nil { - panic("coding error: knownObjects must be provided") + if opts.AtomicEvents { + // If we are emitting atomic events, we must not rely on the known objects store + // as it is a requirement to be able to release the lock while processing events. + if opts.KnownObjects != nil { + panic("coding error: knownObjects must not be provided when AtomicEvents is true") + } + } else { + if opts.KnownObjects == nil { + panic("coding error: knownObjects must be provided when AtomicEvents is false") + } } f := &RealFIFO{ diff --git a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go index c5984ef9a03..18363f1a090 100644 --- a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go +++ b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go @@ -1438,6 +1438,60 @@ func TestRealFIFO_ReplaceAtomicPop(t *testing.T) { <-done } +func TestRealFIFO_ResyncAtomic(t *testing.T) { + obj := mkFifoObj("foo", 2) + table := []struct { + name string + operations func(f *RealFIFO) + expectedDeltas Deltas + }{ + { + name: "Base resync", + operations: func(f *RealFIFO) { + f.resyncTest(t) + }, + expectedDeltas: Deltas{ + {Type: SyncAll, Object: SyncAllInfo{}}, + }, + }, + { + name: "Added object exists in deltas on Atomic Resync", + operations: func(f *RealFIFO) { + f.addTest(t, obj) + f.resyncTest(t) + }, + expectedDeltas: Deltas{ + {Type: Added, Object: obj}, + {Type: SyncAll, Object: SyncAllInfo{}}, + }, + }, + } + for _, tt := range table { + t.Run(tt.name, func(t *testing.T) { + // Test with a RealFIFO with a backing KnownObjects + f := NewRealFIFO( + testFifoObjectKeyFunc, + literalListerGetter(func() []testFifoObject { + return []testFifoObject{} + }), + nil, + ) + f.emitAtomicEvents = true + tt.operations(f) + actualDeltasWithKnownObjects := popN(f, len(f.getItems())) + actualAsDeltas := collapseDeltas(actualDeltasWithKnownObjects) + + // Resync appends to the list, so we enable checking preserving the order + if !reflect.DeepEqual(tt.expectedDeltas, actualAsDeltas) { + t.Errorf("expected %#v, got %#v", tt.expectedDeltas, actualAsDeltas) + } + if len(f.items) != 0 { + t.Errorf("expected no extra deltas (empty map), got %#v", f.items) + } + }) + } +} + func (f *RealFIFO) addTest(t *testing.T, obj interface{}) { err := f.Add(obj) if err != nil { @@ -1451,3 +1505,10 @@ func (f *RealFIFO) replaceTest(t *testing.T, objs []interface{}, rv string) { t.Fatalf("Test error on RealFIFO replace: %s", err) } } + +func (f *RealFIFO) resyncTest(t *testing.T) { + err := f.Resync() + if err != nil { + t.Fatalf("Test error on RealFIFO resync: %s", err) + } +}