Repository: kubernetes-sigs/kueue (go)

Issue:
Flaky integration test for the scheduler?
**What happened**:

The integration tests failed with a extraneous message in the logs:

```
ERROR	scheduler	Updating QueuedWorkload status	{"error": "resource name may not be empty"}
    sigs.k8s.io/kueue/pkg/scheduler.(*Scheduler).schedule
    	/home/prow/go/src/sigs.k8s.io/kueue/pkg/scheduler/scheduler.go:128
    k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext.func1
    	/home/prow/go/pkg/mod/k8s.io/apimachinery@v0.23.4/pkg/util/wait/wait.go:188
    k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1
    	/home/prow/go/pkg/mod/k8s.io/apimachinery@v0.23.4/pkg/util/wait/wait.go:155
    k8s.io/apimachinery/pkg/util/wait.BackoffUntil
    	/home/prow/go/pkg/mod/k8s.io/apimachinery@v0.23.4/pkg/util/wait/wait.go:156
    k8s.io/apimachinery/pkg/util/wait.JitterUntil
    	/home/prow/go/pkg/mod/k8s.io/apimachinery@v0.23.4/pkg/util/wait/wait.go:133
    k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext
    	/home/prow/go/pkg/mod/k8s.io/apimachinery@v0.23.4/pkg/util/wait/wait.go:188
    k8s.io/apimachinery/pkg/util/wait.UntilWithContext
    	/home/prow/go/pkg/mod/k8s.io/apimachinery@v0.23.4/pkg/util/wait/wait.go:99
    sigs.k8s.io/kueue/pkg/scheduler.(*Scheduler).Start
    	/home/prow/go/src/sigs.k8s.io/kueue/pkg/scheduler/scheduler.go:66
    sigs.k8s.io/kueue/test/integration/scheduler.managerAndSchedulerSetup.func1
    	/home/prow/go/src/sigs.k8s.io/kueue/test/integration/scheduler/suite_test.go:87
```

Full error log: https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/157/pull-kueue-test-integration-main/1508537482113716224

So somehow we got a QW without a name? The only way I could see this possibly happen is that we used a QW object without cloning it from the informer cache.

**What you expected to happen**:

No error

**How to reproduce it (as minimally and precisely as possible)**:

No idea. I will try to run the test multiple times.

**Anything else we need to know?**:

**Environment**:
- Kubernetes version (use `kubectl version`): main
- Kueue version (use `git describe --tags --dirty --always`):
- Cloud provider or hardware configuration:
- OS (e.g: `cat /etc/os-release`):
- Kernel (e.g. `uname -a`):
- Install tools:
- Others:

Pull request description:
Fix scheduler getting stuck when there are admission failures
#### What type of PR is this?

/kind bug
/kind failing-test
/kind flake

#### What this PR does / why we need it:

When requeueing, we need to notify the condition.

Also fixed queue manager to keep the most up-to-date object in a Queue

#### Which issue(s) this PR fixes:

Fixes #158 

#### Special notes for your reviewer:

Used #161 to see that the scheduler issued a requeue, but there was no following scheduling attempts, suggesting that the scheduler was hanging waiting for kueues.

Patch:
diff --git a/pkg/controller/core/clusterqueue_controller.go b/pkg/controller/core/clusterqueue_controller.go
index 38040c99d..cc6492619 100644
--- a/pkg/controller/core/clusterqueue_controller.go
+++ b/pkg/controller/core/clusterqueue_controller.go
@@ -144,7 +144,7 @@ func (r *ClusterQueueReconciler) Update(e event.UpdateEvent) bool {
 }
 
 func (r *ClusterQueueReconciler) Generic(e event.GenericEvent) bool {
-	r.log.V(3).Info("Ignore generic event", "obj", klog.KObj(e.Object), "kind", e.Object.GetObjectKind().GroupVersionKind())
+	r.log.V(3).Info("Got QueuedWorkload event", "queuedWorkload", klog.KObj(e.Object))
 	return true
 }
 
diff --git a/pkg/controller/core/queue_controller.go b/pkg/controller/core/queue_controller.go
index a11d84941..5c825ef95 100644
--- a/pkg/controller/core/queue_controller.go
+++ b/pkg/controller/core/queue_controller.go
@@ -129,7 +129,7 @@ func (r *QueueReconciler) Update(e event.UpdateEvent) bool {
 }
 
 func (r *QueueReconciler) Generic(e event.GenericEvent) bool {
-	r.log.V(3).Info("Ignore generic event", "obj", klog.KObj(e.Object), "kind", e.Object.GetObjectKind().GroupVersionKind())
+	r.log.V(3).Info("Got QueuedWorkload event", "queuedWorkload", klog.KObj(e.Object))
 	return true
 }
 
diff --git a/pkg/queue/cluster_queue_impl.go b/pkg/queue/cluster_queue_impl.go
index 17e9c61ee..b4b044a68 100644
--- a/pkg/queue/cluster_queue_impl.go
+++ b/pkg/queue/cluster_queue_impl.go
@@ -106,3 +106,11 @@ func (c *ClusterQueueImpl) Dump() (sets.String, bool) {
 	}
 	return elements, true
 }
+
+func (c *ClusterQueueImpl) Info(key string) *workload.Info {
+	item := c.heap.items[key]
+	if item != nil {
+		return &item.obj
+	}
+	return nil
+}
diff --git a/pkg/queue/cluster_queue_interface.go b/pkg/queue/cluster_queue_interface.go
index a86ee496d..81d088b2b 100644
--- a/pkg/queue/cluster_queue_interface.go
+++ b/pkg/queue/cluster_queue_interface.go
@@ -56,6 +56,9 @@ type ClusterQueue interface {
 	// this ClusterQueue. It returns false if the queue is empty.
 	// Otherwise returns true.
 	Dump() (sets.String, bool)
+	// Info returns workload.Info for the workload key.
+	// Users of this method should not modify the returned object.
+	Info(string) *workload.Info
 }
 
 var registry = map[kueue.QueueingStrategy]func(cq *kueue.ClusterQueue) (ClusterQueue, error){
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index 2f154c9a5..0089d6063 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -24,6 +24,7 @@ import (
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kueue "sigs.k8s.io/kueue/api/v1alpha1"
@@ -271,14 +272,16 @@ func (m *Manager) RequeueWorkload(ctx context.Context, info *workload.Info) bool
 		return false
 	}
 
-	key := workload.Key(info.Obj)
-	q.items[key] = info
-
+	q.AddIfNotPresent(info)
 	cq := m.clusterQueues[q.ClusterQueue]
 	if cq == nil {
 		return false
 	}
-	return cq.PushIfNotPresent(info)
+	added := cq.PushIfNotPresent(info)
+	if added {
+		m.cond.Broadcast()
+	}
+	return added
 }
 
 func (m *Manager) DeleteWorkload(w *kueue.QueuedWorkload) {
@@ -306,19 +309,8 @@ func (m *Manager) UpdateWorkload(oldW, w *kueue.QueuedWorkload) bool {
 	defer m.Unlock()
 	if oldW.Spec.QueueName != w.Spec.QueueName {
 		m.deleteWorkloadFromQueueAndClusterQueue(w, queueKeyForWorkload(oldW))
-		return m.addOrUpdateWorkload(w)
 	}
-
-	q := m.queues[queueKeyForWorkload(w)]
-	if q == nil {
-		return false
-	}
-	cq := m.clusterQueues[q.ClusterQueue]
-	if cq != nil {
-		cq.PushOrUpdate(w)
-		return true
-	}
-	return false
+	return m.addOrUpdateWorkload(w)
 }
 
 // CleanUpOnContext tracks the context. When closed, it wakes routines waiting
@@ -334,8 +326,10 @@ func (m *Manager) CleanUpOnContext(ctx context.Context) {
 func (m *Manager) Heads(ctx context.Context) []workload.Info {
 	m.Lock()
 	defer m.Unlock()
+	log := ctrl.LoggerFrom(ctx)
 	for {
 		workloads := m.heads()
+		log.V(3).Info("Obtained ClusterQueue heads", "count", len(workloads))
 		if len(workloads) != 0 {
 			return workloads
 		}
diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go
index 9bca1f54f..3b95c1b5e 100644
--- a/pkg/queue/queue.go
+++ b/pkg/queue/queue.go
@@ -54,14 +54,19 @@ func (q *Queue) update(apiQueue *kueue.Queue) {
 
 func (q *Queue) AddOrUpdate(w *kueue.QueuedWorkload) {
 	key := workload.Key(w)
-	info := q.items[key]
-	if info != nil {
-		info.Obj = w
-		return
-	}
 	q.items[key] = workload.NewInfo(w)
 }
 
+func (q *Queue) AddIfNotPresent(w *workload.Info) bool {
+	key := workload.Key(w.Obj)
+	_, ok := q.items[key]
+	if !ok {
+		q.items[key] = w
+		return true
+	}
+	return false
+}
+
 // heap.Interface implementation inspired by
 // https://github.com/kubernetes/kubernetes/blob/master/pkg/scheduler/internal/heap/heap.go
 
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 7b0b1ed2d..763cc5337 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -165,6 +165,7 @@ func (s *Scheduler) nominate(ctx context.Context, workloads []workload.Info, sna
 			log.V(2).Info("Workload didn't fit in remaining clusterQueue even when borrowing")
 			continue
 		}
+		log.V(3).Info("Nominated workload for admission")
 		entries = append(entries, e)
 	}
 	return entries
@@ -397,8 +398,8 @@ func (e entryOrdering) Less(i, j int) bool {
 }
 
 func (s *Scheduler) requeueAndUpdate(log logr.Logger, ctx context.Context, w *workload.Info, message string) {
-	existed := s.queues.RequeueWorkload(ctx, w)
-	log.V(2).Info("Workload re-queued", "queuedWorkload", klog.KObj(w.Obj), "queue", klog.KRef(w.Obj.Namespace, w.Obj.Spec.QueueName), "existed", existed)
+	added := s.queues.RequeueWorkload(ctx, w)
+	log.V(2).Info("Workload re-queued", "queuedWorkload", klog.KObj(w.Obj), "queue", klog.KRef(w.Obj.Namespace, w.Obj.Spec.QueueName), "added", added)
 	err := workload.UpdateWorkloadStatus(ctx, s.client, w.Obj, kueue.QueuedWorkloadAdmitted, corev1.ConditionFalse, "Pending", message)
 	if err != nil {
 		log.Error(err, "Could not update QueuedWorkload status")


Test patch:
diff --git a/pkg/queue/manager_test.go b/pkg/queue/manager_test.go
index b35d3a07b..177c35e3e 100644
--- a/pkg/queue/manager_test.go
+++ b/pkg/queue/manager_test.go
@@ -559,6 +559,25 @@ func TestUpdateWorkload(t *testing.T) {
 			if updated := manager.UpdateWorkload(tc.workloads[0], wl); updated != tc.wantUpdated {
 				t.Errorf("UpdatedWorkload returned %t, want %t", updated, tc.wantUpdated)
 			}
+			q := manager.queues[queueKeyForWorkload(wl)]
+			if q != nil {
+				key := workload.Key(wl)
+				item := q.items[key]
+				if item == nil {
+					t.Errorf("Object not stored in queue")
+				} else if diff := cmp.Diff(wl, item.Obj); diff != "" {
+					t.Errorf("Object stored in queue differs (-want,+got):\n%s", diff)
+				}
+				cq := manager.clusterQueues[q.ClusterQueue]
+				if cq != nil {
+					item := cq.Info(key)
+					if item == nil {
+						t.Errorf("Object not stored in clusterQueue")
+					} else if diff := cmp.Diff(wl, item.Obj); diff != "" {
+						t.Errorf("Object stored in clusterQueue differs (-want,+got):\n%s", diff)
+					}
+				}
+			}
 			queueOrder := make(map[string][]string)
 			for name, cq := range manager.clusterQueues {
 				queueOrder[name] = popNamesFromCQ(cq)
@@ -741,6 +760,37 @@ func TestHeadsAsync(t *testing.T) {
 				}()
 			},
 		},
+		"UpdateWorkload": {
+			op: func(ctx context.Context, mgr *Manager) {
+				if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+					t.Errorf("Failed adding clusterQueue: %v", err)
+				}
+				if err := mgr.AddQueue(ctx, &q); err != nil {
+					t.Errorf("Failed adding queue: %s", err)
+				}
+				go func() {
+					wlCopy := wl.DeepCopy()
+					wlCopy.ResourceVersion = "old"
+					mgr.UpdateWorkload(wlCopy, &wl)
+				}()
+			},
+		},
+		"RequeueWorkload": {
+			initialObjs: []client.Object{&wl},
+			op: func(ctx context.Context, mgr *Manager) {
+				if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+					t.Errorf("Failed adding clusterQueue: %v", err)
+				}
+				if err := mgr.AddQueue(ctx, &q); err != nil {
+					t.Errorf("Failed adding queue: %s", err)
+				}
+				// Remove the initial workload from the manager.
+				mgr.Heads(ctx)
+				go func() {
+					mgr.RequeueWorkload(ctx, workload.NewInfo(&wl))
+				}()
+			},
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {