diff --git a/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go b/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go index a40c515cd14..41ccd8094ba 100644 --- a/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go +++ b/pkg/scheduler/framework/plugins/nodevolumelimits/csi.go @@ -57,11 +57,11 @@ type InTreeToCSITranslator interface { // CSILimits is a plugin that checks node volume limits. type CSILimits struct { - csiNodeLister fwk.CSINodeLister - pvLister corelisters.PersistentVolumeLister - pvcLister corelisters.PersistentVolumeClaimLister - scLister storagelisters.StorageClassLister - vaLister storagelisters.VolumeAttachmentLister + csiManager fwk.CSIManager + pvLister corelisters.PersistentVolumeLister + pvcLister corelisters.PersistentVolumeClaimLister + scLister storagelisters.StorageClassLister + vaLister storagelisters.VolumeAttachmentLister enableCSIMigrationPortworx bool randomVolumeIDPrefix string @@ -261,7 +261,7 @@ func (pl *CSILimits) Filter(ctx context.Context, _ fwk.CycleState, pod *v1.Pod, logger := klog.FromContext(ctx) - csiNode, err := pl.csiNodeLister.Get(node.Name) + csiNode, err := pl.csiManager.CSINodes().Get(node.Name) if err != nil { // TODO: return the error once CSINode is created by default (2 releases) logger.V(5).Info("Could not get a CSINode object for the node", "node", klog.KObj(node), "err", err) @@ -554,10 +554,8 @@ func NewCSI(_ context.Context, _ runtime.Object, handle fwk.Handle, fts feature. vaLister := informerFactory.Storage().V1().VolumeAttachments().Lister() csiTranslator := csitrans.New() - csiNodesLister := handle.SharedCSINodeLister() - return &CSILimits{ - csiNodeLister: csiNodesLister, + csiManager: handle.SharedCSIManager(), pvLister: pvLister, pvcLister: pvcLister, scLister: scLister, diff --git a/pkg/scheduler/framework/plugins/nodevolumelimits/csi_manager.go b/pkg/scheduler/framework/plugins/nodevolumelimits/csi_manager.go new file mode 100644 index 00000000000..a15cf899591 --- /dev/null +++ b/pkg/scheduler/framework/plugins/nodevolumelimits/csi_manager.go @@ -0,0 +1,41 @@ +package nodevolumelimits + +import ( + storagev1 "k8s.io/api/storage/v1" + "k8s.io/apimachinery/pkg/labels" + storagelisters "k8s.io/client-go/listers/storage/v1" + fwk "k8s.io/kube-scheduler/framework" +) + +// CSIManagerImpl is a implementation of the CSIManager interface. +type CSIManagerImpl struct { + csiNodeListerImpl *csiNodeLister +} + +var _ fwk.CSIManager = &CSIManagerImpl{} + +func NewCSIManager(csinodeLister storagelisters.CSINodeLister) *CSIManagerImpl { + return &CSIManagerImpl{csiNodeListerImpl: NewCsiNodeLister(csinodeLister)} +} + +func (m *CSIManagerImpl) CSINodes() fwk.CSINodeLister { + return m.csiNodeListerImpl +} + +type csiNodeLister struct { + csinodeLister storagelisters.CSINodeLister +} + +var _ fwk.CSINodeLister = &csiNodeLister{} + +func NewCsiNodeLister(csinodeLister storagelisters.CSINodeLister) *csiNodeLister { + return &csiNodeLister{csinodeLister: csinodeLister} +} + +func (l *csiNodeLister) List() ([]*storagev1.CSINode, error) { + return l.csinodeLister.List(labels.Everything()) +} + +func (l *csiNodeLister) Get(name string) (*storagev1.CSINode, error) { + return l.csinodeLister.Get(name) +} diff --git a/pkg/scheduler/framework/plugins/nodevolumelimits/csi_node_lister.go b/pkg/scheduler/framework/plugins/nodevolumelimits/csi_node_lister.go deleted file mode 100644 index a8fd073745d..00000000000 --- a/pkg/scheduler/framework/plugins/nodevolumelimits/csi_node_lister.go +++ /dev/null @@ -1,26 +0,0 @@ -package nodevolumelimits - -import ( - storagev1 "k8s.io/api/storage/v1" - "k8s.io/apimachinery/pkg/labels" - storagelisters "k8s.io/client-go/listers/storage/v1" - fwk "k8s.io/kube-scheduler/framework" -) - -type CSINodeListerImpl struct { - csinodeLister storagelisters.CSINodeLister -} - -var _ fwk.CSINodeLister = &CSINodeListerImpl{} - -func NewCSINodeLister(csinodeLister storagelisters.CSINodeLister) *CSINodeListerImpl { - return &CSINodeListerImpl{csinodeLister: csinodeLister} -} - -func (l *CSINodeListerImpl) List() ([]*storagev1.CSINode, error) { - return l.csinodeLister.List(labels.Everything()) -} - -func (l *CSINodeListerImpl) Get(name string) (*storagev1.CSINode, error) { - return l.csinodeLister.Get(name) -} diff --git a/pkg/scheduler/framework/runtime/framework.go b/pkg/scheduler/framework/runtime/framework.go index afae5fd7bb1..e527f8c20b0 100644 --- a/pkg/scheduler/framework/runtime/framework.go +++ b/pkg/scheduler/framework/runtime/framework.go @@ -82,7 +82,7 @@ type frameworkImpl struct { logger klog.Logger // for tracking CSI node limits - sharedCSINodeLister fwk.CSINodeLister + sharedCSIManager fwk.CSIManager metricsRecorder *metrics.MetricAsyncRecorder profileName string @@ -137,7 +137,7 @@ type frameworkOptions struct { eventRecorder events.EventRecorder informerFactory informers.SharedInformerFactory sharedDRAManager fwk.SharedDRAManager - sharedCSINodeLister fwk.CSINodeLister + sharedCSIManager fwk.CSIManager snapshotSharedLister fwk.SharedLister metricsRecorder *metrics.MetricAsyncRecorder podNominator fwk.PodNominator @@ -198,10 +198,10 @@ func WithSharedDRAManager(sharedDRAManager fwk.SharedDRAManager) Option { } } -// WithSharedCSINodeLister sets SharedCSINodeLister for the framework. -func WithSharedCSINodeLister(sharedCSINodeLister fwk.CSINodeLister) Option { +// WithSharedCSIManager sets SharedCSIManager for the framework. +func WithSharedCSIManager(sharedCSIManager fwk.CSIManager) Option { return func(o *frameworkOptions) { - o.sharedCSINodeLister = sharedCSINodeLister + o.sharedCSIManager = sharedCSIManager } } @@ -301,7 +301,7 @@ func NewFramework(ctx context.Context, r Registry, profile *config.KubeScheduler f := &frameworkImpl{ registry: r, snapshotSharedLister: options.snapshotSharedLister, - sharedCSINodeLister: options.sharedCSINodeLister, + sharedCSIManager: options.sharedCSIManager, scorePluginWeight: make(map[string]int), waitingPods: options.waitingPods, clientSet: options.clientSet, @@ -1734,9 +1734,9 @@ func (f *frameworkImpl) SharedDRAManager() fwk.SharedDRAManager { return f.sharedDRAManager } -// SharedCSINodeLister returns the SharedCSINodeLister of the framework. -func (f *frameworkImpl) SharedCSINodeLister() fwk.CSINodeLister { - return f.sharedCSINodeLister +// SharedCSIManager returns the SharedCSIManager of the framework. +func (f *frameworkImpl) SharedCSIManager() fwk.CSIManager { + return f.sharedCSIManager } func (f *frameworkImpl) pluginsNeeded(plugins *config.Plugins) sets.Set[string] { diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index a5d1fc31641..75a374f679e 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -344,8 +344,8 @@ func New(ctx context.Context, } draManager = dynamicresources.NewDRAManager(ctx, resourceClaimCache, resourceSliceTracker, informerFactory) } - var sharedCSINodeLister fwk.CSINodeLister - sharedCSINodeLister = nodevolumelimits.NewCSINodeLister(informerFactory.Storage().V1().CSINodes().Lister()) + var sharedCSIManager fwk.CSIManager + sharedCSIManager = nodevolumelimits.NewCSIManager(informerFactory.Storage().V1().CSINodes().Lister()) var apiDispatcher *apidispatcher.APIDispatcher if feature.DefaultFeatureGate.Enabled(features.SchedulerAsyncAPICalls) { @@ -365,7 +365,7 @@ func New(ctx context.Context, frameworkruntime.WithMetricsRecorder(metricsRecorder), frameworkruntime.WithWaitingPods(waitingPods), frameworkruntime.WithAPIDispatcher(apiDispatcher), - frameworkruntime.WithSharedCSINodeLister(sharedCSINodeLister), + frameworkruntime.WithSharedCSIManager(sharedCSIManager), ) if err != nil { return nil, fmt.Errorf("initializing profiles: %v", err) diff --git a/staging/src/k8s.io/kube-scheduler/framework/interface.go b/staging/src/k8s.io/kube-scheduler/framework/interface.go index 8742a46a57c..00310ba4437 100644 --- a/staging/src/k8s.io/kube-scheduler/framework/interface.go +++ b/staging/src/k8s.io/kube-scheduler/framework/interface.go @@ -672,8 +672,9 @@ type Handle interface { // A non-default implementation can be plugged into the framework to simulate the state of DRA objects. SharedDRAManager() SharedDRAManager - // SharedCSINodeLister returns the SharedCSINodeLister of the framework. - SharedCSINodeLister() CSINodeLister + // SharedCSIManager can be used to obtain CSINode objects, and track changes to them in-memory. + // A non-default implementation can be plugged into the framework to simulate the state of CSINode objects. + SharedCSIManager() CSIManager // RunFilterPluginsWithNominatedPods runs the set of configured filter plugins for nominated pod on the given node. RunFilterPluginsWithNominatedPods(ctx context.Context, state CycleState, pod *v1.Pod, info NodeInfo) *Status diff --git a/staging/src/k8s.io/kube-scheduler/framework/listers.go b/staging/src/k8s.io/kube-scheduler/framework/listers.go index 2888323577f..05cf96febd0 100644 --- a/staging/src/k8s.io/kube-scheduler/framework/listers.go +++ b/staging/src/k8s.io/kube-scheduler/framework/listers.go @@ -125,3 +125,11 @@ type SharedDRAManager interface { ResourceSlices() ResourceSliceLister DeviceClasses() DeviceClassLister } + +// CSIManager can be used to obtain CSINode objects, and track changes to CSINode objects in-memory. +// The plugin's default implementation obtains the objects from the API. A different implementation can be +// plugged into the framework in order to simulate the state of CSINode objects. For example, Cluster Autoscaler +// can use this to provide the correct CSINode object state to the CSINode plugin when simulating scheduling changes in-memory. +type CSIManager interface { + CSINodes() CSINodeLister +}