[Feat][RayJob] UserMode SubmissionMode (#2364)

Author: MortalHappiness

helm-chart/kuberay-operator/crds/ray.io_rayjobs.yaml

@@ -42,6 +42,7 @@ const (
 	JobDeploymentStatusSuspending   JobDeploymentStatus = "Suspending"
 	JobDeploymentStatusSuspended    JobDeploymentStatus = "Suspended"
 	JobDeploymentStatusRetrying     JobDeploymentStatus = "Retrying"
+	JobDeploymentStatusWaiting      JobDeploymentStatus = "Waiting"
 )
 
 // JobFailedReason indicates the reason the RayJob changes its JobDeploymentStatus to 'Failed'
@@ -58,6 +59,7 @@ type JobSubmissionMode string
 const (
 	K8sJobMode JobSubmissionMode = "K8sJobMode" // Submit job via Kubernetes Job
 	HTTPMode   JobSubmissionMode = "HTTPMode"   // Submit job via HTTP request
+	UserMode   JobSubmissionMode = "UserMode"   // Don't submit job in KubeRay. Instead, wait for user to submit job and provide the job submission ID
 )
 
 type SubmitterConfig struct {
@@ -86,7 +88,7 @@ type RayJobSpec struct {
 	SubmitterConfig *SubmitterConfig `json:"submitterConfig,omitempty"`
 	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
 	// Important: Run "make" to regenerate code after modifying this file
-	Entrypoint string `json:"entrypoint"`
+	Entrypoint string `json:"entrypoint,omitempty"`
 	// RuntimeEnvYAML represents the runtime environment configuration
 	// provided as a multi-line YAML string.
 	RuntimeEnvYAML string `json:"runtimeEnvYAML,omitempty"`

ray-operator/apis/ray/v1/rayjob_types.go

@@ -129,6 +129,11 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 	}
 
+	if err := validateRayJobStatus(rayJobInstance); err != nil {
+		logger.Error(err, "The RayJob status is invalid")
+		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
+	}
+
 	// Please do NOT modify `originalRayJobInstance` in the following code.
 	originalRayJobInstance := rayJobInstance.DeepCopy()
 
@@ -177,15 +182,30 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 			rayJobInstance.Status.DashboardURL = clientURL
 		}
 
+		if rayJobInstance.Spec.SubmissionMode == rayv1.UserMode {
+			logger.Info("SubmissionMode is UserMode and the RayCluster is created. Transition the status from `Initializing` to `Waiting`.")
+			rayJobInstance.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusWaiting
+			break
+		}
+
 		if rayJobInstance.Spec.SubmissionMode == rayv1.K8sJobMode {
 			if err := r.createK8sJobIfNeed(ctx, rayJobInstance, rayClusterInstance); err != nil {
 				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 			}
 		}
 
-		logger.Info("Both RayCluster and the submitter K8s Job are created. Transition the status from `Initializing` to `Running`.",
+		logger.Info("Both RayCluster and the submitter K8s Job are created. Transition the status from `Initializing` to `Running`.", "SubmissionMode", rayJobInstance.Spec.SubmissionMode,
 			"RayJob", rayJobInstance.Name, "RayCluster", rayJobInstance.Status.RayClusterName)
 		rayJobInstance.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusRunning
+	case rayv1.JobDeploymentStatusWaiting:
+		// Try to get the Ray job id from the Ray job annotations.
+		rayJobId, found := rayJobInstance.ObjectMeta.Annotations[utils.RayJobSubmissionIdLabelKey]
+		logger.Info("Get Ray job id from the Ray job annotations", "RayJobId", rayJobId, "Found", found)
+		if !found {
+			return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, nil
+		}
+		rayJobInstance.Status.JobId = rayJobId
+		rayJobInstance.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusRunning
 	case rayv1.JobDeploymentStatusRunning:
 		if shouldUpdate := r.updateStatusToSuspendingIfNeeded(ctx, rayJobInstance); shouldUpdate {
 			break
@@ -606,6 +626,7 @@ func (r *RayJobReconciler) SetupWithManager(mgr ctrl.Manager, reconcileConcurren
 // This function is the sole place where `JobDeploymentStatusInitializing` is defined. It initializes `Status.JobId` and `Status.RayClusterName`
 // prior to job submissions and RayCluster creations. This is used to avoid duplicate job submissions and cluster creations. In addition, this
 // function also sets `Status.StartTime` to support `ActiveDeadlineSeconds`.
+// This function will set or generate JobId if SubmissionMode is not UserMode.
 func (r *RayJobReconciler) initRayJobStatusIfNeed(ctx context.Context, rayJob *rayv1.RayJob) error {
 	logger := ctrl.LoggerFrom(ctx)
 	shouldUpdateStatus := rayJob.Status.JobId == "" || rayJob.Status.RayClusterName == "" || rayJob.Status.JobStatus == ""
@@ -615,7 +636,7 @@ func (r *RayJobReconciler) initRayJobStatusIfNeed(ctx context.Context, rayJob *r
 		return nil
 	}
 
-	if rayJob.Status.JobId == "" {
+	if rayJob.Spec.SubmissionMode != rayv1.UserMode && rayJob.Status.JobId == "" {
 		if rayJob.Spec.JobId != "" {
 			rayJob.Status.JobId = rayJob.Spec.JobId
 		} else {
@@ -811,3 +832,11 @@ func validateRayJobSpec(rayJob *rayv1.RayJob) error {
 	}
 	return nil
 }
+
+func validateRayJobStatus(rayJob *rayv1.RayJob) error {
+	if rayJob.Status.JobDeploymentStatus == rayv1.JobDeploymentStatusWaiting && rayJob.Spec.SubmissionMode != rayv1.UserMode {
+		return fmt.Errorf("invalid RayJob State: JobDeploymentStatus cannot be `Waiting` when SubmissionMode is not UserMode")
+	}
+
+	return nil
+}