Repository: kubernetes-sigs/kueue

Issue:
Flaky Test: MultiKueue when Creating a multikueue admission check Should run an appwrapper containing a job on worker if admitted
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

If the matter is security related, please disclose it privately via https://kubernetes.io/security/
-->


**What happened**:
The below E2E test case failed in unrelated PR:

`End To End MultiKueue Suite: kindest/node:v1.31.2: [It] MultiKueue when Creating a multikueue admission check Should run an appwrapper containing a job on worker if admitted`

```shell
{Expected
    <string>: Internal error occurred: error executing command in container: failed to exec in container: failed to create exec "0c0eab3acc210c78029f12abd60c3ac7a6149ce617809ac123da928f9fffa2b5": task 6ac8586efc6285b09f2d01bb6dfbc669f23b4164cacbdf623aae114fb13f346e not found: not found
to contain substring
    <string>: 137 failed [FAILED] Expected
    <string>: Internal error occurred: error executing command in container: failed to exec in container: failed to create exec "0c0eab3acc210c78029f12abd60c3ac7a6149ce617809ac123da928f9fffa2b5": task 6ac8586efc6285b09f2d01bb6dfbc669f23b4164cacbdf623aae114fb13f346e not found: not found
to contain substring
    <string>: 137
In [It] at: /home/prow/go/src/sigs.k8s.io/kueue/test/e2e/multikueue/e2e_test.go:609 @ 02/24/25 12:39:23.83
}
```

**What you expected to happen**:
Never errors

**How to reproduce it (as minimally and precisely as possible)**:
https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/4375/pull-kueue-test-e2e-multikueue-main/1893999933191622656

**Anything else we need to know?**:

**Environment**:
- Kubernetes version (use `kubectl version`):
- 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:

Hints from issue comments:
@tenzen-y: /kind flake@tenzen-y: cc @dgrove-oss 
@mimowo: x-ref with potentially the same underlying bug: https://github.com/kubernetes-sigs/kueue/issues/4378@mimowo: /close
Doing reset of e2e-related flakes as agreed in https://github.com/kubernetes-sigs/kueue/issues/4674#issuecomment-2734095182.

The reason is that we recently bumped up the job resources, and it is expected to help for most of the flakes were attributed to long termination of a job. So, this way we can avoid people looking into an already solved problem.

For more details check the PR [kubernetes/test-infra#34529](https://github.com/kubernetes/test-infra/pull/34529) as discussed here: [#4669](https://github.com/kubernetes-sigs/kueue/issues/4669).

If the failure re-occurs feel free to re-open or open a new one.

Also, feel free to re-open if you have some evidence / hints that constrained resources is not the reason for the failure.@k8s-ci-robot: @mimowo: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/4376#issuecomment-2735682017):

>/close
>Doing reset of e2e-related flakes as agreed in https://github.com/kubernetes-sigs/kueue/issues/4674#issuecomment-2734095182.
>
>The reason is that we recently bumped up the job resources, and it is expected to help for most of the flakes were attributed to long termination of a job. So, this way we can avoid people looking into an already solved problem.
>
>For more details check the PR [kubernetes/test-infra#34529](https://github.com/kubernetes/test-infra/pull/34529) as discussed here: [#4669](https://github.com/kubernetes-sigs/kueue/issues/4669).
>
>If the failure re-occurs feel free to re-open or open a new one.
>
>Also, feel free to re-open if you have some evidence / hints that constrained resources is not the reason for the failure.


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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

Solution patch:
diff --git a/pkg/controller/tas/topology_controller.go b/pkg/controller/tas/topology_controller.go
index 6b9fa6f28..0cc35b077 100644
--- a/pkg/controller/tas/topology_controller.go
+++ b/pkg/controller/tas/topology_controller.go
@@ -84,14 +84,14 @@ func (r *topologyReconciler) setupWithManager(mgr ctrl.Manager, cfg *configapi.C
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=topologies,verbs=get;list;watch;update
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=topologies/finalizers,verbs=update
 
-func (r topologyReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
+func (r *topologyReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	topology := &kueuealpha.Topology{}
 	if err := r.client.Get(ctx, req.NamespacedName, topology); err != nil {
 		// we'll ignore not-found errors, since there is nothing to do.
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	log := r.log.WithValues("name", req.NamespacedName.Name)
+	log := ctrl.LoggerFrom(ctx)
 	log.V(2).Info("Reconcile Topology")
 
 	if !topology.DeletionTimestamp.IsZero() {



Test patch:
diff --git a/pkg/cache/cache_test.go b/pkg/cache/cache_test.go
index 660a9062a..100378790 100644
--- a/pkg/cache/cache_test.go
+++ b/pkg/cache/cache_test.go
@@ -1156,7 +1156,7 @@ func TestCacheWorkloadOperations(t *testing.T) {
 			Request(corev1.ResourceCPU, "5m").
 			Obj(),
 	}
-	psAssignment := []kueue.PodSetAssignment{
+	psAssignments := []kueue.PodSetAssignment{
 		{
 			Name: "driver",
 			Flavors: map[corev1.ResourceName]kueue.ResourceFlavorReference{
@@ -1179,7 +1179,7 @@ func TestCacheWorkloadOperations(t *testing.T) {
 	cl := utiltesting.NewFakeClient(
 		utiltesting.MakeWorkload("a", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 			ClusterQueue:      "one",
-			PodSetAssignments: psAssignment,
+			PodSetAssignments: psAssignments,
 		}).Obj(),
 		utiltesting.MakeWorkload("b", "").ReserveQuota(&kueue.Admission{
 			ClusterQueue: "one",
@@ -1206,7 +1206,7 @@ func TestCacheWorkloadOperations(t *testing.T) {
 				workloads := []*kueue.Workload{
 					utiltesting.MakeWorkload("a", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 						ClusterQueue:      "one",
-						PodSetAssignments: psAssignment,
+						PodSetAssignments: psAssignments,
 					}).Obj(),
 					utiltesting.MakeWorkload("d", "").ReserveQuota(&kueue.Admission{
 						ClusterQueue: "two",
@@ -1288,7 +1288,7 @@ func TestCacheWorkloadOperations(t *testing.T) {
 				}).Obj()
 				latest := utiltesting.MakeWorkload("a", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 					ClusterQueue:      "two",
-					PodSetAssignments: psAssignment,
+					PodSetAssignments: psAssignments,
 				}).Obj()
 				return cache.UpdateWorkload(old, latest)
 			},
@@ -1492,11 +1492,11 @@ func TestCacheWorkloadOperations(t *testing.T) {
 				workloads := []*kueue.Workload{
 					utiltesting.MakeWorkload("d", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 						ClusterQueue:      "one",
-						PodSetAssignments: psAssignment,
+						PodSetAssignments: psAssignments,
 					}).Obj(),
 					utiltesting.MakeWorkload("e", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 						ClusterQueue:      "two",
-						PodSetAssignments: psAssignment,
+						PodSetAssignments: psAssignments,
 					}).Obj(),
 				}
 				for i := range workloads {
@@ -1559,11 +1559,11 @@ func TestCacheWorkloadOperations(t *testing.T) {
 				workloads := []*kueue.Workload{
 					utiltesting.MakeWorkload("d", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 						ClusterQueue:      "one",
-						PodSetAssignments: psAssignment,
+						PodSetAssignments: psAssignments,
 					}).Obj(),
 					utiltesting.MakeWorkload("e", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 						ClusterQueue:      "two",
-						PodSetAssignments: psAssignment,
+						PodSetAssignments: psAssignments,
 					}).Obj(),
 				}
 				for i := range workloads {
@@ -1626,11 +1626,11 @@ func TestCacheWorkloadOperations(t *testing.T) {
 				workloads := []*kueue.Workload{
 					utiltesting.MakeWorkload("d", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 						ClusterQueue:      "one",
-						PodSetAssignments: psAssignment,
+						PodSetAssignments: psAssignments,
 					}).Obj(),
 					utiltesting.MakeWorkload("e", "").PodSets(podSets...).ReserveQuota(&kueue.Admission{
 						ClusterQueue:      "two",
-						PodSetAssignments: psAssignment,
+						PodSetAssignments: psAssignments,
 					}).Obj(),
 				}
 				for i := range workloads {
diff --git a/test/e2e/singlecluster/e2e_test.go b/test/e2e/singlecluster/e2e_test.go
index 7a0750e07..32952ce93 100644
--- a/test/e2e/singlecluster/e2e_test.go
+++ b/test/e2e/singlecluster/e2e_test.go
@@ -130,7 +130,7 @@ var _ = ginkgo.Describe("Kueue", func() {
 			createdWorkload := &kueue.Workload{}
 
 			// The job might have finished at this point. That shouldn't be a problem for the purpose of this test
-			expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 				"instance-type": "on-demand",
 			})
 			wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(sampleJob.Name, sampleJob.UID), Namespace: ns.Name}
@@ -182,7 +182,7 @@ var _ = ginkgo.Describe("Kueue", func() {
 			})
 
 			ginkgo.By("Verify the job is running", func() {
-				expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 					"instance-type": "on-demand",
 				})
 			})
@@ -223,7 +223,7 @@ var _ = ginkgo.Describe("Kueue", func() {
 			})
 
 			ginkgo.By("Job is admitted using the first flavor", func() {
-				expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 					"instance-type": "on-demand",
 				})
 			})
@@ -237,13 +237,13 @@ var _ = ginkgo.Describe("Kueue", func() {
 					Obj()
 				gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
 
-				expectJobUnsuspendedWithNodeSelectors(client.ObjectKeyFromObject(job), map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, client.ObjectKeyFromObject(job), map[string]string{
 					"instance-type": "on-demand",
 				})
 			})
 
 			ginkgo.By("Job is re-admitted using the second flavor", func() {
-				expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 					"instance-type": "spot",
 				})
 			})
@@ -259,7 +259,7 @@ var _ = ginkgo.Describe("Kueue", func() {
 			})
 
 			ginkgo.By("Job is admitted using the first flavor", func() {
-				expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 					"instance-type": "on-demand",
 				})
 			})
@@ -273,13 +273,13 @@ var _ = ginkgo.Describe("Kueue", func() {
 					Obj()
 				gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
 
-				expectJobUnsuspendedWithNodeSelectors(client.ObjectKeyFromObject(job), map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, client.ObjectKeyFromObject(job), map[string]string{
 					"instance-type": "on-demand",
 				})
 			})
 
 			ginkgo.By("Job is re-admitted using the second flavor", func() {
-				expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 					"instance-type": "spot",
 				})
 			})
@@ -299,7 +299,7 @@ var _ = ginkgo.Describe("Kueue", func() {
 			// The job might have finished at this point. That shouldn't be a problem for the purpose of this test
 			ginkgo.By("Wait for the job to start and check the updated Parallelism and Completions", func() {
 				jobKey := client.ObjectKeyFromObject(job)
-				expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+				util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 					"instance-type": "on-demand",
 				})
 
@@ -406,7 +406,7 @@ var _ = ginkgo.Describe("Kueue", func() {
 			})
 
 			// The job might have finished at this point. That shouldn't be a problem for the purpose of this test
-			expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 				"instance-type": "on-demand",
 			})
 			gomega.Eventually(func(g gomega.Gomega) {
@@ -443,7 +443,7 @@ var _ = ginkgo.Describe("Kueue", func() {
 				}, util.Timeout, util.Interval).Should(gomega.Succeed())
 			})
 
-			expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 				"instance-type": "on-demand",
 			})
 
@@ -481,15 +481,6 @@ func expectJobUnsuspended(key types.NamespacedName) {
 	}, util.Timeout, util.Interval).Should(gomega.Succeed())
 }
 
-func expectJobUnsuspendedWithNodeSelectors(key types.NamespacedName, ns map[string]string) {
-	job := &batchv1.Job{}
-	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
-		g.Expect(k8sClient.Get(ctx, key, job)).To(gomega.Succeed())
-		g.Expect(job.Spec.Suspend).Should(gomega.Equal(ptr.To(false)))
-		g.Expect(job.Spec.Template.Spec.NodeSelector).Should(gomega.Equal(ns))
-	}, util.Timeout, util.Interval).Should(gomega.Succeed())
-}
-
 func defaultOwnerReferenceForJob(name string) []metav1.OwnerReference {
 	return []metav1.OwnerReference{
 		{
diff --git a/test/e2e/singlecluster/tas_test.go b/test/e2e/singlecluster/tas_test.go
index acc7978d1..75f21039c 100644
--- a/test/e2e/singlecluster/tas_test.go
+++ b/test/e2e/singlecluster/tas_test.go
@@ -109,7 +109,7 @@ var _ = ginkgo.Describe("TopologyAwareScheduling", func() {
 			createdWorkload := &kueue.Workload{}
 
 			// The job might have finished at this point. That shouldn't be a problem for the purpose of this test
-			expectJobUnsuspendedWithNodeSelectors(jobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, jobKey, map[string]string{
 				"instance-type": "on-demand",
 			})
 			wlLookupKey := types.NamespacedName{Name: workloadjob.GetWorkloadNameForJob(sampleJob.Name, sampleJob.UID), Namespace: ns.Name}
diff --git a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
index 5b1d4ccec..39fb130ff 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -1696,7 +1696,7 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 				Obj()
 			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
 
-			expectJobUnsuspendedWithNodeSelectors(lowJobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, lowJobKey, map[string]string{
 				instanceKey: "spot-untainted",
 			})
 		})
@@ -1712,13 +1712,13 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
 
 			highJobKey := types.NamespacedName{Name: "high", Namespace: ns.Name}
-			expectJobUnsuspendedWithNodeSelectors(highJobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, highJobKey, map[string]string{
 				instanceKey: "spot-untainted",
 			})
 		})
 
 		ginkgo.By("Preempted job should be admitted on second flavor", func() {
-			expectJobUnsuspendedWithNodeSelectors(lowJobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, lowJobKey, map[string]string{
 				instanceKey: "on-demand",
 			})
 		})
@@ -1743,7 +1743,7 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 				Obj()
 			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
 
-			expectJobUnsuspendedWithNodeSelectors(lowJobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, lowJobKey, map[string]string{
 				instanceKey: "spot-untainted",
 			})
 		})
@@ -1759,13 +1759,13 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
 
 			highJobKey := types.NamespacedName{Name: "high", Namespace: ns.Name}
-			expectJobUnsuspendedWithNodeSelectors(highJobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, highJobKey, map[string]string{
 				instanceKey: "spot-untainted",
 			})
 		})
 
 		ginkgo.By("Preempted job should be admitted on second flavor", func() {
-			expectJobUnsuspendedWithNodeSelectors(lowJobKey, map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, lowJobKey, map[string]string{
 				instanceKey: "on-demand",
 			})
 		})
@@ -1867,7 +1867,7 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 				Parallelism(2).
 				Obj()
 			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
-			expectJobUnsuspendedWithNodeSelectors(client.ObjectKeyFromObject(job), map[string]string{
+			util.ExpectJobUnsuspendedWithNodeSelectors(ctx, k8sClient, client.ObjectKeyFromObject(job), map[string]string{
 				instanceKey: "on-demand",
 			})
 		})
@@ -2414,12 +2414,3 @@ var _ = ginkgo.Describe("Job controller when TopologyAwareScheduling enabled", g
 		})
 	})
 })
-
-func expectJobUnsuspendedWithNodeSelectors(key types.NamespacedName, ns map[string]string) {
-	job := &batchv1.Job{}
-	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
-		g.Expect(k8sClient.Get(ctx, key, job)).To(gomega.Succeed())
-		g.Expect(job.Spec.Suspend).Should(gomega.Equal(ptr.To(false)))
-		g.Expect(job.Spec.Template.Spec.NodeSelector).Should(gomega.Equal(ns))
-	}, util.Timeout, util.Interval).Should(gomega.Succeed())
-}
diff --git a/test/util/util.go b/test/util/util.go
index 3d0a7e18d..c6908fdc4 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -879,6 +879,15 @@ func ExpectLocalQueuesToBeActive(ctx context.Context, c client.Client, lqs ...*k
 	}, Timeout, Interval).Should(gomega.Succeed())
 }
 
+func ExpectJobUnsuspendedWithNodeSelectors(ctx context.Context, c client.Client, key types.NamespacedName, nodeSelector map[string]string) {
+	job := &batchv1.Job{}
+	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
+		g.Expect(c.Get(ctx, key, job)).To(gomega.Succeed())
+		g.Expect(job.Spec.Suspend).Should(gomega.Equal(ptr.To(false)))
+		g.Expect(job.Spec.Template.Spec.NodeSelector).Should(gomega.Equal(nodeSelector))
+	}, Timeout, Interval).Should(gomega.Succeed())
+}
+
 func CreateNodesWithStatus(ctx context.Context, c client.Client, nodes []corev1.Node) {
 	for _, node := range nodes {
 		// 1. Create a node