[cherry-pick for release-1.8]fix: task scheduling latancy metrics is not accurate

pkg/scheduler/cache/cache.go

@@ -91,9 +91,9 @@ func New(config *rest.Config, schedulerNames []string, defaultQueue string, node
 type SchedulerCache struct {
 	sync.Mutex
 
-	kubeClient   *kubernetes.Clientset
+	kubeClient   kubernetes.Interface
 	restConfig   *rest.Config
-	vcClient     *vcclient.Clientset
+	vcClient     vcclient.Interface
 	defaultQueue string
 	// schedulerName is the name for volcano scheduler
 	schedulerNames     []string
@@ -174,6 +174,8 @@ func (db *DefaultBinder) Bind(kubeClient kubernetes.Interface, tasks []*scheduli
 			metav1.CreateOptions{}); err != nil {
 			klog.Errorf("Failed to bind pod <%v/%v> to node %s : %#v", p.Namespace, p.Name, task.NodeName, err)
 			errTasks = append(errTasks, task)
+		} else {
+			metrics.UpdateTaskScheduleDuration(metrics.Duration(p.CreationTimestamp.Time)) // update metrics as soon as pod is bind
 		}
 	}
 
@@ -189,7 +191,7 @@ func NewBinder() *DefaultBinder {
 }
 
 type defaultEvictor struct {
-	kubeclient *kubernetes.Clientset
+	kubeclient kubernetes.Interface
 	recorder   record.EventRecorder
 }
 
@@ -228,8 +230,8 @@ func (de *defaultEvictor) Evict(p *v1.Pod, reason string) error {
 
 // defaultStatusUpdater is the default implementation of the StatusUpdater interface
 type defaultStatusUpdater struct {
-	kubeclient *kubernetes.Clientset
-	vcclient   *vcclient.Clientset
+	kubeclient kubernetes.Interface
+	vcclient   vcclient.Interface
 }
 
 // following the same logic as podutil.UpdatePodCondition
@@ -347,8 +349,8 @@ func (dvb *defaultVolumeBinder) BindVolumes(task *schedulingapi.TaskInfo, podVol
 }
 
 type podgroupBinder struct {
-	kubeclient *kubernetes.Clientset
-	vcclient   *vcclient.Clientset
+	kubeclient kubernetes.Interface
+	vcclient   vcclient.Interface
 }
 
 // Bind will add silo cluster annotaion on pod and podgroup
@@ -794,11 +796,12 @@ func (sc *SchedulerCache) Evict(taskInfo *schedulingapi.TaskInfo, reason string)
 }
 
 // Bind binds task to the target host.
-func (sc *SchedulerCache) Bind(tasks []*schedulingapi.TaskInfo) error {
+func (sc *SchedulerCache) Bind(tasks []*schedulingapi.TaskInfo) {
 	tmp := time.Now()
 	errTasks, err := sc.Binder.Bind(sc.kubeClient, tasks)
 	if err == nil {
 		klog.V(3).Infof("bind ok, latency %v", time.Since(tmp))
+		// TODO: need to move this event recording into Bind so that record it as soon as pod is bind
 		for _, task := range tasks {
 			sc.Recorder.Eventf(task.Pod, v1.EventTypeNormal, "Scheduled", "Successfully assigned %v/%v to %v",
 				task.Namespace, task.Name, task.NodeName)
@@ -810,7 +813,6 @@ func (sc *SchedulerCache) Bind(tasks []*schedulingapi.TaskInfo) error {
 			sc.resyncTask(task)
 		}
 	}
-	return nil
 }
 
 // BindPodGroup binds job to silo cluster
@@ -982,6 +984,7 @@ func (sc *SchedulerCache) processResyncTask() {
 	}
 }
 
+// AddBindTask add task to be bind to a cache which consumes by go runtime
 func (sc *SchedulerCache) AddBindTask(taskInfo *schedulingapi.TaskInfo) error {
 	klog.V(5).Infof("add bind task %v/%v", taskInfo.Namespace, taskInfo.Name)
 	sc.Mutex.Lock()
@@ -1052,6 +1055,7 @@ func (sc *SchedulerCache) processBindTask() {
 	sc.BindTask()
 }
 
+// BindTask do k8s binding with a goroutine
 func (sc *SchedulerCache) BindTask() {
 	klog.V(5).Infof("batch bind task count %d", len(sc.bindCache))
 	var tmpBindCache []*schedulingapi.TaskInfo = make([]*schedulingapi.TaskInfo, len(sc.bindCache))
@@ -1071,14 +1075,7 @@ func (sc *SchedulerCache) BindTask() {
 
 		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 successfulTasks {
-			metrics.UpdateTaskScheduleDuration(metrics.Duration(task.Pod.CreationTimestamp.Time))
-		}
+		sc.Bind(bindTasks)
 	}(tmpBindCache)
 	sc.bindCache = sc.bindCache[0:0]
 }

pkg/scheduler/cache/cache_test.go

@@ -17,16 +17,25 @@ limitations under the License.
 package cache
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"testing"
+	"time"
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
 
 	"volcano.sh/volcano/pkg/scheduler/api"
+	volumescheduling "volcano.sh/volcano/pkg/scheduler/capabilities/volumebinding"
 	"volcano.sh/volcano/pkg/scheduler/util"
 )
 
@@ -314,3 +323,79 @@ func TestNodeOperation(t *testing.T) {
 		}
 	}
 }
+
+func TestBindTasks(t *testing.T) {
+	owner := buildOwnerReference("j1")
+	scheduler := "fake-scheduler"
+
+	fakeKube := fake.NewSimpleClientset()
+	informerFactory := informers.NewSharedInformerFactory(fakeKube, 0)
+	sc := &SchedulerCache{
+		Jobs:            make(map[api.JobID]*api.JobInfo),
+		Nodes:           make(map[string]*api.NodeInfo),
+		kubeClient:      fakeKube,
+		schedulerNames:  []string{scheduler},
+		Recorder:        record.NewFakeRecorder(10),
+		BindFlowChannel: make(chan *api.TaskInfo, 5000),
+		podInformer:     informerFactory.Core().V1().Pods(),
+		nodeInformer:    informerFactory.Core().V1().Nodes(),
+		csiNodeInformer: informerFactory.Storage().V1().CSINodes(),
+		pvcInformer:     informerFactory.Core().V1().PersistentVolumeClaims(),
+		pvInformer:      informerFactory.Core().V1().PersistentVolumes(),
+		scInformer:      informerFactory.Storage().V1().StorageClasses(),
+		errTasks:        workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+	}
+
+	sc.Binder = &DefaultBinder{}
+	sc.VolumeBinder = &defaultVolumeBinder{
+		volumeBinder: volumescheduling.NewVolumeBinder(
+			sc.kubeClient,
+			sc.podInformer,
+			sc.nodeInformer,
+			sc.csiNodeInformer,
+			sc.pvcInformer,
+			sc.pvInformer,
+			sc.scInformer,
+			nil,
+			100*time.Millisecond,
+		)}
+
+	sc.podInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    sc.AddPod,
+			UpdateFunc: sc.UpdatePod,
+			DeleteFunc: sc.DeletePod,
+		},
+	)
+	sc.nodeInformer.Informer().AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    sc.AddNode,
+			UpdateFunc: sc.UpdateNode,
+		},
+	)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go wait.Until(sc.processBindTask, time.Millisecond*5, ctx.Done())
+	pod := buildPod("c1", "p1", "", v1.PodPending, buildResourceList("1000m", "1G"), []metav1.OwnerReference{owner}, make(map[string]string))
+	node := buildNode("n1", buildResourceList("2000m", "10G"))
+	pod.Annotations = map[string]string{"scheduling.k8s.io/group-name": "j1"}
+	pod.Spec.SchedulerName = scheduler
+
+	// make sure pod exist when calling fake client binding
+	fakeKube.CoreV1().Pods(pod.Namespace).Create(ctx, pod, metav1.CreateOptions{})
+	fakeKube.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{})
+	informerFactory.Start(ctx.Done())
+	informerFactory.WaitForCacheSync(ctx.Done())
+
+	task := api.NewTaskInfo(pod)
+	task.NodeName = "n1"
+	err := sc.AddBindTask(task)
+	if err != nil {
+		t.Errorf("failed to bind pod to node: %v", err)
+	}
+	time.Sleep(200 * time.Millisecond)
+	r := sc.Recorder.(*record.FakeRecorder)
+	if len(r.Events) != 1 {
+		t.Fatalf("succesfully binding task should have 1 event")
+	}
+}

pkg/scheduler/cache/factory.go

@@ -8,6 +8,7 @@ func RegisterBindMethod(binder Binder) {
 	bindMethodMap = binder
 }
 
+// GetBindMethod get the registered Binder
 func GetBindMethod() Binder {
 	return bindMethodMap
 }

pkg/scheduler/capabilities/volumebinding/binder.go

@@ -522,6 +522,11 @@ func (b *volumeBinder) BindPodVolumes(ctx context.Context, assumedPod *v1.Pod, p
 		}
 	}()
 
+	if podVolumes == nil {
+		klog.Infof("BindPodVolumes for pod(%s): pod volumes is nil", assumedPod.Name)
+		return nil
+	}
+
 	bindings := podVolumes.StaticBindings
 	claimsToProvision := podVolumes.DynamicProvisions