Repository: kubernetes-sigs/kueue
Pull request #245 (2022-05-03 08:52:01)

fix: update failure in requeue
Signed-off-by: kerthcet <kerthcet@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 bug

<!--
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`_*
-->
Part of https://github.com/kubernetes-sigs/kueue/issues/241

#### 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/245#" title="Author self-approved">kerthcet</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **ahg-g** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @ahg-g` 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":["ahg-g"]} -->
kerthcet: /test pull-kueue-test-integration-main
kerthcet: /test pull-kueue-test-integration-main
kerthcet: /test pull-kueue-test-integration-main
k8s-ci-robot: @kerthcet: 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 | dba9d51de6a4ce206f9e7b0ec301f07255b33ecf | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kueue/245/pull-kueue-test-unit-main/1521443790114525184) | true | `/test pull-kueue-test-unit-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kueue&pr=245). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Akerthcet). 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: @kerthcet: PR needs rebase.

<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.
</details>
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/245#issuecomment-1119648379" title="Approved">alculquicondor</a>*, *<a href="https://github.com/kubernetes-sigs/kueue/pull/245#" title="Author self-approved">kerthcet</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 12167d3d8..f0c07db88 100644
--- a/pkg/queue/manager.go
+++ b/pkg/queue/manager.go
@@ -293,25 +293,27 @@ func (m *Manager) RequeueWorkload(ctx context.Context, info *workload.Info, imme
 	m.Lock()
 	defer m.Unlock()
 
-	q := m.queues[queueKeyForWorkload(info.Obj)]
-	if q == nil {
-		return false
-	}
-
 	var w kueue.Workload
+	// Always get the newest workload to avoid requeuing the out-of-date obj.
 	err := m.client.Get(ctx, client.ObjectKeyFromObject(info.Obj), &w)
 	// Since the client is cached, the only possible error is NotFound
 	if apierrors.IsNotFound(err) || w.Spec.Admission != nil {
 		return false
 	}
 
-	q.AddIfNotPresent(info)
+	q := m.queues[queueKeyForWorkload(&w)]
+	if q == nil {
+		return false
+	}
+	q.AddOrUpdate(&w)
 	q.reportPendingWorkloads()
 	cq := m.clusterQueues[q.ClusterQueue]
 	if cq == nil {
 		return false
 	}
 
+	// TODO(#162) Update the info object instead of constructing one.
+	info = workload.NewInfo(&w)
 	added := cq.RequeueIfNotPresent(info, immediate)
 	if added {
 		m.cond.Broadcast()


Test patch:
diff --git a/pkg/queue/manager_test.go b/pkg/queue/manager_test.go
index 26bc385ed..acb4c4bfb 100644
--- a/pkg/queue/manager_test.go
+++ b/pkg/queue/manager_test.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -700,6 +701,8 @@ func TestHeads(t *testing.T) {
 	}
 }
 
+var ignoreTypeMeta = cmpopts.IgnoreTypes(metav1.TypeMeta{})
+
 // TestHeadAsync ensures that Heads call is blocked until the queues are filled
 // asynchronously.
 func TestHeadsAsync(t *testing.T) {
@@ -708,7 +711,10 @@ func TestHeadsAsync(t *testing.T) {
 		t.Fatalf("Failed adding kueue scheme: %s", err)
 	}
 	now := time.Now().Truncate(time.Second)
-	cq := utiltesting.MakeClusterQueue("fooCq").Obj()
+	clusterQueues := []*kueue.ClusterQueue{
+		utiltesting.MakeClusterQueue("fooCq").Obj(),
+		utiltesting.MakeClusterQueue("barCq").Obj(),
+	}
 	wl := kueue.Workload{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:              "a",
@@ -716,68 +722,90 @@ func TestHeadsAsync(t *testing.T) {
 		},
 		Spec: kueue.WorkloadSpec{QueueName: "foo"},
 	}
-	q := kueue.Queue{
-		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
-		Spec: kueue.QueueSpec{
-			ClusterQueue: "fooCq",
+	var newWl kueue.Workload
+	queues := []kueue.Queue{
+		{
+			ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+			Spec: kueue.QueueSpec{
+				ClusterQueue: "fooCq",
+			},
 		},
-	}
-	wantHeads := []workload.Info{
 		{
-			Obj:          &wl,
-			ClusterQueue: "fooCq",
+			ObjectMeta: metav1.ObjectMeta{Name: "bar"},
+			Spec: kueue.QueueSpec{
+				ClusterQueue: "barCq",
+			},
 		},
 	}
 	cases := map[string]struct {
 		initialObjs []client.Object
 		op          func(context.Context, *Manager)
+		wantHeads   []workload.Info
 	}{
 		"AddClusterQueue": {
-			initialObjs: []client.Object{&wl, &q},
+			initialObjs: []client.Object{&wl, &queues[0]},
 			op: func(ctx context.Context, mgr *Manager) {
-				if err := mgr.AddQueue(ctx, &q); err != nil {
+				if err := mgr.AddQueue(ctx, &queues[0]); err != nil {
 					t.Errorf("Failed adding queue: %s", err)
 				}
 				mgr.AddOrUpdateWorkload(&wl)
 				go func() {
-					if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+					if err := mgr.AddClusterQueue(ctx, clusterQueues[0]); err != nil {
 						t.Errorf("Failed adding clusterQueue: %v", err)
 					}
 				}()
 			},
+			wantHeads: []workload.Info{
+				{
+					Obj:          &wl,
+					ClusterQueue: "fooCq",
+				},
+			},
 		},
 		"AddQueue": {
 			initialObjs: []client.Object{&wl},
 			op: func(ctx context.Context, mgr *Manager) {
-				if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+				if err := mgr.AddClusterQueue(ctx, clusterQueues[0]); err != nil {
 					t.Errorf("Failed adding clusterQueue: %v", err)
 				}
 				go func() {
-					if err := mgr.AddQueue(ctx, &q); err != nil {
+					if err := mgr.AddQueue(ctx, &queues[0]); err != nil {
 						t.Errorf("Failed adding queue: %s", err)
 					}
 				}()
 			},
+			wantHeads: []workload.Info{
+				{
+					Obj:          &wl,
+					ClusterQueue: "fooCq",
+				},
+			},
 		},
 		"AddWorkload": {
 			op: func(ctx context.Context, mgr *Manager) {
-				if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+				if err := mgr.AddClusterQueue(ctx, clusterQueues[0]); err != nil {
 					t.Errorf("Failed adding clusterQueue: %v", err)
 				}
-				if err := mgr.AddQueue(ctx, &q); err != nil {
+				if err := mgr.AddQueue(ctx, &queues[0]); err != nil {
 					t.Errorf("Failed adding queue: %s", err)
 				}
 				go func() {
 					mgr.AddOrUpdateWorkload(&wl)
 				}()
 			},
+			wantHeads: []workload.Info{
+				{
+					Obj:          &wl,
+					ClusterQueue: "fooCq",
+				},
+			},
 		},
 		"UpdateWorkload": {
 			op: func(ctx context.Context, mgr *Manager) {
-				if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+				if err := mgr.AddClusterQueue(ctx, clusterQueues[0]); err != nil {
 					t.Errorf("Failed adding clusterQueue: %v", err)
 				}
-				if err := mgr.AddQueue(ctx, &q); err != nil {
+				if err := mgr.AddQueue(ctx, &queues[0]); err != nil {
 					t.Errorf("Failed adding queue: %s", err)
 				}
 				go func() {
@@ -786,14 +814,20 @@ func TestHeadsAsync(t *testing.T) {
 					mgr.UpdateWorkload(wlCopy, &wl)
 				}()
 			},
+			wantHeads: []workload.Info{
+				{
+					Obj:          &wl,
+					ClusterQueue: "fooCq",
+				},
+			},
 		},
 		"RequeueWorkload": {
 			initialObjs: []client.Object{&wl},
 			op: func(ctx context.Context, mgr *Manager) {
-				if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+				if err := mgr.AddClusterQueue(ctx, clusterQueues[0]); err != nil {
 					t.Errorf("Failed adding clusterQueue: %v", err)
 				}
-				if err := mgr.AddQueue(ctx, &q); err != nil {
+				if err := mgr.AddQueue(ctx, &queues[0]); err != nil {
 					t.Errorf("Failed adding queue: %s", err)
 				}
 				// Remove the initial workload from the manager.
@@ -802,6 +836,72 @@ func TestHeadsAsync(t *testing.T) {
 					mgr.RequeueWorkload(ctx, workload.NewInfo(&wl), true)
 				}()
 			},
+			wantHeads: []workload.Info{
+				{
+					Obj:          &wl,
+					ClusterQueue: "fooCq",
+				},
+			},
+		},
+		"RequeueWithOutOfDateWorkload": {
+			initialObjs: []client.Object{&wl},
+			op: func(ctx context.Context, mgr *Manager) {
+				if err := mgr.AddClusterQueue(ctx, clusterQueues[0]); err != nil {
+					t.Errorf("Failed adding clusterQueue: %v", err)
+				}
+				if err := mgr.AddQueue(ctx, &queues[0]); err != nil {
+					t.Errorf("Failed adding queue: %s", err)
+				}
+
+				newWl = wl
+				newWl.Annotations = map[string]string{"foo": "bar"}
+				if err := mgr.client.Update(ctx, &newWl, &client.UpdateOptions{}); err != nil {
+					t.Errorf("Failed to update the workload; %s", err)
+				}
+				// Remove the initial workload from the manager.
+				mgr.Heads(ctx)
+				go func() {
+					mgr.RequeueWorkload(ctx, workload.NewInfo(&wl), true)
+				}()
+			},
+			wantHeads: []workload.Info{
+				{
+					Obj:          &newWl,
+					ClusterQueue: "fooCq",
+				},
+			},
+		},
+		"RequeueWithQueueChangedWorkload": {
+			initialObjs: []client.Object{&wl},
+			op: func(ctx context.Context, mgr *Manager) {
+				for _, cq := range clusterQueues {
+					if err := mgr.AddClusterQueue(ctx, cq); err != nil {
+						t.Errorf("Failed adding clusterQueue: %v", err)
+					}
+				}
+				for _, q := range queues {
+					if err := mgr.AddQueue(ctx, &q); err != nil {
+						t.Errorf("Failed adding queue: %s", err)
+					}
+				}
+
+				newWl = wl
+				newWl.Spec.QueueName = "bar"
+				if err := mgr.client.Update(ctx, &newWl, &client.UpdateOptions{}); err != nil {
+					t.Errorf("Failed to update the workload; %s", err)
+				}
+				// Remove the initial workload from the manager.
+				mgr.Heads(ctx)
+				go func() {
+					mgr.RequeueWorkload(ctx, workload.NewInfo(&wl), true)
+				}()
+			},
+			wantHeads: []workload.Info{
+				{
+					Obj:          &newWl,
+					ClusterQueue: "barCq",
+				},
+			},
 		},
 	}
 	for name, tc := range cases {
@@ -813,7 +913,7 @@ func TestHeadsAsync(t *testing.T) {
 			go manager.CleanUpOnContext(ctx)
 			tc.op(ctx, manager)
 			heads := manager.Heads(ctx)
-			if diff := cmp.Diff(wantHeads, heads); diff != "" {
+			if diff := cmp.Diff(tc.wantHeads, heads, ignoreTypeMeta); diff != "" {
 				t.Errorf("GetHeads returned wrong heads (-want,+got):\n%s", diff)
 			}
 		})