scheduler/volumebinding: passive assume cache

Currently volume and dynamic-resource plugin shares an AssumeCache implementation. However, they features significantly different use case. DRA call Assume() on objects returned by APIServer, but volume call Assume() on objects yet to be sent to the APIServer. VolumeBinding plugin only makes one update request, while DynamicResource makes 2 requests (add finalizer then update allocation status). Taking advantage of this, currently the volume cache is simpler: 1. Reserve: assume PV/PVC will be updated 2. PreBind: really send the update request 3. AssumeCache receives an update from informer and overwrite the assumed state. a. if Prebind succeeded, this will surely include the update from step 2. b. if PreBind is not finished yet, and this is an irrelevant update, it is safe to overwrite the assumed state, because our update in PreBind will surely fail with Conflict. While for DynamicResource: 1. Reserve: add devices to inFlightAllocations 2. PreBind: a. send the 2 update requests b. add the returned object into AssumeCache c. AssumeCache dispatch events synchronously to update allocatedDevices d. remove devices from inFlightAllocations DynamicResource needs some features from AssumeCache that is not necessary for VolumeBinding: 1. DynamicResource needs strictly ordered update events to update allocatedDevices, including those from Assume() and Restore() 2. DynamicResource needs to compare ResourceVersion to prevent the assumed state from being overwritten by older version from informer. While this works, the doc[1] says: "you must not compare resource versions for greater-than or less-than relationships". Given so much difference, it can be beneficial to fork another simpler AssumeCache for VolumeBinding plugin. Because of no need to send events, the lite AssumeCache is a passive component. It only record the assumed version without copying all objects from informer into its local cache. When reading, we read from both informer and local cache. So it will always be up-to-date with informer, no need to wait for event handler. This resolves a race condition where AssumeCache and scheduler queue both receive events from informer. When a pod is scheduling due to PV update event, the PVCache may be not updated yet because it has not processed the relevant event. The passive version still listens events from informer, but only for cleaning up its local cache to save memory. [1]: https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions
2026-07-20 16:19:33 +00:00 · 2025-09-12 15:03:25 +08:00
parent c385a229d4
commit ed19492dc2
6 changed files with 461 additions and 139 deletions
--- a/pkg/scheduler/framework/plugins/volumebinding/assume_cache.go
+++ b/pkg/scheduler/framework/plugins/volumebinding/assume_cache.go
@@ -20,112 +20,73 @@ import (
 	"fmt"

 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/tools/cache"
 	storagehelpers "k8s.io/component-helpers/storage/volume"
 	"k8s.io/klog/v2"
-	"k8s.io/kubernetes/pkg/scheduler/util/assumecache"
 )

 // PVAssumeCache is a AssumeCache for PersistentVolume objects
 type PVAssumeCache struct {
-	*assumecache.AssumeCache
-	logger klog.Logger
+	*passiveAssumeCache[*v1.PersistentVolume]
 }

 func pvStorageClassIndexFunc(obj interface{}) ([]string, error) {
 	if pv, ok := obj.(*v1.PersistentVolume); ok {
 		return []string{storagehelpers.GetPersistentVolumeClass(pv)}, nil
 	}
-	return []string{""}, fmt.Errorf("object is not a v1.PersistentVolume: %v", obj)
+	return nil, fmt.Errorf("object is not a v1.PersistentVolume: %v", obj)
 }

+const storageClassIndex = "storageclass"
+
 // NewPVAssumeCache creates a PV assume cache.
-func NewPVAssumeCache(logger klog.Logger, informer assumecache.Informer) *PVAssumeCache {
-	logger = klog.LoggerWithName(logger, "PV Cache")
-	return &PVAssumeCache{
-		AssumeCache: assumecache.NewAssumeCache(logger, informer, "v1.PersistentVolume", "storageclass", pvStorageClassIndexFunc),
-		logger:      logger,
-	}
-}
-
-func (c *PVAssumeCache) GetPV(pvName string) (*v1.PersistentVolume, error) {
-	obj, err := c.Get(pvName)
-	if err != nil {
-		return nil, err
-	}
-
-	pv, ok := obj.(*v1.PersistentVolume)
-	if !ok {
-		return nil, &assumecache.WrongTypeError{TypeName: "v1.PersistentVolume", Object: obj}
-	}
-	return pv, nil
-}
-
-func (c *PVAssumeCache) GetAPIPV(pvName string) (*v1.PersistentVolume, error) {
-	obj, err := c.GetAPIObj(pvName)
-	if err != nil {
-		return nil, err
-	}
-	pv, ok := obj.(*v1.PersistentVolume)
-	if !ok {
-		return nil, &assumecache.WrongTypeError{TypeName: "v1.PersistentVolume", Object: obj}
-	}
-	return pv, nil
-}
-
-func (c *PVAssumeCache) ListPVs(storageClassName string) []*v1.PersistentVolume {
-	objs := c.List(&v1.PersistentVolume{
-		Spec: v1.PersistentVolumeSpec{
-			StorageClassName: storageClassName,
-		},
+func NewPVAssumeCache(logger klog.Logger, informer informer) (PVAssumeCache, error) {
+	logger = klog.LoggerWithName(logger, "pv-cache")
+	err := informer.GetIndexer().AddIndexers(map[string]cache.IndexFunc{
+		storageClassIndex: pvStorageClassIndexFunc,
 	})
-	pvs := []*v1.PersistentVolume{}
-	for _, obj := range objs {
-		pv, ok := obj.(*v1.PersistentVolume)
-		if !ok {
-			c.logger.Error(&assumecache.WrongTypeError{TypeName: "v1.PersistentVolume", Object: obj}, "ListPVs")
-			continue
+	if err != nil {
+		// Ignore the error if the index already exists. This can happen if
+		// the same informer is shared among multiple PVAssumeCache, maybe created from multiple profiles.
+		if informer.GetIndexer().GetIndexers()[storageClassIndex] == nil {
+			return PVAssumeCache{}, err
 		}
-		pvs = append(pvs, pv)
 	}
-	return pvs
+	cache, err := newAssumeCache[*v1.PersistentVolume](logger, informer, schema.GroupResource{Resource: "persistentvolumes"})
+	return PVAssumeCache{cache}, err
+}
+
+func (c PVAssumeCache) GetPV(pvName string) (*v1.PersistentVolume, error) {
+	return c.Get(pvName)
+}
+
+func (c PVAssumeCache) GetAPIPV(pvName string) (*v1.PersistentVolume, error) {
+	return c.GetAPIObj(pvName)
+}
+
+func (c PVAssumeCache) ListPVs(storageClassName string) ([]*v1.PersistentVolume, error) {
+	// This works because we will never change the storage class in scheduler
+	// Assumed PVs needs to be included here to ensure the same PVC will not be bound to another PV in the next scheduling cycle.
+	return c.ByIndex(storageClassIndex, storageClassName)
 }

 // PVCAssumeCache is a AssumeCache for PersistentVolumeClaim objects
 type PVCAssumeCache struct {
-	*assumecache.AssumeCache
-	logger klog.Logger
+	*passiveAssumeCache[*v1.PersistentVolumeClaim]
 }

 // NewPVCAssumeCache creates a PVC assume cache.
-func NewPVCAssumeCache(logger klog.Logger, informer assumecache.Informer) *PVCAssumeCache {
-	logger = klog.LoggerWithName(logger, "PVC Cache")
-	return &PVCAssumeCache{
-		AssumeCache: assumecache.NewAssumeCache(logger, informer, "v1.PersistentVolumeClaim", "", nil),
-		logger:      logger,
-	}
+func NewPVCAssumeCache(logger klog.Logger, informer informer) (PVCAssumeCache, error) {
+	logger = klog.LoggerWithName(logger, "pvc-cache")
+	cache, err := newAssumeCache[*v1.PersistentVolumeClaim](logger, informer, schema.GroupResource{Resource: "persistentvolumeclaims"})
+	return PVCAssumeCache{cache}, err
 }

-func (c *PVCAssumeCache) GetPVC(pvcKey string) (*v1.PersistentVolumeClaim, error) {
-	obj, err := c.Get(pvcKey)
-	if err != nil {
-		return nil, err
-	}
-
-	pvc, ok := obj.(*v1.PersistentVolumeClaim)
-	if !ok {
-		return nil, &assumecache.WrongTypeError{TypeName: "v1.PersistentVolumeClaim", Object: obj}
-	}
-	return pvc, nil
+func (c PVCAssumeCache) GetPVC(pvName string) (*v1.PersistentVolumeClaim, error) {
+	return c.Get(pvName)
 }

-func (c *PVCAssumeCache) GetAPIPVC(pvcKey string) (*v1.PersistentVolumeClaim, error) {
-	obj, err := c.GetAPIObj(pvcKey)
-	if err != nil {
-		return nil, err
-	}
-	pvc, ok := obj.(*v1.PersistentVolumeClaim)
-	if !ok {
-		return nil, &assumecache.WrongTypeError{TypeName: "v1.PersistentVolumeClaim", Object: obj}
-	}
-	return pvc, nil
+func (c PVCAssumeCache) GetAPIPVC(pvName string) (*v1.PersistentVolumeClaim, error) {
+	return c.GetAPIObj(pvName)
 }
--- a/pkg/scheduler/framework/plugins/volumebinding/assume_cache_test.go
+++ b/pkg/scheduler/framework/plugins/volumebinding/assume_cache_test.go
@@ -30,6 +30,8 @@ import (
 // sufficient for one assume cache.
 type testInformer struct {
 	handler cache.ResourceEventHandler
+	indexer cache.Indexer
+	t       *testing.T
 }

 func (i *testInformer) AddEventHandler(handler cache.ResourceEventHandler) (cache.ResourceEventHandlerRegistration, error) {
@@ -37,27 +39,49 @@ func (i *testInformer) AddEventHandler(handler cache.ResourceEventHandler) (cach
 	return nil, nil
 }

+func (i *testInformer) GetIndexer() cache.Indexer {
+	return i.indexer
+}
+
 func (i *testInformer) add(obj interface{}) {
+	if err := i.indexer.Add(obj); err != nil {
+		i.t.Fatalf("failed to add object into indexer: %v", err)
+	}
 	i.handler.OnAdd(obj, false)
 }

 func (i *testInformer) update(oldObj, obj interface{}) {
+	if err := i.indexer.Update(obj); err != nil {
+		i.t.Fatalf("failed to update object to indexer: %v", err)
+	}
 	i.handler.OnUpdate(oldObj, obj)
 }

 func (i *testInformer) delete(obj interface{}) {
+	if err := i.indexer.Delete(obj); err != nil {
+		i.t.Fatalf("failed to delete object from indexer: %v", err)
+	}
 	i.handler.OnDelete(obj)
 }

-func newTestPVCache(t *testing.T) (*testInformer, *PVAssumeCache) {
+func newTestPVCache(t *testing.T) (*testInformer, PVAssumeCache) {
 	logger, _ := ktesting.NewTestContext(t)
-	informer := &testInformer{}
-	cache := NewPVAssumeCache(logger, informer)
+	informer := &testInformer{
+		indexer: cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}),
+		t:       t,
+	}
+	cache, err := NewPVAssumeCache(logger, informer)
+	if err != nil {
+		t.Fatalf("NewPVAssumeCache() failed: %v", err)
+	}
 	return informer, cache
 }

-func verifyListPVs(t *testing.T, cache *PVAssumeCache, expectedPVs map[string]*v1.PersistentVolume, storageClassName string) {
-	pvList := cache.ListPVs(storageClassName)
+func verifyListPVs(t *testing.T, cache PVAssumeCache, expectedPVs map[string]*v1.PersistentVolume, storageClassName string) {
+	pvList, err := cache.ListPVs(storageClassName)
+	if err != nil {
+		t.Errorf("ListPVs() failed: %v", err)
+	}
 	if len(pvList) != len(expectedPVs) {
 		t.Errorf("ListPVs() returned %v PVs, expected %v", len(pvList), len(expectedPVs))
 	}
@@ -72,7 +96,7 @@ func verifyListPVs(t *testing.T, cache *PVAssumeCache, expectedPVs map[string]*v
 	}
 }

-func verifyPV(cache *PVAssumeCache, name string, expectedPV *v1.PersistentVolume) error {
+func verifyPV(cache PVAssumeCache, name string, expectedPV *v1.PersistentVolume) error {
 	pv, err := cache.GetPV(name)
 	if err != nil {
 		return err
@@ -99,10 +123,10 @@ func TestAssumePV(t *testing.T) {
 			newPV:         makePV("pv1", "class1").withVersion("5").PersistentVolume,
 			shouldSucceed: true,
 		},
-		"success-new-higher-version": {
+		"fail-new-higher-version": {
 			oldPV:         makePV("pv1", "").withVersion("5").PersistentVolume,
 			newPV:         makePV("pv1", "").withVersion("6").PersistentVolume,
-			shouldSucceed: true,
+			shouldSucceed: false,
 		},
 		"fail-old-not-found": {
 			oldPV:         makePV("pv2", "").withVersion("5").PersistentVolume,
@@ -163,7 +187,7 @@ func TestRestorePV(t *testing.T) {
 	newPV := makePV("pv1", "").withVersion("5").PersistentVolume

 	// Restore PV that doesn't exist
-	cache.Restore("nothing")
+	cache.Restore(&v1.PersistentVolume{})

 	// Add oldPV to cache
 	informer.add(oldPV)
@@ -172,7 +196,7 @@ func TestRestorePV(t *testing.T) {
 	}

 	// Restore PV
-	cache.Restore(oldPV.Name)
+	cache.Restore(oldPV)
 	if err := verifyPV(cache, oldPV.Name, oldPV); err != nil {
 		t.Fatalf("Failed to GetPV() after initial restore: %v", err)
 	}
@@ -186,7 +210,7 @@ func TestRestorePV(t *testing.T) {
 	}

 	// Restore PV
-	cache.Restore(oldPV.Name)
+	cache.Restore(newPV)
 	if err := verifyPV(cache, oldPV.Name, oldPV); err != nil {
 		t.Fatalf("Failed to GetPV() after restore: %v", err)
 	}
@@ -314,7 +338,7 @@ func makeClaim(name, version, namespace string) *v1.PersistentVolumeClaim {
 	}
 }

-func verifyPVC(cache *PVCAssumeCache, pvcKey string, expectedPVC *v1.PersistentVolumeClaim) error {
+func verifyPVC(cache PVCAssumeCache, pvcKey string, expectedPVC *v1.PersistentVolumeClaim) error {
 	pvc, err := cache.GetPVC(pvcKey)
 	if err != nil {
 		return err
@@ -325,10 +349,16 @@ func verifyPVC(cache *PVCAssumeCache, pvcKey string, expectedPVC *v1.PersistentV
 	return nil
 }

-func newTestPVCCache(t *testing.T) (*testInformer, *PVCAssumeCache) {
+func newTestPVCCache(t *testing.T) (*testInformer, PVCAssumeCache) {
 	logger, _ := ktesting.NewTestContext(t)
-	informer := &testInformer{}
-	cache := NewPVCAssumeCache(logger, informer)
+	informer := &testInformer{
+		indexer: cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}),
+		t:       t,
+	}
+	cache, err := NewPVCAssumeCache(logger, informer)
+	if err != nil {
+		t.Fatalf("NewPVCAssumeCache() failed: %v", err)
+	}
 	return informer, cache
 }

@@ -343,10 +373,10 @@ func TestAssumePVC(t *testing.T) {
 			newPVC:        makeClaim("pvc1", "5", "ns1"),
 			shouldSucceed: true,
 		},
-		"success-new-higher-version": {
+		"fail-new-higher-version": {
 			oldPVC:        makeClaim("pvc1", "5", "ns1"),
 			newPVC:        makeClaim("pvc1", "6", "ns1"),
-			shouldSucceed: true,
+			shouldSucceed: false,
 		},
 		"fail-old-not-found": {
 			oldPVC:        makeClaim("pvc2", "5", "ns1"),
@@ -408,7 +438,7 @@ func TestRestorePVC(t *testing.T) {
 	newPVC := makeClaim("pvc1", "5", "ns1")

 	// Restore PVC that doesn't exist
-	cache.Restore("nothing")
+	cache.Restore(&v1.PersistentVolumeClaim{})

 	// Add oldPVC to cache
 	informer.add(oldPVC)
@@ -417,7 +447,7 @@ func TestRestorePVC(t *testing.T) {
 	}

 	// Restore PVC
-	cache.Restore(getPVCName(oldPVC))
+	cache.Restore(oldPVC)
 	if err := verifyPVC(cache, getPVCName(oldPVC), oldPVC); err != nil {
 		t.Fatalf("Failed to GetPVC() after initial restore: %v", err)
 	}
@@ -431,12 +461,45 @@ func TestRestorePVC(t *testing.T) {
 	}

 	// Restore PVC
-	cache.Restore(getPVCName(oldPVC))
+	cache.Restore(newPVC)
 	if err := verifyPVC(cache, getPVCName(oldPVC), oldPVC); err != nil {
 		t.Fatalf("Failed to GetPVC() after restore: %v", err)
 	}
 }

+func TestConcurrentAssumePVC(t *testing.T) {
+	informer, cache := newTestPVCCache(t)
+
+	pvc1 := makeClaim("pvc1", "5", "ns1")
+	pvc1Update := makeClaim("pvc1", "5", "ns1")
+	// Add PVC to cache
+	informer.add(pvc1)
+
+	// Update PVC 1
+	if err := cache.Assume(pvc1Update); err != nil {
+		t.Fatalf("Assume() returned error %v", err)
+	}
+	if err := verifyPVC(cache, getPVCName(pvc1Update), pvc1Update); err != nil {
+		t.Fatalf("Failed to GetPVC() after Assume: %v", err)
+	}
+
+	pvc2 := makeClaim("pvc1", "7", "ns1")
+	pvc2Update := makeClaim("pvc1", "7", "ns1")
+	// PVC updated externally
+	informer.add(pvc2)
+
+	// Update PVC 2
+	if err := cache.Assume(pvc2Update); err != nil {
+		t.Fatalf("Assume() returned error %v", err)
+	}
+	// PVC 1 failed with conflict
+	cache.Restore(pvc1Update)
+	// Should still have pvc 2 in cache
+	if err := verifyPVC(cache, getPVCName(pvc2Update), pvc2Update); err != nil {
+		t.Fatalf("Failed to GetPVC() after restore: %v", err)
+	}
+}
+
 func TestAssumeUpdatePVCCache(t *testing.T) {
 	informer, cache := newTestPVCCache(t)

@@ -466,3 +529,32 @@ func TestAssumeUpdatePVCCache(t *testing.T) {
 		t.Fatalf("failed to get PVC after old PVC added: %v", err)
 	}
 }
+
+func TestDelayedInformerEvent(t *testing.T) {
+	informer, cache := newTestPVCCache(t)
+
+	pvcName := "test-pvc0"
+	pvcNamespace := "test-ns"
+
+	pvc1 := makeClaim(pvcName, "1", pvcNamespace)
+	pvc2 := makeClaim(pvcName, "2", pvcNamespace)
+	// Only add indexer, simulating delayed informer event
+	if err := informer.indexer.Add(pvc2); err != nil {
+		t.Fatalf("failed to add PVC: %v", err)
+	}
+
+	newPVC := pvc2.DeepCopy()
+	newPVC.Annotations[volume.AnnSelectedNode] = "test-node"
+	if err := cache.Assume(newPVC); err != nil {
+		t.Fatalf("failed to assume PVC: %v", err)
+	}
+
+	// Send the delayed event
+	informer.handler.OnAdd(pvc1, false)
+	informer.handler.OnDelete(pvc1)
+	informer.handler.OnAdd(pvc2, false)
+	// Expect assumed version not overwritten
+	if err := verifyPVC(cache, getPVCName(newPVC), newPVC); err != nil {
+		t.Fatalf("failed to get PVC after assume: %v", err)
+	}
+}
--- a/pkg/scheduler/framework/plugins/volumebinding/binder.go
+++ b/pkg/scheduler/framework/plugins/volumebinding/binder.go
@@ -46,7 +46,6 @@ import (
 	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/feature"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/volumebinding/metrics"
-	"k8s.io/kubernetes/pkg/scheduler/util/assumecache"
 )

 // ConflictReason is used for the special strings which explain why
@@ -217,8 +216,8 @@ type volumeBinder struct {
 	nodeLister    corelisters.NodeLister
 	csiNodeLister storagelisters.CSINodeLister

-	pvcCache *PVCAssumeCache
-	pvCache  *PVAssumeCache
+	pvcCache PVCAssumeCache
+	pvCache  PVAssumeCache

 	// Amount of time to wait for the bind operation to succeed
 	bindTimeout time.Duration
@@ -253,7 +252,13 @@ func NewVolumeBinder(
 	pvInformer coreinformers.PersistentVolumeInformer,
 	storageClassInformer storageinformers.StorageClassInformer,
 	capacityCheck CapacityCheck,
-	bindTimeout time.Duration) SchedulerVolumeBinder {
+	bindTimeout time.Duration) (SchedulerVolumeBinder, error) {
+
+	pvcCache, err1 := NewPVCAssumeCache(logger, pvcInformer.Informer())
+	pvCache, err2 := NewPVAssumeCache(logger, pvInformer.Informer())
+	if err := errors.Join(err1, err2); err != nil {
+		return nil, err
+	}
 	b := &volumeBinder{
 		kubeClient:                  kubeClient,
 		enableVolumeAttributesClass: fts.EnableVolumeAttributesClass,
@@ -262,8 +267,8 @@ func NewVolumeBinder(
 		classLister:                 storageClassInformer.Lister(),
 		nodeLister:                  nodeInformer.Lister(),
 		csiNodeLister:               csiNodeInformer.Lister(),
-		pvcCache:                    NewPVCAssumeCache(logger, pvcInformer.Informer()),
-		pvCache:                     NewPVAssumeCache(logger, pvInformer.Informer()),
+		pvcCache:                    pvcCache,
+		pvCache:                     pvCache,
 		bindTimeout:                 bindTimeout,
 		translator:                  csitrans.New(),
 	}
@@ -271,7 +276,7 @@ func NewVolumeBinder(
 	b.csiDriverLister = capacityCheck.CSIDriverInformer.Lister()
 	b.csiStorageCapacityLister = capacityCheck.CSIStorageCapacityInformer.Lister()

-	return b
+	return b, nil
 }

 // FindPodVolumes finds the matching PVs for PVCs and nodes to provision PVs
@@ -685,7 +690,7 @@ func (b *volumeBinder) checkBindings(logger klog.Logger, pod *v1.Pod, bindings [
 		if pvc.Spec.VolumeName != "" {
 			pv, err := b.pvCache.GetAPIPV(pvc.Spec.VolumeName)
 			if err != nil {
-				if errors.Is(err, assumecache.ErrNotFound) {
+				if apierrors.IsNotFound(err) {
 					// We tolerate NotFound error here, because PV is possibly
 					// not found because of API delay, we can check next time.
 					// And if PV does not exist because it's deleted, PVC will
@@ -822,7 +827,11 @@ func (b *volumeBinder) GetPodVolumeClaims(logger klog.Logger, pod *v1.Pod) (podV
 	for _, pvc := range podVolumeClaims.unboundClaimsDelayBinding {
 		// Get storage class name from each PVC
 		storageClassName := volume.GetPersistentVolumeClaimClass(pvc)
-		podVolumeClaims.unboundVolumesDelayBinding[storageClassName] = b.pvCache.ListPVs(storageClassName)
+		pvs, err := b.pvCache.ListPVs(storageClassName)
+		if err != nil {
+			return nil, err
+		}
+		podVolumeClaims.unboundVolumesDelayBinding[storageClassName] = pvs
 	}
 	return podVolumeClaims, nil
 }
@@ -838,7 +847,7 @@ func (b *volumeBinder) checkBoundClaims(logger klog.Logger, claims []*v1.Persist
 		pvName := pvc.Spec.VolumeName
 		pv, err := b.pvCache.GetPV(pvName)
 		if err != nil {
-			if errors.Is(err, assumecache.ErrNotFound) {
+			if apierrors.IsNotFound(err) {
 				err = nil
 			}
 			return true, false, err
@@ -954,13 +963,13 @@ func (b *volumeBinder) checkVolumeProvisions(logger klog.Logger, pod *v1.Pod, cl

 func (b *volumeBinder) revertAssumedPVs(bindings []*BindingInfo) {
 	for _, BindingInfo := range bindings {
-		b.pvCache.Restore(BindingInfo.pv.Name)
+		b.pvCache.Restore(BindingInfo.pv)
 	}
 }

 func (b *volumeBinder) revertAssumedPVCs(claims []*v1.PersistentVolumeClaim) {
 	for _, claim := range claims {
-		b.pvcCache.Restore(getPVCName(claim))
+		b.pvcCache.Restore(claim)
 	}
 }

--- a/pkg/scheduler/framework/plugins/volumebinding/binder_test.go
+++ b/pkg/scheduler/framework/plugins/volumebinding/binder_test.go
@@ -46,7 +46,6 @@ import (
 	"k8s.io/kubernetes/pkg/controller"
 	pvtesting "k8s.io/kubernetes/pkg/controller/volume/persistentvolume/testing"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/feature"
-	"k8s.io/kubernetes/pkg/scheduler/util/assumecache"
 )

 var (
@@ -170,7 +169,7 @@ func newTestBinder(t *testing.T, ctx context.Context) *testEnv {
 		CSIDriverInformer:          csiDriverInformer,
 		CSIStorageCapacityInformer: csiStorageCapacityInformer,
 	}
-	binder := NewVolumeBinder(
+	binder, err := NewVolumeBinder(
 		logger,
 		client,
 		feature.Features{},
@@ -182,6 +181,9 @@ func newTestBinder(t *testing.T, ctx context.Context) *testEnv {
 		classInformer,
 		capacityCheck,
 		10*time.Second)
+	if err != nil {
+		t.Fatalf("Failed to create VolumeBinder: %v", err)
+	}

 	// Wait for informers cache sync
 	informerFactory.Start(ctx.Done())
@@ -298,7 +300,9 @@ func (env *testEnv) addCSIStorageCapacities(capacities []*storagev1.CSIStorageCa

 func (env *testEnv) initClaims(t *testing.T, cachedPVCs []*v1.PersistentVolumeClaim, apiPVCs []*v1.PersistentVolumeClaim) {
 	for _, pvc := range cachedPVCs {
-		assumecache.AddTestObject(env.internalBinder.pvcCache.AssumeCache, pvc)
+		if err := env.internalBinder.pvcCache.store.Add(pvc); err != nil {
+			t.Fatalf("error adding PVC %s/%s to cache: %v", pvc.Namespace, pvc.Name, err)
+		}
 		if apiPVCs == nil {
 			env.reactor.AddClaim(pvc)
 		}
@@ -310,7 +314,9 @@ func (env *testEnv) initClaims(t *testing.T, cachedPVCs []*v1.PersistentVolumeCl

 func (env *testEnv) initVolumes(t *testing.T, cachedPVs []*v1.PersistentVolume, apiPVs []*v1.PersistentVolume) {
 	for _, pv := range cachedPVs {
-		assumecache.AddTestObject(env.internalBinder.pvCache.AssumeCache, pv)
+		if err := env.internalBinder.pvCache.store.Add(pv); err != nil {
+			t.Fatalf("error adding PV %s to cache: %v", pv.Name, err)
+		}
 		if apiPVs == nil {
 			env.reactor.AddVolume(pv)
 		}
@@ -318,7 +324,6 @@ func (env *testEnv) initVolumes(t *testing.T, cachedPVs []*v1.PersistentVolume,
 	for _, pv := range apiPVs {
 		env.reactor.AddVolume(pv)
 	}
-
 }

 func (env *testEnv) updateVolumes(ctx context.Context, pvs []*v1.PersistentVolume) error {
@@ -331,14 +336,10 @@ func (env *testEnv) updateVolumes(ctx context.Context, pvs []*v1.PersistentVolum
 	}
 	return wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 3*time.Second, false, func(ctx context.Context) (bool, error) {
 		for _, pv := range pvs {
-			obj, err := env.internalBinder.pvCache.GetAPIObj(pv.Name)
-			if obj == nil || err != nil {
+			pvInCache, err := env.internalBinder.pvCache.GetAPIPV(pv.Name)
+			if pvInCache == nil || err != nil {
 				return false, nil
 			}
-			pvInCache, ok := obj.(*v1.PersistentVolume)
-			if !ok {
-				return false, fmt.Errorf("PV %s invalid object", pvInCache.Name)
-			}
 			if versioner.CompareResourceVersion(pvInCache, pv) != 0 {
 				return false, nil
 			}
@@ -357,14 +358,10 @@ func (env *testEnv) updateClaims(ctx context.Context, pvcs []*v1.PersistentVolum
 	}
 	return wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 3*time.Second, false, func(ctx context.Context) (bool, error) {
 		for _, pvc := range pvcs {
-			obj, err := env.internalBinder.pvcCache.GetAPIObj(getPVCName(pvc))
-			if obj == nil || err != nil {
+			pvcInCache, err := env.internalBinder.pvcCache.GetAPIPVC(getPVCName(pvc))
+			if pvcInCache == nil || err != nil {
 				return false, nil
 			}
-			pvcInCache, ok := obj.(*v1.PersistentVolumeClaim)
-			if !ok {
-				return false, fmt.Errorf("PVC %s invalid object", pvcInCache.Name)
-			}
 			if versioner.CompareResourceVersion(pvcInCache, pvc) != 0 {
 				return false, nil
 			}
@@ -375,13 +372,17 @@ func (env *testEnv) updateClaims(ctx context.Context, pvcs []*v1.PersistentVolum

 func (env *testEnv) deleteVolumes(t *testing.T, pvs []*v1.PersistentVolume) {
 	for _, pv := range pvs {
-		assumecache.DeleteTestObject(env.internalBinder.pvCache.AssumeCache, pv)
+		if err := env.internalBinder.pvCache.store.Delete(pv); err != nil {
+			t.Fatalf("Error deleting PV %s: %v", pv.Name, err)
+		}
 	}
 }

 func (env *testEnv) deleteClaims(t *testing.T, pvcs []*v1.PersistentVolumeClaim) {
 	for _, pvc := range pvcs {
-		assumecache.DeleteTestObject(env.internalBinder.pvcCache.AssumeCache, pvc)
+		if err := env.internalBinder.pvcCache.store.Delete(pvc); err != nil {
+			t.Fatalf("Error deleting PVC %s/%s: %v", pvc.Namespace, pvc.Name, err)
+		}
 	}
 }

@@ -489,8 +490,8 @@ func (env *testEnv) validateCacheRestored(t *testing.T, pod *v1.Pod, bindings []
 	// All PVs have been unmodified in cache
 	pvCache := env.internalBinder.pvCache
 	for _, b := range bindings {
-		pv, _ := pvCache.GetPV(b.pv.Name)
-		apiPV, _ := pvCache.GetAPIPV(b.pv.Name)
+		pv, _ := pvCache.Get(b.pv.Name)
+		apiPV, _ := pvCache.GetAPIObj(b.pv.Name)
 		// PV could be nil if it's missing from cache
 		if pv != nil && pv != apiPV {
 			t.Errorf("PV %q was modified in cache", b.pv.Name)
@@ -2141,20 +2142,20 @@ func TestFindAssumeVolumes(t *testing.T) {
 	// 1. Find matching PVs
 	podVolumes, reasons, err := findPodVolumes(logger, testEnv.binder, pod, testNode)
 	if err != nil {
-		t.Errorf("Test failed: FindPodVolumes returned error: %v", err)
+		t.Fatalf("Test failed: FindPodVolumes returned error: %v", err)
 	}
 	if len(reasons) > 0 {
-		t.Errorf("Test failed: couldn't find PVs for all PVCs: %v", reasons)
+		t.Fatalf("Test failed: couldn't find PVs for all PVCs: %v", reasons)
 	}
 	expectedBindings := podVolumes.StaticBindings

 	// 2. Assume matches
 	allBound, err := testEnv.binder.AssumePodVolumes(logger, pod, testNode.Name, podVolumes)
 	if err != nil {
-		t.Errorf("Test failed: AssumePodVolumes returned error: %v", err)
+		t.Fatalf("Test failed: AssumePodVolumes returned error: %v", err)
 	}
 	if allBound {
-		t.Errorf("Test failed: detected unbound volumes as bound")
+		t.Fatalf("Test failed: detected unbound volumes as bound")
 	}
 	testEnv.validateAssume(t, pod, expectedBindings, nil)

@@ -2167,10 +2168,10 @@ func TestFindAssumeVolumes(t *testing.T) {
 	for i := 0; i < 50; i++ {
 		podVolumes, reasons, err := findPodVolumes(logger, testEnv.binder, pod, testNode)
 		if err != nil {
-			t.Errorf("Test failed: FindPodVolumes returned error: %v", err)
+			t.Fatalf("Test failed: FindPodVolumes returned error: %v", err)
 		}
 		if len(reasons) > 0 {
-			t.Errorf("Test failed: couldn't find PVs for all PVCs: %v", reasons)
+			t.Fatalf("Test failed: couldn't find PVs for all PVCs: %v", reasons)
 		}
 		testEnv.validatePodCache(t, testNode.Name, pod, podVolumes, expectedBindings, nil)
 	}
--- a/pkg/scheduler/framework/plugins/volumebinding/passive_assume_cache.go
+++ b/pkg/scheduler/framework/plugins/volumebinding/passive_assume_cache.go
@@ -0,0 +1,256 @@
+/*
+Copyright 2025 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 volumebinding
+
+import (
+	"fmt"
+	"sync"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+)
+
+// informer is the subset of [cache.SharedInformer] that newAssumeCache depends upon.
+type informer interface {
+	AddEventHandler(handler cache.ResourceEventHandler) (cache.ResourceEventHandlerRegistration, error)
+	GetIndexer() cache.Indexer
+}
+
+// passiveAssumeCache is a cache on top of the informer that allows for updating
+// objects outside of informer events and also restoring the informer
+// cache's version of the object.
+//
+// An informer update always overrides the assumed object.
+//
+// This is different from pkg/scheduler/util/assumecache in:
+//   - this does not dispatch events
+//   - this is always up-to-date with the informer
+//   - this only allow assuming objects yet to be sent to the apiserver,
+//     not the ones returned from the apiserver
+type passiveAssumeCache[T v1.Object] struct {
+	// The logger that was chosen when setting up the cache.
+	// Will be used for all operations.
+	logger klog.Logger
+	gr     schema.GroupResource
+
+	// Synchronizes updates to all fields below.
+	// Although [store] have its own lock, we still need to hold our lock
+	// before reading from either [store] or [assumed] if we compare
+	// ResourceVersion between them.
+	rwMutex sync.RWMutex
+
+	// Objects from informer
+	store   cache.Indexer
+	assumed map[string]T
+}
+
+// newAssumeCache creates an assume cache for objects of type T.
+func newAssumeCache[T v1.Object](logger klog.Logger, informer informer, gr schema.GroupResource) (*passiveAssumeCache[T], error) {
+	c := &passiveAssumeCache[T]{
+		logger:  logger,
+		gr:      gr,
+		store:   informer.GetIndexer(),
+		assumed: make(map[string]T),
+	}
+
+	_, err := informer.AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    c.add,
+			UpdateFunc: c.update,
+			DeleteFunc: c.delete,
+		},
+	)
+	return c, err
+}
+
+// Receives events from informer. May expire the assumed object if it is older.
+func (c *passiveAssumeCache[T]) mayExpire(key string) {
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+
+	assumed, ok := c.assumed[key]
+	if !ok {
+		return
+	}
+
+	// Get the latest version to avoid overwriting newer object from [Assume]
+	obj, exists, err := c.store.GetByKey(key)
+	if err != nil {
+		utilruntime.HandleErrorWithLogger(c.logger, err, "mayExpire get", "key", key)
+		return
+	}
+
+	expire := true
+	if exists {
+		newMeta, err := meta.Accessor(obj)
+		if err != nil {
+			utilruntime.HandleErrorWithLogger(c.logger, err, "mayExpire meta", "key", key)
+			return
+		}
+
+		// Only overwrite assumed object if version is newer (not resync).
+		if assumed.GetResourceVersion() == newMeta.GetResourceVersion() {
+			c.logger.V(10).Info("ignoring resync of assumed object", "key", key, "version", assumed.GetResourceVersion())
+			expire = false
+		} else {
+			c.logger.V(4).Info("assumed object expired", "newVersion", newMeta.GetResourceVersion(),
+				"key", key, "version", assumed.GetResourceVersion())
+		}
+	} else {
+		c.logger.V(4).Info("assumed object expired", "key", key, "version", assumed.GetResourceVersion())
+	}
+	if expire {
+		delete(c.assumed, key)
+	}
+}
+
+func (c *passiveAssumeCache[T]) add(obj any) {
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		utilruntime.HandleErrorWithLogger(c.logger, err, "Add object get key")
+		return
+	}
+	c.mayExpire(key)
+}
+
+func (c *passiveAssumeCache[T]) update(_, obj any) {
+	c.add(obj)
+}
+
+func (c *passiveAssumeCache[T]) delete(obj any) {
+	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		utilruntime.HandleErrorWithLogger(c.logger, err, "Delete object get key")
+		return
+	}
+	c.mayExpire(key)
+}
+
+// ByIndex returns the stored objects whose set of indexed values for the named index includes the given indexed value
+//
+// The index is evaluated on the object from store. Objects from [Assume] will present in the result but will not affect the index.
+func (c *passiveAssumeCache[T]) ByIndex(indexName, indexedValue string) ([]T, error) {
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
+
+	objs, err := c.store.ByIndex(indexName, indexedValue)
+	if err != nil {
+		return nil, err
+	}
+	return c.replaceAssumed(objs), nil
+}
+
+// Get the object by its key.
+func (c *passiveAssumeCache[T]) Get(key string) (T, error) {
+	c.rwMutex.RLock()
+	defer c.rwMutex.RUnlock()
+
+	obj, err := c.GetAPIObj(key)
+	if err != nil {
+		return obj, err
+	}
+
+	assumed, ok := c.assumed[key]
+	if !ok || assumed.GetResourceVersion() != obj.GetResourceVersion() { // not assumed or Informer object is newer
+		return obj, nil
+	}
+	return assumed, nil
+}
+
+// GetAPIObj gets the informer cache's version by its key.
+func (c *passiveAssumeCache[T]) GetAPIObj(key string) (T, error) {
+	obj, ok, err := c.store.GetByKey(key)
+	var zero T
+	if err != nil {
+		return zero, err
+	}
+	if !ok {
+		return zero, apierrors.NewNotFound(c.gr, key)
+	}
+	v, ok := obj.(T)
+	if !ok {
+		return zero, fmt.Errorf("object is not of type %T", zero)
+	}
+	return v, nil
+}
+
+func keyOf[T v1.Object](obj T) string {
+	return cache.MetaObjectToName(obj).String()
+}
+
+func (c *passiveAssumeCache[T]) replaceAssumed(objs []any) []T {
+	allObjs := make([]T, 0, len(objs))
+	for _, obj := range objs {
+		v, ok := obj.(T)
+		if !ok {
+			utilruntime.HandleErrorWithLogger(c.logger, nil, "listed object has wrong type", "type", fmt.Sprintf("%T", obj))
+			continue
+		}
+		assumed, ok := c.assumed[keyOf(v)]
+		if ok && assumed.GetResourceVersion() == v.GetResourceVersion() {
+			// assumed object is not in informer yet
+			v = assumed
+		}
+		allObjs = append(allObjs, v)
+	}
+	return allObjs
+}
+
+// Assume updates the object in-memory only.
+//
+// The version of the object must be equal to
+// the current object, otherwise an error is returned.
+// If an update is received via the informer while such an
+// object is assumed, it gets dropped in favor of the
+// newer object from the apiserver.
+func (c *passiveAssumeCache[T]) Assume(obj T) error {
+	key := keyOf(obj)
+
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+
+	stored, err := c.GetAPIObj(key)
+	if err != nil {
+		return err
+	}
+
+	if stored.GetResourceVersion() != obj.GetResourceVersion() {
+		return fmt.Errorf("%q is out of sync (stored: %s, assume: %s)", key, stored.GetResourceVersion(), obj.GetResourceVersion())
+	}
+	c.assumed[key] = obj
+	c.logger.V(4).Info("Assumed object", "key", key, "version", obj.GetResourceVersion())
+	return nil
+}
+
+// Restore the informer cache's version of the object.
+func (c *passiveAssumeCache[T]) Restore(obj T) {
+	key := keyOf(obj)
+
+	c.rwMutex.Lock()
+	defer c.rwMutex.Unlock()
+
+	assumed, ok := c.assumed[key]
+	if ok && assumed.GetResourceVersion() == obj.GetResourceVersion() {
+		delete(c.assumed, key)
+		c.logger.V(4).Info("Restored object", "key", key, "version", obj.GetResourceVersion())
+	}
+}
--- a/pkg/scheduler/framework/plugins/volumebinding/volume_binding.go
+++ b/pkg/scheduler/framework/plugins/volumebinding/volume_binding.go
@@ -627,7 +627,10 @@ func New(ctx context.Context, plArgs runtime.Object, fh framework.Handle, fts fe
 		CSIDriverInformer:          fh.SharedInformerFactory().Storage().V1().CSIDrivers(),
 		CSIStorageCapacityInformer: fh.SharedInformerFactory().Storage().V1().CSIStorageCapacities(),
 	}
-	binder := NewVolumeBinder(klog.FromContext(ctx), fh.ClientSet(), fts, podInformer, nodeInformer, csiNodeInformer, pvcInformer, pvInformer, storageClassInformer, capacityCheck, time.Duration(args.BindTimeoutSeconds)*time.Second)
+	binder, err := NewVolumeBinder(klog.FromContext(ctx), fh.ClientSet(), fts, podInformer, nodeInformer, csiNodeInformer, pvcInformer, pvInformer, storageClassInformer, capacityCheck, time.Duration(args.BindTimeoutSeconds)*time.Second)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build volume binder: %w", err)
+	}

 	// build score function
 	var scorer volumeCapacityScorer