[RayJob][Status][4/n] Remove some JobDeploymentStatus and updateState function calls (#1743)

Author: kevin85421

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

@@ -33,14 +33,10 @@ func IsJobTerminal(status JobStatus) bool {
 type JobDeploymentStatus string
 
 const (
-	JobDeploymentStatusInitializing                  JobDeploymentStatus = "Initializing"
-	JobDeploymentStatusFailedToGetOrCreateRayCluster JobDeploymentStatus = "FailedToGetOrCreateRayCluster"
-	JobDeploymentStatusWaitForK8sJob                 JobDeploymentStatus = "WaitForK8sJob"
-	JobDeploymentStatusFailedJobDeploy               JobDeploymentStatus = "FailedJobDeploy"
-	JobDeploymentStatusRunning                       JobDeploymentStatus = "Running"
-	JobDeploymentStatusFailedToGetJobStatus          JobDeploymentStatus = "FailedToGetJobStatus"
-	JobDeploymentStatusComplete                      JobDeploymentStatus = "Complete"
-	JobDeploymentStatusSuspended                     JobDeploymentStatus = "Suspended"
+	JobDeploymentStatusInitializing JobDeploymentStatus = "Initializing"
+	JobDeploymentStatusRunning      JobDeploymentStatus = "Running"
+	JobDeploymentStatusComplete     JobDeploymentStatus = "Complete"
+	JobDeploymentStatusSuspended    JobDeploymentStatus = "Suspended"
 )
 
 // RayJobSpec defines the desired state of RayJob

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

@@ -33,14 +33,10 @@ func IsJobTerminal(status JobStatus) bool {
 type JobDeploymentStatus string
 
 const (
-	JobDeploymentStatusInitializing                  JobDeploymentStatus = "Initializing"
-	JobDeploymentStatusFailedToGetOrCreateRayCluster JobDeploymentStatus = "FailedToGetOrCreateRayCluster"
-	JobDeploymentStatusWaitForK8sJob                 JobDeploymentStatus = "WaitForK8sJob"
-	JobDeploymentStatusFailedJobDeploy               JobDeploymentStatus = "FailedJobDeploy"
-	JobDeploymentStatusRunning                       JobDeploymentStatus = "Running"
-	JobDeploymentStatusFailedToGetJobStatus          JobDeploymentStatus = "FailedToGetJobStatus"
-	JobDeploymentStatusComplete                      JobDeploymentStatus = "Complete"
-	JobDeploymentStatusSuspended                     JobDeploymentStatus = "Suspended"
+	JobDeploymentStatusInitializing JobDeploymentStatus = "Initializing"
+	JobDeploymentStatusRunning      JobDeploymentStatus = "Running"
+	JobDeploymentStatusComplete     JobDeploymentStatus = "Complete"
+	JobDeploymentStatusSuspended    JobDeploymentStatus = "Suspended"
 )
 
 // RayJobSpec defines the desired state of RayJob

ray-operator/controllers/ray/rayjob_controller.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
-	fmtErrors "github.com/pkg/errors"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -146,14 +145,14 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 			if !errors.IsNotFound(err) {
 				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 			}
-			if err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusComplete, nil); err != nil {
+			if err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusComplete); err != nil {
 				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 			}
 			return ctrl.Result{}, nil
 		}
 
 		if rayClusterInstance.DeletionTimestamp != nil {
-			if err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusComplete, nil); err != nil {
+			if err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusComplete); err != nil {
 				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 			}
 			return ctrl.Result{}, nil
@@ -168,7 +167,6 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 
 	var rayClusterInstance *rayv1.RayCluster
 	if rayClusterInstance, err = r.getOrCreateRayClusterInstance(ctx, rayJobInstance); err != nil {
-		err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusFailedToGetOrCreateRayCluster, err)
 		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 	}
 	// If there is no cluster instance and no error suspend the job deployment
@@ -177,7 +175,7 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 		if rayJobInstance.Status.JobDeploymentStatus == rayv1.JobDeploymentStatusSuspended {
 			return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 		}
-		err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusSuspended, err)
+		err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusSuspended)
 		if err != nil {
 			return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 		}
@@ -216,7 +214,6 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 	// Ensure k8s job has been created
 	jobName, wasJobCreated, err := r.getOrCreateK8sJob(ctx, rayJobInstance, rayClusterInstance)
 	if err != nil {
-		err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusFailedJobDeploy, err)
 		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 	}
 
@@ -234,27 +231,29 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			r.Log.Info("Job not found", "RayJob", rayJobInstance.Name, "jobId", jobName)
-			err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusWaitForK8sJob, err)
 			return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 		}
 		r.Log.Error(err, "failed to get k8s job")
-		err = r.updateState(ctx, rayJobInstance, nil, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusFailedToGetJobStatus, err)
 		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 	}
 
 	// Check the current status of ray jobs
 	jobInfo, err := rayDashboardClient.GetJobInfo(ctx, rayJobInstance.Status.JobId)
 	if err != nil {
 		r.Log.Error(err, "failed to get job info", "jobId", rayJobInstance.Status.JobId)
-		err = r.updateState(ctx, rayJobInstance, jobInfo, rayJobInstance.Status.JobStatus, rayv1.JobDeploymentStatusFailedToGetJobStatus, err)
 		// Dashboard service in head pod takes time to start, it's possible we get connection refused error.
 		// Requeue after few seconds to avoid continuous connection errors.
 		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 	}
 
-	// Update RayJob.Status (Kubernetes CR) from Ray Job Status from Dashboard service
-	if r.shouldUpdateJobStatus(rayJobInstance.Status.JobStatus, rayJobInstance.Status.JobDeploymentStatus, jobInfo) {
-		err = r.updateState(ctx, rayJobInstance, jobInfo, jobInfo.JobStatus, rayv1.JobDeploymentStatusRunning, nil)
+	if jobInfo != nil {
+		// TODO (kevin85421): `GetJobInfo` should not return both JobInfo and error with nil values,
+		// but it does when the job is not found. This check is a workaround to avoid dereferencing
+		// a nil pointer.
+		err = r.updateState(ctx, rayJobInstance, jobInfo, jobInfo.JobStatus, rayv1.JobDeploymentStatusRunning)
+	}
+
+	if err != nil {
 		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 	}
 
@@ -288,7 +287,7 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 			rayJobInstance.Status.DashboardURL = ""
 			rayJobInstance.Status.JobId = ""
 			rayJobInstance.Status.Message = ""
-			err = r.updateState(ctx, rayJobInstance, jobInfo, rayv1.JobStatusStopped, rayv1.JobDeploymentStatusSuspended, nil)
+			err = r.updateState(ctx, rayJobInstance, jobInfo, rayv1.JobStatusStopped, rayv1.JobDeploymentStatusSuspended)
 			if err != nil {
 				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 			}
@@ -520,22 +519,11 @@ func (r *RayJobReconciler) initRayJobStatusIfNeed(ctx context.Context, rayJob *r
 		rayJob.Status.JobStatus = rayv1.JobStatusPending
 	}
 
-	return r.updateState(ctx, rayJob, nil, rayJob.Status.JobStatus, rayv1.JobDeploymentStatusInitializing, nil)
-}
-
-func (r *RayJobReconciler) shouldUpdateJobStatus(oldJobStatus rayv1.JobStatus, oldJobDeploymentStatus rayv1.JobDeploymentStatus, jobInfo *utils.RayJobInfo) bool {
-	if jobInfo != nil {
-		jobStatusChanged := (oldJobStatus != jobInfo.JobStatus)
-		// If the status changed, or if we didn't have the status before and now we have it, update the status and deployment status.
-		if jobStatusChanged || oldJobDeploymentStatus == rayv1.JobDeploymentStatusFailedToGetJobStatus {
-			return true
-		}
-	}
-	return false
+	return r.updateState(ctx, rayJob, nil, rayJob.Status.JobStatus, rayv1.JobDeploymentStatusInitializing)
 }
 
 // make sure the priority is correct
-func (r *RayJobReconciler) updateState(ctx context.Context, rayJob *rayv1.RayJob, jobInfo *utils.RayJobInfo, jobStatus rayv1.JobStatus, jobDeploymentStatus rayv1.JobDeploymentStatus, err error) error {
+func (r *RayJobReconciler) updateState(ctx context.Context, rayJob *rayv1.RayJob, jobInfo *utils.RayJobInfo, jobStatus rayv1.JobStatus, jobDeploymentStatus rayv1.JobDeploymentStatus) error {
 	r.Log.Info("UpdateState", "oldJobStatus", rayJob.Status.JobStatus, "newJobStatus", jobStatus, "oldJobDeploymentStatus", rayJob.Status.JobDeploymentStatus, "newJobDeploymentStatus", jobDeploymentStatus)
 
 	// Let's skip update the APIServer if it's synced.
@@ -558,10 +546,10 @@ func (r *RayJobReconciler) updateState(ctx context.Context, rayJob *rayv1.RayJob
 	// TODO (kevin85421): ObservedGeneration should be used to determine whether update this CR or not.
 	rayJob.Status.ObservedGeneration = rayJob.ObjectMeta.Generation
 
-	if errStatus := r.Status().Update(ctx, rayJob); errStatus != nil {
-		return fmtErrors.Errorf("combined error: %v %v", err, errStatus)
+	if err := r.Status().Update(ctx, rayJob); err != nil {
+		return err
 	}
-	return err
+	return nil
 }
 
 // TODO: select existing rayclusters by ClusterSelector
@@ -645,11 +633,6 @@ func (r *RayJobReconciler) getOrCreateRayClusterInstance(ctx context.Context, ra
 			return nil, err
 		}
 
-		// special case: is the job is complete status and cluster has been recycled.
-		if isJobSucceedOrFail(rayJobInstance.Status.JobStatus) && rayJobInstance.Status.JobDeploymentStatus == rayv1.JobDeploymentStatusComplete {
-			r.Log.Info("The cluster has been recycled for the job, skip duplicate creation", "rayjob", rayJobInstance.Name)
-			return nil, err
-		}
 		// special case: don't create a cluster instance and don't return an error if the suspend flag of the job is true
 		if rayJobInstance.Spec.Suspend {
 			return nil, nil

ray-operator/controllers/ray/rayjob_controller_unit_test.go

@@ -176,57 +176,3 @@ func TestGetSubmitterTemplate(t *testing.T) {
 	}
 	assert.True(t, found)
 }
-
-func TestShouldUpdateJobStatus(t *testing.T) {
-	r := &RayJobReconciler{}
-
-	tests := []struct {
-		name                   string
-		oldJobStatus           rayv1.JobStatus
-		oldJobDeploymentStatus rayv1.JobDeploymentStatus
-		jobInfo                *utils.RayJobInfo
-		expectedShouldUpdate   bool
-	}{
-		{
-			name:                   "jobInfo is nil",
-			oldJobStatus:           rayv1.JobStatusPending,
-			oldJobDeploymentStatus: rayv1.JobDeploymentStatusRunning,
-			jobInfo:                nil,
-			expectedShouldUpdate:   false,
-		},
-		{
-			name:                   "job status changed",
-			oldJobStatus:           rayv1.JobStatusRunning,
-			oldJobDeploymentStatus: rayv1.JobDeploymentStatusRunning,
-			jobInfo: &utils.RayJobInfo{
-				JobStatus: rayv1.JobStatusStopped,
-			},
-			expectedShouldUpdate: true,
-		},
-		{
-			name:                   "job status same but JobDeploymentStatus failed to get status",
-			oldJobStatus:           rayv1.JobStatusRunning,
-			oldJobDeploymentStatus: rayv1.JobDeploymentStatusFailedToGetJobStatus,
-			jobInfo: &utils.RayJobInfo{
-				JobStatus: rayv1.JobStatusRunning,
-			},
-			expectedShouldUpdate: true,
-		},
-		{
-			name:                   "job status same and JobDeploymentStatus not failed to get status",
-			oldJobStatus:           rayv1.JobStatusRunning,
-			oldJobDeploymentStatus: rayv1.JobDeploymentStatusRunning,
-			jobInfo: &utils.RayJobInfo{
-				JobStatus: rayv1.JobStatusRunning,
-			},
-			expectedShouldUpdate: false,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result := r.shouldUpdateJobStatus(tt.oldJobStatus, tt.oldJobDeploymentStatus, tt.jobInfo)
-			assert.Equal(t, tt.expectedShouldUpdate, result)
-		})
-	}
-}

ray-operator/controllers/ray/utils/dashboard_httpclient.go

@@ -481,7 +481,7 @@ func (r *RayDashboardClient) GetJobLog(ctx context.Context, jobName string, log
 	var jobLog RayJobLogsResponse
 	if err = json.Unmarshal(body, &jobLog); err != nil {
 		// Maybe body is not valid json, raise an error with the body.
-		return nil, fmt.Errorf("GetJobInfo fail: %s", string(body))
+		return nil, fmt.Errorf("GetJobLog fail: %s", string(body))
 	}
 
 	return &jobLog.Logs, nil