diff --git a/pkg/scheduler/framework/plugins/volumebinding/assume_cache.go b/pkg/scheduler/framework/plugins/volumebinding/assume_cache.go index e12fb08a0d4..b4e6063b149 100644 --- 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) } diff --git a/pkg/scheduler/framework/plugins/volumebinding/assume_cache_test.go b/pkg/scheduler/framework/plugins/volumebinding/assume_cache_test.go index 407c284d638..8af8e024e02 100644 --- 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) + } +} diff --git a/pkg/scheduler/framework/plugins/volumebinding/binder.go b/pkg/scheduler/framework/plugins/volumebinding/binder.go index 9d376b644e0..738718f19e1 100644 --- 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) } } diff --git a/pkg/scheduler/framework/plugins/volumebinding/binder_test.go b/pkg/scheduler/framework/plugins/volumebinding/binder_test.go index 2e867770e45..5fb2bb26944 100644 --- 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) } diff --git a/pkg/scheduler/framework/plugins/volumebinding/passive_assume_cache.go b/pkg/scheduler/framework/plugins/volumebinding/passive_assume_cache.go new file mode 100644 index 00000000000..e27d74b40c8 --- /dev/null +++ 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()) + } +} diff --git a/pkg/scheduler/framework/plugins/volumebinding/volume_binding.go b/pkg/scheduler/framework/plugins/volumebinding/volume_binding.go index a3cb861ed24..287fd910952 100644 --- 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