[RayCluster] support suspending worker groups

Signed-off-by: Andrew Sy Kim <[email protected]>

docs/reference/api.md

@@ -298,6 +298,7 @@ _Appears in:_
 | `template` _[PodTemplateSpec](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.28/#podtemplatespec-v1-core)_ | Template is a pod template for the worker |  |  |
 | `scaleStrategy` _[ScaleStrategy](#scalestrategy)_ | ScaleStrategy defines which pods to remove |  |  |
 | `numOfHosts` _integer_ | NumOfHosts denotes the number of hosts to create per replica. The default value is 1. | 1 |  |
+| `suspend` _boolean_ | Suspend indicates whether a worker group should be suspended.<br />A suspended worker group will have all pods deleted. |  |  |
 
 
 

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

@@ -77,6 +77,9 @@ type WorkerGroupSpec struct {
 	// NumOfHosts denotes the number of hosts to create per replica. The default value is 1.
 	// +kubebuilder:default:=1
 	NumOfHosts int32 `json:"numOfHosts,omitempty"`
+	// Suspend indicates whether a worker group should be suspended.
+	// A suspended worker group will have all pods deleted.
+	Suspend *bool `json:"suspend,omitempty"`
 }
 
 // ScaleStrategy to remove workers

ray-operator/controllers/ray/raycluster_controller.go

@@ -776,6 +776,18 @@ func (r *RayClusterReconciler) reconcilePods(ctx context.Context, instance *rayv
 			return err
 		}
 
+		// Delete all workers if worker group is suspended
+		if len(workerPods.Items) > 0 && worker.Suspend != nil && *worker.Suspend {
+			for _, workerPod := range workerPods.Items {
+				if err := r.Delete(ctx, &workerPod); err != nil {
+					r.Recorder.Eventf(instance, corev1.EventTypeWarning, string(utils.FailedToDeleteWorkerPod),
+						"Failed deleting worker Pod %s/%s; Pod status: %s; Pod restart policy: %s; Ray container terminated status: %v, %v",
+						workerPod.Namespace, workerPod.Name, workerPod.Status.Phase, workerPod.Spec.RestartPolicy, getRayContainerStateTerminated(workerPod), err)
+					return errstd.Join(utils.ErrFailedDeleteWorkerPod, err)
+				}
+			}
+		}
+
 		// Delete unhealthy worker Pods.
 		deletedWorkers := make(map[string]struct{})
 		deleted := struct{}{}