Plumb bookmark RVs to controller for observation

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
 		}

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

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)
 			}

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),

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)
+	}
+}