Merge pull request #77529 from draveness/feature/add-queuesort-extension-point

pkg/scheduler/core/extender_test.go

@@ -536,7 +536,7 @@ func TestGenericSchedulerWithExtenders(t *testing.T) {
 			for _, name := range test.nodes {
 				cache.AddNode(createNode(name))
 			}
-			queue := internalqueue.NewSchedulingQueue(nil)
+			queue := internalqueue.NewSchedulingQueue(nil, nil)
 			scheduler := NewGenericScheduler(
 				cache,
 				queue,

pkg/scheduler/core/generic_scheduler_test.go

@@ -452,7 +452,7 @@ func TestGenericScheduler(t *testing.T) {
 
 			scheduler := NewGenericScheduler(
 				cache,
-				internalqueue.NewSchedulingQueue(nil),
+				internalqueue.NewSchedulingQueue(nil, nil),
 				test.predicates,
 				algorithmpredicates.EmptyPredicateMetadataProducer,
 				test.prioritizers,
@@ -488,7 +488,7 @@ func makeScheduler(predicates map[string]algorithmpredicates.FitPredicate, nodes
 
 	s := NewGenericScheduler(
 		cache,
-		internalqueue.NewSchedulingQueue(nil),
+		internalqueue.NewSchedulingQueue(nil, nil),
 		predicates,
 		algorithmpredicates.EmptyPredicateMetadataProducer,
 		prioritizers,
@@ -1491,7 +1491,7 @@ func TestPreempt(t *testing.T) {
 			}
 			scheduler := NewGenericScheduler(
 				cache,
-				internalqueue.NewSchedulingQueue(nil),
+				internalqueue.NewSchedulingQueue(nil, nil),
 				map[string]algorithmpredicates.FitPredicate{"matches": algorithmpredicates.PodFitsResources},
 				algorithmpredicates.EmptyPredicateMetadataProducer,
 				[]priorities.PriorityConfig{{Function: numericPriority, Weight: 1}},

pkg/scheduler/factory/factory.go

@@ -262,7 +262,7 @@ func NewConfigFactory(args *ConfigFactoryArgs) Configurator {
 	c := &configFactory{
 		client:                         args.Client,
 		podLister:                      schedulerCache,
-		podQueue:                       internalqueue.NewSchedulingQueue(stopEverything),
+		podQueue:                       internalqueue.NewSchedulingQueue(stopEverything, framework),
 		nodeLister:                     args.NodeInformer.Lister(),
 		pVLister:                       args.PvInformer.Lister(),
 		pVCLister:                      args.PvcInformer.Lister(),

pkg/scheduler/factory/factory_test.go

@@ -256,7 +256,7 @@ func TestDefaultErrorFunc(t *testing.T) {
 	defer close(stopCh)
 
 	timestamp := time.Now()
-	queue := internalqueue.NewPriorityQueueWithClock(nil, clock.NewFakeClock(timestamp))
+	queue := internalqueue.NewPriorityQueueWithClock(nil, clock.NewFakeClock(timestamp), nil)
 	schedulerCache := internalcache.New(30*time.Second, stopCh)
 	errFunc := MakeDefaultErrorFunc(client, queue, schedulerCache, stopCh)
 

pkg/scheduler/framework/v1alpha1/framework.go

@@ -34,6 +34,7 @@ type framework struct {
 	nodeInfoSnapshot *cache.NodeInfoSnapshot
 	waitingPods      *waitingPodsMap
 	plugins          map[string]Plugin // a map of initialized plugins. Plugin name:plugin instance.
+	queueSortPlugins []QueueSortPlugin
 	reservePlugins   []ReservePlugin
 	prebindPlugins   []PrebindPlugin
 	unreservePlugins []UnreservePlugin
@@ -69,6 +70,10 @@ func NewFramework(r Registry, _ *runtime.Unknown) (Framework, error) {
 		// TODO: For now, we assume any plugins that implements an extension
 		// point wants to be called at that extension point. We should change this
 		// later and add these plugins based on the configuration.
+		if qsp, ok := p.(QueueSortPlugin); ok {
+			f.queueSortPlugins = append(f.queueSortPlugins, qsp)
+		}
+
 		if rp, ok := p.(ReservePlugin); ok {
 			f.reservePlugins = append(f.reservePlugins, rp)
 		}
@@ -85,6 +90,16 @@ func NewFramework(r Registry, _ *runtime.Unknown) (Framework, error) {
 	return f, nil
 }
 
+// QueueSortFunc returns the function to sort pods in scheduling queue
+func (f *framework) QueueSortFunc() LessFunc {
+	if len(f.queueSortPlugins) == 0 {
+		return nil
+	}
+
+	// Only one QueueSort plugin can be enabled.
+	return f.queueSortPlugins[0].Less
+}
+
 // RunPrebindPlugins runs the set of configured prebind plugins. It returns a
 // failure (bool) if any of the plugins returns an error. It also returns an
 // error containing the rejection message or the error occurred in the plugin.

pkg/scheduler/framework/v1alpha1/interface.go

@@ -107,6 +107,25 @@ type Plugin interface {
 	Name() string
 }
 
+// PodInfo is minimum cell in the scheduling queue.
+type PodInfo struct {
+	Pod *v1.Pod
+	// The time pod added to the scheduling queue.
+	Timestamp time.Time
+}
+
+// LessFunc is the function to sort pod info
+type LessFunc func(podInfo1, podInfo2 *PodInfo) bool
+
+// QueueSortPlugin is an interface that must be implemented by "QueueSort" plugins.
+// These plugins are used to sort pods in the scheduling queue. Only one queue sort
+// plugin may be enabled at a time.
+type QueueSortPlugin interface {
+	Plugin
+	// Less are used to sort pods in the scheduling queue.
+	Less(*PodInfo, *PodInfo) bool
+}
+
 // ReservePlugin is an interface for Reserve plugins. These plugins are called
 // at the reservation point. These are meant to update the state of the plugin.
 // This concept used to be called 'assume' in the original scheduler.
@@ -157,6 +176,9 @@ type PermitPlugin interface {
 // Configured plugins are called at specified points in a scheduling context.
 type Framework interface {
 	FrameworkHandle
+	// QueueSortFunc returns the function to sort pods in scheduling queue
+	QueueSortFunc() LessFunc
+
 	// RunPrebindPlugins runs the set of configured prebind plugins. It returns
 	// *Status and its code is set to non-success if any of the plugins returns
 	// anything but Success. If the Status code is "Unschedulable", it is

pkg/scheduler/internal/queue/BUILD

@@ -11,6 +11,7 @@ go_library(
     deps = [
         "//pkg/scheduler/algorithm/predicates:go_default_library",
         "//pkg/scheduler/algorithm/priorities/util:go_default_library",
+        "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/metrics:go_default_library",
         "//pkg/scheduler/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
@@ -31,6 +32,8 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//pkg/api/v1/pod:go_default_library",
+        "//pkg/scheduler/framework/v1alpha1:go_default_library",
+        "//pkg/scheduler/internal/cache:go_default_library",
         "//pkg/scheduler/metrics:go_default_library",
         "//pkg/scheduler/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",

pkg/scheduler/scheduler_test.go

@@ -641,7 +641,7 @@ func setupTestScheduler(queuedPodStore *clientcache.FIFO, scache internalcache.C
 	framework, _ := framework.NewFramework(EmptyPluginRegistry, nil)
 	algo := core.NewGenericScheduler(
 		scache,
-		internalqueue.NewSchedulingQueue(nil),
+		internalqueue.NewSchedulingQueue(nil, nil),
 		predicateMap,
 		predicates.EmptyPredicateMetadataProducer,
 		[]priorities.PriorityConfig{},
@@ -694,7 +694,7 @@ func setupTestSchedulerLongBindingWithRetry(queuedPodStore *clientcache.FIFO, sc
 	framework, _ := framework.NewFramework(EmptyPluginRegistry, nil)
 	algo := core.NewGenericScheduler(
 		scache,
-		internalqueue.NewSchedulingQueue(nil),
+		internalqueue.NewSchedulingQueue(nil, nil),
 		predicateMap,
 		predicates.EmptyPredicateMetadataProducer,
 		[]priorities.PriorityConfig{},