Merge pull request #140063 from pohly/fix-dra-resync-after-mutation-cache-ttl

DRA resourceslice controller: fix update after quick delete Kubernetes-commit: 67b6825a74f1485bb5246d67f9bea10cb665a827
2026-07-16 00:10:39 +00:00 · 2026-07-10 22:07:43 +00:00
parent 9236c74747 ead00aa2cf
commit 2bdf2d3bf8
3 changed files with 461 additions and 2 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,23 @@ Go API changes are typically not included in the Kubernetes release notes, so
 noteworthy Go API changes *may* be documented here. This is currently not
 *required*, so consult the git history to see all changes.

+### mutation cache: support informer events
+
+Calling OnAddOrUpdate and OnDelete from event handlers is optional. Calling
+them has the advantage that the mutation cache is updated sooner. It also fixes
+incorrectly reporting a locally updated item when a) using "include adds" and
+b) the updated item got deleted in the apiserver and informer cache.
+
+Although formally a Go API break because the MutationCache interface gets
+extended, in practice that interface is expected to have only the single
+implementation in client-go itself, so no-one should be affected by this.
+
+```
+- ./tools/cache.MutationCache.OnAddOrUpdate: added
+- ./tools/cache.MutationCache.OnDelete: added
+- ./tools/cache.MutationCache.internal: added unexported method
+```
+
 ### restmapper + discovery: add context-aware APIs

 See [PR #129109](https://github.com/kubernetes/kubernetes/pull/129109).
--- a/tools/cache/mutation_cache.go
+++ b/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,19 @@ 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)
+
+	// This interface is not meant to be implemented elsewhere.
+	// Marking it as internal enables future changes without
+	// triggering apidiff.
+	internal()
 }

 // ResourceVersionComparator is able to compare object versions.
@@ -60,6 +71,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 +134,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 +196,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 +255,114 @@ 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)
+}
+
+func (c *mutationCache) internal() {}
+
 // etcdObjectVersioner implements versioning and extracting etcd node information
 // for objects that have an embedded ObjectMeta or ListMeta field.
 type etcdObjectVersioner struct{}
@@ -262,3 +403,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")
+}
--- a/tools/cache/mutation_cache_test.go
+++ b/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")
+}