github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-25 14:06:46 +00:00
Code Issues Projects Releases Wiki Activity

use indexer to acclerate volume limit plugin

Browse Source
This commit is contained in:
KunWuLuan
2025-08-20 17:44:51 +08:00
parent 028015267e
commit edf132e175
1 changed files with 17 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,11 +23,11 @@ import (
v1 "k8s.io/api/core/v1"
storagev1 "k8s.io/api/storage/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/util/rand"
corelisters "k8s.io/client-go/listers/core/v1"
storagelisters "k8s.io/client-go/listers/storage/v1"
"k8s.io/client-go/tools/cache"
ephemeral "k8s.io/component-helpers/storage/ephemeral"
storagehelpers "k8s.io/component-helpers/storage/volume"
csitrans "k8s.io/csi-translation-lib"
@@ -63,6 +63,7 @@ type CSILimits struct {
scLister storagelisters.StorageClassLister
vaLister storagelisters.VolumeAttachmentLister
csiDriverLister storagelisters.CSIDriverLister
vaindexer cache.Indexer
randomVolumeIDPrefix string
enableVolumeLimitScaling bool
@@ -590,6 +591,14 @@ func NewCSI(_ context.Context, _ runtime.Object, handle fwk.Handle, fts feature.
scLister := informerFactory.Storage().V1().StorageClasses().Lister()
vaLister := informerFactory.Storage().V1().VolumeAttachments().Lister()
csiDriverLister := informerFactory.Storage().V1().CSIDrivers().Lister()
vaindexer := informerFactory.Storage().V1().VolumeAttachments().Informer().GetIndexer()
informerFactory.Storage().V1().VolumeAttachments().Informer().AddIndexers(cache.Indexers{"nodename": func(obj interface{}) ([]string, error) {
va, ok := obj.(*storagev1.VolumeAttachment)
if !ok {
return []string{}, nil
}
return []string{va.Spec.NodeName}, nil
}})
csiTranslator := csitrans.New()
return &CSILimits{
@@ -602,6 +611,7 @@ func NewCSI(_ context.Context, _ runtime.Object, handle fwk.Handle, fts feature.
enableVolumeLimitScaling: fts.EnableVolumeLimitScaling,
randomVolumeIDPrefix: rand.String(32),
translator: csiTranslator,
vaindexer: vaindexer,
}, nil
}
@@ -624,11 +634,15 @@ func getVolumeLimits(csiNode *storagev1.CSINode) map[string]int64 {
// getNodeVolumeAttachmentInfo returns a map of volumeID to driver name for the given node.
func (pl *CSILimits) getNodeVolumeAttachmentInfo(logger klog.Logger, nodeName string) (map[string]string, error) {
volumeAttachments := make(map[string]string)
vas, err := pl.vaLister.List(labels.Everything())
vas, err := pl.vaindexer.ByIndex("nodename", nodeName)
if err != nil {
return nil, err
}
for _, va := range vas {
for _, vao := range vas {
va, ok := vao.(*storagev1.VolumeAttachment)
if !ok {
continue
}
if va.Spec.NodeName == nodeName {
if va.Spec.Attacher == "" {
logger.V(5).Info("VolumeAttachment has no attacher", "VolumeAttachment", klog.KObj(va))