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 3a4c4cb5cb6..55ab62c763d 100644 --- a/staging/src/k8s.io/client-go/tools/cache/controller.go +++ b/staging/src/k8s.io/client-go/tools/cache/controller.go @@ -653,6 +653,12 @@ func processDeltas( return err } handler.OnDelete(obj) + case Bookmark: + info, ok := obj.(BookmarkInfo) + if !ok { + return fmt.Errorf("bookmark delta did not contain BookmarkInfo: %T", obj) + } + clientState.Bookmark(info.ResourceVersion) } } return nil @@ -853,6 +859,7 @@ func newQueueFIFO(logger klog.Logger, objectType any, clientState Store, transfo if clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.AtomicFIFO) { options.AtomicEvents = true options.UnlockWhileProcessing = clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.UnlockWhileProcessingFIFO) + options.EmitDeltaTypeBookmark = true } else { options.KnownObjects = clientState } 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 b57cc472831..b2a191f78a5 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 @@ -202,6 +202,9 @@ const ( // SyncAll indicates all known objects should be reprocessed. // This event contains an object of type SyncAllInfo. SyncAll DeltaType = "SyncAll" + // Bookmark is emitted on Bookmark calls and Replace calls to pass resource + // version information to the consumer. + Bookmark DeltaType = "Bookmark" ) // Delta is a member of Deltas (a list of Delta objects) which diff --git a/staging/src/k8s.io/client-go/tools/cache/reflector.go b/staging/src/k8s.io/client-go/tools/cache/reflector.go index 62ea68c72ca..66af9febc3a 100644 --- a/staging/src/k8s.io/client-go/tools/cache/reflector.go +++ b/staging/src/k8s.io/client-go/tools/cache/reflector.go @@ -76,6 +76,12 @@ type ReflectorStore interface { Resync() error } +// ReflectorBookmarkStore is an optional interface that allows a store +// to be informed of bookmark events received by the reflector. +type ReflectorBookmarkStore interface { + Bookmark(resourceVersion string) error +} + // TransformingStore is an optional interface that can be implemented by the provided store. // If implemented on the provided store reflector will use the same transformer in its internal stores. type TransformingStore interface { @@ -1006,6 +1012,13 @@ loop: if meta.GetAnnotations()[metav1.InitialEventsAnnotationKey] == "true" { watchListBookmarkReceived = true } + // Propagate the resource version from the bookmark event to stores which indicate they want it + if bookmarkStore, ok := store.(ReflectorBookmarkStore); ok { + err := bookmarkStore.Bookmark(resourceVersion) + if err != nil { + utilruntime.HandleErrorWithContext(ctx, err, "Unable to send bookmark event to store", "reflector", name, "object", event.Object) + } + } default: utilruntime.HandleErrorWithContext(ctx, err, "Unknown watch event", "reflector", name, "event", event) } 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 4856a9bafff..c07c4eef8c4 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 @@ -73,6 +73,10 @@ type RealFIFOOptions struct { // MetricsProvider is used to create metrics for the FIFO. MetricsProvider FIFOMetricsProvider + + // EmitDeltaTypeBookmark is used to specify whether the RealFIFO will emit + // bookmark deltas or not. This can only be set if AtomicEvents is true. + EmitDeltaTypeBookmark bool } const ( @@ -148,6 +152,11 @@ type RealFIFO struct { // metrics holds all metrics for this FIFO. metrics *fifoMetrics + + // emitDeltaTypeBookmark defines whether bookmark deltas should be emitted. + // This may only be set if emitAtomicEvents is true, which avoids events + // propagating out of RV order during Replace and Resync. + emitDeltaTypeBookmark bool } // ReplacedAllInfo is the object associated with a Delta of type=ReplacedAll @@ -159,6 +168,12 @@ type ReplacedAllInfo struct { Objects []interface{} } +// BookmarkInfo is the object associated with a Delta of type=Bookmark +type BookmarkInfo struct { + // ResourceVersion is the resource version passed to the Bookmark() call that created this Delta + ResourceVersion string +} + // 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{} @@ -564,6 +579,21 @@ func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProc }) } +func (f *RealFIFO) Bookmark(resourceVersion string) error { + if !f.emitDeltaTypeBookmark { + return nil + } + f.lock.Lock() + defer f.lock.Unlock() + + f.items = append(f.items, Delta{ + Type: Bookmark, + Object: BookmarkInfo{ResourceVersion: resourceVersion}, + }) + f.cond.Broadcast() + return nil +} + // Replace // 1. finds those items in f.items that are not in newItems and creates synthetic deletes for them // 2. finds items in knownObjects that are not in newItems and creates synthetic deletes for them @@ -790,6 +820,10 @@ func NewRealFIFOWithOptions(opts RealFIFOOptions) *RealFIFO { if opts.KnownObjects == nil { panic("coding error: knownObjects must be provided when AtomicEvents is false") } + // If we are not emitting atomic events, we must not emit bookmark deltas. + if opts.EmitDeltaTypeBookmark { + panic("coding error: EmitDeltaTypeBookmark must be false when AtomicEvents is false") + } } f := &RealFIFO{ @@ -802,6 +836,7 @@ func NewRealFIFOWithOptions(opts RealFIFOOptions) *RealFIFO { transformer: opts.Transformer, batchSize: defaultBatchSize, emitAtomicEvents: opts.AtomicEvents, + emitDeltaTypeBookmark: opts.EmitDeltaTypeBookmark, unlockWhileProcessing: opts.UnlockWhileProcessing, identifier: opts.Identifier, metrics: newFIFOMetrics(opts.Identifier, opts.MetricsProvider), 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 18363f1a090..6b9efbd6e02 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 @@ -41,6 +41,7 @@ func (f *RealFIFO) getItems() []Delta { const closedFIFOName = "FIFO WAS CLOSED" const isAtomic = "ATOMIC REPLACED OBJ" +const isBookmark = "BOOKMARK OBJ" func popN(queue Queue, count int) []interface{} { result := []interface{}{} @@ -66,6 +67,9 @@ func testRealFIFOPop(f *RealFIFO) testFifoObject { } return testFifoObject{name: isAtomic, val: objs} } + if val.(Deltas).Newest().Type == Bookmark { + return testFifoObject{name: isBookmark} + } return val.(Deltas).Newest().Object.(testFifoObject) } @@ -231,6 +235,15 @@ func TestRealFIFOW_ReplaceMakesDeletionsForObjectsOnlyInQueue(t *testing.T) { {Deleted, DeletedFinalStateUnknown{Key: "foo", Obj: objV2}}, }, }, + { + name: "Bookmark object should not be added without atomic", + operations: func(f *RealFIFO) { + f.bookmarkTest(t, "123") + f.replaceTest(t, []interface{}{}, "0") + f.replaceTest(t, []interface{}{}, "1") + }, + expectedDeltas: Deltas{}, + }, } for _, tt := range table { tt := tt @@ -1131,7 +1144,7 @@ func TestRealFIFO_PopMultipleDeltaInBatch(t *testing.T) { for i, item := range tc.initialItems { initialItems[i] = item } - _ = f.Replace(initialItems, "") + _ = f.Replace(initialItems, "123") for _, action := range tc.actions { action(f) } @@ -1365,6 +1378,20 @@ func TestRealFIFO_ReplaceAtomic(t *testing.T) { }}, }, }, + { + name: "Bookmark object should not be included in Replace", + operations: func(f *RealFIFO) { + f.bookmarkTest(t, "123") + f.replaceTest(t, []interface{}{}, "1234") + }, + expectedDeltas: Deltas{ + {Type: Bookmark, Object: BookmarkInfo{ResourceVersion: "123"}}, + {Type: ReplacedAll, Object: ReplacedAllInfo{ + ResourceVersion: "1234", + Objects: []interface{}{}, + }}, + }, + }, } for _, tt := range table { tt := tt @@ -1379,6 +1406,7 @@ func TestRealFIFO_ReplaceAtomic(t *testing.T) { nil, ) f.emitAtomicEvents = true + f.emitDeltaTypeBookmark = true tt.operations(f) actualDeltasWithKnownObjects := popN(f, len(f.getItems())) actualAsDeltas := collapseDeltas(actualDeltasWithKnownObjects) @@ -1512,3 +1540,10 @@ func (f *RealFIFO) resyncTest(t *testing.T) { t.Fatalf("Test error on RealFIFO resync: %s", err) } } + +func (f *RealFIFO) bookmarkTest(t *testing.T, bookmark string) { + err := f.Bookmark(bookmark) + if err != nil { + t.Fatalf("Test error on RealFIFO bookmark: %s", err) + } +}