Repository: kubernetes-sigs/kueue
Pull request #240 (2022-04-30 22:11:44)

Add docs on how to change the configuration
#### What type of PR is this?

/kind documentation

#### What this PR does / why we need it:

Documents how to change the default configuration of a released version

#### Which issue(s) this PR fixes:
Part of https://github.com/kubernetes-sigs/kueue/issues/169

#### Special notes for your reviewer:



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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kueue/pull/240#" title="Author self-approved">ahg-g</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)~~ [ahg-g]

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":[]} -->
ahg-g: /hold
ahg-g: 
cc @cortespao



Patch:
diff --git a/CHANGELOG/CHANGELOG-0.2.md b/CHANGELOG/CHANGELOG-0.2.md
new file mode 100644
index 000000000..efd7f5ca1
--- /dev/null
+++ b/CHANGELOG/CHANGELOG-0.2.md
@@ -0,0 +1,6 @@
+## v0.2.0
+
+Changes since `v0.1.0`:
+
+- Fixed bug in a BestEffortFIFO ClusterQueue where a workload might not be
+  retried after a transient error.
diff --git a/docs/setup/install.md b/docs/setup/install.md
index 7ee0d41f8..3afe4a443 100644
--- a/docs/setup/install.md
+++ b/docs/setup/install.md
@@ -18,6 +18,48 @@ VERSION=v0.1.0
 kubectl apply -f https://github.com/kubernetes-sigs/kueue/releases/download/$VERSION/manifests.yaml
 ```
 
+## Install a custom-configured released version
+
+To install a custom-configured released version of Kueue in your cluster, execute the following steps: 
+
+1. Download the release's `manifests.yaml` file:
+
+```shell
+VERSION=v0.1.0
+wget https://github.com/kubernetes-sigs/kueue/releases/download/$VERSION/manifests.yaml
+```
+2. With an editor of your preference, open `manifests.yaml`.
+3. In the `kueue-manager-config` ConfigMap manifest, edit the 
+`controller_manager_config.yaml` data entry. The entry represents
+the default Kueue Configuration 
+struct ([v1alpha1@v0.1.0](https://pkg.go.dev/sigs.k8s.io/kueue@v0.1.0/apis/config/v1alpha1#Configuration)).
+The contents of the ConfigMap are similar to the following:
+
+```yaml
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: kueue-manager-config
+  namespace: kueue-system
+data:
+  controller_manager_config.yaml: |
+    apiVersion: config.kueue.x-k8s.io/v1alpha1
+    kind: Configuration
+    health:
+      healthProbeBindAddress: :8081
+    metrics:
+      bindAddress: :8080
+    webhook:
+      port: 9443
+    manageJobsWithoutQueueName: true
+```
+
+3. Apply the customized manifests to the cluster:
+
+```shell
+kubectl apply -f manifests.yaml
+```
+
 ## Install the latest development version
 
 To install the latest development version of Kueue in your cluster, run the
diff --git a/go.mod b/go.mod
index a409a487b..4728562dc 100644
--- a/go.mod
+++ b/go.mod
@@ -12,6 +12,7 @@ require (
 	k8s.io/api v0.23.4
 	k8s.io/apimachinery v0.23.4
 	k8s.io/client-go v0.23.4
+	k8s.io/component-base v0.23.3
 	k8s.io/component-helpers v0.23.4
 	k8s.io/klog/v2 v2.40.1
 	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
@@ -27,6 +28,7 @@ require (
 	github.com/Azure/go-autorest/logger v0.2.1 // indirect
 	github.com/Azure/go-autorest/tracing v0.6.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/blang/semver v3.5.1+incompatible // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/evanphx/json-patch v5.6.0+incompatible // indirect
@@ -66,7 +68,6 @@ require (
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 	k8s.io/apiextensions-apiserver v0.23.3 // indirect
-	k8s.io/component-base v0.23.3 // indirect
 	k8s.io/kube-openapi v0.0.0-20220124234850-424119656bbf // indirect
 	sigs.k8s.io/json v0.0.0-20211208200746-9f7c6b3444d2 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect
diff --git a/go.sum b/go.sum
index dce906f49..5c8aa213f 100644
--- a/go.sum
+++ b/go.sum
@@ -95,6 +95,7 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
 github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
 github.com/bketelsen/crypt v0.0.4/go.mod h1:aI6NrJ0pMGgvZKL1iVgXLnfIFJtfV+bKCoqOes/6LfM=
+github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdnnjpJbkM4JQ=
 github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/certifi/gocertifi v0.0.0-20191021191039-0944d244cd40/go.mod h1:sGbDF6GwGcLpkNXPUTkMRoywsNa/ol15pxFe6ERfguA=
diff --git a/main.go b/main.go
index b4a85e543..f47e4bc25 100644
--- a/main.go
+++ b/main.go
@@ -41,6 +41,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/core"
 	"sigs.k8s.io/kueue/pkg/controller/workload/job"
+	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/queue"
 	"sigs.k8s.io/kueue/pkg/scheduler"
 	//+kubebuilder:scaffold:imports
@@ -95,6 +96,7 @@ func main() {
 		}
 		setupLog.Info("Successfully loaded config file", "config", cfgStr)
 	}
+	metrics.Register()
 
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), options)
 	if err != nil {
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 062e5059d..71ac5e97d 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -114,7 +114,7 @@ func (r *WorkloadReconciler) Create(e event.CreateEvent) bool {
 	}
 
 	wlCopy := wl.DeepCopy()
-	handlePodOverhead(wlCopy, r.client)
+	handlePodOverhead(r.log, wlCopy, r.client)
 
 	if wl.Spec.Admission == nil {
 		if !r.queues.AddOrUpdateWorkload(wlCopy) {
@@ -186,7 +186,7 @@ func (r *WorkloadReconciler) Update(e event.UpdateEvent) bool {
 
 	wlCopy := wl.DeepCopy()
 	// We do not handle old workload here as it will be deleted or replaced by new one anyway.
-	handlePodOverhead(wlCopy, r.client)
+	handlePodOverhead(r.log, wlCopy, r.client)
 
 	switch {
 	case status == finished:
@@ -264,14 +264,14 @@ func workloadStatus(w *kueue.Workload) string {
 // As a result, the pod's Overhead is not always correct. E.g. if we set a non-existent runtime class name to
 // `pod.Spec.RuntimeClassName` and we also set the `pod.Spec.Overhead`, in real world, the pod creation will be
 // rejected due to the mismatch with RuntimeClass. However, in the future we assume that they are correct.
-func handlePodOverhead(wl *kueue.Workload, c client.Client) {
+func handlePodOverhead(log logr.Logger, wl *kueue.Workload, c client.Client) {
 	ctx := context.Background()
 
 	for i, pod := range wl.Spec.PodSets {
 		if pod.Spec.RuntimeClassName != nil && len(pod.Spec.Overhead) == 0 {
 			var runtimeClass nodev1.RuntimeClass
 			if err := c.Get(ctx, types.NamespacedName{Name: *pod.Spec.RuntimeClassName}, &runtimeClass); err != nil {
-				klog.Error(err, "Could not get RuntimeClass")
+				log.Error(err, "Could not get RuntimeClass")
 				continue
 			}
 			if runtimeClass.Overhead != nil {
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index 026dbc69f..d830fada8 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -48,6 +48,13 @@ var (
 			Help:      "Latency of an admission attempt, broken down by result.",
 		}, []string{"result"},
 	)
+
+	PendingWorkloads = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: subsystemName,
+			Name:      "pending_workloads",
+			Help:      "Number of pending workloads, per queue and cluster_queue.",
+		}, []string{"cluster_queue", "queue"})
 )
 
 func AdmissionAttempt(result AdmissionResult, duration time.Duration) {
@@ -55,9 +62,10 @@ func AdmissionAttempt(result AdmissionResult, duration time.Duration) {
 	admissionAttemptLatency.WithLabelValues(string(result)).Observe(duration.Seconds())
 }
 
-func init() {
+func Register() {
 	metrics.Registry.MustRegister(
 		admissionAttempts,
 		admissionAttemptLatency,
+		PendingWorkloads,
 	)
 }
diff --git a/pkg/queue/cluster_queue_best_effort_fifo.go b/pkg/queue/cluster_queue_best_effort_fifo.go
index 20cf5bbca..2d380472d 100644
--- a/pkg/queue/cluster_queue_best_effort_fifo.go
+++ b/pkg/queue/cluster_queue_best_effort_fifo.go
@@ -74,11 +74,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
 	}
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index 338be868c..12167d3d8 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -178,6 +178,7 @@ func (m *Manager) AddQueue(ctx context.Context, q *kueue.Queue) error {
 	if cq != nil && cq.AddFromQueue(qImpl) {
 		m.cond.Broadcast()
 	}
+	qImpl.reportPendingWorkloads()
 	return nil
 }
 
@@ -189,6 +190,7 @@ func (m *Manager) UpdateQueue(q *kueue.Queue) error {
 		return errQueueDoesNotExist
 	}
 	if qImpl.ClusterQueue != string(q.Spec.ClusterQueue) {
+		qImpl.resetPendingWorkloads()
 		oldCQ := m.clusterQueues[qImpl.ClusterQueue]
 		if oldCQ != nil {
 			oldCQ.DeleteFromQueue(qImpl)
@@ -199,6 +201,7 @@ func (m *Manager) UpdateQueue(q *kueue.Queue) error {
 		}
 	}
 	qImpl.update(q)
+	qImpl.reportPendingWorkloads()
 	return nil
 }
 
@@ -215,6 +218,7 @@ func (m *Manager) DeleteQueue(q *kueue.Queue) {
 		cq.DeleteFromQueue(qImpl)
 	}
 	delete(m.queues, key)
+	qImpl.resetPendingWorkloads()
 }
 
 func (m *Manager) PendingWorkloads(q *kueue.Queue) (int32, error) {
@@ -272,6 +276,7 @@ func (m *Manager) addOrUpdateWorkload(w *kueue.Workload) bool {
 		return false
 	}
 	q.AddOrUpdate(w)
+	q.reportPendingWorkloads()
 	cq := m.clusterQueues[q.ClusterQueue]
 	if cq == nil {
 		return false
@@ -301,6 +306,7 @@ func (m *Manager) RequeueWorkload(ctx context.Context, info *workload.Info, imme
 	}
 
 	q.AddIfNotPresent(info)
+	q.reportPendingWorkloads()
 	cq := m.clusterQueues[q.ClusterQueue]
 	if cq == nil {
 		return false
@@ -325,6 +331,7 @@ func (m *Manager) deleteWorkloadFromQueueAndClusterQueue(w *kueue.Workload, qKey
 		return
 	}
 	delete(q.items, workload.Key(w))
+	q.reportPendingWorkloads()
 	cq := m.clusterQueues[q.ClusterQueue]
 	if cq != nil {
 		cq.Delete(w)
@@ -450,6 +457,7 @@ func (m *Manager) heads() []workload.Info {
 		workloads = append(workloads, wlCopy)
 		q := m.queues[queueKeyForWorkload(wl.Obj)]
 		delete(q.items, workload.Key(wl.Obj))
+		q.reportPendingWorkloads()
 	}
 	return workloads
 }
diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go
index e7babb668..c9fe89641 100644
--- a/pkg/queue/queue.go
+++ b/pkg/queue/queue.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	"sigs.k8s.io/kueue/pkg/metrics"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
@@ -40,6 +41,7 @@ func queueKeyForWorkload(w *kueue.Workload) string {
 
 // Queue is the internal implementation of kueue.Queue.
 type Queue struct {
+	Key          string
 	ClusterQueue string
 
 	items map[string]*workload.Info
@@ -47,6 +49,7 @@ type Queue struct {
 
 func newQueue(q *kueue.Queue) *Queue {
 	qImpl := &Queue{
+		Key:   Key(q),
 		items: make(map[string]*workload.Info),
 	}
 	qImpl.update(q)
@@ -71,3 +74,11 @@ func (q *Queue) AddIfNotPresent(w *workload.Info) bool {
 	}
 	return false
 }
+
+func (q *Queue) reportPendingWorkloads() {
+	metrics.PendingWorkloads.WithLabelValues(q.ClusterQueue, q.Key).Set(float64(len(q.items)))
+}
+
+func (q *Queue) resetPendingWorkloads() {
+	metrics.PendingWorkloads.DeleteLabelValues(q.ClusterQueue, q.Key)
+}


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 9b2574c6b..4ef842fda 100644
--- a/pkg/queue/cluster_queue_best_effort_fifo_test.go
+++ b/pkg/queue/cluster_queue_best_effort_fifo_test.go
@@ -41,55 +41,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,
 		},
 	}
 
@@ -104,9 +106,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)
diff --git a/test/integration/controller/core/queue_controller_test.go b/test/integration/controller/core/queue_controller_test.go
index 75ebbe75e..5d54363cd 100644
--- a/test/integration/controller/core/queue_controller_test.go
+++ b/test/integration/controller/core/queue_controller_test.go
@@ -82,6 +82,7 @@ var _ = ginkgo.Describe("Queue controller", func() {
 			gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(queue), &updatedQueue)).To(gomega.Succeed())
 			return updatedQueue.Status
 		}, framework.Timeout, framework.Interval).Should(testing.Equal(kueue.QueueStatus{PendingWorkloads: 3}))
+		framework.ExpectPendingWorkloadsMetric(queue, 3)
 
 		ginkgo.By("Admitting workloads")
 		for _, w := range workloads {
@@ -98,6 +99,7 @@ var _ = ginkgo.Describe("Queue controller", func() {
 			gomega.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(queue), &updatedQueue)).To(gomega.Succeed())
 			return updatedQueue.Status
 		}, framework.Timeout, framework.Interval).Should(testing.Equal(kueue.QueueStatus{PendingWorkloads: 0}))
+		framework.ExpectPendingWorkloadsMetric(queue, 0)
 
 		ginkgo.By("Finishing workloads")
 		for _, w := range workloads {
diff --git a/test/integration/framework/framework.go b/test/integration/framework/framework.go
index f772f1abb..9ae9a5009 100644
--- a/test/integration/framework/framework.go
+++ b/test/integration/framework/framework.go
@@ -33,6 +33,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/component-base/metrics/testutil"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -40,6 +41,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
+	"sigs.k8s.io/kueue/pkg/metrics"
+	"sigs.k8s.io/kueue/pkg/queue"
 	"sigs.k8s.io/kueue/pkg/workload"
 	// +kubebuilder:scaffold:imports
 )
@@ -215,6 +218,15 @@ func ExpectWorkloadsToBePending(ctx context.Context, k8sClient client.Client, wl
 	}, Timeout, Interval).Should(gomega.Equal(len(wls)), "Not enough workloads are pending")
 }
 
+func ExpectPendingWorkloadsMetric(q *kueue.Queue, v int) {
+	metric := metrics.PendingWorkloads.WithLabelValues(string(q.Spec.ClusterQueue), queue.Key(q))
+	gomega.EventuallyWithOffset(1, func() int {
+		v, err := testutil.GetGaugeMetricValue(metric)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+		return int(v)
+	}, Timeout, Interval).Should(gomega.Equal(v))
+}
+
 func UpdateWorkloadStatus(ctx context.Context, k8sClient client.Client, wl *kueue.Workload, update func(*kueue.Workload)) {
 	gomega.EventuallyWithOffset(1, func() error {
 		var updatedWl kueue.Workload
diff --git a/test/integration/scheduler/scheduler_test.go b/test/integration/scheduler/scheduler_test.go
index a67c2461f..bf65f4390 100644
--- a/test/integration/scheduler/scheduler_test.go
+++ b/test/integration/scheduler/scheduler_test.go
@@ -124,6 +124,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdProdJob1.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdProdJob1.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 
 		ginkgo.By("checking a second no-fit prod job does not start")
 		prodJob2 := testing.MakeJob("prod-job2", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "5").Obj()
@@ -134,6 +135,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, lookupKey2, createdProdJob2)).Should(gomega.Succeed())
 			return createdProdJob2.Spec.Suspend
 		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 1)
 
 		ginkgo.By("checking a dev job starts")
 		devJob := testing.MakeJob("dev-job", ns.Name).Queue(devQueue.Name).Request(corev1.ResourceCPU, "5").Obj()
@@ -145,6 +147,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdDevJob.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdDevJob.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(spotUntaintedFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(devQueue, 0)
 
 		ginkgo.By("checking the second prod job starts when the first finishes")
 		createdProdJob1.Status.Conditions = append(createdProdJob1.Status.Conditions,
@@ -160,6 +163,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdProdJob2.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdProdJob2.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 	})
 
 	ginkgo.It("Should schedule jobs on tolerated flavors", func() {
@@ -173,6 +177,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob1.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob1.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 
 		// TODO(#8): uncomment the following once we have proper re-queueing.
 		// ginkgo.By("checking a second job without toleration doesn't start")
@@ -195,6 +200,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob3.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob3.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(spotTaintedFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 	})
 
 	ginkgo.It("Should schedule jobs using borrowed ClusterQueue", func() {
@@ -208,6 +214,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob)).Should(gomega.Succeed())
 			return createdJob.Spec.Suspend
 		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 1)
 
 		ginkgo.By("checking the job starts when a fallback ClusterQueue gets added")
 		fallbackClusterQueue := testing.MakeClusterQueue("fallback-cq").
@@ -226,6 +233,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 	})
 
 	ginkgo.It("Should schedule jobs with affinity to specific flavor", func() {
@@ -239,6 +247,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob1.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob1.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(spotUntaintedFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(devQueue, 0)
 
 		ginkgo.By("checking a second job with affinity to on-demand")
 		job2 := testing.MakeJob("affinity-job", ns.Name).Queue(devQueue.Name).
@@ -253,6 +262,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		}, framework.Timeout, framework.Interval).Should(gomega.BeTrue())
 		gomega.Expect(len(createdJob2.Spec.Template.Spec.NodeSelector)).Should(gomega.Equal(2))
 		gomega.Expect(createdJob2.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(devQueue, 0)
 	})
 
 	ginkgo.It("Should schedule jobs from the selected namespaces", func() {
@@ -285,6 +295,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: job.Name, Namespace: job.Namespace}, createdJob)).Should(gomega.Succeed())
 			return createdJob.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(true)), "Job should be suspended")
+		framework.ExpectPendingWorkloadsMetric(queue, 1)
 
 		ginkgo.By("checking the job starts after updating namespace labels to match QC selector")
 		ns.Labels = map[string]string{"dep": "eng"}
@@ -293,6 +304,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: job.Name, Namespace: job.Namespace}, createdJob)).Should(gomega.Succeed())
 			return createdJob.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)), "Job should be unsuspended")
+		framework.ExpectPendingWorkloadsMetric(queue, 0)
 	})
 
 	ginkgo.It("Should schedule jobs according to their priorities", func() {
@@ -344,6 +356,8 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, types.NamespacedName{Name: jobHighPriority.Name, Namespace: ns.Name}, createdJob2)).Should(gomega.Succeed())
 			return createdJob2.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
+
+		framework.ExpectPendingWorkloadsMetric(queue, 1)
 	})
 
 	ginkgo.It("Should re-enqueue by the delete event of workload belonging to the same ClusterQueue", func() {
@@ -357,6 +371,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob1.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob1.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 
 		job2 := testing.MakeJob("on-demand-job2", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "4").Obj()
 		gomega.Expect(k8sClient.Create(ctx, job2)).Should(gomega.Succeed())
@@ -366,6 +381,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob2)).Should(gomega.Succeed())
 			return createdJob2.Spec.Suspend
 		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 1)
 
 		job3 := testing.MakeJob("on-demand-job3", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "1").Obj()
 		gomega.Expect(k8sClient.Create(ctx, job3)).Should(gomega.Succeed())
@@ -376,9 +392,10 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob3.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob3.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 1)
 
 		ginkgo.By("deleting job1")
-		gomega.Expect(k8sClient.Delete(ctx, job1)).Should(gomega.Succeed())
+		gomega.Expect(k8sClient.Delete(ctx, job1, client.PropagationPolicy(metav1.DeletePropagationBackground))).Should(gomega.Succeed())
 		// Simulated deletion of workload by ownerReference of job1.
 		gomega.Expect(k8sClient.Delete(ctx,
 			&kueue.Workload{ObjectMeta: metav1.ObjectMeta{
@@ -390,8 +407,8 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob2)).Should(gomega.Succeed())
 			return createdJob2.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
-
 		gomega.Expect(createdJob2.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 	})
 
 	ginkgo.It("Should re-enqueue by the delete event of workload belonging to the same Cohort", func() {
@@ -421,6 +438,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob1.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob1.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(queue, 0)
 
 		job2 := testing.MakeJob("on-demand-job2",
 			ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "8").Obj()
@@ -431,6 +449,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob2)).Should(gomega.Succeed())
 			return createdJob2.Spec.Suspend
 		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 1)
 
 		job3 := testing.MakeJob("on-demand-job3", ns.Name).Queue(queue.Name).Request(corev1.ResourceCPU, "2").Obj()
 		gomega.Expect(k8sClient.Create(ctx, job3)).Should(gomega.Succeed())
@@ -441,9 +460,10 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob3.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob3.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(queue, 0)
 
 		ginkgo.By("deleting job1")
-		gomega.Expect(k8sClient.Delete(ctx, job1, client.GracePeriodSeconds(-1))).Should(gomega.Succeed())
+		gomega.Expect(k8sClient.Delete(ctx, job1, client.PropagationPolicy(metav1.DeletePropagationBackground))).Should(gomega.Succeed())
 		// Simulated deletion of workload by ownerReference of job1.
 		gomega.Expect(k8sClient.Delete(ctx,
 			&kueue.Workload{ObjectMeta: metav1.ObjectMeta{
@@ -457,9 +477,10 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob2.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob2.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 	})
 
-	ginkgo.It("Should re-enqueue by the updated event of ClusterQueue", func() {
+	ginkgo.It("Should re-enqueue by the update event of ClusterQueue", func() {
 		devBEClusterQ := testing.MakeClusterQueue("dev-be-cq").
 			Cohort("be").
 			Resource(testing.MakeResource(corev1.ResourceCPU).
@@ -485,6 +506,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob1.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob1.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 
 		job2 := testing.MakeJob("on-demand-job2", ns.Name).Queue(devBEQueue.Name).Request(corev1.ResourceCPU, "6").Obj()
 		gomega.Expect(k8sClient.Create(ctx, job2)).Should(gomega.Succeed())
@@ -494,6 +516,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob2)).Should(gomega.Succeed())
 			return createdJob2.Spec.Suspend
 		}, framework.ConsistentDuration, framework.Interval).Should(gomega.Equal(pointer.Bool(true)))
+		framework.ExpectPendingWorkloadsMetric(devBEQueue, 1)
 
 		job3 := testing.MakeJob("on-demand-job3", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "2").Obj()
 		gomega.Expect(k8sClient.Create(ctx, job3)).Should(gomega.Succeed())
@@ -504,6 +527,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob3.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob3.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 
 		ginkgo.By("updating ClusterQueue")
 		devCq := &kueue.ClusterQueue{}
@@ -519,6 +543,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 			return createdJob2.Spec.Suspend
 		}, framework.Timeout, framework.Interval).Should(gomega.Equal(pointer.Bool(false)))
 		gomega.Expect(createdJob2.Spec.Template.Spec.NodeSelector[instanceKey]).Should(gomega.Equal(onDemandFlavor.Name))
+		framework.ExpectPendingWorkloadsMetric(devBEQueue, 0)
 	})
 
 	ginkgo.It("Should admit two small workloads after a big one finishes", func() {
@@ -527,6 +552,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		gomega.Expect(k8sClient.Create(ctx, bigWl)).Should(gomega.Succeed())
 
 		framework.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, prodClusterQ.Name, bigWl)
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 
 		smallWl1 := testing.MakeWorkload("small-wl-1", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "2.5").Obj()
 		smallWl2 := testing.MakeWorkload("small-wl-2", ns.Name).Queue(prodQueue.Name).Request(corev1.ResourceCPU, "2.5").Obj()
@@ -535,6 +561,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		gomega.Expect(k8sClient.Create(ctx, smallWl2)).Should(gomega.Succeed())
 
 		framework.ExpectWorkloadsToBePending(ctx, k8sClient, smallWl1, smallWl2)
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 2)
 
 		ginkgo.By("Marking the big workload as finished")
 		framework.UpdateWorkloadStatus(ctx, k8sClient, bigWl, func(wl *kueue.Workload) {
@@ -545,6 +572,7 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		})
 
 		framework.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, prodClusterQ.Name, smallWl1, smallWl2)
+		framework.ExpectPendingWorkloadsMetric(prodQueue, 0)
 	})
 
 	ginkgo.It("Should schedule workloads borrowing quota from ClusterQueues in the same Cohort", func() {
@@ -577,6 +605,8 @@ var _ = ginkgo.Describe("Scheduler", func() {
 		gomega.Expect(k8sClient.Create(ctx, wl1)).Should(gomega.Succeed())
 		gomega.Expect(k8sClient.Create(ctx, wl2)).Should(gomega.Succeed())
 		framework.ExpectWorkloadsToBePending(ctx, k8sClient, wl1, wl2)
+		framework.ExpectPendingWorkloadsMetric(prodBEQueue, 1)
+		framework.ExpectPendingWorkloadsMetric(devBEQueue, 1)
 
 		// Make sure workloads are in the same scheduling cycle.
 		testBEClusterQ := testing.MakeClusterQueue("test-be-cq").
@@ -595,6 +625,8 @@ var _ = ginkgo.Describe("Scheduler", func() {
 
 		framework.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, prodBEClusterQ.Name, wl1)
 		framework.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, devBEClusterQ.Name, wl2)
+		framework.ExpectPendingWorkloadsMetric(prodBEQueue, 0)
+		framework.ExpectPendingWorkloadsMetric(devBEQueue, 0)
 	})
 
 	ginkgo.It("Should schedule workloads by their priority strictly in StrictFIFO", func() {