diff --git a/ray-operator/controllers/ray/rayjob_controller.go b/ray-operator/controllers/ray/rayjob_controller.go index ac13b360c4d..6487b730aa2 100644 --- a/ray-operator/controllers/ray/rayjob_controller.go +++ b/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. @@ -1059,6 +1060,16 @@ func updateStatusToSuspendingIfNeeded(ctx context.Context, rayJob *rayv1.RayJob) return true } +// rayJobIsRunning reports whether the Ray job has been observed running. +// +// A submitter that dies while the job is running proves the submission itself succeeded, so failing +// the RayJob with SubmissionFailed would be doubly wrong: it discards a job that is still going +// (#2314) and reports a submission failure that did not happen. Leave those to the submitter-finished +// timeout, which decides on live cluster state once the grace period elapses. +func rayJobIsRunning(rayJob *rayv1.RayJob) bool { + return rayJob.Status.JobStatus == rayv1.JobStatusRunning +} + func (r *RayJobReconciler) checkSubmitterAndUpdateStatusIfNeeded(ctx context.Context, rayJob *rayv1.RayJob) (shouldUpdate bool, finishedAt *time.Time, err error) { logger := ctrl.LoggerFrom(ctx) shouldUpdate = false @@ -1097,6 +1108,7 @@ func (r *RayJobReconciler) checkSubmitterAndUpdateStatusIfNeeded(ctx context.Con // so a terminated container is transient — not a permanent failure. if !features.Enabled(features.SidecarSubmitterRestart) { shouldUpdate, submitterContainerStatus = checkSidecarContainerStatus(headPod) + shouldUpdate = shouldUpdate && !rayJobIsRunning(rayJob) if shouldUpdate { logger.Info("The submitter sidecar container has failed. Attempting to transition the status to `Failed`.", "Submitter sidecar container", submitterContainerStatus.Name, "Reason", submitterContainerStatus.State.Terminated.Reason, "Message", submitterContainerStatus.State.Terminated.Message) @@ -1118,6 +1130,7 @@ func (r *RayJobReconciler) checkSubmitterAndUpdateStatusIfNeeded(ctx context.Con submitterBackoffLimit = *rayJob.Spec.SubmitterConfig.BackoffLimit } shouldUpdate, submitterContainerStatus = checkIsRestartCountExceeded(headPod, submitterBackoffLimit) + shouldUpdate = shouldUpdate && !rayJobIsRunning(rayJob) if shouldUpdate { logger.Info("The submitter sidecar container has exceeded the max restart count. Attempting to transition the status to `Failed`.", "Submitter sidecar container", submitterContainerStatus.Name, @@ -1149,6 +1162,7 @@ func (r *RayJobReconciler) checkSubmitterAndUpdateStatusIfNeeded(ctx context.Con } shouldUpdate, condition = checkK8sJobStatus(job) + shouldUpdate = shouldUpdate && !rayJobIsRunning(rayJob) if shouldUpdate { logger.Info("The submitter Kubernetes Job has failed. Attempting to transition the status to `Failed`.", "Submitter K8s Job", job.Name, "Reason", condition.Reason, "Message", condition.Message) @@ -1260,7 +1274,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 +1300,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()) diff --git a/ray-operator/controllers/ray/rayjob_controller_unit_test.go b/ray-operator/controllers/ray/rayjob_controller_unit_test.go index b6cd9dfee6f..a7172841daf 100644 --- a/ray-operator/controllers/ray/rayjob_controller_unit_test.go +++ b/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 @@ -1272,3 +1405,175 @@ func TestBatchSchedulerCleanupCalledWhenRayJobSuspendingOrRetrying(t *testing.T) }) } } + +// TestCheckSubmitterAndUpdateStatusIfNeeded_K8sJobMode covers a submitter Kubernetes Job failing +// underneath a job that is still running: an infrastructure-killed submitter exhausts its backoff +// and fails the Job, but the driver runs on the cluster and carries on (#2314). +func TestCheckSubmitterAndUpdateStatusIfNeeded_SidecarMode(t *testing.T) { + newRayJob := func(jobStatus rayv1.JobStatus) *rayv1.RayJob { + return &rayv1.RayJob{ + ObjectMeta: metav1.ObjectMeta{Name: "rayjob-sample", Namespace: "default"}, + Spec: rayv1.RayJobSpec{ + SubmissionMode: rayv1.SidecarMode, + RayClusterSpec: &rayv1.RayClusterSpec{}, + }, + Status: rayv1.RayJobStatus{JobStatus: jobStatus, RayClusterName: "raycluster-sample"}, + } + } + + rayCluster := &rayv1.RayCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "raycluster-sample", Namespace: "default"}, + } + + headPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "head-pod", + Namespace: "default", + Labels: map[string]string{ + utils.RayClusterLabelKey: "raycluster-sample", + utils.RayNodeTypeLabelKey: string(rayv1.HeadNode), + }, + }, + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{{ + Name: utils.SubmitterContainerName, + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + ExitCode: 1, + Reason: "Error", + FinishedAt: metav1.NewTime(time.Now()), + }}, + }}, + }, + } + + tests := []struct { + name string + jobStatus rayv1.JobStatus + wantReason rayv1.JobFailedReason + wantUpdate bool + }{ + { + // Same rule as K8sJobMode: a sidecar that exits non-zero while the job is + // running is not evidence about the job, so the timeout decides it instead. + name: "job running is left to the submitter-finished timeout", + jobStatus: rayv1.JobStatusRunning, + }, + { + name: "job never observed is a genuine submission failure", + jobStatus: rayv1.JobStatusNew, + wantUpdate: true, + wantReason: rayv1.SubmissionFailed, + }, + { + name: "job already failed keeps the application failure reason", + jobStatus: rayv1.JobStatusFailed, + wantUpdate: true, + wantReason: rayv1.AppFailed, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + newScheme := runtime.NewScheme() + require.NoError(t, rayv1.AddToScheme(newScheme)) + require.NoError(t, corev1.AddToScheme(newScheme)) + require.NoError(t, batchv1.AddToScheme(newScheme)) + + rayJob := newRayJob(tc.jobStatus) + fakeClient := clientFake.NewClientBuilder().WithScheme(newScheme). + WithRuntimeObjects(rayJob, rayCluster, headPod).Build() + r := &RayJobReconciler{Client: fakeClient, Scheme: newScheme} + + shouldUpdate, _, err := r.checkSubmitterAndUpdateStatusIfNeeded(context.Background(), rayJob) + require.NoError(t, err) + assert.Equal(t, tc.wantUpdate, shouldUpdate) + if !tc.wantUpdate { + assert.Equal(t, tc.jobStatus, rayJob.Status.JobStatus) + assert.Empty(t, rayJob.Status.JobDeploymentStatus) + assert.Empty(t, rayJob.Status.Reason) + return + } + assert.Equal(t, rayv1.JobDeploymentStatusFailed, rayJob.Status.JobDeploymentStatus) + assert.Equal(t, tc.wantReason, rayJob.Status.Reason) + }) + } +} + +func TestCheckSubmitterAndUpdateStatusIfNeeded_K8sJobMode(t *testing.T) { + failedAt := metav1.NewTime(time.Now()) + + newRayJob := func(jobStatus rayv1.JobStatus) *rayv1.RayJob { + return &rayv1.RayJob{ + ObjectMeta: metav1.ObjectMeta{Name: "rayjob-sample", Namespace: "default"}, + Spec: rayv1.RayJobSpec{SubmissionMode: rayv1.K8sJobMode}, + Status: rayv1.RayJobStatus{JobStatus: jobStatus, RayClusterName: "raycluster-sample"}, + } + } + + failedSubmitterJob := func(rayJob *rayv1.RayJob) *batchv1.Job { + return &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{Name: rayJob.Name, Namespace: rayJob.Namespace}, + Status: batchv1.JobStatus{Conditions: []batchv1.JobCondition{{ + Type: batchv1.JobFailed, + Status: corev1.ConditionTrue, + Reason: "BackoffLimitExceeded", + Message: "Job has reached the specified backoff limit", + LastTransitionTime: failedAt, + }}}, + } + } + + tests := []struct { + name string + jobStatus rayv1.JobStatus + wantReason rayv1.JobFailedReason + wantUpdate bool + }{ + { + // The submission plainly succeeded, so SubmissionFailed would be wrong. The + // submitter-finished timeout decides this one on live cluster state instead. + name: "job running is left to the submitter-finished timeout", + jobStatus: rayv1.JobStatusRunning, + }, + { + name: "job never observed is a genuine submission failure", + jobStatus: rayv1.JobStatusNew, + wantUpdate: true, + wantReason: rayv1.SubmissionFailed, + }, + { + name: "job already failed keeps the application failure reason", + jobStatus: rayv1.JobStatusFailed, + wantUpdate: true, + wantReason: rayv1.AppFailed, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + newScheme := runtime.NewScheme() + require.NoError(t, rayv1.AddToScheme(newScheme)) + require.NoError(t, corev1.AddToScheme(newScheme)) + require.NoError(t, batchv1.AddToScheme(newScheme)) + + rayJob := newRayJob(tc.jobStatus) + fakeClient := clientFake.NewClientBuilder().WithScheme(newScheme). + WithRuntimeObjects(rayJob, failedSubmitterJob(rayJob)).Build() + r := &RayJobReconciler{Client: fakeClient, Scheme: newScheme} + + shouldUpdate, finishedAt, err := r.checkSubmitterAndUpdateStatusIfNeeded(context.Background(), rayJob) + require.NoError(t, err) + // A failed submitter Job is terminal, so the timeout path always gets a reference point. + assert.NotNil(t, finishedAt) + assert.Equal(t, tc.wantUpdate, shouldUpdate) + if !tc.wantUpdate { + assert.Equal(t, tc.jobStatus, rayJob.Status.JobStatus) + assert.Empty(t, rayJob.Status.JobDeploymentStatus) + assert.Empty(t, rayJob.Status.Reason) + return + } + assert.Equal(t, rayv1.JobDeploymentStatusFailed, rayJob.Status.JobDeploymentStatus) + assert.Equal(t, tc.wantReason, rayJob.Status.Reason) + }) + } +} diff --git a/ray-operator/controllers/ray/utils/dashboardclient/dashboard_httpclient.go b/ray-operator/controllers/ray/utils/dashboardclient/dashboard_httpclient.go index aecb32adc64..3e72be21dc4 100644 --- a/ray-operator/controllers/ray/utils/dashboardclient/dashboard_httpclient.go +++ b/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 { diff --git a/ray-operator/controllers/ray/utils/fake_serve_httpclient.go b/ray-operator/controllers/ray/utils/fake_serve_httpclient.go index 35d15bdf07c..31142eafbc5 100644 --- a/ray-operator/controllers/ray/utils/fake_serve_httpclient.go +++ b/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) diff --git a/ray-operator/controllers/ray/utils/types/dashboard_httpclient.go b/ray-operator/controllers/ray/utils/types/dashboard_httpclient.go index 95a3c73f015..94eadf36203 100644 --- a/ray-operator/controllers/ray/utils/types/dashboard_httpclient.go +++ b/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"` +} diff --git a/ray-operator/test/e2erayjob/rayjob_test.go b/ray-operator/test/e2erayjob/rayjob_test.go index b862439129c..1273043fcac 100644 --- a/ray-operator/test/e2erayjob/rayjob_test.go +++ b/ray-operator/test/e2erayjob/rayjob_test.go @@ -6,7 +6,6 @@ import ( "time" . "github.com/onsi/gomega" - "github.com/stretchr/testify/assert" batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" k8serrors "k8s.io/apimachinery/pkg/api/errors" @@ -373,7 +372,7 @@ env_vars: LogWithTimestamp(test.T(), "Deleted RayJob %s/%s successfully", *rayJobAC.Namespace, *rayJobAC.Name) }) - test.T().Run("RayJob has exceed SubmitterFinishedTimeout", func(_ *testing.T) { + test.T().Run("RayJob survives the submitter finishing while the job is still running", func(_ *testing.T) { rayJobAC := rayv1ac.RayJob("submitter-timeout", namespace.Name). WithSpec(rayv1ac.RayJobSpec(). WithRayClusterSpec(NewRayClusterSpec(MountConfigMap[rayv1ac.RayClusterSpecApplyConfiguration](jobs, "/home/ray/jobs"))). @@ -430,28 +429,16 @@ env_vars: g.Expect(err).NotTo(HaveOccurred()) LogWithTimestamp(test.T(), "Successfully marked submitter job as completed at %v", now.Time) - // Record the start time for timeout measurement - timeoutStartTime := time.Now() - - // Wait for the timeout to trigger - LogWithTimestamp(test.T(), "Waiting for RayJob %s/%s to exceed SubmitterFinishedTimeout", rayJob.Namespace, rayJob.Name) - g.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutShort). - Should(WithTransform(RayJobDeploymentStatus, Equal(rayv1.JobDeploymentStatusFailed))) - - // Measure the actual timeout duration and verify the timeout duration is close to DefaultSubmitterFinishedTimeout - actualTimeoutDuration := time.Since(timeoutStartTime) - expectedTimeout := ray.DefaultSubmitterFinishedTimeout - assert.InDelta(test.T(), expectedTimeout.Seconds(), actualTimeoutDuration.Seconds(), 5.0, - "Actual timeout duration should be close to DefaultSubmitterFinishedTimeout") + // The submitter is gone but the head never restarted, so the RUNNING status it reports is + // current and the job must be left to run. Watch well past the grace period. + LogWithTimestamp(test.T(), "Asserting RayJob %s/%s outlives the submitter", rayJob.Namespace, rayJob.Name) + g.Consistently(RayJob(test, rayJob.Namespace, rayJob.Name), 3*ray.DefaultSubmitterFinishedTimeout). + Should(WithTransform(RayJobDeploymentStatus, Equal(rayv1.JobDeploymentStatusRunning))) - // Get the updated rayJob rayJob, err = GetRayJob(test, rayJob.Namespace, rayJob.Name) g.Expect(err).NotTo(HaveOccurred()) - - reason := rayJob.Status.Reason - message := rayJob.Status.Message - g.Expect(reason).To(Equal(rayv1.JobDeploymentStatusTransitionGracePeriodExceeded)) - g.Expect(message).To(MatchRegexp(`The RayJob submitter finished at .* but the ray job did not reach terminal state within .*`)) + g.Expect(rayJob.Status.JobStatus).To(Equal(rayv1.JobStatusRunning)) + g.Expect(rayJob.Status.Reason).NotTo(Equal(rayv1.JobDeploymentStatusTransitionGracePeriodExceeded)) }) test.T().Run("RayCluster status update propagates to RayJob", func(_ *testing.T) {