Automated cherry pick of kubeflow#2079: fix volcano podgroup update i…

…ssue kubeflow#2130: Refine the integration tests for the immutable PyTorchJob (kubeflow#2139) Signed-off-by: Yuki Iwai <[email protected]> Co-authored-by: Weiyu Yen <[email protected]>
andreyvelich · Jun 10, 2024 · 498e66e · 498e66e
1 parent 643af3d
commit 498e66e
Show file tree

Hide file tree

Showing 9 changed files with 48 additions and 1 deletion.
diff --git a/manifests/base/crds/kubeflow.org_mpijobs.yaml b/manifests/base/crds/kubeflow.org_mpijobs.yaml
@@ -7330,6 +7330,9 @@ spec:
                         type: string
                       queue:
                         type: string
+                        x-kubernetes-validations:
+                        - message: spec.runPolicy.schedulingPolicy.queue is immutable
+                          rule: self == oldSelf
                       scheduleTimeoutSeconds:
                         format: int32
                         type: integer

diff --git a/manifests/base/crds/kubeflow.org_mxjobs.yaml b/manifests/base/crds/kubeflow.org_mxjobs.yaml
@@ -7333,6 +7333,9 @@ spec:
                         type: string
                       queue:
                         type: string
+                        x-kubernetes-validations:
+                        - message: spec.runPolicy.schedulingPolicy.queue is immutable
+                          rule: self == oldSelf
                       scheduleTimeoutSeconds:
                         format: int32
                         type: integer

diff --git a/manifests/base/crds/kubeflow.org_paddlejobs.yaml b/manifests/base/crds/kubeflow.org_paddlejobs.yaml
@@ -7812,6 +7812,9 @@ spec:
                         type: string
                       queue:
                         type: string
+                        x-kubernetes-validations:
+                        - message: spec.runPolicy.schedulingPolicy.queue is immutable
+                          rule: self == oldSelf
                       scheduleTimeoutSeconds:
                         format: int32
                         type: integer

diff --git a/manifests/base/crds/kubeflow.org_pytorchjobs.yaml b/manifests/base/crds/kubeflow.org_pytorchjobs.yaml
@@ -7849,6 +7849,9 @@ spec:
                         type: string
                       queue:
                         type: string
+                        x-kubernetes-validations:
+                        - message: spec.runPolicy.schedulingPolicy.queue is immutable
+                          rule: self == oldSelf
                       scheduleTimeoutSeconds:
                         format: int32
                         type: integer

diff --git a/manifests/base/crds/kubeflow.org_tfjobs.yaml b/manifests/base/crds/kubeflow.org_tfjobs.yaml
@@ -90,6 +90,9 @@ spec:
                         type: string
                       queue:
                         type: string
+                        x-kubernetes-validations:
+                        - message: spec.runPolicy.schedulingPolicy.queue is immutable
+                          rule: self == oldSelf
                       scheduleTimeoutSeconds:
                         format: int32
                         type: integer

diff --git a/manifests/base/crds/kubeflow.org_xgboostjobs.yaml b/manifests/base/crds/kubeflow.org_xgboostjobs.yaml
@@ -86,6 +86,9 @@ spec:
                         type: string
                       queue:
                         type: string
+                        x-kubernetes-validations:
+                        - message: spec.runPolicy.schedulingPolicy.queue is immutable
+                          rule: self == oldSelf
                       scheduleTimeoutSeconds:
                         format: int32
                         type: integer

diff --git a/pkg/apis/kubeflow.org/v1/common_types.go b/pkg/apis/kubeflow.org/v1/common_types.go
@@ -226,7 +226,8 @@ type RunPolicy struct {
 // SchedulingPolicy encapsulates various scheduling policies of the distributed training
 // job, for example `minAvailable` for gang-scheduling.
 type SchedulingPolicy struct {
-	MinAvailable           *int32                                 `json:"minAvailable,omitempty"`
+	MinAvailable *int32 `json:"minAvailable,omitempty"`
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf", message="spec.runPolicy.schedulingPolicy.queue is immutable"
 	Queue                  string                                 `json:"queue,omitempty"`
 	MinResources           *map[v1.ResourceName]resource.Quantity `json:"minResources,omitempty"`
 	PriorityClass          string                                 `json:"priorityClass,omitempty"`

diff --git a/pkg/controller.v1/common/job.go b/pkg/controller.v1/common/job.go
@@ -284,6 +284,11 @@ func (jc *JobController) ReconcileJobs(
 					if !match {
 						return fmt.Errorf("unable to recognize PodGroup: %v", klog.KObj(pg))
 					}
+
+					if q := volcanoPodGroup.Spec.Queue; len(q) > 0 {
+						queue = q
+					}
+
 					volcanoPodGroup.Spec = volcanov1beta1.PodGroupSpec{
 						MinMember:         minMember,
 						Queue:             queue,

diff --git a/pkg/controller.v1/pytorch/pytorchjob_controller_test.go b/pkg/controller.v1/pytorch/pytorchjob_controller_test.go
@@ -195,6 +195,29 @@ var _ = Describe("PyTorchJob controller", func() {
 			cond := getCondition(created.Status, kubeflowv1.JobSucceeded)
 			Expect(cond.Status).To(Equal(corev1.ConditionTrue))
 		})
+		It("Shouldn't be updated resources if spec.runPolicy.schedulingPolicy.queue is changed after the job is created", func() {
+			By("Creating a PyTorchJob with a specific queue")
+			job.Spec.RunPolicy.SchedulingPolicy = &kubeflowv1.SchedulingPolicy{}
+			job.Spec.RunPolicy.SchedulingPolicy.Queue = "initial-queue"
+			Expect(testK8sClient.Create(ctx, job)).Should(Succeed())
+
+			By("Attempting to update the PyTorchJob with a different queue value")
+			Eventually(func(g Gomega) {
+				updatedJob := &kubeflowv1.PyTorchJob{}
+				g.Expect(testK8sClient.Get(ctx, jobKey, updatedJob)).Should(Succeed(), "Failed to get PyTorchJob")
+				updatedJob.Spec.RunPolicy.SchedulingPolicy.Queue = "test"
+				err := testK8sClient.Update(ctx, updatedJob)
+				g.Expect(err).To(HaveOccurred(), "Expected an error when updating the queue, but update succeeded")
+				By("Checking that the queue update fails")
+				Expect(err).To(MatchError(ContainSubstring("spec.runPolicy.schedulingPolicy.queue is immutable"), "The error message did not contain the expected message"))
+
+			}, testutil.Timeout, testutil.Interval).Should(Succeed())
+
+			By("Validating the queue was not updated")
+			freshJob := &kubeflowv1.PyTorchJob{}
+			Expect(testK8sClient.Get(ctx, client.ObjectKeyFromObject(job), freshJob)).Should(Succeed(), "Failed to get PyTorchJob after update attempt")
+			Expect(freshJob.Spec.RunPolicy.SchedulingPolicy.Queue).To(Equal("initial-queue"), "The queue should remain as the initial value since it should be immutable")
+		})
 
 		It("Shouldn't create resources if PyTorchJob is suspended", func() {
 			By("By creating a new PyTorchJob with suspend=true")