[Fix][RayJob SidecarMode] Prevent premature job termination during transient head node spikes (#4399)

* [Fix][1/N] defer sidecar submitter failure until dashboard check Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Feature][SidecarMode] Add SidecarSubmitterRestart feature gate for per-container restart Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Feature][SidecarMode] Add SidecarSubmitterRestart feature gate unit test Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Feature][SidecarMode] Add K8s version check if SidecarSubmitterRestart feature gate is enabled at operator startup Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Chore] better explainations Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Chore] remove incorrect nil check Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix] SidecarMode should not fail job based on container exit status Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * Trigger CI Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix][SidecarMode] add e2e tests Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix][SidecarMode] clarify RestartPolicyRules exit code rationale Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix][Sidecar Mode] Use ParseGeneric + GitVersion, and add patch version for more robust k8s version comparison Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * Apply suggestions Co-authored-by: Jia-Wei Jiang <36886416+JiangJiaWei1103@users.noreply.github.com> Signed-off-by: JustinYeh <justinyeh1995@gmail.com> * [Fix] use ContainerRestartPolicyOnFailure to express the intention better; update the comment to explain the intention better Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix] bump version check to v1.35 and update sidecar mode test to check onFailure restart policy Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix] check the existence of the feature gate instead of relying on the derivate result Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * improve the why for using K8s v1.35 Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix] Revert checkSidecarContainerStatus removal; scope PR to K8s v1.35+ use case Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix] use correct health command constant and argument order in the feature gate test. Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix] remove leftover changes Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * [Fix] prevent panicing due to failing to fetch head pod Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * add warning about the risk of having kubelet version skew Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * document ray v2.54.0+ is needed Co-authored-by: Jun-Hao Wan <ken89@kimo.com> Signed-off-by: JustinYeh <justinyeh1995@gmail.com> * add max restart count annotaion, envVar, and default max restart count Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * do not fail the job if the submitter container exited successfully Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * fix BasePythonHealthCommand argument Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * trigger ci Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * address suggestion part 1 Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * k8s 1.35 version Signed-off-by: Future-Outlier <eric901201@gmail.com> * add override Signed-off-by: Future-Outlier <eric901201@gmail.com> * update Signed-off-by: Future-Outlier <eric901201@gmail.com> * update Signed-off-by: Future-Outlier <eric901201@gmail.com> * upd for helm and kustomization Signed-off-by: Future-Outlier <eric901201@gmail.com> * update Signed-off-by: Future-Outlier <eric901201@gmail.com> * fix typo Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * address comment, assert no resubmission Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * bump kind version to the latest version which supports v1.35.0 by default Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * incremental upgrade use kind-config-buildkite.yml (v1.35) Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> * replace hardcoded value to use rayJob.Spec.SubmitterConfig.BackoffLimit Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> --------- Signed-off-by: justinyeh1995 <justinyeh1995@gmail.com> Signed-off-by: JustinYeh <justinyeh1995@gmail.com> Signed-off-by: Future-Outlier <eric901201@gmail.com> Co-authored-by: Jia-Wei Jiang <36886416+JiangJiaWei1103@users.noreply.github.com> Co-authored-by: Jun-Hao Wan <ken89@kimo.com> Co-authored-by: Future-Outlier <eric901201@gmail.com>
2026-07-27 06:55:10 +00:00 · 2026-06-01 21:20:44 -07:00
parent ff0e074694
commit a284b829f8
16 changed files with 448 additions and 28 deletions
@@ -7,7 +7,7 @@ export PATH=$PATH:/usr/local/go/bin
 export DOCKER_API_VERSION=1.43

 # Install kind
-curl -Lo ./kind https://kind.sigs.k8s.io/dl/v0.22.0/kind-linux-amd64
+curl -Lo ./kind https://kind.sigs.k8s.io/dl/v0.31.0/kind-linux-amd64
 chmod +x ./kind
 mv ./kind /usr/local/bin/kind

@@ -83,7 +83,7 @@
  image: golang:1.26-bookworm
  commands:
    - source .buildkite/setup-env.sh
-    - kind create cluster --wait 900s --config ./ci/kind-config-buildkite-1-29.yml
+    - kind create cluster --wait 900s --config ./ci/kind-config-buildkite.yml
    - kubectl config set clusters.kind-kind.server https://docker:6443

    # Install MetalLB for LoadBalancer IPs on Kind
@@ -20,3 +20,5 @@ featureGates:
    enabled: true
  - name: RayServiceIncrementalUpgrade
    enabled: true
+  - name: SidecarSubmitterRestart
+    enabled: true
@@ -9,7 +9,7 @@ networking:
 # https://blog.scottlowe.org/2019/07/30/adding-a-name-to-kubernetes-api-server-certificate/
 nodes:
 - role: control-plane
-  image: kindest/node:v1.29.0
+  image: kindest/node:v1.35.0
  kubeadmConfigPatches:
  - |
    kind: ClusterConfiguration
@@ -593,7 +593,7 @@ _Appears in:_

 | Field | Description | Default | Validation |
 | --- | --- | --- | --- |
-| `backoffLimit` _integer_ | BackoffLimit of the submitter k8s job. |  |  |
+| `backoffLimit` _integer_ | BackoffLimit of the submitter. In K8sJobMode, this is the K8s Job backoffLimit.<br />In SidecarMode with SidecarSubmitterRestart enabled, this is the maximum container restart count. |  |  |


 #### UpscalingMode
@@ -187,7 +187,8 @@ const (
 )

 type SubmitterConfig struct {
-	// BackoffLimit of the submitter k8s job.
+	// BackoffLimit of the submitter. In K8sJobMode, this is the K8s Job backoffLimit.
+	// In SidecarMode with SidecarSubmitterRestart enabled, this is the maximum container restart count.
 	// +optional
 	BackoffLimit *int32 `json:"backoffLimit,omitempty"`
 }
@@ -9,4 +9,4 @@ spec:
      containers:
      - name: kuberay-operator
        args:
-        - --feature-gates=RayClusterStatusConditions=true,RayJobDeletionPolicy=true,RayMultiHostIndexing=true,RayCronJob=true,RayServiceIncrementalUpgrade=true
+        - --feature-gates=RayClusterStatusConditions=true,RayJobDeletionPolicy=true,RayMultiHostIndexing=true,RayCronJob=true,RayServiceIncrementalUpgrade=true,SidecarSubmitterRestart=true
@@ -14,6 +14,7 @@ import (

 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
+	"github.com/ray-project/kuberay/ray-operator/pkg/features"
 	pkgutils "github.com/ray-project/kuberay/ray-operator/pkg/utils"
 )

@@ -150,10 +151,13 @@ func BuildJobSubmitCommand(rayJobInstance *rayv1.RayJob, submissionMode rayv1.Jo
 	}
 	cmd = append(cmd, waitLoop...)

-	// In Sidecar mode, we only support RayJob level retry, which means that the submitter retry won't happen,
+	// In Sidecar mode without SidecarSubmitterRestart feature gate enabled, we only support RayJob level retry, which means that the submitter retry won't happen,
 	// so we won't have to check if the job has been submitted.
-	if submissionMode == rayv1.K8sJobMode {
-		// Only check job status in K8s mode to handle duplicated submission gracefully
+	// In K8sJobMode (submitter Job may retry) or Sidecar mode with SidecarSubmitterRestart feature gate enabled (submitter container may restart on failure).
+	// we check job status before submitting to handle duplicated submission gracefully.
+	needsStatusCheck := submissionMode == rayv1.K8sJobMode || (submissionMode == rayv1.SidecarMode && features.Enabled(features.SidecarSubmitterRestart))
+
+	if needsStatusCheck {
 		cmd = append(cmd, "if", "!")
 		cmd = append(cmd, jobStatusCommand...)
 		cmd = append(cmd, ";", "then")
@@ -161,7 +165,7 @@ func BuildJobSubmitCommand(rayJobInstance *rayv1.RayJob, submissionMode rayv1.Jo

 	cmd = append(cmd, jobSubmitCommand...)

-	if submissionMode == rayv1.K8sJobMode {
+	if needsStatusCheck {
 		cmd = append(cmd, "--no-wait")
 	}

@@ -199,7 +203,7 @@ func BuildJobSubmitCommand(rayJobInstance *rayv1.RayJob, submissionMode rayv1.Jo

 	// "--" is used to separate the entrypoint from the Ray Job CLI command and its arguments.
 	cmd = append(cmd, "--", entrypoint, ";")
-	if submissionMode == rayv1.K8sJobMode {
+	if needsStatusCheck {
 		cmd = append(cmd, "fi", ";")
 		cmd = append(cmd, jobFollowCommand...)
 	}
@@ -12,6 +12,7 @@ import (

 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
+	"github.com/ray-project/kuberay/ray-operator/pkg/features"
 )

 func rayJobTemplate() *rayv1.RayJob {
@@ -213,6 +214,54 @@ func TestBuildJobSubmitCommandWithK8sJobModeHealthWaitLoop(t *testing.T) {
 	assert.NotContains(t, command[1], "wget")
 }

+func TestBuildJobSubmitCommandWithSidecarModeAndFeatureGate(t *testing.T) {
+	// Enable the SidecarSubmitterRestart feature gate for this test
+	features.SetFeatureGateDuringTest(t, features.SidecarSubmitterRestart, true)
+
+	testRayJob := rayJobTemplate()
+	testRayJob.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.Containers = []corev1.Container{
+		{
+			Ports: []corev1.ContainerPort{
+				{
+					Name:          utils.DashboardPortName,
+					ContainerPort: utils.DefaultDashboardPort,
+				},
+			},
+		},
+	}
+
+	// With SidecarSubmitterRestart enabled, the command should include:
+	// - status check (if ! ray job status ...)
+	// - --no-wait flag
+	// - job logs follow at the end
+	healthURL := fmt.Sprintf("http://localhost:%d/%s", utils.DefaultDashboardPort, utils.RayDashboardGCSHealthPath)
+	expected := []string{
+		"until",
+		fmt.Sprintf(
+			utils.BasePythonHealthCommand,
+			healthURL,
+			utils.RayDashboardGCSHealthCheckTimeoutSeconds,
+		),
+		">/dev/null", "2>&1", ";",
+		"do", "echo", strconv.Quote("Waiting for Ray Dashboard GCS to become healthy at http://127.0.0.1:8265 ..."), ";", "sleep", "2", ";", "done", ";",
+		"if", "!", "ray", "job", "status", "--address", "http://127.0.0.1:8265", "testJobId", ">/dev/null", "2>&1", ";", "then",
+		"ray", "job", "submit", "--address", "http://127.0.0.1:8265", "--no-wait",
+		"--runtime-env-json", strconv.Quote(`{"test":"test"}`),
+		"--metadata-json", strconv.Quote(`{"testKey":"testValue"}`),
+		"--submission-id", "testJobId",
+		"--entrypoint-num-cpus", "1.000000",
+		"--entrypoint-num-gpus", "0.500000",
+		"--entrypoint-resources", strconv.Quote(`{"Custom_1": 1, "Custom_2": 5.5}`),
+		"--",
+		"echo no quote 'single quote' \"double quote\"",
+		";", "fi", ";",
+		"ray", "job", "logs", "--address", "http://127.0.0.1:8265", "--follow", "testJobId",
+	}
+	command, err := BuildJobSubmitCommand(testRayJob, rayv1.SidecarMode)
+	require.NoError(t, err)
+	assert.Equal(t, expected, command)
+}
+
 func TestBuildJobSubmitCommandWithK8sJobModeAndYAML(t *testing.T) {
 	rayJobWithYAML := &rayv1.RayJob{
 		Spec: rayv1.RayJobSpec{
@@ -588,6 +588,19 @@ func getSubmitterContainer(rayJobInstance *rayv1.RayJob, rayClusterInstance *ray
 		return corev1.Container{}, err
 	}

+	// When SidecarSubmitterRestart feature gate is enabled, configure per-container restart rules.
+	// This requires Kubernetes 1.35+ with ContainerRestartRules feature gate enabled.
+	if features.Enabled(features.SidecarSubmitterRestart) {
+		// OnFailure restarts only the submitter container (not all containers in the pod) on non-zero exit.
+		// The non-zero exit can come from `ray job submit --no-wait` or `ray job logs --follow`.
+		// The key case is `ray job logs --follow` exiting non-zero on a transient
+		// WebSocket closure even when the Ray job is still running.
+		// On restart, the submitter checks ray job status first.
+		// Since the job is still running, the submitter simply reattaches to the log stream.
+		// See BuildJobSubmitCommand in ray-operator/controllers/ray/common/job.go for more details.
+		submitterContainer.RestartPolicy = ptr.To(corev1.ContainerRestartPolicyOnFailure)
+	}
+
 	return submitterContainer, nil
 }

@@ -992,8 +1005,8 @@ func (r *RayJobReconciler) constructRayClusterForRayJob(rayJobInstance *rayv1.Ra
 			rayCluster.Spec.HeadGroupSpec.Template.Spec.Containers, sidecar)
 		// In K8sJobMode, the submitter Job relies on the K8s Job backoffLimit API to restart if it fails.
 		// This mainly handles WebSocket connection failures caused by transient network issues.
-		// In SidecarMode, however, the submitter container shares the same network namespace as the Ray dashboard,
-		// so restarts are no longer needed.
+		// In SidecarMode, the pod-level RestartPolicy is set to Never.
+		// The submitter container may override this with per-container restart rules when the SidecarSubmitterRestart feature gate is enabled.
 		rayCluster.Spec.HeadGroupSpec.Template.Spec.RestartPolicy = corev1.RestartPolicyNever
 	}

@@ -1052,20 +1065,44 @@ func (r *RayJobReconciler) checkSubmitterAndUpdateStatusIfNeeded(ctx context.Con
 			return
 		}

-		shouldUpdate, submitterContainerStatus = checkSidecarContainerStatus(headPod)
-		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)
-			rayJob.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusFailed
-			// The submitter sidecar container needs to wait for the user code to finish and retrieve its logs.
-			// Therefore, a failed Submitter sidecar container indicates that the submission itself has failed or the user code has thrown an error.
-			// If the failure is due to user code, the JobStatus and Job message will be updated accordingly from the previous reconciliation.
-			if rayJob.Status.JobStatus == rayv1.JobStatusFailed {
-				rayJob.Status.Reason = rayv1.AppFailed
-			} else {
-				rayJob.Status.Reason = rayv1.SubmissionFailed
-				rayJob.Status.Message = fmt.Sprintf("Ray head pod container %s terminated with exit code %d: %s",
-					submitterContainerStatus.Name, submitterContainerStatus.State.Terminated.ExitCode, submitterContainerStatus.State.Terminated.Reason)
+		// Only check exit code when the feature gate is disabled.
+		// When SidecarSubmitterRestart is enabled, the container restarts on non-zero exit,
+		// so a terminated container is transient — not a permanent failure.
+		if !features.Enabled(features.SidecarSubmitterRestart) {
+			shouldUpdate, submitterContainerStatus = checkSidecarContainerStatus(headPod)
+			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)
+				rayJob.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusFailed
+				// The submitter sidecar container needs to wait for the user code to finish and retrieve its logs.
+				// Therefore, a failed Submitter sidecar container indicates that the submission itself has failed or the user code has thrown an error.
+				// If the failure is due to user code, the JobStatus and Job message will be updated accordingly from the previous reconciliation.
+				if rayJob.Status.JobStatus == rayv1.JobStatusFailed {
+					rayJob.Status.Reason = rayv1.AppFailed
+				} else {
+					rayJob.Status.Reason = rayv1.SubmissionFailed
+					rayJob.Status.Message = fmt.Sprintf("Ray head pod container %s terminated with exit code %d: %s",
+						submitterContainerStatus.Name, submitterContainerStatus.State.Terminated.ExitCode, submitterContainerStatus.State.Terminated.Reason)
+				}
+			}
+		} else {
+			submitterBackoffLimit := int32(2)
+			if rayJob.Spec.SubmitterConfig != nil && rayJob.Spec.SubmitterConfig.BackoffLimit != nil {
+				submitterBackoffLimit = *rayJob.Spec.SubmitterConfig.BackoffLimit
+			}
+			shouldUpdate, submitterContainerStatus = checkIsRestartCountExceeded(headPod, submitterBackoffLimit)
+			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,
+					"RestartCount", submitterContainerStatus.RestartCount)
+				rayJob.Status.JobDeploymentStatus = rayv1.JobDeploymentStatusFailed
+				if rayJob.Status.JobStatus == rayv1.JobStatusFailed {
+					rayJob.Status.Reason = rayv1.AppFailed
+				} else {
+					rayJob.Status.Reason = rayv1.SubmissionFailed
+					rayJob.Status.Message = fmt.Sprintf("Ray head pod submitter container %s terminated after exceeding the maximum restart count",
+						submitterContainerStatus.Name)
+				}
 			}
 		}

@@ -1149,6 +1186,24 @@ func checkSidecarContainerStatus(headPod *corev1.Pod) (bool, *corev1.ContainerSt
 	return false, nil
 }

+func checkIsRestartCountExceeded(headPod *corev1.Pod, backoffLimit int32) (bool, *corev1.ContainerStatus) {
+	for _, containerStatus := range headPod.Status.ContainerStatuses {
+		if containerStatus.Name == utils.SubmitterContainerName {
+			// Only check when the container has been terminated at least once.
+			// When the submitter container fails in a CrashLoopBackOff fashion, LastTerminationState.Terminated is populated
+			if containerStatus.LastTerminationState.Terminated != nil {
+				// If the container exited successfully, we do not fail the job.
+				if containerStatus.State.Terminated != nil && containerStatus.State.Terminated.ExitCode == 0 {
+					break
+				}
+				return containerStatus.RestartCount >= backoffLimit, &containerStatus
+			}
+			break
+		}
+	}
+	return false, nil
+}
+
 func checkActiveDeadlineAndUpdateStatusIfNeeded(ctx context.Context, rayJob *rayv1.RayJob) bool {
 	logger := ctrl.LoggerFrom(ctx)
 	if rayJob.Spec.ActiveDeadlineSeconds == nil || time.Now().Before(rayJob.Status.StartTime.Add(time.Duration(*rayJob.Spec.ActiveDeadlineSeconds)*time.Second)) {
@@ -30,6 +30,7 @@ import (
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/metrics/mocks"
 	utils "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 	"github.com/ray-project/kuberay/ray-operator/pkg/client/clientset/versioned/scheme"
+	"github.com/ray-project/kuberay/ray-operator/pkg/features"
 )

 // fakeBatchScheduler implements schedulerinterface.BatchScheduler for testing.
@@ -224,6 +225,118 @@ func TestGetSubmitterTemplate(t *testing.T) {
 	assert.Equal(t, "test-job-id", envVar.Value)
 }

+func TestGetSubmitterContainerWithFeatureGate(t *testing.T) {
+	// Enable the SidecarSubmitterRestart feature gate for this test
+	features.SetFeatureGateDuringTest(t, features.SidecarSubmitterRestart, true)
+
+	rayJobInstance := &rayv1.RayJob{
+		Spec: rayv1.RayJobSpec{
+			Entrypoint:     "echo test",
+			SubmissionMode: rayv1.SidecarMode,
+			RayClusterSpec: &rayv1.RayClusterSpec{
+				HeadGroupSpec: rayv1.HeadGroupSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Image: "rayproject/ray:test",
+									Ports: []corev1.ContainerPort{
+										{
+											Name:          utils.DashboardPortName,
+											ContainerPort: utils.DefaultDashboardPort,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		Status: rayv1.RayJobStatus{
+			DashboardURL: "http://127.0.0.1:8265",
+			JobId:        "test-job-id",
+		},
+	}
+
+	rayClusterInstance := &rayv1.RayCluster{
+		Spec: *rayJobInstance.Spec.RayClusterSpec,
+	}
+
+	container, err := getSubmitterContainer(rayJobInstance, rayClusterInstance)
+	require.NoError(t, err)
+
+	// Verify restart policy is set to OnFailure for the submitter container
+	require.NotNil(t, container.RestartPolicy)
+	assert.Equal(t, corev1.ContainerRestartPolicyOnFailure, *container.RestartPolicy)
+}
+
+func TestCheckIsRestartCountExceeded(t *testing.T) {
+	tests := []struct {
+		name                 string
+		restartCount         int32
+		existCode            int32
+		hasTermState         bool
+		hasLastTermState     bool
+		expectedShouldUpdate bool
+	}{
+		{
+			name:                 "restart count below limit of 2",
+			restartCount:         1,
+			existCode:            1,
+			hasTermState:         true,
+			hasLastTermState:     true,
+			expectedShouldUpdate: false,
+		},
+		{
+			name:                 "restart count at limit of 2",
+			restartCount:         2,
+			existCode:            1,
+			hasTermState:         true,
+			hasLastTermState:     true,
+			expectedShouldUpdate: true,
+		},
+		{
+			name:                 "successful exit should not trigger failure",
+			restartCount:         3,
+			existCode:            0,
+			hasTermState:         true,
+			hasLastTermState:     true,
+			expectedShouldUpdate: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			containerStatus := corev1.ContainerStatus{
+				Name:         utils.SubmitterContainerName,
+				RestartCount: tt.restartCount,
+			}
+
+			if tt.hasTermState {
+				containerStatus.State = corev1.ContainerState{
+					Terminated: &corev1.ContainerStateTerminated{ExitCode: tt.existCode},
+				}
+			}
+
+			if tt.hasLastTermState {
+				containerStatus.LastTerminationState = corev1.ContainerState{
+					Terminated: &corev1.ContainerStateTerminated{ExitCode: 1},
+				}
+			}
+
+			headPod := &corev1.Pod{
+				Status: corev1.PodStatus{
+					ContainerStatuses: []corev1.ContainerStatus{containerStatus},
+				},
+			}
+
+			shouldUpdate, _ := checkIsRestartCountExceeded(headPod, 2)
+			assert.Equal(t, tt.expectedShouldUpdate, shouldUpdate)
+		})
+	}
+}
+
 func TestUpdateStatusToSuspendingIfNeeded(t *testing.T) {
 	newScheme := runtime.NewScheme()
 	_ = rayv1.AddToScheme(newScheme)
@@ -23,6 +23,9 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/apimachinery/pkg/util/rand"
+	utilversion "k8s.io/apimachinery/pkg/util/version"
+	"k8s.io/apimachinery/pkg/version"
+	"k8s.io/client-go/discovery"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -885,6 +888,39 @@ func GetWeightsFromHTTPRoute(httpRoute *gwv1.HTTPRoute, rayServiceInstance *rayv
 	return
 }

+// GetKubernetesVersion returns the API server version
+func GetKubernetesVersion() (*version.Info, error) {
+	config, err := ctrl.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+
+	serverVersion, err := discoveryClient.ServerVersion()
+	if err != nil {
+		return nil, err
+	}
+	return serverVersion, nil
+}
+
+// IsK8sVersionAtLeast checks the API server version is at least v{major}.{minor}.{patch}
+func IsK8sVersionAtLeast(serverVersion *version.Info, major, minor, patch int) (bool, error) {
+	requiredVersionString := fmt.Sprintf("%d.%d.%d", major, minor, patch)
+	currentVersion, err := utilversion.ParseGeneric(serverVersion.GitVersion)
+	if err != nil {
+		return false, err
+	}
+	requiredVersion, err := utilversion.ParseGeneric(requiredVersionString)
+	if err != nil {
+		return false, err
+	}
+	return currentVersion.AtLeast(requiredVersion), nil
+}
+
 func GetContainerCommand(additionalOptions []string) []string {
 	bashOptions := []string{"c"}
 	bashOptions = append(bashOptions, additionalOptions...)
@@ -194,6 +194,30 @@ func main() {
 		utilruntime.Must(gwv1.Install(scheme))
 	}

+	// Although ContainerRestartPolicy is introduced in Kubernetes v1.34 as an alpha feature, which requires
+	// manually enabling the ContainerRestartRules feature gate, it becomes beta in v1.35.
+	// We require v1.35+ to avoid asking users to opt in to a separate K8s feature gate.
+	// NOTE: this checks the API server version only. Please make sure worker nodes are also v1.35+,
+	// as kubelets are allowed to be up to 3 minor versions older per the Kubernetes version skew policy.
+	// If ContainerRestartRules is not enabled on a kubelet, the per-container restart policy on the
+	// submitter container will be ignored. In this scenario, if the submitter container exit on failure,
+	// a 30-second timeout will be applied by the operator and could incorrectly mark the RayJob as Failed
+	// even if the Ray job is still running.
+	if features.Enabled(features.SidecarSubmitterRestart) {
+		serverVersion, err := utils.GetKubernetesVersion()
+		if err != nil {
+			exitOnError(err, "SidecarSubmitterRestart feature gate enabled but unable to detect K8s version. Feature requires K8s 1.35+.")
+		}
+		isAtLeast, err := utils.IsK8sVersionAtLeast(serverVersion, 1, 35, 0)
+		if err != nil {
+			exitOnError(err, "Failed to compare K8s version.")
+		}
+		if !isAtLeast {
+			exitOnError(fmt.Errorf("current version %s is below 1.35", serverVersion.GitVersion),
+				"SidecarSubmitterRestart feature gate requires K8s 1.35+")
+		}
+	}
+
 	// Manager options
 	options := ctrl.Options{
 		Cache: cache.Options{
@@ -5,7 +5,8 @@ package v1
 // SubmitterConfigApplyConfiguration represents a declarative configuration of the SubmitterConfig type for use
 // with apply.
 type SubmitterConfigApplyConfiguration struct {
-	// BackoffLimit of the submitter k8s job.
+	// BackoffLimit of the submitter. In K8sJobMode, this is the K8s Job backoffLimit.
+	// In SidecarMode with SidecarSubmitterRestart enabled, this is the maximum container restart count.
 	BackoffLimit *int32 `json:"backoffLimit,omitempty"`
 }

@@ -47,6 +47,14 @@ const (
 	//
 	// Enables RayCronJob controller for scheduled RayJob execution.
 	RayCronJob featuregate.Feature = "RayCronJob"
+
+	// owner: @justinyeh1995
+	// rep: N/A
+	// alpha: v1.7
+	//
+	// Enables per-container restart policy for SidecarMode submitter to handle transient failures.
+	// Requires Kubernetes v1.35+ since it supports ContainerRestartPolicy by default starting in v1.35 and Ray v2.54.0+.
+	SidecarSubmitterRestart featuregate.Feature = "SidecarSubmitterRestart"
 )

 func init() {
@@ -59,6 +67,7 @@ var defaultFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{
 	RayMultiHostIndexing:         {Default: true, PreRelease: featuregate.Beta},
 	RayServiceIncrementalUpgrade: {Default: false, PreRelease: featuregate.Alpha},
 	RayCronJob:                   {Default: false, PreRelease: featuregate.Alpha},
+	SidecarSubmitterRestart:      {Default: false, PreRelease: featuregate.Alpha},
 }

 // SetFeatureGateDuringTest is a helper method to override feature gates in tests.
@@ -1,6 +1,8 @@
 package e2erayjob

 import (
+	"os/exec"
+	"strings"
 	"testing"
 	"time"

@@ -301,4 +303,128 @@ env_vars:

 		LogWithTimestamp(test.T(), "RayJob %s/%s completed successfully with auth token", rayJob.Namespace, rayJob.Name)
 	})
+
+	test.T().Run("RayJob Sidecar Mode should restart submitter container on non-zero exit", func(t *testing.T) {
+		// Check if the SidecarSubmitterRestart feature gate is enabled by inspecting the operator's args directly, rather than inferring from the container spec.
+		// The k8s version >= 1.35 is already verified at operator startup (see main.go), so we only need to check the feature gate here.
+		var sidecarSubmitterRestartEnabled bool
+		// Search all namespaces since the operator may be deployed in any namespace
+		operatorDeploymentList, err := test.Client().Core().AppsV1().Deployments("").List(t.Context(), metav1.ListOptions{
+			LabelSelector: "app.kubernetes.io/component=kuberay-operator",
+		})
+		g.Expect(err).NotTo(HaveOccurred())
+		// Assuming there should only be one kuberay-operator in the cluster
+		g.Expect(operatorDeploymentList.Items).To(HaveLen(1), "expected exactly one kuberay-operator deployment during the test")
+
+		for _, container := range operatorDeploymentList.Items[0].Spec.Template.Spec.Containers {
+			if container.Name != "kuberay-operator" {
+				continue
+			}
+			for _, arg := range container.Args {
+				if strings.Contains(arg, "SidecarSubmitterRestart=true") {
+					sidecarSubmitterRestartEnabled = true
+					break
+				}
+			}
+
+			if sidecarSubmitterRestartEnabled {
+				break
+			}
+		}
+
+		if !sidecarSubmitterRestartEnabled {
+			t.Skip("SidecarSubmitterRestart feature gate is not active. SidecarSubmitterRestart is not set to true")
+		}
+
+		rayJobAC := rayv1ac.RayJob("submitter-container-should-restart", namespace.Name).
+			WithSpec(rayv1ac.RayJobSpec().
+				WithSubmissionMode(rayv1.SidecarMode).
+				WithRayClusterSpec(NewRayClusterSpec()).
+				WithEntrypoint("python -c \"import time; time.sleep(60)\"").
+				WithShutdownAfterJobFinishes(true))
+
+		rayJob, err := test.Client().Ray().RayV1().RayJobs(namespace.Name).Apply(test.Ctx(), rayJobAC, TestApplyOptions)
+		g.Expect(err).NotTo(HaveOccurred())
+		LogWithTimestamp(test.T(), "Created RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)
+
+		// Wait until the RayJob to become Running
+		LogWithTimestamp(test.T(), "Waiting for RayJob %s/%s to be 'Running'", rayJob.Namespace, rayJob.Name)
+		g.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutMedium).
+			Should(WithTransform(RayJobStatus, Equal(rayv1.JobStatusRunning)))
+
+		// Get RayCluster name
+		rayJob, err = GetRayJob(test, rayJob.Namespace, rayJob.Name)
+		g.Expect(err).NotTo(HaveOccurred())
+		rayClusterName := rayJob.Status.RayClusterName
+
+		// Get RayCluster
+		rayCluster, err := GetRayCluster(test, namespace.Name, rayClusterName)
+		g.Expect(err).NotTo(HaveOccurred())
+
+		// Get headPod
+		headPod, err := GetHeadPod(test, rayCluster)
+		g.Expect(err).NotTo(HaveOccurred())
+		g.Expect(headPod).NotTo(BeNil())
+
+		// Get Submitter Container id
+		var containerID string
+		for _, containerStatus := range headPod.Status.ContainerStatuses {
+			if containerStatus.Name == utils.SubmitterContainerName {
+				containerID = strings.TrimPrefix(containerStatus.ContainerID, "containerd://")
+				break
+			}
+		}
+		g.Expect(containerID).ToNot(BeEmpty(), "Submitter's container ID should not be empty")
+		LogWithTimestamp(test.T(), "Found submitter container ID: %s", containerID)
+
+		// Record job id before restart
+		initialJobID := rayJob.Status.JobId
+
+		// stop the container with docker exec
+		kindNodeName := headPod.Spec.NodeName
+		cmd := exec.CommandContext(test.Ctx(), "docker", "exec", kindNodeName, "crictl", "stop", containerID)
+		err = cmd.Run()
+		g.Expect(err).NotTo(HaveOccurred())
+
+		// Verify RayJob is still running
+		g.Expect(GetRayJob(test, rayJob.Namespace, rayJob.Name)).
+			To(WithTransform(RayJobStatus, Equal(rayv1.JobStatusRunning)))
+
+		// Verify the restart count should > 0
+		g.Eventually(func() int32 {
+			headPod, err = GetHeadPod(test, rayCluster)
+			if err != nil || headPod == nil {
+				return 0
+			}
+			for _, cs := range headPod.Status.ContainerStatuses {
+				if cs.Name == utils.SubmitterContainerName {
+					return cs.RestartCount
+				}
+			}
+			return 0
+		}, TestTimeoutMedium, 2*time.Second).Should(BeNumerically(">", 0))
+
+		// Verify RayJob is still running
+		rayJob, err = GetRayJob(test, rayJob.Namespace, rayJob.Name)
+		g.Expect(err).NotTo(HaveOccurred())
+		LogWithTimestamp(test.T(), "RayJob status after stopping submitter: jobStatus=%s deploymentStatus=%s", rayJob.Status.JobStatus, rayJob.Status.JobDeploymentStatus)
+
+		// Verify there is no duplicate submission by checking the job id with initialJobID
+		g.Consistently(func() string {
+			rayjob, err := GetRayJob(test, rayJob.Namespace, rayJob.Name)
+			g.Expect(err).NotTo(HaveOccurred())
+			return rayjob.Status.JobId
+		}, TestTimeoutShort, 2*time.Second).Should(Equal(initialJobID))
+		LogWithTimestamp(test.T(), "RayJob JobID remains %s, there is no duplicate submission", initialJobID)
+
+		// Verify RayJob does not transition to Failed
+		g.Consistently(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutShort, 2*time.Second).
+			Should(WithTransform(RayJobDeploymentStatus, Not(Equal(rayv1.JobDeploymentStatusFailed))))
+
+		// Verify RayJob eventually completed.
+		g.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutLong).
+			Should(WithTransform(RayJobDeploymentStatus, Equal(rayv1.JobDeploymentStatusComplete)))
+
+		LogWithTimestamp(test.T(), "RayJob %s/%s completed successfully", rayJob.Namespace, rayJob.Name)
+	})
 }