Merge pull request #2458 from jiangkaihua/master

Add nodeVolumeLimits plugin.

pkg/scheduler/plugins/predicates/predicates.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodeaffinity"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodeports"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodeunschedulable"
+	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodevolumelimits"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/tainttoleration"
 
 	"volcano.sh/volcano/pkg/scheduler/api"
@@ -311,6 +312,9 @@ func (pp *predicatesPlugin) OnSessionOpen(ssn *framework.Session) {
 	features := feature.Features{}
 	plugin, _ = interpodaffinity.New(plArgs, handle, features)
 	podAffinityFilter := plugin.(*interpodaffinity.InterPodAffinity)
+	// 6. NodeVolumeLimits
+	plugin, _ = nodevolumelimits.NewCSI(nil, handle, features)
+	nodeVolumeLimitsCSIFilter := plugin.(*nodevolumelimits.CSILimits)
 
 	ssn.AddPredicateFn(pp.Name(), func(task *api.TaskInfo, node *api.NodeInfo) error {
 		nodeInfo, found := nodeMap[node.Name]
@@ -386,6 +390,12 @@ func (pp *predicatesPlugin) OnSessionOpen(ssn *framework.Session) {
 			return fmt.Errorf("plugin %s predicates failed %s", interpodaffinity.Name, status.Message())
 		}
 
+		// Check NodeVolumeLimits
+		status = nodeVolumeLimitsCSIFilter.Filter(context.TODO(), state, task.Pod, nodeInfo)
+		if !status.IsSuccess() {
+			return fmt.Errorf("plugin %s predicates failed %s", nodeVolumeLimitsCSIFilter.Name(), status.Message())
+		}
+
 		if predicate.gpuSharingEnable {
 			// CheckGPUSharingPredicate
 			fit, err := checkNodeGPUSharingPredicate(task.Pod, node)

pkg/scheduler/plugins/predicates/predicates_test.go

@@ -5,6 +5,8 @@ import (
 	"testing"
 
 	"github.com/agiledragon/gomonkey/v2"
+	"github.com/spf13/pflag"
+
 	apiv1 "k8s.io/api/core/v1"
 	schedulingv1 "k8s.io/api/scheduling/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -13,6 +15,7 @@ import (
 
 	schedulingv1beta1 "volcano.sh/apis/pkg/apis/scheduling/v1beta1"
 	"volcano.sh/volcano/cmd/scheduler/app/options"
+	"volcano.sh/volcano/pkg/kube"
 	"volcano.sh/volcano/pkg/scheduler/actions/allocate"
 	"volcano.sh/volcano/pkg/scheduler/api"
 	"volcano.sh/volcano/pkg/scheduler/cache"
@@ -58,6 +61,18 @@ func TestEventHandler(t *testing.T) {
 	options.ServerOpts = options.NewServerOption()
 	defer framework.CleanupPluginBuilders()
 
+	option := options.NewServerOption()
+	option.AddFlags(pflag.CommandLine)
+	option.RegisterOptions()
+
+	config, err := kube.BuildConfig(option.KubeClientOptions)
+	if err != nil {
+		return
+	}
+
+	sc := cache.New(config, option.SchedulerNames, option.DefaultQueue, option.NodeSelector)
+	schedulerCache := sc.(*cache.SchedulerCache)
+
 	// pending pods
 	w1 := util.BuildPod("ns1", "worker-1", "", apiv1.PodPending, util.BuildResourceList("3", "3k"), "pg1", map[string]string{"role": "worker"}, map[string]string{"selector": "worker"})
 	w2 := util.BuildPod("ns1", "worker-2", "", apiv1.PodPending, util.BuildResourceList("5", "5k"), "pg1", map[string]string{"role": "worker"}, map[string]string{})
@@ -141,16 +156,6 @@ func TestEventHandler(t *testing.T) {
 				t.Logf("%s: [Event] %s", test.name, event)
 			}
 		}()
-		schedulerCache := &cache.SchedulerCache{
-			Nodes:           make(map[string]*api.NodeInfo),
-			Jobs:            make(map[api.JobID]*api.JobInfo),
-			PriorityClasses: make(map[string]*schedulingv1.PriorityClass),
-			Queues:          make(map[api.QueueID]*api.QueueInfo),
-			Binder:          binder,
-			StatusUpdater:   &util.FakeStatusUpdater{},
-			VolumeBinder:    &util.FakeVolumeBinder{},
-			Recorder:        recorder,
-		}
 		for _, node := range test.nodes {
 			schedulerCache.AddNode(node)
 		}