DRA resourceslice controller: fix update + delete sequence

When updating a ResourceSlice, a more recent copy than in the underlying store
gets cached by the MutationCache. Then when that ResourceSlice gets deleted on
the apiserver and then the store, the MutationCache kept the stale copy and
returned it when the controller synced again, causing the controller to
erroneously not recreate the deleted ResourceSlice.

Depending on timing, this could have happened while updating the DRA driver:
- Driver restarts, updating the existing ResourceSlices.
- kubelet deletes the slices before realizing that the driver
  is running again.

The fix is to address a long-standing TODO in the MutationCache: it needs to
react to changes in the store. This allows it to drop cached objects sooner and
let's it remove cached objects that are known to be deleted. Users of a
MutationCache must inform the cache in their own informer event
handlers *before* triggering sync operations.

The integration test demonstrates the problem when run against master:

    core.go:560: I0629 09:43:22.656382] Deleting slice...
    resourceslicecontroller.go:610: I0629 09:43:22.659446] ResourceSlice delete slice="00000-ga-publishresourceslices-5s47j-slice"
    resourceslicecontroller.go:612: I0629 09:43:22.659496] Scheduled sync poolName="global" at="2026-06-29 09:43:25.659481942 +0200 CEST m=+5.886096424"
    resourceslicecontroller.go:755: I0629 09:43:24.639116] Matched existing slice by index poolName="global" slice="00000-ga-publishresourceslices-5s47j-slice" matchIndex=0
    resourceslicecontroller.go:788: I0629 09:43:24.639317] Completed comparison poolName="global" numObsolete=0 numMatchedSlices=1 numChangedMatchedSlices=0 numNewSlices=0
    resourceslicecontroller.go:804: I0629 09:43:24.639391] Kept generation because at most one update API call is necessary poolName="global" generation=1
    core.go:566: FATAL ERROR: I0629 09:43:30.661588]
        	Timed out after 8.002s.
        	Expected
        	    <resourceslice.Stats>: {NumCreates: 0, NumUpdates: 1, NumDeletes: 0}
        	to equal
        	    <resourceslice.Stats>: {NumCreates: 1, NumUpdates: 1, NumDeletes: 0}

The "matched existing slice" doesn't actually exist anymore, so the controller
does nothing and the test fails.

Kubernetes-commit: 495f81169d8b2fe9e6e5dafd6b31fca1ae0483fa

tools/cache/mutation_cache.go

@@ -25,7 +25,10 @@ import (
 	"k8s.io/klog/v2"
 
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	utilcache "k8s.io/apimachinery/pkg/util/cache"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -35,11 +38,14 @@ import (
 // that can be used to provide a more current view of a requested object.  It requires interpreting
 // resourceVersions for comparisons.
 // Implementations must be thread-safe.
-// TODO find a way to layer this into an informer/lister
+// OnAddOrupdate and OnDelete should be called from informer event handlers to increase
+// the accuracy of the cache.
 type MutationCache interface {
 	GetByKey(key string) (interface{}, bool, error)
 	ByIndex(indexName, indexKey string) ([]interface{}, error)
 	Mutation(interface{})
+	OnAddOrUpdate(obj runtime.Object)
+	OnDelete(obj runtime.Object)
 }
 
 // ResourceVersionComparator is able to compare object versions.
@@ -60,6 +66,21 @@ type ResourceVersionComparator interface {
 // If includeAdds is true, objects in the mutation cache will be returned even if they don't exist
 // in the underlying store. This is only safe if your use of the cache can handle mutation entries
 // remaining in the cache for up to ttl when mutations and deletes occur very closely in time.
+//
+// Note that this also applies to objects updated by the caller, not just
+// objects added by it.  That's because the MutationCache may hold an updated
+// object at a time when the underlying store already removed it because it was
+// deleted in the apiserver after the update. To address this, the caller
+// can call OnAddOrUpdate and OnDelete from informer event handlers.
+//
+// This informs the MutationCache about all changes observed by the store
+// and enables it to remove a locally added or updated object immediately once
+// it appears in the store, which prevents the "stale updated object" problem.
+//
+// This does not solve the "stale added object" problem reliably because
+// the informer might never see the added object when the add is followed
+// quickly by a delete. The caller has to handle stale added objects by
+// checking whether they still exist once the TTL is over.
 func NewIntegerResourceVersionMutationCache(logger klog.Logger, backingCache Store, indexer Indexer, ttl time.Duration, includeAdds bool) MutationCache {
 	return &mutationCache{
 		backingCache:  backingCache,
@@ -108,11 +129,24 @@ func (c *mutationCache) GetByKey(key string) (interface{}, bool, error) {
 		if !exists {
 			return nil, false, nil
 		}
+		// Don't return the tombstone.
+		if _, ok := obj.(*tombstone); ok {
+			return nil, false, nil
+		}
 	}
 	objRuntime, ok := obj.(runtime.Object)
 	if !ok {
 		return obj, true, nil
 	}
+	// If the object is from the store,
+	// then this will remove the obsolete tombstone.
+	//
+	// The newer object can never be the tombstone
+	// because a) we don't get here if only the tombstone
+	// exists (early return above) and b) the store's
+	// object must be more recent than the tombstone
+	// (the tombstone was created by an earlier store
+	// deletion).
 	return c.newerObject(key, objRuntime), true, nil
 }
 
@@ -157,6 +191,9 @@ func (c *mutationCache) ByIndex(name string, indexKey string) ([]interface{}, er
 			if keySet.Has(key.(string)) {
 				continue
 			}
+			if _, ok := updated.(*tombstone); ok {
+				continue
+			}
 			elements, err := fn(updated)
 			if err != nil {
 				c.logger.V(4).Info("Unable to calculate an index entry for mutation cache entry", "key", key, "err", err)
@@ -213,15 +250,112 @@ func (c *mutationCache) Mutation(obj interface{}) {
 	if objRuntime, ok := obj.(runtime.Object); ok {
 		if mutatedObj, exists := c.mutationCache.Get(key); exists {
 			if mutatedObjRuntime, ok := mutatedObj.(runtime.Object); ok {
-				if c.comparator.CompareResourceVersion(objRuntime, mutatedObjRuntime) < 0 {
+				cmp := c.comparator.CompareResourceVersion(objRuntime, mutatedObjRuntime)
+				if cmp < 0 {
 					return
 				}
+				if t, ok := mutatedObj.(*tombstone); ok {
+					if cmp == 0 {
+						// Exactly this object instance is known to be deleted.
+						// Don't store it, keep the tombstone.
+						return
+					}
+					objMeta, err := meta.Accessor(obj)
+					if err == nil && t.GetUID() == objMeta.GetUID() {
+						// Some other revision of this object instance
+						// is known to be deleted. Also don't store it.
+						return
+					}
+				}
 			}
 		}
 	}
 	c.mutationCache.Add(key, obj, c.ttl)
 }
 
+// OnAddOrUpdate can be called to informer the cache about an object added to
+// the store or updated in in. If the object is as recent as the cached object
+// or newer, the cached object gets removed because it is no longer
+// needed. This keeps the cache smaller.
+func (c *mutationCache) OnAddOrUpdate(obj runtime.Object) {
+	key, err := DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		// this is a "nice to have", so failures shouldn't do anything weird
+		utilruntime.HandleErrorWithLogger(c.logger, err, "DeletionHandlingMetaNamespaceKeyFunc")
+		return
+	}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if mutatedObj, exists := c.mutationCache.Get(key); exists {
+		if mutatedObj, ok := mutatedObj.(runtime.Object); ok {
+			// No need to compare the UID here: resource versions can be compared
+			// between different instances. If it's newer or equal, then the mutation
+			// cache is out-dated.
+			if c.comparator.CompareResourceVersion(obj, mutatedObj) >= 0 {
+				c.mutationCache.Remove(key)
+			}
+		}
+	}
+}
+
+// OnDelete can be called to informer the cache about an object deleted in the store.
+// If there is a cached object with the same UID or an older RV, it gets removed.
+func (c *mutationCache) OnDelete(obj runtime.Object) {
+	objMeta, err := meta.Accessor(obj)
+	if err != nil {
+		return
+	}
+
+	key, err := DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		// this is a "nice to have", so failures shouldn't do anything weird
+		utilruntime.HandleErrorWithLogger(c.logger, err, "DeletionHandlingMetaNamespaceKeyFunc")
+		return
+	}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if mutatedObj, exists := c.mutationCache.Get(key); exists {
+		if mutatedObj, ok := mutatedObj.(runtime.Object); ok {
+			mutatedObjMeta, err := meta.Accessor(mutatedObj)
+			if err != nil {
+				return
+			}
+			// If the deleted object is known to be more recent than the
+			// mutated one, then the mutated one must have been removed
+			// because resource versions are incremented by type, not by
+			// object instance.
+			//
+			// If the UID matches, then we can also be sure that it got
+			// removed.
+			//
+			// If neither of this is true, then the caller has added
+			// a mutated object that may or may not still exist. They
+			// have to check after the TTL.
+			if c.comparator.CompareResourceVersion(obj, mutatedObj) >= 0 ||
+				mutatedObjMeta.GetUID() == objMeta.GetUID() {
+				c.mutationCache.Remove(key)
+			} else {
+				return
+			}
+		}
+	}
+
+	// Store a tombstone object in the mutation cache.
+	// A future Mutation call is outdated if it has a RV
+	// which is lower or equal or has the same UID.
+	t := &tombstone{
+		namespace: objMeta.GetNamespace(),
+		name:      objMeta.GetName(),
+		uid:       objMeta.GetUID(),
+		rv:        objMeta.GetResourceVersion(),
+	}
+	c.mutationCache.Add(key, t, c.ttl)
+}
+
 // etcdObjectVersioner implements versioning and extracting etcd node information
 // for objects that have an embedded ObjectMeta or ListMeta field.
 type etcdObjectVersioner struct{}
@@ -262,3 +396,57 @@ func (a etcdObjectVersioner) CompareResourceVersion(lhs, rhs runtime.Object) int
 
 	return 1
 }
+
+// tombstone is a replacement for a deleted object. It has the same key as it
+// and the ResourceVersion at which the deletion was detected.
+//
+// It implements GetName, GetNamespace, GetResourceVersion and GetUID and thus
+// the code above can compare it against other, normal objects. The difference
+// is that it never gets returned by ByIndex or GetByKey.
+type tombstone struct {
+	namespace, name, rv string
+	uid                 types.UID
+}
+
+var _ metav1.Object = &tombstone{}
+var _ runtime.Object = &tombstone{}
+
+func (t *tombstone) DeepCopyObject() runtime.Object {
+	clone := *t
+	return &clone
+}
+
+func (t *tombstone) GetObjectKind() schema.ObjectKind { panic("not implemented") }
+
+func (t *tombstone) GetNamespace() string                          { return t.namespace }
+func (t *tombstone) SetNamespace(namespace string)                 { panic("not implemented") }
+func (t *tombstone) GetName() string                               { return t.name }
+func (t *tombstone) SetName(name string)                           { panic("not implemented") }
+func (t *tombstone) GetGenerateName() string                       { panic("not implemented") }
+func (t *tombstone) SetGenerateName(name string)                   { panic("not implemented") }
+func (t *tombstone) GetUID() types.UID                             { return t.uid }
+func (t *tombstone) SetUID(uid types.UID)                          { panic("not implemented") }
+func (t *tombstone) GetResourceVersion() string                    { return t.rv }
+func (t *tombstone) SetResourceVersion(version string)             { panic("not implemented") }
+func (t *tombstone) GetGeneration() int64                          { panic("not implemented") }
+func (t *tombstone) SetGeneration(generation int64)                { panic("not implemented") }
+func (t *tombstone) GetSelfLink() string                           { panic("not implemented") }
+func (t *tombstone) SetSelfLink(selfLink string)                   { panic("not implemented") }
+func (t *tombstone) GetCreationTimestamp() metav1.Time             { panic("not implemented") }
+func (t *tombstone) SetCreationTimestamp(timestamp metav1.Time)    { panic("not implemented") }
+func (t *tombstone) GetDeletionTimestamp() *metav1.Time            { panic("not implemented") }
+func (t *tombstone) SetDeletionTimestamp(timestamp *metav1.Time)   { panic("not implemented") }
+func (t *tombstone) GetDeletionGracePeriodSeconds() *int64         { panic("not implemented") }
+func (t *tombstone) SetDeletionGracePeriodSeconds(*int64)          { panic("not implemented") }
+func (t *tombstone) GetLabels() map[string]string                  { panic("not implemented") }
+func (t *tombstone) SetLabels(labels map[string]string)            { panic("not implemented") }
+func (t *tombstone) GetAnnotations() map[string]string             { panic("not implemented") }
+func (t *tombstone) SetAnnotations(annotations map[string]string)  { panic("not implemented") }
+func (t *tombstone) GetFinalizers() []string                       { panic("not implemented") }
+func (t *tombstone) SetFinalizers(finalizers []string)             { panic("not implemented") }
+func (t *tombstone) GetOwnerReferences() []metav1.OwnerReference   { panic("not implemented") }
+func (t *tombstone) SetOwnerReferences([]metav1.OwnerReference)    { panic("not implemented") }
+func (t *tombstone) GetManagedFields() []metav1.ManagedFieldsEntry { panic("not implemented") }
+func (t *tombstone) SetManagedFields(managedFields []metav1.ManagedFieldsEntry) {
+	panic("not implemented")
+}

tools/cache/mutation_cache_test.go

@@ -0,0 +1,247 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cache
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/klog/v2"
+)
+
+func makeMutationTestPod(name string, uid types.UID, rv string) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            name,
+			UID:             uid,
+			ResourceVersion: rv,
+		},
+	}
+}
+
+func TestMutationCacheOnAddOrUpdate(t *testing.T) {
+	tests := map[string]struct {
+		mutationRV  string
+		storeUID    types.UID // UID passed to OnAddOrUpdate
+		storeRV     string    // RV passed to OnAddOrUpdate
+		wantCleared bool
+	}{
+		"equal-rv-clears": {
+			mutationRV:  "2",
+			storeUID:    "uid-1",
+			storeRV:     "2",
+			wantCleared: true,
+		},
+		"newer-store-clears": {
+			mutationRV:  "2",
+			storeUID:    "uid-1",
+			storeRV:     "3",
+			wantCleared: true,
+		},
+		"older-store-keeps": {
+			mutationRV:  "2",
+			storeUID:    "uid-1",
+			storeRV:     "1",
+			wantCleared: false,
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			store := NewStore(MetaNamespaceKeyFunc)
+			indexer := NewIndexer(MetaNamespaceKeyFunc, Indexers{})
+			mc := NewIntegerResourceVersionMutationCache(klog.Background(), store, indexer, time.Minute, false)
+
+			mutated := makeMutationTestPod("pod", "uid-1", tc.mutationRV)
+			mc.Mutation(mutated)
+
+			mc.OnAddOrUpdate(makeMutationTestPod("pod", tc.storeUID, tc.storeRV))
+
+			// Add the pod to the backing store at RV "1" (always older than the
+			// mutation's RV "2"), so we can tell whether GetByKey returns the
+			// mutation or the backing copy.
+			require.NoError(t, store.Add(makeMutationTestPod("pod", "uid-1", "1")))
+
+			got, exists, err := mc.GetByKey("pod")
+			require.NoError(t, err)
+			require.True(t, exists)
+
+			gotRV := got.(*v1.Pod).ResourceVersion
+			if tc.wantCleared {
+				assert.Equal(t, "1", gotRV, "backing store version expected after mutation cleared")
+			} else {
+				assert.Equal(t, tc.mutationRV, gotRV, "mutation version expected")
+			}
+		})
+	}
+}
+
+// TestMutationCacheOnDelete checks the behavior of OnDelete
+// when invoked after Mutation.
+func TestMutationCacheOnDelete(t *testing.T) {
+	tests := map[string]struct {
+		deleteUID  types.UID
+		deleteRV   string
+		wantExists bool
+	}{
+		"old-rv-same-UID": {
+			deleteUID:  "uid-1",
+			deleteRV:   "1", // Could be from a stale object in DeletedFinalStateUnknown.
+			wantExists: false,
+		},
+		"new-rv-different-UID": {
+			deleteUID:  "uid-2",
+			deleteRV:   "3",
+			wantExists: false,
+		},
+		"old-rv-different-uid": {
+			deleteUID:  "uid-other",
+			deleteRV:   "1",
+			wantExists: true,
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			store := NewStore(MetaNamespaceKeyFunc)
+			indexer := NewIndexer(MetaNamespaceKeyFunc, Indexers{})
+			mc := NewIntegerResourceVersionMutationCache(klog.Background(), store, indexer, time.Minute, true)
+
+			// Backing store has the pod at RV "2".
+			require.NoError(t, store.Add(makeMutationTestPod("pod", "uid-1", "2")))
+
+			// Mutation brings it to RV "3".
+			mc.Mutation(makeMutationTestPod("pod", "uid-1", "3"))
+			deletedPod := makeMutationTestPod("pod", tc.deleteUID, tc.deleteRV)
+			require.NoError(t, store.Delete(deletedPod))
+			mc.OnDelete(deletedPod)
+
+			_, exists, err := mc.GetByKey("pod")
+			require.NoError(t, err)
+			require.Equal(t, tc.wantExists, exists)
+		})
+	}
+}
+
+// TestMutationCacheUpdateConcurrentDelete checks the behavior when
+// Mutation is called after some informer events.
+func TestMutationCacheUpdateConcurrentDelete(t *testing.T) {
+	store := NewStore(MetaNamespaceKeyFunc)
+	indexer := NewIndexer(MetaNamespaceKeyFunc, Indexers{})
+	mc := NewIntegerResourceVersionMutationCache(klog.Background(), store, indexer, time.Hour, true)
+
+	// Backing store has the pod at RV "1".
+	require.NoError(t, store.Add(makeMutationTestPod("pod", "uid-1", "1")))
+
+	// Client starts an update leading to RV "2", which immediately gets
+	// deleted by some other client. That deletion is received before the
+	// client finishes its update.
+	updatedPod := makeMutationTestPod("pod", "uid-1", "2")
+	require.NoError(t, store.Update(updatedPod))
+	require.NoError(t, store.Delete(updatedPod))
+
+	// Informer events get delivered with a delay.
+	mc.OnAddOrUpdate(updatedPod)
+	mc.OnDelete(updatedPod)
+
+	// This Mutation call is stale, which gets detected because
+	// the mutation cache contains a tombstone object.
+	mc.Mutation(updatedPod)
+
+	_, exists, err := mc.GetByKey("pod")
+	require.NoError(t, err)
+	require.False(t, exists)
+	require.Equal(t, []any{"pod"}, mc.(*mutationCache).mutationCache.Keys())
+}
+
+// TestMutationCacheUpdateConcurrentRecreate checks the behavior when
+// Mutation is called after some informer events.
+func TestMutationCacheUpdateConcurrentRecreate(t *testing.T) {
+	store := NewStore(MetaNamespaceKeyFunc)
+	indexer := NewIndexer(MetaNamespaceKeyFunc, Indexers{})
+	mc := NewIntegerResourceVersionMutationCache(klog.Background(), store, indexer, time.Hour, true)
+
+	// Backing store has the pod at RV "1".
+	require.NoError(t, store.Add(makeMutationTestPod("pod", "uid-1", "1")))
+
+	// Client starts an update leading to RV "2", which immediately gets
+	// replaced by some other pod using the same name. Those changes are
+	// received before the client finishes its update.
+	updatedPod := makeMutationTestPod("pod", "uid-1", "2")
+	require.NoError(t, store.Update(updatedPod))
+	require.NoError(t, store.Delete(updatedPod))
+	replacementPod := makeMutationTestPod("pod", "uid-2", "3")
+	require.NoError(t, store.Add(replacementPod))
+
+	// Informer events get delivered with a delay.
+	mc.OnAddOrUpdate(updatedPod)
+	mc.OnDelete(updatedPod)
+	mc.OnAddOrUpdate(replacementPod)
+
+	// This Mutation call is stale, which gets detected because there is a more
+	// recent object in the store.
+	mc.Mutation(updatedPod)
+
+	got, exists, err := mc.GetByKey("pod")
+	require.NoError(t, err)
+	require.True(t, exists)
+	require.Equal(t, replacementPod, got)
+	require.Empty(t, mc.(*mutationCache).mutationCache.Keys())
+}
+
+// TestMutationCacheOnDeleteClearsStaleMutation exercises the core bug that motivated
+// the OnDelete method: a mutation stored after an update must not survive once the
+// informer reports the object deleted.
+//
+// The scenario:
+//  1. The object is added by the caller (includeAdds=true) so mutation cache
+//     returns it even when the backing store is empty.
+//  2. The object is deleted on the server; the backing store is now empty.
+//  3. Without OnDelete the mutation would linger in the cache and ByIndex
+//     would still return it, preventing the caller from recreating it.
+func TestMutationCacheOnDeleteClearsStaleMutation(t *testing.T) {
+	store := NewStore(MetaNamespaceKeyFunc)
+	byNameIndex := "by-name"
+	indexer := NewIndexer(MetaNamespaceKeyFunc, Indexers{
+		byNameIndex: func(obj interface{}) ([]string, error) {
+			return []string{obj.(*v1.Pod).Name}, nil
+		},
+	})
+	mc := NewIntegerResourceVersionMutationCache(klog.Background(), store, indexer, time.Minute, true /* includeAdds */)
+
+	// Simulate an update: mutation cache holds the pod at RV "2" while the
+	// backing store is still empty (informer hasn't caught up yet).
+	mc.Mutation(makeMutationTestPod("pod", "uid-1", "2"))
+
+	items, err := mc.ByIndex(byNameIndex, "pod")
+	require.NoError(t, err)
+	assert.Len(t, items, 1, "mutation should be visible via ByIndex when backing store is empty")
+
+	// Informer reports the delete: OnDelete must clear the stale mutation so
+	// that the next ByIndex call returns nothing and lets the caller recreate
+	// the object.
+	mc.OnDelete(makeMutationTestPod("pod", "uid-1", "1"))
+
+	items, err = mc.ByIndex(byNameIndex, "pod")
+	require.NoError(t, err)
+	assert.Empty(t, items, "OnDelete must clear the mutation; ByIndex must return nothing")
+}