Add atomic resync and remove usage of store in FIFO

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,

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{

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

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{

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