Merge pull request #142353 from supahero1/optimize-realfifo-batching

Informer cache: do not cut a batch after duplicate key

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

@@ -110,7 +110,7 @@ type ShouldResyncFunc func() bool
 type ProcessFunc func(obj interface{}, isInInitialList bool) error
 
 // ProcessBatchFunc processes multiple objects in batch.
-// The deltas must not contain multiple entries for the same object.
+// The deltas are ordered from oldest to newest and may contain multiple entries for the same object.
 type ProcessBatchFunc func(deltas []Delta, isInInitialList bool) error
 
 // `*controller` implements Controller
@@ -887,8 +887,15 @@ func processDeltas(
 // atomically in a single transaction and corresponding handler callbacks are
 // executed afterward. Otherwise, each Delta is processed individually.
 //
-// Returns an error if any Delta or transaction fails. For TransactionError,
-// only successful operations trigger callbacks.
+// The batch may contain multiple Deltas for the same object. Each of them
+// triggers its own callback, which receives the object as of the previous Delta
+// as the old object. Since callbacks run after the whole transaction, the Store
+// may already reflect later Deltas of the batch when a callback reads it.
+//
+// Returns an error if any Delta or transaction fails. Like with individual
+// processing, a Delta whose object key cannot be computed is skipped without
+// a callback, and the rest of the batch is still processed. For
+// TransactionError, only successful operations trigger callbacks.
 func processDeltasInBatch(
 	logger klog.Logger,
 	handler ResourceEventHandler,
@@ -913,13 +920,26 @@ func processDeltasInBatch(
 		}
 		return nil
 	}
-	// deltasList is a list of unique objects
+	// inBatch holds the newest object of every key seen so far in the batch, or nil if it was
+	// deleted. The store only changes after the loop, so it cannot provide the old object of a
+	// key that repeats within the batch.
+	inBatch := make(map[string]interface{}, len(deltas))
+	var keyErrs []error
 	for _, d := range deltas {
 		obj := d.Object
+		key, err := keyFunc(obj)
+		if err != nil {
+			keyErrs = append(keyErrs, KeyError{obj, err})
+			continue
+		}
 		switch d.Type {
 		case Sync, Replaced, Added, Updated:
-			// it will only return one old object for each because items are unique
-			if old, exists, err := clientState.Get(obj); err == nil && exists {
+			old, seen := inBatch[key]
+			if !seen {
+				old, _, _ = clientState.GetByKey(key)
+			}
+			inBatch[key] = obj
+			if old != nil {
 				txn := Transaction{
 					Type:   TransactionTypeUpdate,
 					Object: obj,
@@ -939,6 +959,7 @@ func processDeltasInBatch(
 				})
 			}
 		case Deleted:
+			inBatch[key] = nil
 			txn := Transaction{
 				Type:   TransactionTypeDelete,
 				Object: obj,
@@ -961,12 +982,12 @@ func processDeltasInBatch(
 			}
 		}
 		// formatting the error so txns doesn't escape and keeps allocated in the stack.
-		return fmt.Errorf("not all items in the batch successfully processed: %s", err.Error())
+		return errors.Join(append(keyErrs, fmt.Errorf("not all items in the batch successfully processed: %s", err.Error()))...)
 	}
 	for _, callback := range callbacks {
 		callback()
 	}
-	return nil
+	return errors.Join(keyErrs...)
 }
 
 func processReplacedAllInfo(logger klog.Logger, handler ResourceEventHandler, info ReplacedAllInfo, clientState Store, isInInitialList bool, keyFunc KeyFunc) error {

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

@@ -1006,6 +1006,157 @@ func (m *mockTxnStore) Transaction(txns ...Transaction) *TransactionError {
 	return nil
 }
 
+func TestProcessDeltasInBatchRepeatedKeys(t *testing.T) {
+	mkCM := func(name, rv string) *v1.ConfigMap {
+		return &v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: name, ResourceVersion: rv}}
+	}
+	a1, a2, a3 := mkCM("a", "1"), mkCM("a", "2"), mkCM("a", "3")
+	b1, b2 := mkCM("b", "1"), mkCM("b", "2")
+
+	testCases := []struct {
+		name           string
+		initialObjects []interface{}
+		deltas         []Delta
+		// StoreRV is read from inside the callbacks, after the whole batch is applied.
+		expectedHistory []eventRecord
+		expectedStore   map[string]string
+	}{
+		{
+			name:   "multiple updates of a new object",
+			deltas: []Delta{{Added, a1}, {Updated, a2}, {Updated, a3}},
+			expectedHistory: []eventRecord{
+				{Action: "add", Key: "ns/a", EventRV: "1", StoreRV: "3"},
+				{Action: "update", Key: "ns/a", OldRV: "1", EventRV: "2", StoreRV: "3"},
+				{Action: "update", Key: "ns/a", OldRV: "2", EventRV: "3", StoreRV: "3"},
+			},
+			expectedStore: map[string]string{"ns/a": "3"},
+		},
+		{
+			name:           "multiple updates of a stored object",
+			initialObjects: []interface{}{a1},
+			deltas:         []Delta{{Updated, a2}, {Updated, a3}},
+			expectedHistory: []eventRecord{
+				{Action: "update", Key: "ns/a", OldRV: "1", EventRV: "2", StoreRV: "3"},
+				{Action: "update", Key: "ns/a", OldRV: "2", EventRV: "3", StoreRV: "3"},
+			},
+			expectedStore: map[string]string{"ns/a": "3"},
+		},
+		{
+			name:   "add then delete",
+			deltas: []Delta{{Added, a1}, {Deleted, a1}},
+			expectedHistory: []eventRecord{
+				{Action: "add", Key: "ns/a", EventRV: "1"},
+				{Action: "delete", Key: "ns/a", EventRV: "1"},
+			},
+			expectedStore: map[string]string{},
+		},
+		{
+			name:   "add then delete then add",
+			deltas: []Delta{{Added, a1}, {Deleted, a1}, {Added, a2}},
+			expectedHistory: []eventRecord{
+				{Action: "add", Key: "ns/a", EventRV: "1", StoreRV: "2"},
+				{Action: "delete", Key: "ns/a", EventRV: "1", StoreRV: "2"},
+				{Action: "add", Key: "ns/a", EventRV: "2", StoreRV: "2"},
+			},
+			expectedStore: map[string]string{"ns/a": "2"},
+		},
+		{
+			name:           "delete then add",
+			initialObjects: []interface{}{a1},
+			deltas:         []Delta{{Deleted, a1}, {Added, a2}},
+			expectedHistory: []eventRecord{
+				{Action: "delete", Key: "ns/a", EventRV: "1", StoreRV: "2"},
+				{Action: "add", Key: "ns/a", EventRV: "2", StoreRV: "2"},
+			},
+			expectedStore: map[string]string{"ns/a": "2"},
+		},
+		{
+			name:           "tombstone delete then add and update",
+			initialObjects: []interface{}{a1},
+			deltas:         []Delta{{Deleted, DeletedFinalStateUnknown{Key: "ns/a", Obj: a1}}, {Added, a2}, {Updated, a3}},
+			expectedHistory: []eventRecord{
+				{Action: "delete", Key: "ns/a", EventRV: "1", StoreRV: "3"},
+				{Action: "add", Key: "ns/a", EventRV: "2", StoreRV: "3"},
+				{Action: "update", Key: "ns/a", OldRV: "2", EventRV: "3", StoreRV: "3"},
+			},
+			expectedStore: map[string]string{"ns/a": "3"},
+		},
+		{
+			name:           "interleaved objects",
+			initialObjects: []interface{}{b1},
+			deltas:         []Delta{{Added, a1}, {Updated, b2}, {Updated, a2}, {Deleted, b2}, {Sync, a2}},
+			expectedHistory: []eventRecord{
+				{Action: "add", Key: "ns/a", EventRV: "1", StoreRV: "2"},
+				{Action: "update", Key: "ns/b", OldRV: "1", EventRV: "2"},
+				{Action: "update", Key: "ns/a", OldRV: "1", EventRV: "2", StoreRV: "2"},
+				{Action: "delete", Key: "ns/b", EventRV: "2"},
+				{Action: "update", Key: "ns/a", OldRV: "2", EventRV: "2", StoreRV: "2"},
+			},
+			expectedStore: map[string]string{"ns/a": "2"},
+		},
+	}
+
+	newStore := func(t *testing.T, objs []interface{}) Store {
+		store := NewStore(DeletionHandlingMetaNamespaceKeyFunc)
+		require.NoError(t, store.Replace(objs, ""))
+		return store
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			logger, _ := ktesting.NewTestContext(t)
+
+			batchStore := newStore(t, tc.initialObjects)
+			batchRecorder := newEventRecorder(batchStore, DeletionHandlingMetaNamespaceKeyFunc)
+			err := processDeltasInBatch(logger, batchRecorder, batchStore, tc.deltas, false, DeletionHandlingMetaNamespaceKeyFunc)
+			require.NoError(t, err)
+
+			individualStore := newStore(t, tc.initialObjects)
+			individualRecorder := newEventRecorder(individualStore, DeletionHandlingMetaNamespaceKeyFunc)
+			for _, d := range tc.deltas {
+				err := processDeltas(logger, individualRecorder, individualStore, Deltas{d}, false, DeletionHandlingMetaNamespaceKeyFunc)
+				require.NoError(t, err)
+			}
+
+			storeContents := storeResourceVersions(t, batchStore)
+
+			assert.Equal(t, tc.expectedHistory, batchRecorder.getHistory())
+			// Callbacks receive the same objects as with individual processing, but may observe
+			// later Deltas of the batch in the store.
+			assert.Equal(t, withoutStoreRV(individualRecorder.getHistory()), withoutStoreRV(batchRecorder.getHistory()))
+			assert.Equal(t, tc.expectedStore, storeContents)
+		})
+	}
+}
+
+func TestProcessDeltasInBatchKeyError(t *testing.T) {
+	logger, _ := ktesting.NewTestContext(t)
+	mkCM := func(name string) *v1.ConfigMap {
+		return &v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: name}}
+	}
+	ok1, broken, ok2 := mkCM("ok1"), mkCM("broken"), mkCM("ok2")
+	keyFunc := func(obj interface{}) (string, error) {
+		if obj == broken {
+			return "", errors.New("test key func error")
+		}
+		return DeletionHandlingMetaNamespaceKeyFunc(obj)
+	}
+
+	store := NewStore(DeletionHandlingMetaNamespaceKeyFunc)
+	callbacks := 0
+	handler := ResourceEventHandlerFuncs{
+		AddFunc: func(interface{}) { callbacks++ },
+	}
+
+	err := processDeltasInBatch(logger, handler, store, []Delta{{Added, ok1}, {Added, broken}, {Added, ok2}}, false, keyFunc)
+
+	var keyErr KeyError
+	require.ErrorAs(t, err, &keyErr)
+	assert.Equal(t, broken, keyErr.Obj)
+	assert.Equal(t, 2, callbacks)
+	assert.ElementsMatch(t, []interface{}{ok1, ok2}, store.List())
+}
+
 // TestProcessLoopUnkeyableObject checks that the handler receives the same events and the store
 // ends up in the same state with and without batch processing, when an object whose key cannot
 // be computed is queued between Deltas of repeated keys.
@@ -1053,7 +1204,7 @@ func TestProcessLoopUnkeyableObject(t *testing.T) {
 
 			expectedNonBatchProcessCalls:   8,
 			expectedBatchProcessCalls:      0,
-			expectedBatchProcessBatchCalls: 4,
+			expectedBatchProcessBatchCalls: 1,
 		},
 		{
 			name:   "unkeyable object first",
@@ -1066,8 +1217,8 @@ func TestProcessLoopUnkeyableObject(t *testing.T) {
 			expectedStoreRV: "3",
 
 			expectedNonBatchProcessCalls:   3,
-			expectedBatchProcessCalls:      3,
-			expectedBatchProcessBatchCalls: 0,
+			expectedBatchProcessCalls:      0,
+			expectedBatchProcessBatchCalls: 1,
 		},
 		{
 			name:   "unkeyable object last",
@@ -1080,7 +1231,7 @@ func TestProcessLoopUnkeyableObject(t *testing.T) {
 			expectedStoreRV: "2",
 
 			expectedNonBatchProcessCalls:   3,
-			expectedBatchProcessCalls:      1,
+			expectedBatchProcessCalls:      0,
 			expectedBatchProcessBatchCalls: 1,
 		},
 		{
@@ -1094,7 +1245,7 @@ func TestProcessLoopUnkeyableObject(t *testing.T) {
 			expectedStoreRV: "4",
 
 			expectedNonBatchProcessCalls:   4,
-			expectedBatchProcessCalls:      2,
+			expectedBatchProcessCalls:      0,
 			expectedBatchProcessBatchCalls: 1,
 		},
 	}
@@ -1152,6 +1303,13 @@ func TestProcessLoopUnkeyableObject(t *testing.T) {
 	}
 }
 
+func withoutStoreRV(history []eventRecord) []eventRecord {
+	for i := range history {
+		history[i].StoreRV = ""
+	}
+	return history
+}
+
 // storeResourceVersions returns the resource version of every object in the store by key.
 func storeResourceVersions(t *testing.T, store Store) map[string]string {
 	rvs := map[string]string{}

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

@@ -499,14 +499,12 @@ func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProc
 	}
 
 	isInInitialList := !f.hasSynced_locked()
-	unique := sets.NewString()
 	deltas := make([]Delta, 0, min(len(f.items), f.batchSize))
 	moveDeltaToProcessList := func(i int) {
 		deltas = append(deltas, f.items[i])
 		// The underlying array still exists and references this object, so the object will not be garbage collected unless we zero the reference.
 		f.items[i] = Delta{}
 	}
-	// only bundle unique items into a batch
 	for i := 0; i < f.batchSize && i < len(f.items); i++ {
 		if f.initialPopulationCount > 0 && i >= f.initialPopulationCount {
 			break
@@ -520,21 +518,6 @@ func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProc
 			// close the batch when an unbatchable delta is encountered
 			break
 		}
-		id, err := f.keyOf(item)
-		if err != nil {
-			// close the batch here if error happens
-			// TODO: log the error when RealFIFOOptions supports passing klog instance like deprecated DeltaFIFO
-			// still pop the broken item out of queue to be compatible with the non-batch behavior it should be safe
-			// when 1st element is broken, however for Nth broken element, there's possible risk that broken item
-			// still can be processed and broke the uniqueness of the batch unexpectedly.
-			moveDeltaToProcessList(i)
-			break
-		}
-		if unique.Has(id) {
-			// close the batch if a duplicate item is encountered
-			break
-		}
-		unique.Insert(id)
 		moveDeltaToProcessList(i)
 	}
 	f.items = f.items[len(deltas):]

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

@@ -1064,7 +1064,7 @@ func TestRealFIFO_PopMultipleDeltaInBatch(t *testing.T) {
 			},
 		},
 		{
-			name: "split due to non-unique#1: update single item for multiple items should have separate batch",
+			name: "non-split due to non-unique#1: update single item for multiple items should be bundled after initial list",
 			initialItems: []testFifoObject{
 				obj1,
 			},
@@ -1075,8 +1075,7 @@ func TestRealFIFO_PopMultipleDeltaInBatch(t *testing.T) {
 			batchSize: unlimitedBatchSize,
 			expectedBatches: [][]Delta{
 				{{Replaced, obj1}},
-				{{Updated, obj1}},
-				{{Updated, obj1}},
+				{{Updated, obj1}, {Updated, obj1}},
 			},
 		},
 		{
@@ -1268,8 +1267,7 @@ func TestRealFIFO_PopBrokenItemsInBatch(t *testing.T) {
 				sucessObj1, sucessObj2, failObj3, failObj4,
 			},
 			expectedBatches: [][]Delta{
-				{{testDeltaType, sucessObj1}, {testDeltaType, sucessObj2}, {testDeltaType, failObj3}},
-				{{testDeltaType, failObj4}},
+				{{testDeltaType, sucessObj1}, {testDeltaType, sucessObj2}, {testDeltaType, failObj3}, {testDeltaType, failObj4}},
 			},
 		},
 		{
@@ -1278,8 +1276,7 @@ func TestRealFIFO_PopBrokenItemsInBatch(t *testing.T) {
 				failObj3, sucessObj1, failObj4,
 			},
 			expectedBatches: [][]Delta{
-				{{testDeltaType, failObj3}},
-				{{testDeltaType, sucessObj1}, {testDeltaType, failObj4}},
+				{{testDeltaType, failObj3}, {testDeltaType, sucessObj1}, {testDeltaType, failObj4}},
 			},
 		},
 	}

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

@@ -273,6 +273,12 @@ func (c *threadSafeMap) Transaction(txns ...ThreadSafeStoreTransaction) {
 	finalObj := txns[len(txns)-1].Object
 	rv, rvErr := rvFromObject(finalObj)
 	rvInt, parseErr := parseRVForMetricsWithTruncation(rv)
+	// A key can occur multiple times in a transaction. Only its last write is applied, which
+	// avoids indexing objects that are overwritten within the same transaction.
+	lastWrite := make(map[string]int, len(txns))
+	for i, txn := range txns {
+		lastWrite[txn.Key] = i
+	}
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	trace := utiltrace.New("ThreadSafeMap Transaction Process",
@@ -280,11 +286,12 @@ func (c *threadSafeMap) Transaction(txns ...ThreadSafeStoreTransaction) {
 		utiltrace.Field{Key: "Reason", Value: "Slow batch process due to too many items"})
 	defer trace.LogIfLong(min(500*time.Millisecond*time.Duration(len(txns)), 5*time.Second))
 
-	for _, txn := range txns {
+	for i, txn := range txns {
+		if lastWrite[txn.Key] != i {
+			continue
+		}
 		switch txn.Type {
-		case TransactionTypeAdd:
-			c.addLocked(txn.Key, txn.Object)
-		case TransactionTypeUpdate:
+		case TransactionTypeAdd, TransactionTypeUpdate:
 			c.updateLocked(txn.Key, txn.Object)
 		case TransactionTypeDelete:
 			c.deleteLocked(txn.Key)
@@ -302,10 +309,6 @@ func (c *threadSafeMap) Add(key string, obj interface{}) {
 	c.Update(key, obj)
 }
 
-func (c *threadSafeMap) addLocked(key string, obj interface{}) {
-	c.updateLocked(key, obj)
-}
-
 func (c *threadSafeMap) Update(key string, obj interface{}) {
 	rv, rvErr := rvFromObject(obj)
 	rvInt, parseErr := parseRVForMetricsWithTruncation(rv)

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

@@ -23,6 +23,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 )
@@ -285,6 +286,96 @@ func TestThreadSafeStoreRV(t *testing.T) {
 	})
 }
 
+func TestThreadSafeStoreTransactionRepeatedKeys(t *testing.T) {
+	newIndexers := func(indexed *[]string) Indexers {
+		return Indexers{
+			"byValue": func(obj interface{}) ([]string, error) {
+				if indexed != nil {
+					*indexed = append(*indexed, obj.(string))
+				}
+				return strings.Split(obj.(string), ","), nil
+			},
+		}
+	}
+	update := func(key, obj string) ThreadSafeStoreTransaction {
+		return ThreadSafeStoreTransaction{Transaction{Type: TransactionTypeUpdate, Object: obj}, key}
+	}
+	del := func(key string) ThreadSafeStoreTransaction {
+		return ThreadSafeStoreTransaction{Transaction{Type: TransactionTypeDelete}, key}
+	}
+
+	testCases := []struct {
+		name    string
+		initial map[string]string
+		txns    []ThreadSafeStoreTransaction
+		// expectedIndexed are the objects passed to the index function by the transaction:
+		// only the original and the final object of every key, never an intermediate one.
+		expectedIndexed []string
+	}{
+		{
+			name:            "repeated updates of a new key",
+			txns:            []ThreadSafeStoreTransaction{update("a", "x"), update("a", "y,z"), update("a", "z")},
+			expectedIndexed: []string{"z"},
+		},
+		{
+			name:            "repeated updates of a stored key back to its original object",
+			initial:         map[string]string{"a": "x"},
+			txns:            []ThreadSafeStoreTransaction{update("a", "y"), update("a", "x")},
+			expectedIndexed: []string{"x", "x"},
+		},
+		{
+			name: "add then delete",
+			txns: []ThreadSafeStoreTransaction{update("a", "x"), del("a")},
+		},
+		{
+			name:            "delete then add",
+			initial:         map[string]string{"a": "x,y"},
+			txns:            []ThreadSafeStoreTransaction{del("a"), update("a", "y,z")},
+			expectedIndexed: []string{"x,y", "y,z"},
+		},
+		{
+			name:            "interleaved keys sharing index values",
+			initial:         map[string]string{"b": "x"},
+			txns:            []ThreadSafeStoreTransaction{update("a", "x"), update("b", "y"), update("a", "y"), del("b"), update("a", "x,y")},
+			expectedIndexed: []string{"x", "x,y"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var indexed []string
+			store := NewThreadSafeStore(newIndexers(&indexed), Indices{}).(*threadSafeMap)
+			individual := NewThreadSafeStore(newIndexers(nil), Indices{}).(*threadSafeMap)
+			for key, obj := range tc.initial {
+				store.Add(key, obj)
+				individual.Add(key, obj)
+			}
+			indexed = nil // ignore the calls made by the initial Adds
+
+			store.Transaction(tc.txns...)
+			for _, txn := range tc.txns {
+				if txn.Type == TransactionTypeDelete {
+					individual.Delete(txn.Key)
+				} else {
+					individual.Update(txn.Key, txn.Object)
+				}
+			}
+
+			assert.ElementsMatch(t, tc.expectedIndexed, indexed)
+			assert.Equal(t, individual.items, store.items)
+			values := individual.ListIndexFuncValues("byValue")
+			assert.ElementsMatch(t, values, store.ListIndexFuncValues("byValue"))
+			for _, value := range values {
+				want, err := individual.IndexKeys("byValue", value)
+				require.NoError(t, err)
+				got, err := store.IndexKeys("byValue", value)
+				require.NoError(t, err)
+				assert.ElementsMatch(t, want, got, "index value %q", value)
+			}
+		})
+	}
+}
+
 func BenchmarkIndexer(b *testing.B) {
 	testIndexer := "testIndexer"