[Feature] Support configurable RayCluster deletion delay in RayService (#3864)

Signed-off-by: machichima <[email protected]>

Author: machichima

docs/reference/api.md

@@ -294,6 +294,7 @@ _Appears in:_
 
 | Field | Description | Default | Validation |
 | --- | --- | --- | --- |
+| `rayClusterDeletionDelaySeconds` _integer_ | RayClusterDeletionDelaySeconds specifies the delay, in seconds, before deleting old RayClusters.<br />The default value is 60 seconds. |  | Minimum: 0 <br /> |
 | `serviceUnhealthySecondThreshold` _integer_ | Deprecated: This field is not used anymore. ref: https://github.com/ray-project/kuberay/issues/1685 |  |  |
 | `deploymentUnhealthySecondThreshold` _integer_ | Deprecated: This field is not used anymore. ref: https://github.com/ray-project/kuberay/issues/1685 |  |  |
 | `serveService` _[Service](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.28/#service-v1-core)_ | ServeService is the Kubernetes service for head node and worker nodes who have healthy http proxy to serve traffics. |  |  |

helm-chart/kuberay-operator/crds/ray.io_rayservices.yaml

@@ -65,6 +65,11 @@ type RayServiceUpgradeStrategy struct {
 
 // RayServiceSpec defines the desired state of RayService
 type RayServiceSpec struct {
+	// RayClusterDeletionDelaySeconds specifies the delay, in seconds, before deleting old RayClusters.
+	// The default value is 60 seconds.
+	// +kubebuilder:validation:Minimum=0
+	// +optional
+	RayClusterDeletionDelaySeconds *int32 `json:"rayClusterDeletionDelaySeconds,omitempty"`
 	// Deprecated: This field is not used anymore. ref: https://github.com/ray-project/kuberay/issues/1685
 	// +optional
 	ServiceUnhealthySecondThreshold *int32 `json:"serviceUnhealthySecondThreshold,omitempty"`

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

@@ -471,16 +471,22 @@ func (r *RayServiceReconciler) cleanUpRayClusterInstance(ctx context.Context, ra
 		return err
 	}
 
-	// Clean up RayCluster instances. Each instance is deleted 60 seconds
+	// Determine the ray cluster deletion delay seconds
+	deletionDelay := RayClusterDeletionDelayDuration
+	if rayServiceInstance.Spec.RayClusterDeletionDelaySeconds != nil {
+		deletionDelay = time.Duration(*rayServiceInstance.Spec.RayClusterDeletionDelaySeconds) * time.Second
+	}
+	// Clean up RayCluster instances. Each instance is deleted after the configured deletion delay.
 	for _, rayClusterInstance := range rayClusterList.Items {
 		if rayClusterInstance.Name != rayServiceInstance.Status.ActiveServiceStatus.RayClusterName && rayClusterInstance.Name != rayServiceInstance.Status.PendingServiceStatus.RayClusterName {
 			cachedTimestamp, exists := r.RayClusterDeletionTimestamps.Get(rayClusterInstance.Name)
 			if !exists {
-				deletionTimestamp := metav1.Now().Add(RayClusterDeletionDelayDuration)
+				deletionTimestamp := metav1.Now().Add(deletionDelay)
 				r.RayClusterDeletionTimestamps.Set(rayClusterInstance.Name, deletionTimestamp)
 				logger.Info(
 					"Scheduled dangling RayCluster for deletion",
 					"rayClusterName", rayClusterInstance.Name,
+					"deletionDelay", deletionDelay.String(),
 					"deletionTimestamp", deletionTimestamp,
 				)
 			} else {

ray-operator/apis/ray/v1/zz_generated.deepcopy.go

@@ -7,7 +7,9 @@ import (
 	"reflect"
 	"strconv"
 	"testing"
+	"time"
 
+	cmap "github.com/orcaman/concurrent-map/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
@@ -1232,3 +1234,86 @@ func TestIsZeroDowntimeUpgradeEnabled(t *testing.T) {
 		})
 	}
 }
+
+func TestRayClusterDeletionDelaySeconds(t *testing.T) {
+	namespace := "test-namespace"
+	rayClusterName := "test-cluster"
+	rayServiceName := "test-rayservice"
+
+	// Helper to create a RayService with optional RayClusterDeletionDelaySeconds
+	createRayService := func(delaySeconds *int32) *rayv1.RayService {
+		return &rayv1.RayService{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      rayServiceName,
+				Namespace: namespace,
+			},
+			Spec: rayv1.RayServiceSpec{
+				RayClusterDeletionDelaySeconds: delaySeconds,
+			},
+		}
+	}
+
+	rayCluster := rayv1.RayCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      rayClusterName,
+			Namespace: namespace,
+			Labels: map[string]string{
+				utils.RayOriginatedFromCRNameLabelKey: rayServiceName,
+				utils.RayOriginatedFromCRDLabelKey:    utils.RayOriginatedFromCRDLabelValue(utils.RayServiceCRD),
+			},
+		},
+	}
+
+	tests := []struct {
+		delaySeconds     *int32
+		name             string
+		expectedDuration time.Duration
+	}{
+		{
+			name:             "Use default delay when not set",
+			delaySeconds:     nil,
+			expectedDuration: RayClusterDeletionDelayDuration,
+		},
+		{
+			name:             "Use custom delay when set to 0",
+			delaySeconds:     ptr.To[int32](0),
+			expectedDuration: 0 * time.Second,
+		},
+		{
+			name:             "Use custom delay when set to positive",
+			delaySeconds:     ptr.To[int32](5),
+			expectedDuration: 5 * time.Second,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			newScheme := runtime.NewScheme()
+			_ = rayv1.AddToScheme(newScheme)
+			ctx := context.TODO()
+
+			rayService := createRayService(tc.delaySeconds)
+
+			// Initialize a fake client with newScheme and runtimeObjects.
+			runtimeObjects := []runtime.Object{rayService, &rayCluster}
+			fakeClient := clientFake.NewClientBuilder().WithScheme(newScheme).WithRuntimeObjects(runtimeObjects...).Build()
+			r := RayServiceReconciler{
+				Client:                       fakeClient,
+				Scheme:                       newScheme,
+				Recorder:                     record.NewFakeRecorder(1),
+				RayClusterDeletionTimestamps: cmap.New[time.Time](),
+			}
+
+			now := time.Now()
+			err := r.cleanUpRayClusterInstance(ctx, rayService)
+			require.NoError(t, err)
+
+			// Check that the deletion timestamp is set and equals to the expected value
+			ts, exists := r.RayClusterDeletionTimestamps.Get(rayClusterName)
+			assert.True(t, exists, "Deletion timestamp should be set for the cluster")
+			expectedTs := now.Add(tc.expectedDuration)
+
+			assert.InDelta(t, expectedTs.Unix(), ts.Unix(), 1, "Deletion timestamp should be within 1 second of expected timestamp")
+		})
+	}
+}

ray-operator/config/crd/bases/ray.io_rayservices.yaml

@@ -256,5 +256,11 @@ func ValidateRayServiceSpec(rayService *rayv1.RayService) error {
 		*rayService.Spec.UpgradeStrategy.Type != rayv1.NewCluster {
 		return fmt.Errorf("Spec.UpgradeStrategy.Type value %s is invalid, valid options are %s or %s", *rayService.Spec.UpgradeStrategy.Type, rayv1.NewCluster, rayv1.None)
 	}
+
+	if rayService.Spec.RayClusterDeletionDelaySeconds != nil &&
+		*rayService.Spec.RayClusterDeletionDelaySeconds < 0 {
+		return fmt.Errorf("Spec.RayClusterDeletionDelaySeconds should be a non-negative integer, got %d", *rayService.Spec.RayClusterDeletionDelaySeconds)
+	}
+
 	return nil
 }

ray-operator/controllers/ray/rayservice_controller.go

@@ -1084,6 +1084,14 @@ func TestValidateRayServiceSpec(t *testing.T) {
 			spec:        rayv1.RayServiceSpec{},
 			expectError: true,
 		},
+		{
+			name: "Spec.RayClusterDeletionDelaySeconds is negative",
+			spec: rayv1.RayServiceSpec{
+				RayClusterSpec:                 *createBasicRayClusterSpec(),
+				RayClusterDeletionDelaySeconds: ptr.To[int32](-1),
+			},
+			expectError: true,
+		},
 	}
 
 	for _, tt := range tests {