Repository: kubernetes-sigs/kueue
Pull request #75 (2022-02-25 22:01:53)

Add status report for kueue.Queue
Signed-off-by: Carlos Eduardo Arango Gutierrez <carangog@redhat.com>

#### What type of PR is this?
/kind feature
/kind api-change
<!--
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:

#### 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`_*
-->
initial implementation for  #5 

#### 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/75#" title="Author self-approved">ArangoGutierrez</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"]} -->
ArangoGutierrez: /assign @alculquicondor
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/75#" title="Author self-approved">ArangoGutierrez</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **alculquicondor** after the PR has been reviewed.

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"]} -->
k8s-ci-robot: @ArangoGutierrez: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kueue-verify-main | c52adfc827a2967be69ad4c592ec4266ac502e1c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-verify-main/1498662773008109568) | true | `/test pull-kueue-verify-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kueue&pr=75). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3AArangoGutierrez). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kueue/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @ArangoGutierrez: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kueue-verify-main | 96f0aa2ded81555b50338c0c6393c109167614a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-verify-main/1498776204965056512) | true | `/test pull-kueue-verify-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kueue&pr=75). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3AArangoGutierrez). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kueue/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @ArangoGutierrez: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kueue-verify-main | 96f0aa2ded81555b50338c0c6393c109167614a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-verify-main/1498776204965056512) | true | `/test pull-kueue-verify-main`
pull-kueue-test-integration-main | 96f0aa2ded81555b50338c0c6393c109167614a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-test-integration-main/1498776204927307776) | true | `/test pull-kueue-test-integration-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kueue&pr=75). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3AArangoGutierrez). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kueue/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @ArangoGutierrez: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kueue-verify-main | 96f0aa2ded81555b50338c0c6393c109167614a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-verify-main/1498776204965056512) | true | `/test pull-kueue-verify-main`
pull-kueue-test-integration-main | 96f0aa2ded81555b50338c0c6393c109167614a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-test-integration-main/1498776204927307776) | true | `/test pull-kueue-test-integration-main`
pull-kueue-test-unit-main | 919ac31c168d3b4aa09e2ac7744d0e87d0fbcde1 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-test-unit-main/1498778061515001856) | true | `/test pull-kueue-test-unit-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kueue&pr=75). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3AArangoGutierrez). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kueue/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @ArangoGutierrez: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kueue-test-unit-main | b704e3a2881f06e1f7fc3bcbbf0d12828be30de6 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-test-unit-main/1498795386771345408) | true | `/test pull-kueue-test-unit-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kueue&pr=75). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3AArangoGutierrez). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kueue/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @ArangoGutierrez: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kueue-test-unit-main | b704e3a2881f06e1f7fc3bcbbf0d12828be30de6 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-test-unit-main/1498795386771345408) | true | `/test pull-kueue-test-unit-main`
pull-kueue-verify-main | b704e3a2881f06e1f7fc3bcbbf0d12828be30de6 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/75/pull-kueue-verify-main/1498795386939117568) | true | `/test pull-kueue-verify-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kueue&pr=75). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3AArangoGutierrez). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kueue/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/api/v1alpha1/queue_types.go b/api/v1alpha1/queue_types.go
index 1972cce1e..bcb007ea3 100644
--- a/api/v1alpha1/queue_types.go
+++ b/api/v1alpha1/queue_types.go
@@ -32,6 +32,9 @@ type CapacityReference string
 
 // QueueStatus defines the observed state of Queue
 type QueueStatus struct {
+	// PendingWorkloads is the number of workloads currently assigned to this
+	// queue not yet assigned to a Capacity.
+	PendingWorkloads int32 `json:"pendingWorkloads"`
 }
 
 //+kubebuilder:object:root=true
diff --git a/config/crd/bases/kueue.x-k8s.io_queues.yaml b/config/crd/bases/kueue.x-k8s.io_queues.yaml
index 0543f6878..274103f86 100644
--- a/config/crd/bases/kueue.x-k8s.io_queues.yaml
+++ b/config/crd/bases/kueue.x-k8s.io_queues.yaml
@@ -42,6 +42,14 @@ spec:
             type: object
           status:
             description: QueueStatus defines the observed state of Queue
+            properties:
+              pendingWorkloads:
+                description: PendingWorkloads is the number of workloads currently
+                  assigned to this queue not yet assigned to a Capacity.
+                format: int32
+                type: integer
+            required:
+            - pendingWorkloads
             type: object
         type: object
     served: true
diff --git a/main.go b/main.go
index 34a9043fe..1fcc632b4 100644
--- a/main.go
+++ b/main.go
@@ -91,7 +91,7 @@ func main() {
 
 	queues := queue.NewManager(mgr.GetClient())
 	cache := capacity.NewCache(mgr.GetClient())
-	if err = core.NewQueueReconciler(queues).SetupWithManager(mgr); err != nil {
+	if err = core.NewQueueReconciler(mgr.GetClient(), queues).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Queue")
 		os.Exit(1)
 	}
diff --git a/pkg/controller/core/queue_controller.go b/pkg/controller/core/queue_controller.go
index d3221a09d..4dc591d57 100644
--- a/pkg/controller/core/queue_controller.go
+++ b/pkg/controller/core/queue_controller.go
@@ -20,9 +20,15 @@ import (
 	"context"
 
 	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	kueue "sigs.k8s.io/kueue/api/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/queue"
@@ -30,14 +36,16 @@ import (
 
 // QueueReconciler reconciles a Queue object
 type QueueReconciler struct {
+	client client.Client
 	log    logr.Logger
 	queues *queue.Manager
 }
 
-func NewQueueReconciler(queues *queue.Manager) *QueueReconciler {
+func NewQueueReconciler(client client.Client, queues *queue.Manager) *QueueReconciler {
 	return &QueueReconciler{
 		log:    ctrl.Log.WithName("queue-reconciler"),
 		queues: queues,
+		client: client,
 	}
 }
 
@@ -47,47 +55,117 @@ func NewQueueReconciler(queues *queue.Manager) *QueueReconciler {
 //+kubebuilder:rbac:groups=kueue.x-k8s.io,resources=queues/finalizers,verbs=update
 
 func (r *QueueReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	// No-op. All work is done in the event handler.
+	var queueObj kueue.Queue
+	if err := r.client.Get(ctx, req.NamespacedName, &queueObj); err != nil {
+		// we'll ignore not-found errors, since there is nothing to do.
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+	log := ctrl.LoggerFrom(ctx).WithValues("queue", klog.KObj(&queueObj))
+	ctx = ctrl.LoggerInto(ctx, log)
+	log.V(2).Info("Reconciling Queue")
+
+	pendingJobs, err := r.queues.Status(&queueObj)
+	if err != nil {
+		r.log.Error(err, "Failed to retrieve queue status")
+		return ctrl.Result{}, err
+	}
+
+	oldStatus := queueObj.Status
+	queueObj.Status.PendingWorkloads = int32(pendingJobs)
+	if !equality.Semantic.DeepEqual(oldStatus, queueObj.Status) {
+		err := r.client.Status().Update(ctx, &queueObj)
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
 	return ctrl.Result{}, nil
 }
 
 func (r *QueueReconciler) Create(e event.CreateEvent) bool {
-	q := e.Object.(*kueue.Queue)
+	q, match := e.Object.(*kueue.Queue)
+	if !match {
+		// No need to interact with the queue manager for other objects.
+		return true
+	}
 	log := r.log.WithValues("queue", klog.KObj(q))
 	log.V(2).Info("Queue create event")
 	ctx := logr.NewContext(context.Background(), log)
 	if err := r.queues.AddQueue(ctx, q); err != nil {
 		log.Error(err, "Failed to add queue to system")
 	}
-	return false
+	return true
 }
 
 func (r *QueueReconciler) Delete(e event.DeleteEvent) bool {
-	q := e.Object.(*kueue.Queue)
+	q, match := e.Object.(*kueue.Queue)
+	if !match {
+		// No need to interact with the queue manager for other objects.
+		return true
+	}
 	r.log.V(2).Info("Queue delete event", "queue", klog.KObj(q))
 	r.queues.DeleteQueue(q)
-	return false
+	return true
 }
 
 func (r *QueueReconciler) Update(e event.UpdateEvent) bool {
-	q := e.ObjectNew.(*kueue.Queue)
+	q, match := e.ObjectNew.(*kueue.Queue)
+	if !match {
+		// No need to interact with the queue manager for other objects.
+		return true
+	}
 	log := r.log.WithValues("queue", klog.KObj(q))
 	log.V(2).Info("Queue update event")
 	if err := r.queues.UpdateQueue(q); err != nil {
 		log.Error(err, "Failed to update queue in system")
 	}
-	return false
+	return true
 }
 
 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())
-	return false
+	return true
+}
+
+type queuedWorkloadHandler struct{}
+
+func (h *queuedWorkloadHandler) Create(e event.CreateEvent, q workqueue.RateLimitingInterface) {
+	qw := e.Object.(*kueue.QueuedWorkload)
+	if qw.Spec.AssignedCapacity != "" {
+		q.Add(requestForQueueStatus(qw))
+	}
+}
+
+func (h *queuedWorkloadHandler) Update(e event.UpdateEvent, q workqueue.RateLimitingInterface) {
+	oldQW := e.ObjectOld.(*kueue.QueuedWorkload)
+	newQW := e.ObjectOld.(*kueue.QueuedWorkload)
+	q.Add(requestForQueueStatus(newQW))
+	if newQW.Spec.QueueName != oldQW.Spec.QueueName {
+		q.Add(requestForQueueStatus(oldQW))
+	}
+}
+
+func (h *queuedWorkloadHandler) Delete(e event.DeleteEvent, q workqueue.RateLimitingInterface) {
+	qw := e.Object.(*kueue.QueuedWorkload)
+	if qw.Spec.AssignedCapacity == "" {
+		q.Add(requestForQueueStatus(qw))
+	}
+}
+
+func (h *queuedWorkloadHandler) Generic(e event.GenericEvent, q workqueue.RateLimitingInterface) {
+}
+
+func requestForQueueStatus(w *kueue.QueuedWorkload) reconcile.Request {
+	return reconcile.Request{
+		NamespacedName: types.NamespacedName{
+			Name:      w.Spec.QueueName,
+			Namespace: w.Namespace,
+		},
+	}
 }
 
 // SetupWithManager sets up the controller with the Manager.
 func (r *QueueReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&kueue.Queue{}).
+		Watches(&source.Kind{Type: &kueue.QueuedWorkload{}}, &queuedWorkloadHandler{}).
 		WithEventFilter(r).
 		Complete(r)
 }
diff --git a/pkg/queue/manager.go b/pkg/queue/manager.go
index 0510ea314..1aea5c0c6 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -32,8 +32,10 @@ import (
 
 const workloadQueueKey = "spec.queueName"
 
+var errQueueDoesntExist = errors.New("queue doesn't exist")
+
 type Manager struct {
-	sync.Mutex
+	sync.RWMutex
 	cond sync.Cond
 
 	client client.Client
@@ -45,7 +47,7 @@ func NewManager(client client.Client) *Manager {
 		client: client,
 		queues: make(map[string]*Queue),
 	}
-	m.cond.L = &m.Mutex
+	m.cond.L = &m.RWMutex
 	return m
 }
 
@@ -86,7 +88,7 @@ func (m *Manager) UpdateQueue(q *kueue.Queue) error {
 	defer m.Unlock()
 	qImpl, ok := m.queues[Key(q)]
 	if !ok {
-		return errors.New("queue doesn't exist")
+		return errQueueDoesntExist
 	}
 	qImpl.setProperties(q)
 	return nil
@@ -103,6 +105,15 @@ func (m *Manager) DeleteQueue(q *kueue.Queue) {
 	delete(m.queues, key)
 }
 
+func (m *Manager) Status(q *kueue.Queue) (int, error) {
+	m.RLock()
+	defer m.RUnlock()
+	if q := m.queues[Key(q)]; q != nil {
+		return q.heap.Len(), nil
+	}
+	return 0, errQueueDoesntExist
+}
+
 // AddOrUpdateWorkload adds or updates workload to the corresponding queue.
 // Returns whether the queue existed.
 func (m *Manager) AddOrUpdateWorkload(w *kueue.QueuedWorkload) bool {
diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go
index 93b99b876..cd83f5cf8 100644
--- a/pkg/queue/queue.go
+++ b/pkg/queue/queue.go
@@ -83,6 +83,7 @@ func (q *Queue) Delete(w *kueue.QueuedWorkload) {
 		heap.Remove(&q.heap, item.index)
 	}
 }
+
 func (q *Queue) Pop() *workload.Info {
 	if q.heap.Len() == 0 {
 		return nil


Test patch:
diff --git a/pkg/queue/manager_test.go b/pkg/queue/manager_test.go
index 424d39118..c11794906 100644
--- a/pkg/queue/manager_test.go
+++ b/pkg/queue/manager_test.go
@@ -153,6 +153,96 @@ func TestAddWorkload(t *testing.T) {
 	}
 }
 
+func TestStatus(t *testing.T) {
+	ctx := context.Background()
+	scheme := runtime.NewScheme()
+	if err := kueue.AddToScheme(scheme); err != nil {
+		t.Fatalf("Failed adding kueue scheme: %s", err)
+	}
+	now := time.Now().Truncate(time.Second)
+
+	queues := []kueue.Queue{
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			Spec: kueue.QueueSpec{
+				Capacity: "fooCap",
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "bar"},
+			Spec: kueue.QueueSpec{
+				Capacity: "barCap",
+			},
+		},
+	}
+	workloads := []kueue.QueuedWorkload{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:              "a",
+				CreationTimestamp: metav1.NewTime(now.Add(time.Hour)),
+			},
+			Spec: kueue.QueuedWorkloadSpec{QueueName: "foo"},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:              "b",
+				CreationTimestamp: metav1.NewTime(now),
+			},
+			Spec: kueue.QueuedWorkloadSpec{QueueName: "bar"},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:              "c",
+				CreationTimestamp: metav1.NewTime(now),
+			},
+			Spec: kueue.QueuedWorkloadSpec{QueueName: "foo"},
+		},
+	}
+	manager := NewManager(fake.NewClientBuilder().WithScheme(scheme).Build())
+	for _, q := range queues {
+		if err := manager.AddQueue(ctx, &q); err != nil {
+			t.Errorf("Failed adding queue: %s", err)
+		}
+	}
+	for _, wl := range workloads {
+		wl := wl
+		manager.AddOrUpdateWorkload(&wl)
+	}
+
+	cases := map[string]struct {
+		queue      *kueue.Queue
+		wantStatus int
+		wantErr    error
+	}{
+		"foo": {
+			queue:      &queues[0],
+			wantStatus: 2,
+			wantErr:    nil,
+		},
+		"bar": {
+			queue:      &queues[1],
+			wantStatus: 1,
+			wantErr:    nil,
+		},
+		"fake": {
+			queue:      &kueue.Queue{ObjectMeta: metav1.ObjectMeta{Name: "fake"}},
+			wantStatus: 0,
+			wantErr:    errQueueDoesntExist,
+		},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			status, err := manager.Status(tc.queue)
+			if err != tc.wantErr {
+				t.Errorf("Should have failed with: %s", err)
+			}
+			if diff := cmp.Diff(tc.wantStatus, status); diff != "" {
+				t.Errorf("Status func returned wrong queue status: %s", diff)
+			}
+		})
+	}
+}
+
 func TestRequeueWorkload(t *testing.T) {
 	scheme := runtime.NewScheme()
 	if err := kueue.AddToScheme(scheme); err != nil {
@@ -427,7 +517,6 @@ func TestHeads(t *testing.T) {
 	if diff := cmp.Diff(wantHeads, heads); diff != "" {
 		t.Errorf("GetHeads returned wrong heads (-want,+got):\n%s", diff)
 	}
-
 }
 
 // TestHeadAsync ensures that Heads call is blocked until the queues are filled
diff --git a/test/integration/scheduler/suite_test.go b/test/integration/scheduler/suite_test.go
index 857803eee..68d40a133 100644
--- a/test/integration/scheduler/suite_test.go
+++ b/test/integration/scheduler/suite_test.go
@@ -76,7 +76,7 @@ func managerAndSchedulerSetup(mgr manager.Manager) {
 	queues := queue.NewManager(mgr.GetClient())
 	cache := capacity.NewCache(mgr.GetClient())
 
-	err = kueuectrl.NewQueueReconciler(queues).SetupWithManager(mgr)
+	err = kueuectrl.NewQueueReconciler(mgr.GetClient(), queues).SetupWithManager(mgr)
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
 	err = kueuectrl.NewCapacityReconciler(mgr.GetClient(), cache).SetupWithManager(mgr)