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 1fc8a40aa6d..bffdbeb0a1c 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 @@ -341,22 +341,52 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error f.lock.Lock() defer f.lock.Unlock() + var err error + err = reconcileReplacement(f.items, f.knownObjects, newItems, f.keyOf, + func(obj DeletedFinalStateUnknown) error { + return f.addToItems_locked(Deleted, true, obj) + }, + func(obj interface{}) error { + return f.addToItems_locked(Replaced, false, obj) + }) + if err != nil { + return err + } + + if !f.populated { + f.populated = true + f.initialPopulationCount = len(f.items) + } + + return nil +} + +// reconcileReplacement takes the items that are already in the queue and the set of new items +// and based upon the state of the items in the queue and known objects will call onDelete and onReplace +// depending upon whether the item is being deleted or replaced/added. +func reconcileReplacement( + queuedItems []Delta, + knownObjects KeyListerGetter, + newItems []interface{}, + keyOf func(obj interface{}) (string, error), + onDelete func(obj DeletedFinalStateUnknown) error, + onReplace func(obj interface{}) error, +) error { // determine the keys of everything we're adding. We cannot add the items until after the synthetic deletes have been // created for items that don't existing in newItems newKeys := sets.Set[string]{} for _, obj := range newItems { - key, err := f.keyOf(obj) + key, err := keyOf(obj) if err != nil { return KeyError{obj, err} } newKeys.Insert(key) } - queuedItems := f.items queuedKeys := []string{} lastQueuedItemForKey := map[string]Delta{} for _, queuedItem := range queuedItems { - queuedKey, err := f.keyOf(queuedItem.Object) + queuedKey, err := keyOf(queuedItem.Object) if err != nil { return KeyError{queuedItem.Object, err} } @@ -388,7 +418,7 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error // if we got here, then the last entry we have for the queued item is *not* a deletion and we need to add a delete deletedObj := lastQueuedItem.Object - retErr := f.addToItems_locked(Deleted, true, DeletedFinalStateUnknown{ + retErr := onDelete(DeletedFinalStateUnknown{ Key: queuedKey, Obj: deletedObj, }) @@ -398,7 +428,7 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error } // Detect deletions for objects not present in the queue, but present in KnownObjects - knownKeys := f.knownObjects.ListKeys() + knownKeys := knownObjects.ListKeys() for _, knownKey := range knownKeys { if newKeys.Has(knownKey) { // still present continue @@ -407,7 +437,7 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error continue } - deletedObj, exists, err := f.knownObjects.GetByKey(knownKey) + deletedObj, exists, err := knownObjects.GetByKey(knownKey) if err != nil { deletedObj = nil utilruntime.HandleError(fmt.Errorf("error during lookup, placing DeleteFinalStateUnknown marker without object: key=%q, err=%w", knownKey, err)) @@ -415,7 +445,7 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error deletedObj = nil utilruntime.HandleError(fmt.Errorf("key does not exist in known objects store, placing DeleteFinalStateUnknown marker without object: key=%q", knownKey)) } - retErr := f.addToItems_locked(Deleted, false, DeletedFinalStateUnknown{ + retErr := onDelete(DeletedFinalStateUnknown{ Key: knownKey, Obj: deletedObj, }) @@ -426,17 +456,11 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error // now that we have the deletes we need for items, we can add the newItems to the items queue for _, obj := range newItems { - retErr := f.addToItems_locked(Replaced, false, obj) - if retErr != nil { - return fmt.Errorf("couldn't enqueue object: %w", retErr) + if err := onReplace(obj); err != nil { + return fmt.Errorf("couldn't enqueue object: %w", err) } } - if !f.populated { - f.populated = true - f.initialPopulationCount = len(f.items) - } - return nil }