Improved RayJob Logic - bugfix on Lifecycled RayJobs causing Webhook to panic

len(spec.ClusterSelector)>0 && spec.RayClusterSpec != nil -> validation error
len(spec.ClusterSelector)>0 && spec.RayClusterSpec == nil - valid
len(spec.ClusterSelector)==0 && spec.ClusterSpec == nil -> validation error
len(spec.ClusterSelector)==0 && spec.ClusterSpec != nil -> valid + perform additional validation

Author: laurafitzgerald

pkg/controller/jobs/rayjob/rayjob_controller.go

@@ -78,6 +78,7 @@ type RayJob rayv1.RayJob
 
 var _ jobframework.GenericJob = (*RayJob)(nil)
 var _ jobframework.JobWithManagedBy = (*RayJob)(nil)
+var _ jobframework.JobWithSkip = (*RayJob)(nil)
 
 func (j *RayJob) Object() client.Object {
 	return (*rayv1.RayJob)(j)
@@ -100,6 +101,12 @@ func (j *RayJob) Suspend() {
 	j.Spec.Suspend = true
 }
 
+func (j *RayJob) Skip(ctx context.Context) bool {
+	// Skip reconciliation for RayJobs that use clusterSelector to reference existing clusters.
+	// These jobs are not managed by Kueue.
+	return len(j.Spec.ClusterSelector) > 0
+}
+
 func (j *RayJob) GVK() schema.GroupVersionKind {
 	return gvk
 }

pkg/controller/jobs/rayjob/rayjob_webhook.go

@@ -110,17 +110,34 @@ func (w *RayJobWebhook) validateCreate(ctx context.Context, job *rayv1.RayJob) (
 	if w.manageJobsWithoutQueueName || jobframework.QueueName(kueueJob) != "" {
 		spec := &job.Spec
 		specPath := field.NewPath("spec")
+		hasClusterSelector := len(spec.ClusterSelector) > 0
+		hasRayClusterSpec := spec.RayClusterSpec != nil
+
+		// Validate the combination of clusterSelector and RayClusterSpec
+		if hasClusterSelector && hasRayClusterSpec {
+			// len(spec.ClusterSelector)>0 && spec.RayClusterSpec != nil -> validation error
+			allErrors = append(allErrors, field.Invalid(specPath.Child("clusterSelector"), spec.ClusterSelector, "a kueue managed job should not use an existing cluster"))
+			return allErrors, nil
+		}
+
+		if hasClusterSelector && !hasRayClusterSpec {
+			// len(spec.ClusterSelector)>0 && spec.RayClusterSpec == nil -> valid (skip validation)
+			// RayJobs using existing clusters are not managed by Kueue
+			return allErrors, nil
+		}
+
+		if !hasClusterSelector && !hasRayClusterSpec {
+			// len(spec.ClusterSelector)==0 && spec.RayClusterSpec == nil -> validation error
+			allErrors = append(allErrors, field.Required(specPath.Child("rayClusterSpec"), "rayClusterSpec is required for Kueue-managed jobs that don't use clusterSelector"))
+			return allErrors, nil
+		}
+		// len(spec.ClusterSelector)==0 && spec.RayClusterSpec != nil -> valid + perform additional validation
 
 		// Should always delete the cluster after the job has ended, otherwise it will continue to the queue's resources.
 		if !spec.ShutdownAfterJobFinishes {
 			allErrors = append(allErrors, field.Invalid(specPath.Child("shutdownAfterJobFinishes"), spec.ShutdownAfterJobFinishes, "a kueue managed job should delete the cluster after finishing"))
 		}
 
-		// Should not want existing cluster. Kueue (workload) should be able to control the admission of the actual work, not only the trigger.
-		if len(spec.ClusterSelector) > 0 {
-			allErrors = append(allErrors, field.Invalid(specPath.Child("clusterSelector"), spec.ClusterSelector, "a kueue managed job should not use an existing cluster"))
-		}
-
 		clusterSpec := spec.RayClusterSpec
 		clusterSpecPath := specPath.Child("rayClusterSpec")
 

pkg/controller/jobs/rayjob/rayjob_webhook_test.go

@@ -144,6 +144,25 @@ func TestValidateCreate(t *testing.T) {
 				Obj(),
 			wantErr: nil,
 		},
+
+		"valid managed - has cluster selector but no RayClusterSpec": {
+			job: testingrayutil.MakeJob("job", "ns").Queue("queue").
+				ClusterSelector(map[string]string{
+					"k1": "v1",
+				}).
+				RayClusterSpec(nil).
+				Obj(),
+			localQueueDefaulting: false,
+			wantErr:              nil,
+		},
+		"invalid managed - no cluster selector and no RayClusterSpec": {
+			job: testingrayutil.MakeJob("job", "ns").Queue("queue").
+				RayClusterSpec(nil).
+				Obj(),
+			wantErr: field.ErrorList{
+				field.Required(field.NewPath("spec", "rayClusterSpec"), "rayClusterSpec is required for Kueue-managed jobs that don't use clusterSelector"),
+			}.ToAggregate(),
+		},
 		"invalid unmanaged - local queue default": {
 			job: testingrayutil.MakeJob("job", "ns").
 				ShutdownAfterJobFinishes(false).
@@ -168,7 +187,7 @@ func TestValidateCreate(t *testing.T) {
 				field.Invalid(field.NewPath("spec", "shutdownAfterJobFinishes"), false, "a kueue managed job should delete the cluster after finishing"),
 			}.ToAggregate(),
 		},
-		"invalid managed - has cluster selector": {
+		"invalid managed - has cluster selector and RayClusterSpec": {
 			job: testingrayutil.MakeJob("job", "ns").Queue("queue").
 				ClusterSelector(map[string]string{
 					"k1": "v1",

pkg/util/testingjobs/rayjob/wrappers.go

@@ -147,6 +147,11 @@ func (j *JobWrapper) WithEnableAutoscaling(value *bool) *JobWrapper {
 	return j
 }
 
+func (j *JobWrapper) RayClusterSpec(spec *rayv1.RayClusterSpec) *JobWrapper {
+	j.Spec.RayClusterSpec = spec
+	return j
+}
+
 func (j *JobWrapper) WithWorkerGroups(workers ...rayv1.WorkerGroupSpec) *JobWrapper {
 	j.Spec.RayClusterSpec.WorkerGroupSpecs = workers
 	return j

test/integration/singlecluster/controller/jobs/rayjob/rayjob_controller_test.go

@@ -681,4 +681,30 @@ var _ = ginkgo.Describe("Job controller with preemption enabled", ginkgo.Ordered
 			g.Expect(createdJob.Spec.Suspend).Should(gomega.BeFalse())
 		}, util.Timeout, util.Interval).Should(gomega.Succeed())
 	})
+
+	ginkgo.It("Should skip reconciliation for RayJobs with clusterSelector", func() {
+		ginkgo.By("Creating a RayJob with clusterSelector and queue label")
+		job := testingrayjob.MakeJob("rayjob-with-selector", ns.Name).
+			Queue("test-queue").
+			ClusterSelector(map[string]string{"ray.io/cluster": "existing-cluster"}).
+			Suspend(false).
+			Obj()
+		util.MustCreate(ctx, k8sClient, job)
+
+		ginkgo.By("Checking that no workload is created for this job and the job remains unchanged")
+		lookupKey := types.NamespacedName{
+			Name:      workloadrayjob.GetWorkloadNameForRayJob(job.Name, job.UID),
+			Namespace: ns.Name,
+		}
+		createdWorkload := &kueue.Workload{}
+		createdJob := &rayv1.RayJob{}
+
+		gomega.Consistently(func(g gomega.Gomega) {
+			err := k8sClient.Get(ctx, lookupKey, createdWorkload)
+			g.Expect(err).Should(gomega.HaveOccurred())
+			g.Expect(client.IgnoreNotFound(err)).Should(gomega.Succeed())
+			g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).Should(gomega.Succeed())
+			g.Expect(createdJob.Spec.Suspend).Should(gomega.BeFalse())
+		}, util.ConsistentDuration, util.Interval).Should(gomega.Succeed())
+	})
 })

test/integration/singlecluster/webhook/jobs/rayjob_webhook_test.go

@@ -61,5 +61,54 @@ var _ = ginkgo.Describe("RayJob Webhook", func() {
 			gomega.Expect(err).Should(gomega.HaveOccurred())
 			gomega.Expect(err).Should(testing.BeForbiddenError())
 		})
+
+		ginkgo.It("should reject RayJob with clusterSelector and RayClusterSpec", func() {
+			job := testingjob.MakeJob("rayjob-with-both", ns.Name).
+				Queue("queue-name").
+				ClusterSelector(map[string]string{"ray.io/cluster": "existing-cluster"}).
+				Obj()
+			// clusterSelector + RayClusterSpec -> validation error
+			err := k8sClient.Create(ctx, job)
+			gomega.Expect(err).Should(gomega.HaveOccurred())
+			gomega.Expect(err).Should(testing.BeForbiddenError())
+		})
+
+		ginkgo.It("should allow RayJob with clusterSelector but no RayClusterSpec", func() {
+			job := testingjob.MakeJob("rayjob-selector-only", ns.Name).
+				Queue("queue-name").
+				ClusterSelector(map[string]string{"ray.io/cluster": "existing-cluster"}).
+				RayClusterSpec(nil).
+				Obj()
+			// clusterSelector + nil RayClusterSpec -> valid (not managed by Kueue)
+			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
+		})
+
+		ginkgo.It("should reject RayJob with queue label but no clusterSelector and no RayClusterSpec", func() {
+			job := testingjob.MakeJob("rayjob-nil-clusterspec", ns.Name).
+				Queue("queue-name").
+				RayClusterSpec(nil).
+				Obj()
+			// no clusterSelector + nil RayClusterSpec -> validation error
+			err := k8sClient.Create(ctx, job)
+			gomega.Expect(err).Should(gomega.HaveOccurred())
+			gomega.Expect(err).Should(testing.BeForbiddenError())
+		})
+
+		ginkgo.It("should allow RayJob with clusterSelector and no queue label", func() {
+			job := testingjob.MakeJob("rayjob-with-selector-no-queue", ns.Name).
+				ClusterSelector(map[string]string{"ray.io/cluster": "existing-cluster"}).
+				Obj()
+			// No queue label -> not managed by Kueue, validation skipped
+			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
+		})
+
+		ginkgo.It("should allow valid RayJob with RayClusterSpec and no clusterSelector", func() {
+			job := testingjob.MakeJob("rayjob-valid", ns.Name).
+				Queue("queue-name").
+				Obj()
+			// no clusterSelector + RayClusterSpec present -> valid, full validation runs
+			// MakeJob() creates a valid RayClusterSpec by default
+			gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
+		})
 	})
 })