github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-18 04:54:54 +00:00
Code Issues Projects Releases Wiki Activity

Implement a csimanager for managing storage related assets

Browse Source
This commit is contained in:
Hemant Kumar
2025-10-31 10:05:55 -04:00
parent 7bbec73192
commit c71e45c735
7 changed files with 71 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
pkg/scheduler/framework/plugins/nodevolumelimits/csi.go
View File

@@ -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,

41
pkg/scheduler/framework/plugins/nodevolumelimits/csi_manager.go Normal file
View File

@@ -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)
}

26
pkg/scheduler/framework/plugins/nodevolumelimits/csi_node_lister.go
View File

@@ -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)
}

18
pkg/scheduler/framework/runtime/framework.go
View File

@@ -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] {

6
pkg/scheduler/scheduler.go
View File

@@ -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)

5
staging/src/k8s.io/kube-scheduler/framework/interface.go
View File

@@ -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

8
staging/src/k8s.io/kube-scheduler/framework/listers.go
View File

@@ -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
}