Repository: kubernetes-sigs/kueue
Pull request #273 (2022-06-08 19:54:04)

cherry-pick: Immediate requeue in BestEffortFIFO should put a workload for retry
This is a cherry-pick of #246


Patch:
diff --git a/CHANGELOG/CHANGELOG-0.1.md b/CHANGELOG/CHANGELOG-0.1.md
index 1312a48a9..603a01db7 100644
--- a/CHANGELOG/CHANGELOG-0.1.md
+++ b/CHANGELOG/CHANGELOG-0.1.md
@@ -2,4 +2,6 @@
 
 Changes since `v0.1.0`:
 
-- Fixed number of pending workloads in a BestEffortFIFO ClusterQueue.
\ No newline at end of file
+- Fixed number of pending workloads in a BestEffortFIFO ClusterQueue.
+- Fixed bug in a BestEffortFIFO ClusterQueue where a workload might not be
+  retried after a transient error.
diff --git a/pkg/queue/cluster_queue_best_effort_fifo.go b/pkg/queue/cluster_queue_best_effort_fifo.go
index 704368205..1a75d03ab 100644
--- a/pkg/queue/cluster_queue_best_effort_fifo.go
+++ b/pkg/queue/cluster_queue_best_effort_fifo.go
@@ -80,11 +80,17 @@ func (cq *ClusterQueueBestEffortFIFO) Delete(w *kueue.Workload) {
 // the workload will be pushed back to heap directly. If not,
 // the workload will be put into the inadmissibleWorkloads.
 func (cq *ClusterQueueBestEffortFIFO) RequeueIfNotPresent(wInfo *workload.Info, immediate bool) bool {
+	key := workload.Key(wInfo.Obj)
 	if immediate {
+		// If the workload was inadmissible, move it back into the queue.
+		inadmissibleWl := cq.inadmissibleWorkloads[key]
+		if inadmissibleWl != nil {
+			wInfo = inadmissibleWl
+			delete(cq.inadmissibleWorkloads, key)
+		}
 		return cq.ClusterQueueImpl.pushIfNotPresent(wInfo)
 	}
 
-	key := workload.Key(wInfo.Obj)
 	if cq.inadmissibleWorkloads[key] != nil {
 		return false
 	}


Test patch:
diff --git a/pkg/queue/cluster_queue_best_effort_fifo_test.go b/pkg/queue/cluster_queue_best_effort_fifo_test.go
index a60061f4f..d7b28f711 100644
--- a/pkg/queue/cluster_queue_best_effort_fifo_test.go
+++ b/pkg/queue/cluster_queue_best_effort_fifo_test.go
@@ -42,55 +42,57 @@ func TestClusterQueueBestEffortFIFO(t *testing.T) {
 	updatedWorkloads[1].Spec.QueueName = "q1"
 
 	tests := map[string]struct {
-		workloadsToAdd             []*kueue.Workload
-		inadmissibleWorkloadsToAdd []*workload.Info
-		workloadsToUpdate          []*kueue.Workload
-		workloadsToDelete          []*kueue.Workload
-		queueInadmissibleWorkloads bool
-		wantActiveWorkloads        sets.String
-		wantPending                int32
+		workloadsToAdd                 []*kueue.Workload
+		inadmissibleWorkloadsToRequeue []*workload.Info
+		admissibleWorkloadsToRequeue   []*workload.Info
+		workloadsToUpdate              []*kueue.Workload
+		workloadsToDelete              []*kueue.Workload
+		queueInadmissibleWorkloads     bool
+		wantActiveWorkloads            sets.String
+		wantPending                    int32
 	}{
 		"add, update, delete workload": {
-			workloadsToAdd:             []*kueue.Workload{workloads[0], workloads[1]},
-			inadmissibleWorkloadsToAdd: []*workload.Info{},
-			workloadsToUpdate:          []*kueue.Workload{updatedWorkloads[0]},
-			workloadsToDelete:          []*kueue.Workload{workloads[0]},
-			wantActiveWorkloads:        sets.NewString(workloads[1].Name),
-			wantPending:                1,
+			workloadsToAdd:                 []*kueue.Workload{workloads[0], workloads[1]},
+			inadmissibleWorkloadsToRequeue: []*workload.Info{},
+			workloadsToUpdate:              []*kueue.Workload{updatedWorkloads[0]},
+			workloadsToDelete:              []*kueue.Workload{workloads[0]},
+			wantActiveWorkloads:            sets.NewString(workloads[1].Name),
+			wantPending:                    1,
 		},
 		"re-queue inadmissible workload": {
-			workloadsToAdd:             []*kueue.Workload{workloads[0]},
-			inadmissibleWorkloadsToAdd: []*workload.Info{workload.NewInfo(workloads[1])},
-			workloadsToUpdate:          []*kueue.Workload{},
-			workloadsToDelete:          []*kueue.Workload{},
-			wantActiveWorkloads:        sets.NewString(workloads[0].Name),
-			wantPending:                2,
+			workloadsToAdd:                 []*kueue.Workload{workloads[0]},
+			inadmissibleWorkloadsToRequeue: []*workload.Info{workload.NewInfo(workloads[1])},
+			wantActiveWorkloads:            sets.NewString(workloads[0].Name),
+			wantPending:                    2,
+		},
+		"re-queue admissible workload that was inadmissible": {
+			workloadsToAdd:                 []*kueue.Workload{workloads[0]},
+			inadmissibleWorkloadsToRequeue: []*workload.Info{workload.NewInfo(workloads[1])},
+			admissibleWorkloadsToRequeue:   []*workload.Info{workload.NewInfo(workloads[1])},
+			wantActiveWorkloads:            sets.NewString(workloads[0].Name, workloads[1].Name),
+			wantPending:                    2,
 		},
 		"re-queue inadmissible workload and flush": {
-			workloadsToAdd:             []*kueue.Workload{workloads[0]},
-			inadmissibleWorkloadsToAdd: []*workload.Info{workload.NewInfo(workloads[1])},
-			workloadsToUpdate:          []*kueue.Workload{},
-			workloadsToDelete:          []*kueue.Workload{},
-			queueInadmissibleWorkloads: true,
-			wantActiveWorkloads:        sets.NewString(workloads[0].Name, workloads[1].Name),
-			wantPending:                2,
+			workloadsToAdd:                 []*kueue.Workload{workloads[0]},
+			inadmissibleWorkloadsToRequeue: []*workload.Info{workload.NewInfo(workloads[1])},
+			queueInadmissibleWorkloads:     true,
+			wantActiveWorkloads:            sets.NewString(workloads[0].Name, workloads[1].Name),
+			wantPending:                    2,
 		},
 		"update inadmissible workload": {
-			workloadsToAdd:             []*kueue.Workload{workloads[0]},
-			inadmissibleWorkloadsToAdd: []*workload.Info{workload.NewInfo(workloads[1])},
-			workloadsToUpdate:          []*kueue.Workload{updatedWorkloads[1]},
-			workloadsToDelete:          []*kueue.Workload{},
-			wantActiveWorkloads:        sets.NewString(workloads[0].Name, workloads[1].Name),
-			wantPending:                2,
+			workloadsToAdd:                 []*kueue.Workload{workloads[0]},
+			inadmissibleWorkloadsToRequeue: []*workload.Info{workload.NewInfo(workloads[1])},
+			workloadsToUpdate:              []*kueue.Workload{updatedWorkloads[1]},
+			wantActiveWorkloads:            sets.NewString(workloads[0].Name, workloads[1].Name),
+			wantPending:                    2,
 		},
 		"delete inadmissible workload": {
-			workloadsToAdd:             []*kueue.Workload{workloads[0]},
-			inadmissibleWorkloadsToAdd: []*workload.Info{workload.NewInfo(workloads[1])},
-			workloadsToUpdate:          []*kueue.Workload{},
-			workloadsToDelete:          []*kueue.Workload{workloads[1]},
-			queueInadmissibleWorkloads: true,
-			wantActiveWorkloads:        sets.NewString(workloads[0].Name),
-			wantPending:                1,
+			workloadsToAdd:                 []*kueue.Workload{workloads[0]},
+			inadmissibleWorkloadsToRequeue: []*workload.Info{workload.NewInfo(workloads[1])},
+			workloadsToDelete:              []*kueue.Workload{workloads[1]},
+			queueInadmissibleWorkloads:     true,
+			wantActiveWorkloads:            sets.NewString(workloads[0].Name),
+			wantPending:                    1,
 		},
 	}
 
@@ -105,9 +107,12 @@ func TestClusterQueueBestEffortFIFO(t *testing.T) {
 				cq.PushOrUpdate(w)
 			}
 
-			for _, w := range test.inadmissibleWorkloadsToAdd {
+			for _, w := range test.inadmissibleWorkloadsToRequeue {
 				cq.RequeueIfNotPresent(w, false)
 			}
+			for _, w := range test.admissibleWorkloadsToRequeue {
+				cq.RequeueIfNotPresent(w, true)
+			}
 
 			for _, w := range test.workloadsToUpdate {
 				cq.PushOrUpdate(w)