[RayJob] Don't fail a running Ray job when the submitter exits

ray-operator/controllers/ray/rayjob_controller.go

@@ -31,6 +31,7 @@ import (
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/metrics"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils/dashboardclient"
+	utiltypes "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils/types"
 	"github.com/ray-project/kuberay/ray-operator/pkg/features"
 )
 
@@ -283,10 +284,6 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 				return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 			}
 
-			if checkSubmitterFinishedTimeoutAndUpdateStatusIfNeeded(ctx, rayJobInstance, finishedAt) {
-				break
-			}
-
 			if shouldUpdate {
 				break
 			}
@@ -328,6 +325,10 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 		// Reset JobStatusCheckFailureStartTime when the job status check succeeds.
 		rayJobInstance.Status.JobStatusCheckFailureStartTime = nil
 
+		if r.checkSubmitterFinishedTimeoutAndUpdateStatusIfNeeded(ctx, rayJobInstance, rayDashboardClient, jobInfo, finishedAt) {
+			break
+		}
+
 		// If the JobStatus is in a terminal status, such as SUCCEEDED, FAILED, or STOPPED, it is impossible for the Ray job
 		// to transition to any other. Additionally, RayJob does not currently support retries. Hence, we can mark the RayJob
 		// as "Complete" or "Failed" to avoid unnecessary reconciliation.
@@ -1260,7 +1261,20 @@ func checkPreRunningDeadlineAndUpdateStatusIfNeeded(ctx context.Context, rayJob
 	return true
 }
 
-func checkSubmitterFinishedTimeoutAndUpdateStatusIfNeeded(ctx context.Context, rayJob *rayv1.RayJob, finishedAt *time.Time) bool {
+// rayJobDriverIsAlive reports whether the Ray node running this job's driver is still up.
+//
+// An active job status is only current while the node reporting it is alive. Once that node is
+// gone the dashboard keeps returning whatever the status last was — the frozen state #4091 added
+// the submitter grace period for. Asking the cluster which nodes are alive settles both cases
+// directly, without inferring anything from Pod lifecycles or from when the submitter exited.
+func rayJobDriverIsAlive(ctx context.Context, dashboardClient dashboardclient.RayDashboardClientInterface, jobInfo *utiltypes.RayJobInfo) (bool, error) {
+	if jobInfo == nil || jobInfo.DriverNodeID == "" {
+		return false, nil
+	}
+	return dashboardClient.IsNodeAlive(ctx, jobInfo.DriverNodeID)
+}
+
+func (r *RayJobReconciler) checkSubmitterFinishedTimeoutAndUpdateStatusIfNeeded(ctx context.Context, rayJob *rayv1.RayJob, dashboardClient dashboardclient.RayDashboardClientInterface, jobInfo *utiltypes.RayJobInfo, finishedAt *time.Time) bool {
 	logger := ctrl.LoggerFrom(ctx)
 
 	// Check if timeout is configured and submitter has finished
@@ -1273,6 +1287,39 @@ func checkSubmitterFinishedTimeoutAndUpdateStatusIfNeeded(ctx context.Context, r
 		return false
 	}
 
+	// The freshly polled status wins over the CR's copy: if the job has already reached a terminal
+	// state, the normal handling below marks the RayJob Complete or Failed and the timeout must not
+	// overwrite that.
+	jobStatus := rayJob.Status.JobStatus
+	if jobInfo != nil && jobInfo.JobStatus != "" {
+		jobStatus = jobInfo.JobStatus
+	}
+	if rayv1.IsJobTerminal(jobStatus) {
+		return false
+	}
+
+	// The submitter exiting is not evidence about the Ray job: `ray job logs --follow` returns 0
+	// whenever the log WebSocket closes with a non-abnormal code, so a submitter can finish under a
+	// perfectly healthy job. Trust a RUNNING status while the driver's node is still alive.
+	//
+	// Only RUNNING: Ray assigns driver_node_id when it schedules the job's supervisor, so a PENDING
+	// job has none and there is nothing to check it against. Those keep the existing behavior.
+	if jobStatus == rayv1.JobStatusRunning {
+		alive, err := rayJobDriverIsAlive(ctx, dashboardClient, jobInfo)
+		if err != nil {
+			// Inconclusive is not dead. Leave the job alone and let a later reconcile decide, rather
+			// than failing it on a dashboard blip or returning early and skipping the status update.
+			logger.Error(err, "Failed to check whether the Ray job's driver node is alive; leaving the RayJob running",
+				"DriverNodeID", jobInfo.DriverNodeID)
+			return false
+		}
+		if alive {
+			logger.Info("The RayJob submitter finished but the Ray job is still active on a live node; not transitioning to terminal.",
+				"SubmitterFinishedTime", finishedAt, "JobStatus", jobStatus, "DriverNodeID", jobInfo.DriverNodeID)
+			return false
+		}
+	}
+
 	logger.Info("The RayJob has passed the submitterFinishedTimeoutSeconds. Transition the status to terminal.",
 		"SubmitterFinishedTime", finishedAt,
 		"SubmitterFinishedTimeoutSeconds", DefaultSubmitterFinishedTimeout.String())

ray-operator/controllers/ray/rayjob_controller_unit_test.go

@@ -31,6 +31,7 @@ import (
 	schedulerinterface "github.com/ray-project/kuberay/ray-operator/controllers/ray/batchscheduler/interface"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/metrics/mocks"
 	utils "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
+	utiltypes "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils/types"
 	"github.com/ray-project/kuberay/ray-operator/pkg/client/clientset/versioned/scheme"
 	"github.com/ray-project/kuberay/ray-operator/pkg/features"
 )
@@ -1132,6 +1133,138 @@ func TestCheckJobStatusCheckTimeoutAndUpdateStatusIfNeeded(t *testing.T) {
 	assert.Contains(t, rayJob.Status.Message, "exceeded timeout of 1s")
 }
 
+func TestCheckSubmitterFinishedTimeoutAndUpdateStatusIfNeeded(t *testing.T) {
+	ctx := context.Background()
+	justFinished := time.Now()
+	expired := time.Now().Add(-DefaultSubmitterFinishedTimeout - time.Second)
+
+	const driverNode = "3bbbda71075d7929d056881d8921592b969874dd65ea3ee2d2067ca8"
+
+	tests := []struct {
+		finishedAt *time.Time
+		jobInfo    *utiltypes.RayJobInfo
+		name       string
+		jobStatus  rayv1.JobStatus
+		aliveNode  string
+		wantDeploy rayv1.JobDeploymentStatus
+		wantReason rayv1.JobFailedReason
+		wantUpdate bool
+		aliveErr   bool
+	}{
+		{
+			name:       "submitter still running",
+			jobStatus:  rayv1.JobStatusRunning,
+			finishedAt: nil,
+			jobInfo:    &utiltypes.RayJobInfo{DriverNodeID: driverNode},
+			aliveNode:  driverNode,
+		},
+		{
+			name:       "grace period not yet elapsed",
+			jobStatus:  rayv1.JobStatusNew,
+			finishedAt: &justFinished,
+			jobInfo:    &utiltypes.RayJobInfo{DriverNodeID: driverNode},
+			aliveNode:  driverNode,
+		},
+		{
+			name:       "job RUNNING on a live driver node is left alone",
+			jobStatus:  rayv1.JobStatusRunning,
+			finishedAt: &expired,
+			jobInfo:    &utiltypes.RayJobInfo{DriverNodeID: driverNode},
+			aliveNode:  driverNode,
+		},
+		{
+			// Ray assigns driver_node_id only once it schedules the supervisor, so a PENDING job has
+			// none to check. These keep the existing timeout behavior rather than guessing.
+			name:       "job PENDING has no driver node yet and times out",
+			jobStatus:  rayv1.JobStatusPending,
+			finishedAt: &expired,
+			jobInfo:    &utiltypes.RayJobInfo{JobStatus: rayv1.JobStatusPending},
+			wantUpdate: true,
+			wantDeploy: rayv1.JobDeploymentStatusFailed,
+			wantReason: rayv1.JobDeploymentStatusTransitionGracePeriodExceeded,
+		},
+		{
+			name:       "job RUNNING but the driver node is gone times out",
+			jobStatus:  rayv1.JobStatusRunning,
+			finishedAt: &expired,
+			jobInfo:    &utiltypes.RayJobInfo{DriverNodeID: driverNode},
+			aliveNode:  "some-other-node",
+			wantUpdate: true,
+			wantDeploy: rayv1.JobDeploymentStatusFailed,
+			wantReason: rayv1.JobDeploymentStatusTransitionGracePeriodExceeded,
+		},
+		{
+			name:       "job RUNNING with no driver node reported times out",
+			jobStatus:  rayv1.JobStatusRunning,
+			finishedAt: &expired,
+			jobInfo:    &utiltypes.RayJobInfo{},
+			wantUpdate: true,
+			wantDeploy: rayv1.JobDeploymentStatusFailed,
+			wantReason: rayv1.JobDeploymentStatusTransitionGracePeriodExceeded,
+		},
+		{
+			name:       "job never observed times out",
+			jobStatus:  rayv1.JobStatusNew,
+			finishedAt: &expired,
+			jobInfo:    &utiltypes.RayJobInfo{DriverNodeID: driverNode},
+			aliveNode:  driverNode,
+			wantUpdate: true,
+			wantDeploy: rayv1.JobDeploymentStatusFailed,
+			wantReason: rayv1.JobDeploymentStatusTransitionGracePeriodExceeded,
+		},
+		{
+			// The dashboard is the fresher source: a job that has already succeeded must not be
+			// overwritten with a grace-period failure just because the CR has not caught up.
+			name:       "fresh job info is terminal, CR is stale",
+			jobStatus:  rayv1.JobStatusRunning,
+			finishedAt: &expired,
+			jobInfo:    &utiltypes.RayJobInfo{JobStatus: rayv1.JobStatusSucceeded, DriverNodeID: driverNode},
+			aliveNode:  "some-other-node",
+		},
+		{
+			// A dashboard error is inconclusive, not proof of a dead driver, and must not fail the
+			// job or return early and skip the reconcile's status update.
+			name:       "node liveness errors leave the job alone",
+			jobStatus:  rayv1.JobStatusRunning,
+			finishedAt: &expired,
+			jobInfo:    &utiltypes.RayJobInfo{JobStatus: rayv1.JobStatusRunning, DriverNodeID: driverNode},
+			aliveErr:   true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			client := &nodeLivenessDashboardClient{aliveNode: tc.aliveNode, err: tc.aliveErr}
+			r := &RayJobReconciler{}
+			rayJob := &rayv1.RayJob{Status: rayv1.RayJobStatus{JobStatus: tc.jobStatus}}
+
+			got := r.checkSubmitterFinishedTimeoutAndUpdateStatusIfNeeded(ctx, rayJob, client, tc.jobInfo, tc.finishedAt)
+			assert.Equal(t, tc.wantUpdate, got)
+			if !tc.wantUpdate {
+				assert.Equal(t, tc.jobStatus, rayJob.Status.JobStatus, "job status must not be overwritten")
+				assert.Empty(t, rayJob.Status.Reason)
+				return
+			}
+			assert.Equal(t, tc.wantDeploy, rayJob.Status.JobDeploymentStatus)
+			assert.Equal(t, tc.wantReason, rayJob.Status.Reason)
+		})
+	}
+}
+
+// nodeLivenessDashboardClient answers IsNodeAlive for exactly one node.
+type nodeLivenessDashboardClient struct {
+	*utils.FakeRayDashboardClient
+	aliveNode string
+	err       bool
+}
+
+func (c *nodeLivenessDashboardClient) IsNodeAlive(_ context.Context, nodeID string) (bool, error) {
+	if c.err {
+		return false, errors.New("dashboard unreachable")
+	}
+	return nodeID != "" && nodeID == c.aliveNode, nil
+}
+
 // TestBatchSchedulerCleanupCalledWhenRayJobSuspending verifies that batch scheduler resources are
 // cleaned up when a RayJob is suspended/retrying. Unlike the terminal (Complete/Failed) path, the
 // suspend path must requeue on cleanup failure rather than swallow the error, otherwise the PodGroup

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

@@ -24,6 +24,9 @@ var (
 	DeployPathV2     = "/api/serve/applications/"
 	// Job URL paths
 	JobPath = "/api/jobs/"
+
+	// NodesPath reports the cluster's Ray nodes and their liveness.
+	NodesPath = "/nodes?view=summary"
 )
 
 type RayDashboardClientInterface interface {
@@ -32,6 +35,7 @@ type RayDashboardClientInterface interface {
 	GetServeDetails(ctx context.Context) (*utiltypes.ServeDetails, error)
 	GetMultiApplicationStatus(context.Context) (map[string]*utiltypes.ServeApplicationStatus, error)
 	GetJobInfo(ctx context.Context, jobId string) (*utiltypes.RayJobInfo, error)
+	IsNodeAlive(ctx context.Context, nodeID string) (bool, error)
 	ListJobs(ctx context.Context) (*[]utiltypes.RayJobInfo, error)
 	SubmitJob(ctx context.Context, rayJob *rayv1.RayJob) (string, error)
 	SubmitJobReq(ctx context.Context, request *utiltypes.RayJobRequest) (string, error)
@@ -151,6 +155,52 @@ func (r *RayDashboardClient) ConvertServeDetailsToApplicationStatuses(serveDetai
 
 // Note that RayJobInfo and error can't be nil at the same time.
 // Please make sure if the Ray job with JobId can't be found. Return a BadRequest error.
+// IsNodeAlive reports whether the given Ray node is currently ALIVE in the cluster.
+//
+// A job status read from the dashboard is only current while the node running its driver is up.
+// Once that node is gone the status is frozen at whatever it last was, so an active JobStatus on a
+// dead node must not be trusted.
+func (r *RayDashboardClient) IsNodeAlive(ctx context.Context, nodeID string) (bool, error) {
+	if nodeID == "" {
+		return false, nil
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.dashboardURL+NodesPath, nil)
+	if err != nil {
+		return false, err
+	}
+	r.setAuthHeader(req)
+
+	resp, err := r.client.Do(req)
+	if err != nil {
+		return false, err
+	}
+	defer resp.Body.Close()
+
+	// A non-2xx body often still parses as JSON with an empty summary, which would read as a dead
+	// driver and fail a healthy job. Reject it as an error so the caller retries instead.
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return false, fmt.Errorf("listing nodes returned status %d", resp.StatusCode)
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return false, fmt.Errorf("failed to read response when listing nodes: %w", err)
+	}
+
+	var nodes utiltypes.RayNodesSummaryResponse
+	if err = json.Unmarshal(body, &nodes); err != nil {
+		return false, fmt.Errorf("IsNodeAlive fail: %s", string(body))
+	}
+
+	for _, node := range nodes.Data.Summary {
+		if node.Raylet.NodeID == nodeID {
+			return node.Raylet.State == "ALIVE", nil
+		}
+	}
+	return false, nil
+}
+
 func (r *RayDashboardClient) GetJobInfo(ctx context.Context, jobId string) (*utiltypes.RayJobInfo, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.dashboardURL+JobPath+jobId, nil)
 	if err != nil {

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

@@ -41,6 +41,10 @@ func (r *FakeRayDashboardClient) SetMultiApplicationStatuses(statuses map[string
 	r.multiAppStatuses = statuses
 }
 
+func (r *FakeRayDashboardClient) IsNodeAlive(_ context.Context, nodeID string) (bool, error) {
+	return nodeID != "", nil
+}
+
 func (r *FakeRayDashboardClient) GetJobInfo(ctx context.Context, jobId string) (*utiltypes.RayJobInfo, error) {
 	if mock := r.GetJobInfoMock.Load(); mock != nil {
 		return (*mock)(ctx, jobId)

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

@@ -17,9 +17,12 @@ type RayJobInfo struct {
 	Entrypoint   string            `json:"entrypoint,omitempty"`
 	JobId        string            `json:"job_id,omitempty"`
 	SubmissionId string            `json:"submission_id,omitempty"`
-	Message      string            `json:"message,omitempty"`
-	StartTime    uint64            `json:"start_time,omitempty"`
-	EndTime      uint64            `json:"end_time,omitempty"`
+	// DriverNodeID is the Ray node the job's driver runs on. Checking it against the cluster's live
+	// nodes tells whether an active JobStatus is current or frozen behind a node that is gone.
+	DriverNodeID string `json:"driver_node_id,omitempty"`
+	Message      string `json:"message,omitempty"`
+	StartTime    uint64 `json:"start_time,omitempty"`
+	EndTime      uint64 `json:"end_time,omitempty"`
 }
 
 // RayJobRequest is the request body to submit.
@@ -46,3 +49,18 @@ type RayJobStopResponse struct {
 type RayJobLogsResponse struct {
 	Logs string `json:"logs,omitempty"`
 }
+
+// RayNodesSummaryResponse is the subset of the dashboard's /nodes response used to check whether a
+// Ray node is still alive.
+type RayNodesSummaryResponse struct {
+	Data struct {
+		Summary []RayNodeSummary `json:"summary,omitempty"`
+	} `json:"data,omitempty"`
+}
+
+type RayNodeSummary struct {
+	Raylet struct {
+		NodeID string `json:"nodeId,omitempty"`
+		State  string `json:"state,omitempty"`
+	} `json:"raylet,omitempty"`
+}