Repository: kubernetes-sigs/kueue
Pull request #126 (2022-03-18 10:08:48)

Use QueueStrategy in ClusterQueue construct
Signed-off-by: Alex Wang <wangqingcan1990@gmail.com>

<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide/first-contribution.md#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

#### What type of PR is this?
/kind feature
<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:
Use QueueStrategy in ClusterQueue construct.

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
1/n of #8 

#### Special notes for your reviewer:





k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kueue/pull/126#" title="Author self-approved">denkensk</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **alculquicondor** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @alculquicondor` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkueue).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-sigs/kueue/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["alculquicondor"]} -->
alculquicondor: /approve
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kueue/pull/126#issuecomment-1072483139" title="Approved">alculquicondor</a>*, *<a href="https://github.com/kubernetes-sigs/kueue/pull/126#" title="Author self-approved">denkensk</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkueue).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/kueue/blob/main/OWNERS)~~ [alculquicondor]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index e1da2cb52..9b9b710ec 100644
--- 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
index d4214f63b..024237434 100644
--- 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)
 


Test patch:
diff --git a/pkg/queue/queue_test.go b/pkg/queue/queue_test.go
index 42186da7f..6535aa2f1 100644
--- 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
index b2788205a..24d228e51 100644
--- 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
index 430b2ad65..18050cb00 100644
--- 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,
 		},
 	}}
 }