Add support for ReplacedAll delta handling

staging/src/k8s.io/client-go/tools/cache/controller.go

@@ -591,6 +591,14 @@ func processDeltas(
 		obj := d.Object
 
 		switch d.Type {
+		case ReplacedAll:
+			info, ok := obj.(ReplacedAllInfo)
+			if !ok {
+				return fmt.Errorf("ReplacedAll did not contain ReplacedAllInfo: %T", obj)
+			}
+			if err := processReplacedAllInfo(handler, info, clientState, isInInitialList, keyFunc); err != nil {
+				return err
+			}
 		case Sync, Replaced, Added, Updated:
 			if old, exists, err := clientState.Get(obj); err == nil && exists {
 				if err := clientState.Update(obj); err != nil {
@@ -701,6 +709,52 @@ func processDeltasInBatch(
 	return nil
 }
 
+func processReplacedAllInfo(handler ResourceEventHandler, info ReplacedAllInfo, clientState Store, isInInitialList bool, keyFunc KeyFunc) error {
+	var deletions []DeletedFinalStateUnknown
+	type replacement struct {
+		oldObj interface{}
+		newObj interface{}
+	}
+	replacements := make([]replacement, 0, len(info.Objects))
+
+	err := reconcileReplacement(nil, clientState, info.Objects, keyFunc,
+		func(obj DeletedFinalStateUnknown) error {
+			deletions = append(deletions, obj)
+			return nil
+		},
+		func(obj interface{}) error {
+			// This behavior matches processDeltas handling of Replace deltas
+			if old, exists, err := clientState.Get(obj); err == nil && exists {
+				replacements = append(replacements, replacement{newObj: obj, oldObj: old})
+			} else {
+				replacements = append(replacements, replacement{newObj: obj})
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	// Replace the client state first so the store reflects the events handlers are given
+	if err := clientState.Replace(info.Objects, info.ResourceVersion); err != nil {
+		return err
+	}
+	// Processing all deletions first matches behavior of RealFIFO#Replace
+	for _, objToDelete := range deletions {
+		handler.OnDelete(objToDelete)
+	}
+	// Processing adds/updates in order observed by reconcileReplacement matches behavior of RealFIFO#Replace
+	for _, r := range replacements {
+		if r.oldObj != nil {
+			handler.OnUpdate(r.oldObj, r.newObj)
+		} else {
+			handler.OnAdd(r.newObj, isInInitialList)
+		}
+	}
+	return nil
+}
+
 // newInformer returns a controller for populating the store while also
 // providing event notifications.
 //

staging/src/k8s.io/client-go/tools/cache/delta_fifo.go

@@ -179,6 +179,12 @@ const (
 	// as well. Hence, Replaced is only emitted when the option
 	// EmitDeltaTypeReplaced is true.
 	Replaced DeltaType = "Replaced"
+	// ReplacedAll is emitted when we encountered watch errors and had to do
+	// a relist, or on initial listing of objects. This is the same reason as
+	// Replaced but will be emitted instead when the FIFO supports atomic
+	// replacement. This event will return the full list of replaced items
+	// instead of a single object.
+	ReplacedAll DeltaType = "ReplacedAll"
 	// Sync is for synthetic events during a periodic resync.
 	Sync DeltaType = "Sync"
 )

staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go

@@ -98,6 +98,15 @@ type RealFIFO struct {
 	emitAtomicEvents bool
 }
 
+// ReplacedAllInfo is the object associated with a Delta of type=ReplacedAll
+type ReplacedAllInfo struct {
+	// ResourceVersion is the resource version passed to the Replace() call that created this Delta
+	ResourceVersion string
+	// Objects are the list of objects passed to the Replace() call that created this Delta,
+	// with any configured transformation already applied.
+	Objects []interface{}
+}
+
 var (
 	_ = Queue(&RealFIFO{})             // RealFIFO is a Queue
 	_ = TransformingStore(&RealFIFO{}) // RealFIFO implements TransformingStore to allow memory optimizations
@@ -183,6 +192,35 @@ func (f *RealFIFO) addToItems_locked(deltaActionType DeltaType, skipTransform bo
 	return nil
 }
 
+// addReplaceToItemsLocked appends to the delta list.
+func (f *RealFIFO) addReplaceToItemsLocked(objs []interface{}, resourceVersion string) error {
+	// Replaced items must be transformed before being added to the queue. These objects must
+	// all be objects that have not been transformed yet.
+	if f.transformer != nil {
+		transformedObjs := make([]interface{}, len(objs))
+		for i, obj := range objs {
+			transformedObj, err := f.transformer(obj)
+			if err != nil {
+				return err
+			}
+			transformedObjs[i] = transformedObj
+		}
+		objs = transformedObjs
+	}
+
+	info := ReplacedAllInfo{
+		ResourceVersion: resourceVersion,
+		Objects:         objs,
+	}
+	f.items = append(f.items, Delta{
+		Type:   ReplacedAll,
+		Object: info,
+	})
+	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 {
@@ -374,13 +412,17 @@ func (f *RealFIFO) Replace(newItems []interface{}, resourceVersion string) error
 	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 f.emitAtomicEvents {
+		err = f.addReplaceToItemsLocked(newItems, resourceVersion)
+	} else {
+		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
 	}

staging/src/k8s.io/client-go/tools/cache/the_real_fifo_test.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+
 	clientfeatures "k8s.io/client-go/features"
 	clientfeaturestesting "k8s.io/client-go/features/testing"
 )
@@ -39,6 +40,7 @@ func (f *RealFIFO) getItems() []Delta {
 }
 
 const closedFIFOName = "FIFO WAS CLOSED"
+const isAtomic = "ATOMIC REPLACED OBJ"
 
 func popN(queue Queue, count int) []interface{} {
 	result := []interface{}{}
@@ -57,6 +59,13 @@ func testRealFIFOPop(f *RealFIFO) testFifoObject {
 	if val == nil {
 		return testFifoObject{name: closedFIFOName}
 	}
+	if val.(Deltas).Newest().Type == ReplacedAll {
+		var objs []testFifoObject
+		for _, obj := range val.(Deltas).Newest().Object.(ReplacedAllInfo).Objects {
+			objs = append(objs, obj.(testFifoObject))
+		}
+		return testFifoObject{name: isAtomic, val: objs}
+	}
 	return val.(Deltas).Newest().Object.(testFifoObject)
 }
 
@@ -1298,3 +1307,147 @@ func TestRealFIFO_PopBrokenItemsInBatch(t *testing.T) {
 		})
 	}
 }
+
+func TestRealFIFO_ReplaceAtomic(t *testing.T) {
+	obj := mkFifoObj("foo", 2)
+	table := []struct {
+		name           string
+		operations     func(f *RealFIFO)
+		expectedDeltas Deltas
+	}{
+		{
+			name: "Base replace",
+			operations: func(f *RealFIFO) {
+				f.replaceTest(t, []interface{}{}, "123")
+			},
+			expectedDeltas: Deltas{
+				{Type: ReplacedAll, Object: ReplacedAllInfo{
+					ResourceVersion: "123",
+					Objects:         []interface{}{},
+				}},
+			},
+		},
+		{
+			name: "Added object exists in deltas on Atomic Replace",
+			operations: func(f *RealFIFO) {
+				f.addTest(t, obj)
+				f.replaceTest(t, []interface{}{}, "123")
+			},
+			expectedDeltas: Deltas{
+				{Type: Added, Object: obj},
+				{Type: ReplacedAll, Object: ReplacedAllInfo{
+					ResourceVersion: "123",
+					Objects:         []interface{}{},
+				}},
+			},
+		},
+		{
+			name: "Multiple replaces run in order",
+			operations: func(f *RealFIFO) {
+				f.addTest(t, obj)
+				f.replaceTest(t, []interface{}{obj}, "10")
+				f.replaceTest(t, []interface{}{obj}, "20")
+				f.replaceTest(t, []interface{}{}, "56")
+			},
+			expectedDeltas: Deltas{
+				{Type: Added, Object: obj},
+				{Type: ReplacedAll, Object: ReplacedAllInfo{
+					ResourceVersion: "10",
+					Objects:         []interface{}{obj},
+				}},
+				{Type: ReplacedAll, Object: ReplacedAllInfo{
+					ResourceVersion: "20",
+					Objects:         []interface{}{obj},
+				}},
+				{Type: ReplacedAll, Object: ReplacedAllInfo{
+					ResourceVersion: "56",
+					Objects:         []interface{}{},
+				}},
+			},
+		},
+	}
+	for _, tt := range table {
+		tt := tt
+
+		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)
+			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 TestRealFIFO_ReplaceAtomicPop(t *testing.T) {
+	f := NewRealFIFO(
+		testFifoObjectKeyFunc,
+		emptyKnownObjects(),
+		nil,
+	)
+	f.emitAtomicEvents = true
+	f.addTest(t, mkFifoObj("foo", 10))
+	f.replaceTest(t, []interface{}{mkFifoObj("foo", 15)}, "20")
+	got := make(chan testFifoObject, 3)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for {
+			obj := testRealFIFOPop(f)
+			if obj.name == closedFIFOName {
+				break
+			}
+			got <- obj
+		}
+	}()
+
+	added := <-got
+	if e, a, v := "foo", added.name, added.val.(int); e != a || v != 10 {
+		t.Errorf("Didn't get expected name (%v vs %v) or value (%v vs %v)", e, a, v, 10)
+	}
+
+	curr := <-got
+	if e, a, v := isAtomic, curr.name, curr.val.([]testFifoObject)[0]; e != a && v.val == 15 {
+		t.Errorf("Didn't get updated value (%v), got %v", e, a)
+	}
+
+	select {
+	case unexpected := <-got:
+		t.Errorf("Got second value %v", unexpected.val)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	if items := f.getItems(); len(items) > 0 {
+		t.Errorf("item did not get removed")
+	}
+	f.Close()
+	<-done
+}
+
+func (f *RealFIFO) addTest(t *testing.T, obj interface{}) {
+	err := f.Add(obj)
+	if err != nil {
+		t.Fatalf("Test error on RealFIFO add: %s", err)
+	}
+}
+
+func (f *RealFIFO) replaceTest(t *testing.T, objs []interface{}, rv string) {
+	err := f.Replace(objs, rv)
+	if err != nil {
+		t.Fatalf("Test error on RealFIFO replace: %s", err)
+	}
+}