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 e0b7812b088..1913fc309fe 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/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. // 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 41f26d87471..c0f7416bd45 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 @@ -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" ) 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 50e283f9fb1..699f87c4b20 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 @@ -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 } 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 a5d26ab4a3f..c5984ef9a03 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 @@ -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) + } +}