support multi-scheduler for k8s workload deployment, etc

Signed-off-by: huone1 <huwanxing@huawei.com>

cmd/controller-manager/app/options/options.go

@@ -45,8 +45,8 @@ type ServerOption struct {
 	// With the current rate-limiter in use (5ms*2^(maxRetries-1)) the following numbers represent the times
 	// a job, queue or command is going to be requeued:
 	// 5ms, 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms, 1.3s, 2.6s, 5.1s, 10.2s, 20.4s, 41s, 82s
-	MaxRequeueNum int
-	SchedulerName string
+	MaxRequeueNum  int
+	SchedulerNames []string
 	// HealthzBindAddress is the IP address and port for the health check server to serve on,
 	// defaulting to 0.0.0.0:11252
 	HealthzBindAddress string
@@ -72,7 +72,7 @@ func (s *ServerOption) AddFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&s.PrintVersion, "version", false, "Show version and quit")
 	fs.Uint32Var(&s.WorkerThreads, "worker-threads", defaultWorkers, "The number of threads syncing job operations concurrently. "+
 		"Larger number = faster job updating, but more CPU load")
-	fs.StringVar(&s.SchedulerName, "scheduler-name", defaultSchedulerName, "Volcano will handle pods whose .spec.SchedulerName is same as scheduler-name")
+	fs.StringArrayVar(&s.SchedulerNames, "scheduler-name", []string{defaultSchedulerName}, "Volcano will handle pods whose .spec.SchedulerName is same as scheduler-name")
 	fs.IntVar(&s.MaxRequeueNum, "max-requeue-num", defaultMaxRequeueNum, "The number of times a job, queue or command will be requeued before it is dropped out of the queue")
 }
 

cmd/controller-manager/app/options/options_test.go

@@ -33,6 +33,8 @@ func TestAddFlags(t *testing.T) {
 	args := []string{
 		"--master=127.0.0.1",
 		"--kube-api-burst=200",
+		"--scheduler-name=volcano",
+		"--scheduler-name=volcano2",
 	}
 	fs.Parse(args)
 
@@ -46,7 +48,7 @@ func TestAddFlags(t *testing.T) {
 		},
 		PrintVersion:       false,
 		WorkerThreads:      defaultWorkers,
-		SchedulerName:      defaultSchedulerName,
+		SchedulerNames:     []string{"volcano", "volcano2"},
 		MaxRequeueNum:      defaultMaxRequeueNum,
 		HealthzBindAddress: ":11252",
 	}

cmd/controller-manager/app/server.go

@@ -113,7 +113,7 @@ func Run(opt *options.ServerOption) error {
 func startControllers(config *rest.Config, opt *options.ServerOption) func(ctx context.Context) {
 	controllerOpt := &framework.ControllerOption{}
 
-	controllerOpt.SchedulerName = opt.SchedulerName
+	controllerOpt.SchedulerNames = opt.SchedulerNames
 	controllerOpt.WorkerNum = opt.WorkerThreads
 	controllerOpt.MaxRequeueNum = opt.MaxRequeueNum
 

pkg/controllers/framework/interface.go

@@ -28,7 +28,7 @@ type ControllerOption struct {
 	KubeClient            kubernetes.Interface
 	VolcanoClient         vcclientset.Interface
 	SharedInformerFactory informers.SharedInformerFactory
-	SchedulerName         string
+	SchedulerNames        []string
 	WorkerNum             uint32
 	MaxRequeueNum         int
 }

pkg/controllers/podgroup/pg_controller.go

@@ -76,7 +76,7 @@ func (pg *pgcontroller) Initialize(opt *framework.ControllerOption) error {
 			FilterFunc: func(obj interface{}) bool {
 				switch v := obj.(type) {
 				case *v1.Pod:
-					if v.Spec.SchedulerName == opt.SchedulerName &&
+					if contains(opt.SchedulerNames, v.Spec.SchedulerName) &&
 						(v.Annotations == nil || v.Annotations[scheduling.KubeGroupNameAnnotationKey] == "") {
 						return true
 					}
@@ -142,3 +142,12 @@ func (pg *pgcontroller) processNextReq() bool {
 
 	return true
 }
+
+func contains(slice []string, element string) bool {
+	for _, item := range slice {
+		if item == element {
+			return true
+		}
+	}
+	return false
+}

pkg/controllers/podgroup/pg_controller_test.go

@@ -42,7 +42,7 @@ func newFakeController() *pgcontroller {
 		KubeClient:            kubeClient,
 		VolcanoClient:         vcClient,
 		SharedInformerFactory: sharedInformers,
-		SchedulerName:         "volcano",
+		SchedulerNames:        []string{"volcano"},
 	}
 
 	controller.Initialize(opt)