Add K8sJobMode check to prevent YuniKorn from adding submitter pod annotations

Signed-off-by: win5923 <[email protected]>

Author: win5923

ray-operator/controllers/ray/batchscheduler/volcano/volcano_scheduler.go

@@ -80,8 +80,8 @@ func (v *VolcanoBatchScheduler) handleRayJob(ctx context.Context, rayJob *rayv1.
 	// submitter's resource requests into MinResources so capacity is reserved.
 	if rayJob.Spec.SubmissionMode == rayv1.K8sJobMode {
 		submitterTemplate := common.GetSubmitterTemplate(&rayJob.Spec, rayJob.Spec.RayClusterSpec)
-		submitResource := utils.CalculatePodResource(submitterTemplate.Spec)
-		totalResourceList = append(totalResourceList, submitResource)
+		submitterResource := utils.CalculatePodResource(submitterTemplate.Spec)
+		totalResourceList = append(totalResourceList, submitterResource)
 	}
 
 	return v.syncPodGroup(ctx, rayJob, minMember, utils.SumResourceList(totalResourceList))

ray-operator/controllers/ray/rayjob_controller.go

@@ -279,14 +279,23 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 		jobInfo, err := rayDashboardClient.GetJobInfo(ctx, rayJobInstance.Status.JobId)
 		if err != nil {
 			// If the Ray job was not found, GetJobInfo returns a BadRequest error.
-			if rayJobInstance.Spec.SubmissionMode == rayv1.HTTPMode && errors.IsBadRequest(err) {
-				logger.Info("The Ray job was not found. Submit a Ray job via an HTTP request.", "JobId", rayJobInstance.Status.JobId)
-				if _, err := rayDashboardClient.SubmitJob(ctx, rayJobInstance); err != nil {
-					logger.Error(err, "Failed to submit the Ray job", "JobId", rayJobInstance.Status.JobId)
-					return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
+			if errors.IsBadRequest(err) {
+				if rayJobInstance.Spec.SubmissionMode == rayv1.HTTPMode {
+					logger.Info("The Ray job was not found. Submit a Ray job via an HTTP request.", "JobId", rayJobInstance.Status.JobId)
+					if _, err := rayDashboardClient.SubmitJob(ctx, rayJobInstance); err != nil {
+						logger.Error(err, "Failed to submit the Ray job", "JobId", rayJobInstance.Status.JobId)
+						return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
+					}
+					return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, nil
+				}
+				if isSubmitterFinished {
+					rayJobInstance.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusFailed
+					rayJobInstance.Status.Reason = rayv1.AppFailed
+					rayJobInstance.Status.Message = "Submitter completed but Ray job not found in RayCluster."
+					break
 				}
-				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, nil
 			}
+
 			logger.Error(err, "Failed to get job info", "JobId", rayJobInstance.Status.JobId)
 			return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 		}
@@ -937,7 +946,7 @@ func (r *RayJobReconciler) getOrCreateRayClusterInstance(ctx context.Context, ra
 			if err != nil {
 				return nil, err
 			}
-			if r.options.BatchSchedulerManager != nil {
+			if r.options.BatchSchedulerManager != nil && rayJobInstance.Spec.SubmissionMode == rayv1.K8sJobMode {
 				if scheduler, err := r.options.BatchSchedulerManager.GetScheduler(); err == nil {
 					// Group name is only used for individual pods to specify their task group ("headgroup", "worker-group-1", etc.).
 					// RayCluster contains multiple groups, so we pass an empty string.
@@ -1050,7 +1059,11 @@ func (r *RayJobReconciler) checkSubmitterAndUpdateStatusIfNeeded(ctx context.Con
 		}
 
 		if headPod == nil {
-			logger.Info("Ray head pod not found, skipping sidecar container status check")
+			// If head pod is deleted, mark the RayJob as failed
+			shouldUpdate = true
+			rayJob.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusFailed
+			rayJob.Status.Reason = rayv1.AppFailed
+			rayJob.Status.Message = "Ray head pod not found."
 			return
 		}