[Feature] Add initializing timeout for RayService (#4143)

* [Feature] Add initializing timeout for RayService

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

* fix lint

* validate the annotation at the beginning

* fix lint

* add e2e

* enhance comment

* stop recreating raycluster after timeout

* refactor

---------

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

Author: seanlaii

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

@@ -200,6 +200,7 @@ const (
 
 const (
 	RayServiceInitializing         RayServiceConditionReason = "Initializing"
+	RayServiceInitializingTimeout  RayServiceConditionReason = "InitializingTimeout"
 	ZeroServeEndpoints             RayServiceConditionReason = "ZeroServeEndpoints"
 	NonZeroServeEndpoints          RayServiceConditionReason = "NonZeroServeEndpoints"
 	BothActivePendingClustersExist RayServiceConditionReason = "BothActivePendingClustersExist"

ray-operator/controllers/ray/rayservice_controller.go

@@ -127,10 +127,24 @@ func (r *RayServiceReconciler) Reconcile(ctx context.Context, request ctrl.Reque
 	}
 
 	r.cleanUpServeConfigCache(ctx, rayServiceInstance)
-	if err = r.cleanUpRayClusterInstance(ctx, rayServiceInstance); err != nil {
+	hasRayClustersToClean, err := r.cleanUpRayClusterInstance(ctx, rayServiceInstance)
+	if err != nil {
 		return ctrl.Result{}, err
 	}
 
+	// If the RayService has timed out during initialization, skip the rest of the reconciliation.
+	// The service is in a terminal failure state - only cleanup (above) is needed.
+	// The user must delete and recreate the RayService to recover.
+	if isInitializingTimeout(rayServiceInstance) {
+		// Requeue only if there are still RayClusters to clean up
+		// This avoids unnecessary reconciliations after all resources are deleted
+		if hasRayClustersToClean {
+			return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, nil
+		}
+		logger.Info("RayService in terminal failure state, all RayClusters cleaned up")
+		return ctrl.Result{}, nil
+	}
+
 	// Find active and pending ray cluster objects given current service name.
 	var activeRayClusterInstance, pendingRayClusterInstance *rayv1.RayCluster
 	if activeRayClusterInstance, pendingRayClusterInstance, err = r.reconcileRayCluster(ctx, rayServiceInstance); err != nil {
@@ -438,7 +452,9 @@ func (r *RayServiceReconciler) calculateStatus(
 	}
 
 	rayServiceInstance.Status.NumServeEndpoints = int32(numServeEndpoints) //nolint:gosec // This is a false positive from gosec. See https://github.com/securego/gosec/issues/1212 for more details.
-	calculateConditions(rayServiceInstance)
+
+	// Calculate conditions based on current state (endpoints, clusters, timeout, etc.)
+	calculateConditions(ctx, r, rayServiceInstance)
 
 	// The definition of `ServiceStatus` is equivalent to the `RayServiceReady` condition
 	rayServiceInstance.Status.ServiceStatus = rayv1.NotRunning
@@ -449,7 +465,7 @@ func (r *RayServiceReconciler) calculateStatus(
 	return nil
 }
 
-func calculateConditions(rayServiceInstance *rayv1.RayService) {
+func calculateConditions(ctx context.Context, r *RayServiceReconciler, rayServiceInstance *rayv1.RayService) {
 	if rayServiceInstance.Status.Conditions == nil {
 		rayServiceInstance.Status.Conditions = []metav1.Condition{}
 	}
@@ -458,10 +474,15 @@ func calculateConditions(rayServiceInstance *rayv1.RayService) {
 		setCondition(rayServiceInstance, rayv1.RayServiceReady, metav1.ConditionFalse, rayv1.RayServiceInitializing, message)
 		setCondition(rayServiceInstance, rayv1.UpgradeInProgress, metav1.ConditionFalse, rayv1.RayServiceInitializing, message)
 	}
+
 	if rayServiceInstance.Status.NumServeEndpoints > 0 {
 		setCondition(rayServiceInstance, rayv1.RayServiceReady, metav1.ConditionTrue, rayv1.NonZeroServeEndpoints, "Number of serve endpoints is greater than 0")
 	} else if meta.IsStatusConditionTrue(rayServiceInstance.Status.Conditions, string(rayv1.RayServiceReady)) {
 		setCondition(rayServiceInstance, rayv1.RayServiceReady, metav1.ConditionFalse, rayv1.ZeroServeEndpoints, "Number of serve endpoints dropped to 0")
+	} else {
+		// Check if initializing timeout has been exceeded
+		// This runs after endpoint check, so if endpoints appear, they take priority
+		markFailedOnInitializingTimeout(ctx, r, rayServiceInstance)
 	}
 
 	activeClusterName := rayServiceInstance.Status.ActiveServiceStatus.RayClusterName
@@ -472,7 +493,8 @@ func calculateConditions(rayServiceInstance *rayv1.RayService) {
 		setCondition(rayServiceInstance, rayv1.UpgradeInProgress, metav1.ConditionFalse, rayv1.NoPendingCluster, "Active Ray cluster exists and no pending Ray cluster")
 	} else {
 		cond := meta.FindStatusCondition(rayServiceInstance.Status.Conditions, string(rayv1.UpgradeInProgress))
-		if cond == nil || cond.Reason != string(rayv1.RayServiceInitializing) {
+		// Don't override the condition if RayService is initializing or has timed out
+		if cond == nil || (cond.Reason != string(rayv1.RayServiceInitializing) && cond.Reason != string(rayv1.RayServiceInitializingTimeout)) {
 			setCondition(rayServiceInstance, rayv1.UpgradeInProgress, metav1.ConditionUnknown, rayv1.NoActiveCluster, "No active Ray cluster exists, and the RayService is not initializing. Please open a GitHub issue in the KubeRay repository.")
 		}
 	}
@@ -893,23 +915,27 @@ func (r *RayServiceReconciler) reconcileRayCluster(ctx context.Context, rayServi
 }
 
 // cleanUpRayClusterInstance cleans up all the dangling RayCluster instances that are owned by the RayService instance.
-func (r *RayServiceReconciler) cleanUpRayClusterInstance(ctx context.Context, rayServiceInstance *rayv1.RayService) error {
+// Returns true if there are still RayCluster instances that need to be cleaned up (either scheduled for deletion or waiting for deletion delay).
+func (r *RayServiceReconciler) cleanUpRayClusterInstance(ctx context.Context, rayServiceInstance *rayv1.RayService) (bool, error) {
 	logger := ctrl.LoggerFrom(ctx)
 	rayClusterList := rayv1.RayClusterList{}
 
 	var err error
 	if err = r.List(ctx, &rayClusterList, common.RayServiceRayClustersAssociationOptions(rayServiceInstance).ToListOptions()...); err != nil {
-		return err
+		return false, err
 	}
 
 	// Determine the ray cluster deletion delay seconds
 	deletionDelay := RayClusterDeletionDelayDuration
 	if rayServiceInstance.Spec.RayClusterDeletionDelaySeconds != nil {
 		deletionDelay = time.Duration(*rayServiceInstance.Spec.RayClusterDeletionDelaySeconds) * time.Second
 	}
+
+	hasRayClustersToClean := false
 	// 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 {
+			hasRayClustersToClean = true
 			cachedTimestamp, exists := r.RayClusterDeletionTimestamps.Get(rayClusterInstance.Name)
 			if !exists {
 				deletionTimestamp := metav1.Now().Add(deletionDelay)
@@ -932,15 +958,15 @@ func (r *RayServiceReconciler) cleanUpRayClusterInstance(ctx context.Context, ra
 					logger.Info("reconcileRayCluster", "delete Ray cluster", rayClusterInstance.Name, "reason", reasonForDeletion)
 					if err := r.Delete(ctx, &rayClusterInstance, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil {
 						r.Recorder.Eventf(rayServiceInstance, corev1.EventTypeWarning, string(utils.FailedToDeleteRayCluster), "Failed to delete the RayCluster %s/%s: %v", rayClusterInstance.Namespace, rayClusterInstance.Name, err)
-						return err
+						return false, err
 					}
 					r.Recorder.Eventf(rayServiceInstance, corev1.EventTypeNormal, string(utils.DeletedRayCluster), "Deleted the RayCluster %s/%s", rayClusterInstance.Namespace, rayClusterInstance.Name)
 				}
 			}
 		}
 	}
 
-	return nil
+	return hasRayClustersToClean, nil
 }
 
 func (r *RayServiceReconciler) getRayClusterByNamespacedName(ctx context.Context, clusterKey client.ObjectKey) (*rayv1.RayCluster, error) {
@@ -1713,6 +1739,99 @@ func (r *RayServiceReconciler) isHeadPodRunningAndReady(ctx context.Context, ins
 	return utils.IsRunningAndReady(headPod), nil
 }
 
+// getInitializingTimeout parses the initializing timeout annotation from RayService.
+// Returns (timeout, true) if valid. Accepts Go duration format (e.g., "5m", "1h") or integer seconds.
+// The annotation is assumed to be already validated by ValidateRayServiceMetadata.
+// If the annotation is absent, returns (0, false).
+func getInitializingTimeout(rs *rayv1.RayService) (time.Duration, bool) {
+	if rs.Annotations == nil {
+		return 0, false
+	}
+
+	timeoutStr, exists := rs.Annotations[utils.RayServiceInitializingTimeoutAnnotation]
+	if !exists || timeoutStr == "" {
+		return 0, false
+	}
+
+	// Try parsing as Go duration first (e.g., "30m", "1h")
+	// Validation already ensures this is valid, so we can ignore errors
+	if timeout, err := time.ParseDuration(timeoutStr); err == nil {
+		return timeout, true
+	}
+
+	// Try parsing as integer seconds
+	// Validation already ensures this is valid if ParseDuration failed
+	if seconds, err := strconv.Atoi(timeoutStr); err == nil {
+		return time.Duration(seconds) * time.Second, true
+	}
+
+	// This should never happen since validation ensures correctness,
+	// but we handle it gracefully by returning false
+	return 0, false
+}
+
+// isInitializingTimeout returns true if RayServiceReady is False with Reason=InitializingTimeout.
+// Once a RayService has timed out, it remains in a terminal failure state regardless of generation changes.
+func isInitializingTimeout(rs *rayv1.RayService) bool {
+	readyCond := meta.FindStatusCondition(rs.Status.Conditions, string(rayv1.RayServiceReady))
+	if readyCond == nil {
+		return false
+	}
+
+	return readyCond.Status == metav1.ConditionFalse &&
+		readyCond.Reason == string(rayv1.RayServiceInitializingTimeout)
+}
+
+// markFailedOnInitializingTimeout checks if the RayService has been initializing for too long.
+// If timeout is configured and exceeded, it marks the service as failed and triggers cleanup.
+func markFailedOnInitializingTimeout(ctx context.Context, r *RayServiceReconciler, rs *rayv1.RayService) {
+	logger := ctrl.LoggerFrom(ctx)
+
+	// Skip if no timeout is configured
+	timeout, ok := getInitializingTimeout(rs)
+	if !ok {
+		return
+	}
+
+	// Check if currently in Initializing state
+	readyCond := meta.FindStatusCondition(rs.Status.Conditions, string(rayv1.RayServiceReady))
+	if readyCond == nil {
+		return
+	}
+
+	if readyCond.Status != metav1.ConditionFalse || readyCond.Reason != string(rayv1.RayServiceInitializing) {
+		// Not in Initializing state
+		return
+	}
+
+	// Check if timeout has been exceeded
+	timeInInitializing := time.Since(readyCond.LastTransitionTime.Time)
+	if timeInInitializing < timeout {
+		// Still within timeout
+		return
+	}
+
+	// Timeout exceeded - mark as failed
+	logger.Info("RayService initializing timeout exceeded",
+		"timeout", timeout,
+		"timeInInitializing", timeInInitializing,
+		"generation", rs.Generation)
+
+	// Clear cluster names to trigger cleanup
+	rs.Status.ActiveServiceStatus.RayClusterName = ""
+	rs.Status.PendingServiceStatus.RayClusterName = ""
+
+	// Set condition to Failed with InitializingTimeout reason
+	message := fmt.Sprintf("RayService failed to become ready within the configured timeout of %s. Time spent initializing: %s",
+		timeout, timeInInitializing)
+	setCondition(rs, rayv1.RayServiceReady, metav1.ConditionFalse, rayv1.RayServiceInitializingTimeout, message)
+
+	// Emit warning event
+	r.Recorder.Eventf(rs, corev1.EventTypeWarning, string(utils.RayServiceInitializingTimeout),
+		"RayService initializing timeout exceeded after %s (configured timeout: %s)",
+		timeInInitializing, timeout)
+}
+
 // reconcilePerClusterServeService reconciles a load-balancing serve service for a given RayCluster.
 func (r *RayServiceReconciler) reconcilePerClusterServeService(ctx context.Context, rayServiceInstance *rayv1.RayService, rayClusterInstance *rayv1.RayCluster) error {
 	if rayClusterInstance == nil {