ray-project · andrewsykim · Jun 17, 2025 · Jun 2, 2025 · Jun 2, 2025 · Jun 3, 2025
diff --git a/docs/reference/api.md b/docs/reference/api.md
@@ -57,15 +57,48 @@ _Appears in:_
 
 #### DeletionPolicy
 
+
+
+
+
+
+
+_Appears in:_
+- [DeletionStrategy](#deletionstrategy)
+
+| Field | Description | Default | Validation |
+| --- | --- | --- | --- |
+| `policy` _[DeletionPolicyType](#deletionpolicytype)_ | Valid values are 'DeleteCluster', 'DeleteWorkers', 'DeleteSelf' or 'DeleteNone'. |  |  |
+
+
+#### DeletionPolicyType
+
 _Underlying type:_ _string_
 
 
 
 
 
+_Appears in:_
+- [DeletionPolicy](#deletionpolicy)
+
+
+
+#### DeletionStrategy
+
+
+
+
+
+
+
 _Appears in:_
 - [RayJobSpec](#rayjobspec)
 
+| Field | Description | Default | Validation |
+| --- | --- | --- | --- |
+| `onSuccess` _[DeletionPolicy](#deletionpolicy)_ |  |  |  |
+| `onFailure` _[DeletionPolicy](#deletionpolicy)_ |  |  |  |
 
 
 
@@ -209,7 +242,7 @@ _Appears in:_
 | `clusterSelector` _object (keys:string, values:string)_ | clusterSelector is used to select running rayclusters by labels |  |  |
 | `submitterConfig` _[SubmitterConfig](#submitterconfig)_ | Configurations of submitter k8s job. |  |  |
 | `managedBy` _string_ | ManagedBy is an optional configuration for the controller or entity that manages a RayJob.<br />The value must be either 'ray.io/kuberay-operator' or 'kueue.x-k8s.io/multikueue'.<br />The kuberay-operator reconciles a RayJob which doesn't have this field at all or<br />the field value is the reserved string 'ray.io/kuberay-operator',<br />but delegates reconciling the RayJob with 'kueue.x-k8s.io/multikueue' to the Kueue.<br />The field is immutable. |  |  |
-| `deletionPolicy` _[DeletionPolicy](#deletionpolicy)_ | DeletionPolicy indicates what resources of the RayJob are deleted upon job completion.<br />Valid values are 'DeleteCluster', 'DeleteWorkers', 'DeleteSelf' or 'DeleteNone'.<br />If unset, deletion policy is based on 'spec.shutdownAfterJobFinishes'.<br />This field requires the RayJobDeletionPolicy feature gate to be enabled. |  |  |
+| `deletionStrategy` _[DeletionStrategy](#deletionstrategy)_ | DeletionStrategy indicates what resources of the RayJob and how they are deleted upon job completion.<br />If unset, deletion policy is based on 'spec.shutdownAfterJobFinishes'.<br />This field requires the RayJobDeletionPolicy feature gate to be enabled. |  |  |
 | `entrypoint` _string_ | Entrypoint represents the command to start execution. |  |  |
 | `runtimeEnvYAML` _string_ | RuntimeEnvYAML represents the runtime environment configuration<br />provided as a multi-line YAML string. |  |  |
 | `jobId` _string_ | If jobId is not set, a new jobId will be auto-generated. |  |  |
@@ -377,7 +410,7 @@ _Appears in:_
 
 | Field | Description | Default | Validation |
 | --- | --- | --- | --- |
-| `suspend` _boolean_ | Suspend indicates whether a worker group should be suspended.<br />A suspended worker group will have all pods deleted.<br />This is not a user-facing API and is only used by RayJob DeletionPolicy. |  |  |
+| `suspend` _boolean_ | Suspend indicates whether a worker group should be suspended.<br />A suspended worker group will have all pods deleted.<br />This is not a user-facing API and is only used by RayJob DeletionStrategy. |  |  |
 | `groupName` _string_ | we can have multiple worker groups, we distinguish them by name |  |  |
 | `replicas` _integer_ | Replicas is the number of desired Pods for this worker group. See https://github.com/ray-project/kuberay/pull/1443 for more details about the reason for making this field optional. | 0 |  |
 | `minReplicas` _integer_ | MinReplicas denotes the minimum number of desired Pods for this worker group. | 0 |  |

diff --git a/helm-chart/kuberay-operator/crds/ray.io_rayjobs.yaml b/helm-chart/kuberay-operator/crds/ray.io_rayjobs.yaml
diff --git a/ray-operator/apis/ray/v1/raycluster_types.go b/ray-operator/apis/ray/v1/raycluster_types.go
@@ -87,7 +87,7 @@ type HeadGroupSpec struct {
 type WorkerGroupSpec struct {
 	// Suspend indicates whether a worker group should be suspended.
 	// A suspended worker group will have all pods deleted.
-	// This is not a user-facing API and is only used by RayJob DeletionPolicy.
+	// This is not a user-facing API and is only used by RayJob DeletionStrategy.
 	// +optional
 	Suspend *bool `json:"suspend,omitempty"`
 	// we can have multiple worker groups, we distinguish them by name

diff --git a/ray-operator/apis/ray/v1/rayjob_types.go b/ray-operator/apis/ray/v1/rayjob_types.go
@@ -84,13 +84,24 @@ const (
 	InteractiveMode JobSubmissionMode = "InteractiveMode" // Don't submit job in KubeRay. Instead, wait for user to submit job and provide the job submission ID.
 )
 
-type DeletionPolicy string
+type DeletionPolicyType string
+
+type DeletionStrategy struct {
+	OnSuccess DeletionPolicy `json:"onSuccess"`
+	OnFailure DeletionPolicy `json:"onFailure"`
+}
+
+type DeletionPolicy struct {
+	// Valid values are 'DeleteCluster', 'DeleteWorkers', 'DeleteSelf' or 'DeleteNone'.
+	// +kubebuilder:validation:XValidation:rule="self in ['DeleteCluster', 'DeleteWorkers', 'DeleteSelf', 'DeleteNone']",message="the policy field value must be either 'DeleteCluster', 'DeleteWorkers', 'DeleteSelf', or 'DeleteNone'"
+	Policy *DeletionPolicyType `json:"policy"`
+}
 
 const (
-	DeleteClusterDeletionPolicy DeletionPolicy = "DeleteCluster" // Deletion policy to delete the entire RayCluster custom resource on job completion.
-	DeleteWorkersDeletionPolicy DeletionPolicy = "DeleteWorkers" // Deletion policy to delete only the workers on job completion.
-	DeleteSelfDeletionPolicy    DeletionPolicy = "DeleteSelf"    // Deletion policy to delete the RayJob custom resource (and all associated resources) on job completion.
-	DeleteNoneDeletionPolicy    DeletionPolicy = "DeleteNone"    // Deletion policy to delete no resources on job completion.
+	DeleteCluster DeletionPolicyType = "DeleteCluster" // To delete the entire RayCluster custom resource on job completion.
+	DeleteWorkers DeletionPolicyType = "DeleteWorkers" // To delete only the workers on job completion.
+	DeleteSelf    DeletionPolicyType = "DeleteSelf"    // To delete the RayJob custom resource (and all associated resources) on job completion.
+	DeleteNone    DeletionPolicyType = "DeleteNone"    // To delete no resources on job completion.
 )
 
 type SubmitterConfig struct {
@@ -144,13 +155,11 @@ type RayJobSpec struct {
 	// +kubebuilder:validation:XValidation:rule="self in ['ray.io/kuberay-operator', 'kueue.x-k8s.io/multikueue']",message="the managedBy field value must be either 'ray.io/kuberay-operator' or 'kueue.x-k8s.io/multikueue'"
 	// +optional
 	ManagedBy *string `json:"managedBy,omitempty"`
-	// DeletionPolicy indicates what resources of the RayJob are deleted upon job completion.
-	// Valid values are 'DeleteCluster', 'DeleteWorkers', 'DeleteSelf' or 'DeleteNone'.
+	// DeletionStrategy indicates what resources of the RayJob and how they are deleted upon job completion.
 	// If unset, deletion policy is based on 'spec.shutdownAfterJobFinishes'.
 	// This field requires the RayJobDeletionPolicy feature gate to be enabled.
-	// +kubebuilder:validation:XValidation:rule="self in ['DeleteCluster', 'DeleteWorkers', 'DeleteSelf', 'DeleteNone']",message="the deletionPolicy field value must be either 'DeleteCluster', 'DeleteWorkers', 'DeleteSelf', or 'DeleteNone'"
 	// +optional
-	DeletionPolicy *DeletionPolicy `json:"deletionPolicy,omitempty"`
+	DeletionStrategy *DeletionStrategy `json:"deletionStrategy,omitempty"`
 	// Entrypoint represents the command to start execution.
 	// +optional
 	Entrypoint string `json:"entrypoint,omitempty"`

diff --git a/ray-operator/apis/ray/v1/zz_generated.deepcopy.go b/ray-operator/apis/ray/v1/zz_generated.deepcopy.go
diff --git a/ray-operator/config/crd/bases/ray.io_rayjobs.yaml b/ray-operator/config/crd/bases/ray.io_rayjobs.yaml
diff --git a/ray-operator/controllers/ray/rayjob_controller.go b/ray-operator/controllers/ray/rayjob_controller.go
@@ -382,24 +382,44 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 			"ShutdownTime", shutdownTime)
 
 		if features.Enabled(features.RayJobDeletionPolicy) &&
-			rayJobInstance.Spec.DeletionPolicy != nil &&
-			*rayJobInstance.Spec.DeletionPolicy != rayv1.DeleteNoneDeletionPolicy &&
+			rayJobInstance.Spec.DeletionStrategy != nil &&
 			len(rayJobInstance.Spec.ClusterSelector) == 0 {
-			logger.Info("Shutdown behavior is defined by the deletion policy", "deletionPolicy", rayJobInstance.Spec.DeletionPolicy)
+
+			if shutdownTime.After(nowTime) {
+				delta := int32(time.Until(shutdownTime.Add(2 * time.Second)).Seconds())
+				logger.Info("shutdownTime not reached, requeue this RayJob for n seconds", "seconds", delta)
+				return ctrl.Result{RequeueAfter: time.Duration(delta) * time.Second}, nil
+			}
+
+			policy := rayv1.DeleteNone
+			if rayJobInstance.Status.JobStatus == rayv1.JobStatusSucceeded {
+				policy = *rayJobInstance.Spec.DeletionStrategy.OnSuccess.Policy
+			} else if rayJobInstance.Status.JobStatus == rayv1.JobStatusFailed {
+				policy = *rayJobInstance.Spec.DeletionStrategy.OnFailure.Policy
+			} else {
+				logger.Info("jobStatus not valid for deletion", "jobStatus", rayJobInstance.Status.JobStatus)
+			}
 // https://docs.ray.io/en/latest/cluster/running-applications/job-submission/jobs-package-ref.html#jobstatus 
 const ( 
 	JobStatusNew       JobStatus = "" 
 	JobStatusPending   JobStatus = "PENDING" 
 	JobStatusRunning   JobStatus = "RUNNING" 
 	JobStatusStopped   JobStatus = "STOPPED" 
 	JobStatusSucceeded JobStatus = "SUCCEEDED" 
 	JobStatusFailed    JobStatus = "FAILED" 
 // https://docs.ray.io/en/latest/cluster/running-applications/job-submission/jobs-package-ref.html#jobstatus 
 const ( 
 	JobStatusNew       JobStatus = "" 
 	JobStatusPending   JobStatus = "PENDING" 
 	JobStatusRunning   JobStatus = "RUNNING" 
 	JobStatusStopped   JobStatus = "STOPPED" 
 	JobStatusSucceeded JobStatus = "SUCCEEDED" 
 	JobStatusFailed    JobStatus = "FAILED" 
+
+			// no need to continue as the selected policy is DeleteNone
+			if policy == rayv1.DeleteNone {
+				break
+			}
+
+			logger.Info("Shutdown behavior is defined by the deletion policy", "deletionPolicy", rayJobInstance.Spec.DeletionStrategy)
 			if shutdownTime.After(nowTime) {
 				delta := int32(time.Until(shutdownTime.Add(2 * time.Second)).Seconds())
 				logger.Info("shutdownTime not reached, requeue this RayJob for n seconds", "seconds", delta)
 				return ctrl.Result{RequeueAfter: time.Duration(delta) * time.Second}, nil
 			}
 
-			switch *rayJobInstance.Spec.DeletionPolicy {
-			case rayv1.DeleteClusterDeletionPolicy:
+			switch policy {
+			case rayv1.DeleteCluster:
 				logger.Info("Deleting RayCluster", "RayCluster", rayJobInstance.Status.RayClusterName)
 				_, err = r.deleteClusterResources(ctx, rayJobInstance)
-			case rayv1.DeleteWorkersDeletionPolicy:
+			case rayv1.DeleteWorkers:
 				logger.Info("Suspending all worker groups", "RayCluster", rayJobInstance.Status.RayClusterName)
 				err = r.suspendWorkerGroups(ctx, rayJobInstance)
-			case rayv1.DeleteSelfDeletionPolicy:
+			case rayv1.DeleteSelf:
 				logger.Info("Deleting RayJob")
 				err = r.Client.Delete(ctx, rayJobInstance)
 			default:
@@ -409,7 +429,7 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 			}
 		}
 
-		if (!features.Enabled(features.RayJobDeletionPolicy) || rayJobInstance.Spec.DeletionPolicy == nil) && rayJobInstance.Spec.ShutdownAfterJobFinishes && len(rayJobInstance.Spec.ClusterSelector) == 0 {
+		if (!features.Enabled(features.RayJobDeletionPolicy) || rayJobInstance.Spec.DeletionStrategy == nil) && rayJobInstance.Spec.ShutdownAfterJobFinishes && len(rayJobInstance.Spec.ClusterSelector) == 0 {
 			logger.Info("Shutdown behavior is defined by the 
506D
`ShutdownAfterJobFinishes` flag", "shutdownAfterJobFinishes", rayJobInstance.Spec.ShutdownAfterJobFinishes)
 			if shutdownTime.After(nowTime) {
 				delta := int32(time.Until(shutdownTime.Add(2 * time.Second)).Seconds())