fix volcano podgroup update issue

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

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

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

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

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

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

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"`

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,

pkg/controller.v1/pytorch/pytorchjob_controller_test.go

@@ -195,6 +195,34 @@ 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")
+ updatedJob := &kubeflowv1.PyTorchJob{}
+ Eventually(func() bool {
+ err := testK8sClient.Get(ctx, jobKey, updatedJob)
+ return err == nil
+ }, testutil.Timeout, testutil.Interval).Should(BeTrue(), "Failed to get PyTorchJob")
+
+ Eventually(func() bool {
+ updatedJob.Spec.RunPolicy.SchedulingPolicy.Queue = "test"
+ err := testK8sClient.Update(ctx, updatedJob)
+ By("Checking that the queue update fails")
+ Expect(err).To(HaveOccurred(), "Expected an error when updating the queue, but update succeeded")
+ Expect(err).To(MatchError(ContainSubstring("spec.runPolicy.schedulingPolicy.queue is immutable"), "The error message did not contain the expected message"))
+ return err != nil
+ }, testutil.Timeout, testutil.Interval).Should(BeTrue())
+
+ 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")