Merge pull request kubernetes#126 from denkensk/use-queueStrategy

Use QueueStrategy in ClusterQueue construct
yaroslava-serdiuk · Mar 18, 2022 · 31f7287 · 31f7287
2 parents 32a2d30 + eb12462
commit 31f7287
Show file tree

Hide file tree

Showing 5 changed files with 34 additions and 7 deletions.
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
@@ -67,7 +67,12 @@ func (m *Manager) AddClusterQueue(ctx context.Context, cq *kueue.ClusterQueue) e
 	if _, ok := m.clusterQueues[cq.Name]; ok {
 		return errClusterQueueAlreadyExists
 	}
-	cqImpl := newClusterQueue(cq)
+
+	cqImpl, err := newClusterQueue(cq)
+	if err != nil {
+		return err
+	}
+
 	m.clusterQueues[cq.Name] = cqImpl
 
 	// Iterate through existing queues, as queues corresponding to this cluster

diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go
@@ -74,15 +74,24 @@ type ClusterQueue struct {
 	heap heapImpl
 }
 
-func newClusterQueue(cq *kueue.ClusterQueue) *ClusterQueue {
+func newClusterQueue(cq *kueue.ClusterQueue) (*ClusterQueue, error) {
+	var less lessFunc
+
+	switch cq.Spec.QueueingStrategy {
+	case kueue.StrictFIFO:
+		less = strictFIFO
+	default:
+		return nil, fmt.Errorf("invalid QueueingStrategy %q", cq.Spec.QueueingStrategy)
+	}
+
 	cqImpl := &ClusterQueue{
 		heap: heapImpl{
-			less:  creationFIFO,
+			less:  less,
 			items: make(map[string]*heapItem),
 		},
 	}
 	cqImpl.update(cq)
-	return cqImpl
+	return cqImpl, nil
 }
 
 func (cq *ClusterQueue) update(apiCQ *kueue.ClusterQueue) {
@@ -140,7 +149,10 @@ func (cq *ClusterQueue) Pop() *workload.Info {
 	return &w
 }
 
-func creationFIFO(a, b workload.Info) bool {
+// strictFIFO is the function used by the clusterQueue heap algorithm to sort
+// workloads. It sorts workloads based on their priority.
+// When priorities are equal, it uses workloads.creationTimestamp.
+func strictFIFO(a, b workload.Info) bool {
 	p1 := utilpriority.Priority(a.Obj)
 	p2 := utilpriority.Priority(b.Obj)
 

diff --git a/pkg/queue/queue_test.go b/pkg/queue/queue_test.go
@@ -32,11 +32,14 @@ const (
 )
 
 func TestFIFOClusterQueue(t *testing.T) {
-	q := newClusterQueue(&kueue.ClusterQueue{
+	q, err := newClusterQueue(&kueue.ClusterQueue{
 		Spec: kueue.ClusterQueueSpec{
 			QueueingStrategy: kueue.StrictFIFO,
 		},
 	})
+	if err != nil {
+		t.Fatalf("Failed creating ClusterQueue %v", err)
+	}
 	now := metav1.Now()
 	ws := []*kueue.QueuedWorkload{
 		{
@@ -166,11 +169,14 @@ func TestStrictFIFO(t *testing.T) {
 		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
-			q := newClusterQueue(&kueue.ClusterQueue{
+			q, err := newClusterQueue(&kueue.ClusterQueue{
 				Spec: kueue.ClusterQueueSpec{
 					QueueingStrategy: kueue.StrictFIFO,
 				},
 			})
+			if err != nil {
+				t.Fatalf("Failed creating ClusterQueue %v", err)
+			}
 
 			q.PushOrUpdate(tt.w1)
 			q.PushOrUpdate(tt.w2)

diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
@@ -62,6 +62,7 @@ func TestSchedule(t *testing.T) {
 						},
 					},
 				},
+				QueueingStrategy: kueue.StrictFIFO,
 				RequestableResources: []kueue.Resource{
 					{
 						Name: corev1.ResourceCPU,
@@ -91,6 +92,7 @@ func TestSchedule(t *testing.T) {
 						},
 					},
 				},
+				QueueingStrategy: kueue.StrictFIFO,
 				RequestableResources: []kueue.Resource{
 					{
 						Name: corev1.ResourceCPU,
@@ -127,6 +129,7 @@ func TestSchedule(t *testing.T) {
 						},
 					},
 				},
+				QueueingStrategy: kueue.StrictFIFO,
 				RequestableResources: []kueue.Resource{
 					{
 						Name: corev1.ResourceCPU,

diff --git a/pkg/util/testing/wrappers.go b/pkg/util/testing/wrappers.go
@@ -251,6 +251,7 @@ func MakeClusterQueue(name string) *ClusterQueueWrapper {
 		},
 		Spec: kueue.ClusterQueueSpec{
 			NamespaceSelector: &metav1.LabelSelector{},
+			QueueingStrategy:  kueue.StrictFIFO,
 		},
 	}}
 }