From a01b5a2a9698c777c454bc498e94b245d4c750f9 Mon Sep 17 00:00:00 2001 From: Odin Ugedal Date: Mon, 13 Feb 2023 11:12:37 +0000 Subject: [PATCH] client-go/cache: rewrite Replace to check queue first This is useful to both reduce the code complexity, and to ensure clients get the "newest" version of an object known when its deleted. This is all best-effort, but for clients it makes more sense giving them the newest object they observed rather than an old one. This is especially useful when an object is recreated. eg. Object A with key K is in the KnownObjects store; - DELETE delta for A is queued with key K - CREATE delta for B is queued with key K - Replace without any object with key K in it. In this situation its better to create a DELETE delta with DeletedFinalStateUnknown with B (with this patch), than it is to give the client an DeletedFinalStateUnknown with A (without this patch). Signed-off-by: Odin Ugedal Signed-off-by: Odin Ugedal Kubernetes-commit: 7bcc3e00fc28b2548886d04639a2e352ab37fb55 --- tools/cache/delta_fifo.go | 88 +++++++++++----------------------- tools/cache/delta_fifo_test.go | 65 ++++++++++++++++++++++++- 2 files changed, 90 insertions(+), 63 deletions(-) diff --git a/tools/cache/delta_fifo.go b/tools/cache/delta_fifo.go index bb861989..b83e9d2d 100644 --- a/tools/cache/delta_fifo.go +++ b/tools/cache/delta_fifo.go @@ -600,72 +600,12 @@ func (f *DeltaFIFO) Replace(list []interface{}, _ string) error { } } - if f.knownObjects == nil { - // Do deletion detection against our own list. - queuedDeletions := 0 - for k, oldItem := range f.items { - if keys.Has(k) { - continue - } - // Delete pre-existing items not in the new list. - // This could happen if watch deletion event was missed while - // disconnected from apiserver. - var deletedObj interface{} - if n := oldItem.Newest(); n != nil { - deletedObj = n.Object - - // if the previous object is a DeletedFinalStateUnknown, we have to extract the actual Object - if d, ok := deletedObj.(DeletedFinalStateUnknown); ok { - deletedObj = d.Obj - } - } - queuedDeletions++ - if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil { - return err - } - } - - if !f.populated { - f.populated = true - // While there shouldn't be any queued deletions in the initial - // population of the queue, it's better to be on the safe side. - f.initialPopulationCount = keys.Len() + queuedDeletions - } - - return nil - } - - // Detect deletions not already in the queue. - knownKeys := f.knownObjects.ListKeys() + // Do deletion detection against objects in the queue queuedDeletions := 0 - for _, k := range knownKeys { - if keys.Has(k) { - continue - } - - deletedObj, exists, err := f.knownObjects.GetByKey(k) - if err != nil { - deletedObj = nil - klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k) - } else if !exists { - deletedObj = nil - klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k) - } - queuedDeletions++ - if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil { - return err - } - } - - // detect deletions for items in the queue that are not yet present in knownObjects for k, oldItem := range f.items { if keys.Has(k) { continue } - _, exists, _ := f.knownObjects.GetByKey(k) - if exists { - continue - } // Delete pre-existing items not in the new list. // This could happen if watch deletion event was missed while // disconnected from apiserver. @@ -684,6 +624,32 @@ func (f *DeltaFIFO) Replace(list []interface{}, _ string) error { } } + if f.knownObjects != nil { + // Detect deletions for objects not present in the queue, but present in KnownObjects + knownKeys := f.knownObjects.ListKeys() + for _, k := range knownKeys { + if keys.Has(k) { + continue + } + if len(f.items[k]) > 0 { + continue + } + + deletedObj, exists, err := f.knownObjects.GetByKey(k) + if err != nil { + deletedObj = nil + klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k) + } else if !exists { + deletedObj = nil + klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k) + } + queuedDeletions++ + if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil { + return err + } + } + } + if !f.populated { f.populated = true f.initialPopulationCount = keys.Len() + queuedDeletions diff --git a/tools/cache/delta_fifo_test.go b/tools/cache/delta_fifo_test.go index ee69192a..9d124165 100644 --- a/tools/cache/delta_fifo_test.go +++ b/tools/cache/delta_fifo_test.go @@ -121,7 +121,7 @@ func TestDeltaFIFO_replaceWithDeleteDeltaIn(t *testing.T) { } } -func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsInQueue(t *testing.T) { +func TestDeltaFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) { obj := mkFifoObj("foo", 2) objV2 := mkFifoObj("foo", 3) table := []struct { @@ -495,7 +495,7 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { expectedList = []Deltas{ {{Added, mkFifoObj("baz", 10)}, - {Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}}, + {Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 10)}}}, {{Sync, mkFifoObj("foo", 5)}}, // Since "bar" didn't have a delete event and wasn't in the Replace list // it should get a tombstone key with the right Obj. @@ -509,6 +509,67 @@ func TestDeltaFIFO_ReplaceMakesDeletions(t *testing.T) { } } + // Now try deleting and recreating the object in the queue, then delete it by a Replace call + f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: literalListerGetter(func() []testFifoObject { + return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)} + }), + }) + f.Delete(mkFifoObj("bar", 6)) + f.Add(mkFifoObj("bar", 100)) + f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0") + + expectedList = []Deltas{ + { + {Deleted, mkFifoObj("bar", 6)}, + {Added, mkFifoObj("bar", 100)}, + // Since "bar" has a newer object in the queue than in the state, + // it should get a tombstone key with the latest object from the queue + {Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}}, + }, + {{Sync, mkFifoObj("foo", 5)}}, + {{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}}, + } + + for _, expected := range expectedList { + cur := Pop(f).(Deltas) + if e, a := expected, cur; !reflect.DeepEqual(e, a) { + t.Errorf("Expected %#v, got %#v", e, a) + } + } + + // Now try syncing it first to ensure the delete use the latest version + f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: literalListerGetter(func() []testFifoObject { + return []testFifoObject{mkFifoObj("foo", 5), mkFifoObj("bar", 6), mkFifoObj("baz", 7)} + }), + }) + f.Replace([]interface{}{mkFifoObj("bar", 100), mkFifoObj("foo", 5)}, "0") + f.Replace([]interface{}{mkFifoObj("foo", 5)}, "0") + + expectedList = []Deltas{ + { + {Sync, mkFifoObj("bar", 100)}, + // Since "bar" didn't have a delete event and wasn't in the Replace list + // it should get a tombstone key with the right Obj. + {Deleted, DeletedFinalStateUnknown{Key: "bar", Obj: mkFifoObj("bar", 100)}}, + }, + { + {Sync, mkFifoObj("foo", 5)}, + {Sync, mkFifoObj("foo", 5)}, + }, + {{Deleted, DeletedFinalStateUnknown{Key: "baz", Obj: mkFifoObj("baz", 7)}}}, + } + + for _, expected := range expectedList { + cur := Pop(f).(Deltas) + if e, a := expected, cur; !reflect.DeepEqual(e, a) { + t.Errorf("Expected %#v, got %#v", e, a) + } + } + // Now try starting without an explicit KeyListerGetter f = NewDeltaFIFOWithOptions(DeltaFIFOOptions{KeyFunction: testFifoObjectKeyFunc}) f.Add(mkFifoObj("baz", 10))