merge pr from release-1.6

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

@@ -57,6 +57,8 @@ type ServerOption struct {
 	// For dependent tasks, there is a detection cycle inside volcano
 	// It indicates how often to detect the status of dependent tasks
 	DetectionPeriodOfDependsOntask time.Duration
+	// To determine whether inherit owner's annotations for pods when create podgroup
+	InheritOwnerAnnotations bool
 }
 
 // NewServerOption creates a new CMServer with a default config.
@@ -82,6 +84,7 @@ func (s *ServerOption) AddFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&s.EnableHealthz, "enable-healthz", false, "Enable the health check; it is false by default")
 	fs.DurationVar(&s.DetectionPeriodOfDependsOntask, "detection-period-of-dependson-task", defaultDetectionPeriodOfDependsOntask, "It indicates how often to detect the status of dependent tasks."+
 		"e.g. --detection-period-of-dependson-task=1s")
+	fs.BoolVar(&s.InheritOwnerAnnotations, "inherit-owner-annotations", true, "Enable inherit owner annotations for pods when create podgroup; it is enabled by default")
 }
 
 // CheckOptionOrDie checks the LockObjectNamespace.

cmd/controller-manager/app/server.go

@@ -126,6 +126,7 @@ func startControllers(config *rest.Config, opt *options.ServerOption) func(ctx c
 	controllerOpt.KubeClient = kubeclientset.NewForConfigOrDie(config)
 	controllerOpt.VolcanoClient = vcclientset.NewForConfigOrDie(config)
 	controllerOpt.SharedInformerFactory = informers.NewSharedInformerFactory(controllerOpt.KubeClient, 0)
+	controllerOpt.InheritOwnerAnnotations = opt.InheritOwnerAnnotations
 
 	return func(ctx context.Context) {
 		framework.ForeachController(func(c framework.Controller) {

installer/helm/chart/volcano/templates/scheduler.yaml

@@ -41,10 +41,10 @@ rules:
     verbs: ["list", "watch", "update"]
   - apiGroups: [""]
     resources: ["persistentvolumes"]
-    verbs: ["list", "watch"]
+    verbs: ["list", "watch", "update"]
   - apiGroups: [""]
-    resources: ["namespaces"]
-    verbs: ["list", "watch"]
+    resources: ["namespaces", "services", "replicationcontrollers"]
+    verbs: ["list", "watch", "get"]
   - apiGroups: [""]
     resources: ["resourcequotas"]
     verbs: ["list", "watch"]
@@ -72,6 +72,9 @@ rules:
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get", "create", "delete", "update"]
+  - apiGroups: ["apps"]
+    resources: ["daemonsets", "replicasets", "statefulsets"]
+    verbs: ["list", "watch", "get"]
 
 ---
 kind: ClusterRoleBinding
@@ -138,7 +141,7 @@ metadata:
     prometheus.io/port: "8080"
     prometheus.io/scrape: "true"
   name: {{ .Release.Name }}-scheduler-service
-  namespace: {{ .Release.Namespace }} 
+  namespace: {{ .Release.Namespace }}
 spec:
   ports:
   - port: 8080

installer/volcano-development.yaml

@@ -8568,10 +8568,10 @@ rules:
     verbs: ["list", "watch", "update"]
   - apiGroups: [""]
     resources: ["persistentvolumes"]
-    verbs: ["list", "watch"]
+    verbs: ["list", "watch", "update"]
   - apiGroups: [""]
-    resources: ["namespaces"]
-    verbs: ["list", "watch"]
+    resources: ["namespaces", "services", "replicationcontrollers"]
+    verbs: ["list", "watch", "get"]
   - apiGroups: [""]
     resources: ["resourcequotas"]
     verbs: ["list", "watch"]
@@ -8599,6 +8599,9 @@ rules:
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get", "create", "delete", "update"]
+  - apiGroups: ["apps"]
+    resources: ["daemonsets", "replicasets", "statefulsets"]
+    verbs: ["list", "watch", "get"]
 ---
 # Source: volcano/templates/scheduler.yaml
 kind: ClusterRoleBinding
@@ -8623,7 +8626,7 @@ metadata:
     prometheus.io/port: "8080"
     prometheus.io/scrape: "true"
   name: volcano-scheduler-service
-  namespace: volcano-system 
+  namespace: volcano-system
 spec:
   ports:
   - port: 8080

pkg/controllers/framework/interface.go

@@ -31,6 +31,8 @@ type ControllerOption struct {
 	SchedulerNames        []string
 	WorkerNum             uint32
 	MaxRequeueNum         int
+
+	InheritOwnerAnnotations bool
 }
 
 // Controller is the interface of all controllers.

pkg/controllers/podgroup/pg_controller.go

@@ -62,6 +62,9 @@ type pgcontroller struct {
 	queue workqueue.RateLimitingInterface
 
 	schedulerNames []string
+
+	// To determine whether inherit owner's annotations for pods when create podgroup
+	inheritOwnerAnnotations bool
 }
 
 func (pg *pgcontroller) Name() string {
@@ -77,6 +80,7 @@ func (pg *pgcontroller) Initialize(opt *framework.ControllerOption) error {
 
 	pg.schedulerNames = make([]string, len(opt.SchedulerNames))
 	copy(pg.schedulerNames, opt.SchedulerNames)
+	pg.inheritOwnerAnnotations = opt.InheritOwnerAnnotations
 
 	pg.informerFactory = opt.SharedInformerFactory
 	pg.podInformer = opt.SharedInformerFactory.Core().V1().Pods()

pkg/controllers/podgroup/pg_controller_handler.go

@@ -129,6 +129,22 @@ func (pg *pgcontroller) getAnnotationsFromUpperRes(kind string, name string, nam
 	}
 }
 
+// Inherit annotations from upper resources.
+func (pg *pgcontroller) inheritUpperAnnotations(pod *v1.Pod, obj *scheduling.PodGroup) {
+	if pg.inheritOwnerAnnotations {
+		for _, reference := range pod.OwnerReferences {
+			if reference.Kind != "" && reference.Name != "" {
+				var upperAnnotations = pg.getAnnotationsFromUpperRes(reference.Kind, reference.Name, pod.Namespace)
+				for k, v := range upperAnnotations {
+					if strings.HasPrefix(k, scheduling.AnnotationPrefix) {
+						obj.Annotations[k] = v
+					}
+				}
+			}
+		}
+	}
+}
+
 func (pg *pgcontroller) createNormalPodPGIfNotExist(pod *v1.Pod) error {
 	pgName := helpers.GeneratePodgroupName(pod)
 
@@ -157,18 +173,7 @@ func (pg *pgcontroller) createNormalPodPGIfNotExist(pod *v1.Pod) error {
 			},
 		}
 
-		// Inherit annotations from upper resources.
-		for _, reference := range pod.OwnerReferences {
-			if reference.Kind != "" && reference.Name != "" {
-				var upperAnnotations = pg.getAnnotationsFromUpperRes(reference.Kind, reference.Name, pod.Namespace)
-				for k, v := range upperAnnotations {
-					if strings.HasPrefix(k, scheduling.AnnotationPrefix) {
-						obj.Annotations[k] = v
-					}
-				}
-			}
-		}
-
+		pg.inheritUpperAnnotations(pod, obj)
 		// Individual annotations on pods would overwrite annotations inherited from upper resources.
 		if queueName, ok := pod.Annotations[scheduling.QueueNameAnnotationKey]; ok {
 			obj.Spec.Queue = queueName

pkg/scheduler/cache/cache.go

@@ -489,6 +489,15 @@ func newSchedulerCache(config *rest.Config, schedulerNames []string, defaultQueu
 	sc.informerFactory = informerFactory
 	mySchedulerPodName, c := getMultiSchedulerInfo()
 
+	// explictly register informers to the factory, otherwise resources listers cannot get anything
+	// even with no erorr returned. `Namespace` informer is used by `InterPodAffinity` plugin,
+	// `SelectorSpread` and `PodTopologySpread` plugins uses the following four so far.
+	informerFactory.Core().V1().Namespaces().Informer()
+	informerFactory.Core().V1().Services().Informer()
+	informerFactory.Core().V1().ReplicationControllers().Informer()
+	informerFactory.Apps().V1().ReplicaSets().Informer()
+	informerFactory.Apps().V1().StatefulSets().Informer()
+
 	// create informer for node information
 	sc.nodeInformer = informerFactory.Core().V1().Nodes()
 	sc.nodeInformer.Informer().AddEventHandlerWithResyncPeriod(
@@ -990,22 +999,26 @@ func (sc *SchedulerCache) processBindTask() {
 
 func (sc *SchedulerCache) BindTask() {
 	klog.V(5).Infof("batch bind task count %d", len(sc.bindCache))
+	successfulTasks := make([]*schedulingapi.TaskInfo, 0)
 	for _, task := range sc.bindCache {
 		if err := sc.VolumeBinder.BindVolumes(task, task.PodVolumes); err != nil {
 			klog.Errorf("task %s/%s bind Volumes failed: %#v", task.Namespace, task.Name, err)
 			sc.VolumeBinder.RevertVolumes(task, task.PodVolumes)
 			sc.resyncTask(task)
-			return
+		} else {
+			successfulTasks = append(successfulTasks, task)
+			klog.V(5).Infof("task %s/%s bind Volumes done", task.Namespace, task.Name)
 		}
 	}
 
-	bindTasks := make([]*schedulingapi.TaskInfo, len(sc.bindCache))
-	copy(bindTasks, sc.bindCache)
+	bindTasks := make([]*schedulingapi.TaskInfo, len(successfulTasks))
+	copy(bindTasks, successfulTasks)
 	if err := sc.Bind(bindTasks); err != nil {
+		klog.Errorf("failed to bind task count %d: %#v", len(bindTasks), err)
 		return
 	}
 
-	for _, task := range sc.bindCache {
+	for _, task := range successfulTasks {
 		metrics.UpdateTaskScheduleDuration(metrics.Duration(task.Pod.CreationTimestamp.Time))
 	}
 

pkg/scheduler/cache/event_handlers.go

@@ -411,24 +411,18 @@ func (sc *SchedulerCache) AddOrUpdateCSINode(obj interface{}) {
 		return
 	}
 
-	var csiNodeStatus *schedulingapi.CSINodeStatusInfo
-	var found bool
+	csiNodeStatus := &schedulingapi.CSINodeStatusInfo{
+		CSINodeName:  csiNode.Name,
+		DriverStatus: make(map[string]bool),
+	}
 	sc.Mutex.Lock()
 	defer sc.Mutex.Unlock()
-	// update nodeVolumeCount
-
-	if csiNodeStatus, found = sc.CSINodesStatus[csiNode.Name]; !found {
-		csiNodeStatus = &schedulingapi.CSINodeStatusInfo{
-			CSINodeName:  csiNode.Name,
-			DriverStatus: make(map[string]bool),
-		}
-		sc.CSINodesStatus[csiNode.Name] = csiNodeStatus
-	}
 
 	for i := range csiNode.Spec.Drivers {
 		d := csiNode.Spec.Drivers[i]
 		csiNodeStatus.DriverStatus[d.Name] = d.Allocatable != nil && d.Allocatable.Count != nil
 	}
+	sc.CSINodesStatus[csiNode.Name] = csiNodeStatus
 }
 
 func (sc *SchedulerCache) UpdateCSINode(oldObj, newObj interface{}) {