[RayService] Add skipGateway option for NewClusterWithIncrementalUpgrade

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

@@ -67,11 +67,31 @@ type ClusterUpgradeOptions struct {
 	// +kubebuilder:default:=100
 	MaxSurgePercent *int32 `json:"maxSurgePercent,omitempty"`
 	// The percentage of traffic to switch to the upgraded RayCluster at a set interval after scaling by MaxSurgePercent.
-	StepSizePercent *int32 `json:"stepSizePercent"`
+	// Not used when SkipGateway is true.
+	// +optional
+	StepSizePercent *int32 `json:"stepSizePercent,omitempty"`
 	// The interval in seconds between transferring StepSize traffic from the old to new RayCluster.
-	IntervalSeconds *int32 `json:"intervalSeconds"`
+	// Not used when SkipGateway is true.
+	// +optional
+	IntervalSeconds *int32 `json:"intervalSeconds,omitempty"`
 	// The name of the Gateway Class installed by the Kubernetes Cluster admin.
-	GatewayClassName string `json:"gatewayClassName"`
+	// Required when SkipGateway is false or unset.
+	// +optional
+	GatewayClassName string `json:"gatewayClassName,omitempty"`
+	// SkipGateway disables Gateway and HTTPRoute reconciliation and readiness checks during
+	// a NewClusterWithIncrementalUpgrade. When true, traffic splitting must be managed by a
+	// client-side load balancer that reads targetCapacity from the RayService status.
+	// target_capacity advances to the next step only after pending cluster Serve deployments
+	// report HEALTHY, ensuring each autoscaling round completes before the next begins.
+	// GatewayClassName, StepSizePercent, and IntervalSeconds are not required when SkipGateway is true.
+	// +optional
+	SkipGateway *bool `json:"skipGateway,omitempty"`
+	// UpgradeTimeoutSeconds is the maximum duration in seconds the upgrade is allowed to
+	// remain in progress. If the pending cluster does not complete the upgrade within this
+	// window, the upgrade is aborted: the pending RayCluster is deleted and a Kubernetes
+	// event is emitted. Defaults to 0 (no timeout).
+	// +optional
+	UpgradeTimeoutSeconds *int32 `json:"upgradeTimeoutSeconds,omitempty"`
 }
 
 type RayServiceUpgradeStrategy struct {

ray-operator/controllers/ray/rayservice_controller.go

@@ -151,6 +151,13 @@ func (r *RayServiceReconciler) Reconcile(ctx context.Context, request ctrl.Reque
 		return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, client.IgnoreNotFound(err)
 	}
 
+	// Check if the upgrade has exceeded UpgradeTimeoutSeconds; abort if so.
+	if aborted, err := r.abortUpgradeIfTimedOut(ctx, rayServiceInstance, pendingRayClusterInstance); err != nil {
+		return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, err
+	} else if aborted {
+		return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, nil
+	}
+
 	// Check both active and pending Ray clusters to see if the head Pod is ready to serve requests.
 	// This is important to ensure the reliability of the serve service because the head Pod cannot
 	// rely on readiness probes to determine serve readiness.
@@ -200,17 +207,19 @@ func (r *RayServiceReconciler) Reconcile(ctx context.Context, request ctrl.Reque
 		if err = r.reconcilePerClusterServeService(ctx, rayServiceInstance, pendingRayClusterInstance); err != nil {
 			return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, err
 		}
-		// Creates or updates a Gateway CR that points to the Serve services of
-		// the active and pending (if it exists) RayClusters. For incremental upgrades,
-		// the Gateway endpoint is used rather than the Serve service.
-		err = r.reconcileGateway(ctx, rayServiceInstance)
-		if err != nil {
-			return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, client.IgnoreNotFound(err)
-		}
-		// Create or update the HTTPRoute for the Gateway, passing in the pending cluster readiness status.
-		httpRouteInstance, err = r.reconcileHTTPRoute(ctx, rayServiceInstance, isPendingClusterReady)
-		if err != nil {
-			return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, client.IgnoreNotFound(err)
+		if !utils.IsSkipGateway(&rayServiceInstance.Spec) {
+			// Creates or updates a Gateway CR that points to the Serve services of
+			// the active and pending (if it exists) RayClusters. For incremental upgrades,
+			// the Gateway endpoint is used rather than the Serve service.
+			err = r.reconcileGateway(ctx, rayServiceInstance)
+			if err != nil {
+				return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, client.IgnoreNotFound(err)
+			}
+			// Create or update the HTTPRoute for the Gateway, passing in the pending cluster readiness status.
+			httpRouteInstance, err = r.reconcileHTTPRoute(ctx, rayServiceInstance, isPendingClusterReady)
+			if err != nil {
+				return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, client.IgnoreNotFound(err)
+			}
 		}
 	}
 
@@ -251,6 +260,7 @@ func (r *RayServiceReconciler) Reconcile(ctx context.Context, request ctrl.Reque
 		activeClusterServeApplications,
 		pendingClusterServeApplications,
 		httpRouteInstance,
+		isPendingClusterReady,
 	); err != nil {
 		return ctrl.Result{RequeueAfter: ServiceDefaultRequeueDuration}, err
 	}
@@ -341,6 +351,7 @@ func (r *RayServiceReconciler) calculateStatus(
 	activeCluster, pendingCluster *rayv1.RayCluster,
 	activeClusterServeApplications, pendingClusterServeApplications map[string]rayv1.AppStatus,
 	httpRoute *gwv1.HTTPRoute,
+	isPendingClusterReady bool,
 ) error {
 	logger := ctrl.LoggerFrom(ctx)
 
@@ -368,20 +379,45 @@ func (r *RayServiceReconciler) calculateStatus(
 			oldActivePercent := ptr.Deref(rayServiceInstance.Status.ActiveServiceStatus.TrafficRoutedPercent, -1)
 			oldPendingPercent := ptr.Deref(rayServiceInstance.Status.PendingServiceStatus.TrafficRoutedPercent, -1)
 
-			// Update TrafficRoutedPercent to each RayService based on current weights from HTTPRoute.
-			activeWeight, pendingWeight := utils.GetWeightsFromHTTPRoute(httpRoute, rayServiceInstance)
 			now := metav1.Time{Time: time.Now()}
+			// Default sentinel: -1 means "don't update TrafficRoutedPercent this reconcile".
+			activeWeight, pendingWeight := int32(-1), int32(-1)
+			if utils.IsSkipGateway(&rayServiceInstance.Spec) {
+				if pendingCluster != nil {
+					// SkipGateway + endpoint gate: set TrafficRoutedPercent = TargetCapacity only
+					// when the pending cluster's Serve deployments are HEALTHY, meaning the
+					// autoscaling round has fully settled. The gate in reconcileServeTargetCapacity
+					// advances target_capacity only when TrafficRoutedPercent == TargetCapacity.
+					// The client-side LB reads targetCapacity from status directly for traffic routing
+					// and does not rely on TrafficRoutedPercent.
+					pendingTargetCap := ptr.Deref(rayServiceInstance.Status.PendingServiceStatus.TargetCapacity, 0)
+					if isPendingClusterCapacityReady(pendingClusterServeApplications, activeCluster, pendingCluster, pendingTargetCap) {
+						pendingWeight = pendingTargetCap
+						activeWeight = 100 - pendingWeight
+						logger.Info("SkipGateway: pending cluster capacity ready, advancing TrafficRoutedPercent to match TargetCapacity.",
+							"pendingTargetCapacity", pendingTargetCap, "pendingWeight", pendingWeight, "activeWeight", activeWeight)
+					} else {
+						logger.Info("SkipGateway: pending cluster not yet capacity-ready, holding TrafficRoutedPercent until deployments are healthy and replicas are sufficient.",
+							"pendingTargetCapacity", pendingTargetCap)
+					}
+				}
+				// No pending cluster: no upgrade in progress, leave TrafficRoutedPercent unchanged.
+			} else {
+				// Update TrafficRoutedPercent to each RayService based on current weights from HTTPRoute.
+				activeWeight, pendingWeight = utils.GetWeightsFromHTTPRoute(httpRoute, rayServiceInstance)
+				logger.Info("Calculated TrafficRoutedPercent from HTTPRoute", "activeWeight", activeWeight, "pendingWeight", pendingWeight)
+			}
 			if activeWeight >= 0 {
 				rayServiceInstance.Status.ActiveServiceStatus.TrafficRoutedPercent = ptr.To(activeWeight)
-				logger.Info("Updated active TrafficRoutedPercent from HTTPRoute", "activeClusterWeight", activeWeight)
+				logger.Info("Updated active TrafficRoutedPercent", "activeClusterWeight", activeWeight)
 				if activeWeight != oldActivePercent {
 					rayServiceInstance.Status.ActiveServiceStatus.LastTrafficMigratedTime = &now
 					logger.Info("Updated LastTrafficMigratedTime of Active Service.")
 				}
 			}
 			if pendingWeight >= 0 {
 				rayServiceInstance.Status.PendingServiceStatus.TrafficRoutedPercent = ptr.To(pendingWeight)
-				logger.Info("Updated pending TrafficRoutedPercent from HTTPRoute", "pendingClusterWeight", pendingWeight)
+				logger.Info("Updated pending TrafficRoutedPercent", "pendingClusterWeight", pendingWeight)
 				if pendingWeight != oldPendingPercent {
 					rayServiceInstance.Status.PendingServiceStatus.LastTrafficMigratedTime = &now
 					logger.Info("Updated LastTrafficMigratedTime of Pending Service.")
@@ -1192,6 +1228,117 @@ func constructRayClusterForRayService(rayService *rayv1.RayService, rayClusterNa
 	return rayCluster, nil
 }
 
+// arePendingServeDeploymentsHealthy returns true when all Serve applications on the pending
+// cluster are RUNNING and all their deployments are HEALTHY. This is the signal that one
+// autoscaling round has completed at the current target_capacity, allowing the controller to
+// advance to the next step. Returns false when apps is empty (not yet initialised).
+func arePendingServeDeploymentsHealthy(apps map[string]rayv1.AppStatus) bool {
+	if len(apps) == 0 {
+		return false
+	}
+	for _, app := range apps {
+		if app.Status != rayv1.ApplicationStatusEnum.RUNNING {
+			return false
+		}
+		for _, deployment := range app.Deployments {
+			if deployment.Status != rayv1.DeploymentStatusEnum.HEALTHY {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+// isPendingClusterCapacityReady returns true when it is safe to advance TrafficRoutedPercent
+// to pendingTargetCap for a skipGateway upgrade. It combines two gates:
+//
+//  1. Serve health: all pending cluster deployments must be HEALTHY (same as the base health gate).
+//  2. Capacity parity: the pending cluster must have enough available worker replicas to handle
+//     the traffic level it is about to receive.
+//
+// Parity check formula: pending.availableWorkers >= ceil(pendingTargetCap/100 * active.availableWorkers)
+//
+// The parity check is skipped (falls back to health-only) when:
+//   - Either cluster object is nil (no information to compare).
+//   - pending.maxReplicas < active.availableWorkers: the upgrade intentionally reduces capacity
+//     (e.g. cost reduction). Requiring parity would deadlock the upgrade since the pending
+//     cluster can never reach the active cluster's current replica count.
+//
+// TODO: Consider snapshotting the active cluster's replica count at upgrade start rather than
+// using the live count. Without a snapshot, a spike that scales the active cluster mid-upgrade
+// raises the parity bar and may stall the upgrade until load subsides.
+func isPendingClusterCapacityReady(
+	apps map[string]rayv1.AppStatus,
+	activeCluster, pendingCluster *rayv1.RayCluster,
+	pendingTargetCap int32,
+) bool {
+	if !arePendingServeDeploymentsHealthy(apps) {
+		return false
+	}
+
+	if activeCluster == nil || pendingCluster == nil {
+		return true
+	}
+
+	activeAvailable := activeCluster.Status.AvailableWorkerReplicas
+	pendingAvailable := pendingCluster.Status.AvailableWorkerReplicas
+
+	// Sum maxReplicas across all worker groups on the pending cluster.
+	var pendingMaxReplicas int32
+	for _, wg := range pendingCluster.Spec.WorkerGroupSpecs {
+		if wg.MaxReplicas != nil {
+			pendingMaxReplicas += *wg.MaxReplicas
+		}
+	}
+
+	// Intentional capacity reduction: skip parity to avoid deadlock.
+	if pendingMaxReplicas < activeAvailable {
+		return true
+	}
+
+	// ceil(pendingTargetCap/100 * activeAvailable) using integer arithmetic.
+	requiredReplicas := (pendingTargetCap*activeAvailable + 99) / 100
+	return pendingAvailable >= requiredReplicas
+}
+
+// abortUpgradeIfTimedOut checks whether the pending cluster has exceeded UpgradeTimeoutSeconds.
+// If so, it deletes the pending RayCluster and emits a Warning event. Returns (true, nil) when the
+// upgrade was aborted, (false, nil) when no timeout has occurred, and (false, err) on delete failure.
+func (r *RayServiceReconciler) abortUpgradeIfTimedOut(ctx context.Context, rayServiceInstance *rayv1.RayService, pendingCluster *rayv1.RayCluster) (bool, error) {
+	if pendingCluster == nil {
+		return false, nil
+	}
+	if !utils.IsIncrementalUpgradeEnabled(&rayServiceInstance.Spec) {
+		return false, nil
+	}
+	options := utils.GetRayServiceClusterUpgradeOptions(&rayServiceInstance.Spec)
+	if options == nil || options.UpgradeTimeoutSeconds == nil || *options.UpgradeTimeoutSeconds <= 0 {
+		return false, nil
+	}
+
+	elapsed := time.Since(pendingCluster.CreationTimestamp.Time)
+	timeout := time.Duration(*options.UpgradeTimeoutSeconds) * time.Second
+	if elapsed <= timeout {
+		return false, nil
+	}
+
+	logger := ctrl.LoggerFrom(ctx)
+	logger.Info("Upgrade timed out, aborting by deleting pending RayCluster.",
+		"pendingCluster", pendingCluster.Name,
+		"elapsed", elapsed.String(),
+		"timeout", timeout.String())
+
+	if err := r.Delete(ctx, pendingCluster, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil {
+		r.Recorder.Eventf(rayServiceInstance, corev1.EventTypeWarning, string(utils.FailedToDeleteRayCluster),
+			"Upgrade timed out but failed to delete pending RayCluster %s/%s: %v", pendingCluster.Namespace, pendingCluster.Name, err)
+		return false, err
+	}
+	r.Recorder.Eventf(rayServiceInstance, corev1.EventTypeWarning, string(utils.UpgradeTimeout),
+		"Upgrade exceeded timeout of %ds; deleted pending RayCluster %s/%s after %s elapsed",
+		*options.UpgradeTimeoutSeconds, pendingCluster.Namespace, pendingCluster.Name, elapsed.Round(time.Second).String())
+	return true, nil
+}
+
 func checkIfNeedSubmitServeApplications(cachedServeConfigV2 string, serveConfigV2 string, serveApplications map[string]rayv1.AppStatus) (bool, string) {
 	// If the Serve config has not been cached, update the Serve config.
 	if cachedServeConfigV2 == "" {
@@ -1286,28 +1433,38 @@ func (r *RayServiceReconciler) updateServeDeployment(ctx context.Context, raySer
 // The function ensures that traffic migration only proceeds when the target cluster has reached
 // its capacity limit, preventing resource conflicts and ensuring upgrade stability.
 func (r *RayServiceReconciler) checkIfNeedTargetCapacityUpdate(ctx context.Context, rayServiceInstance *rayv1.RayService) (bool, string) {
+	logger := ctrl.LoggerFrom(ctx)
 	activeRayServiceStatus := rayServiceInstance.Status.ActiveServiceStatus
 	pendingRayServiceStatus := rayServiceInstance.Status.PendingServiceStatus
 
 	if activeRayServiceStatus.RayClusterName == "" || pendingRayServiceStatus.RayClusterName == "" {
 		return false, "Both active and pending RayCluster instances are required for NewClusterWithIncrementalUpgrade."
 	}
 
-	// Validate Gateway and HTTPRoute objects are ready
-	gatewayInstance := &gwv1.Gateway{}
-	if err := r.Get(ctx, common.RayServiceGatewayNamespacedName(rayServiceInstance), gatewayInstance); err != nil {
-		return false, fmt.Sprintf("Failed to retrieve Gateway for RayService: %v", err)
-	}
-	if !utils.IsGatewayReady(gatewayInstance) {
-		return false, "Gateway for RayService NewClusterWithIncrementalUpgrade is not ready."
-	}
+	if utils.IsSkipGateway(&rayServiceInstance.Spec) {
+		// Gateway and HTTPRoute reconciliation is disabled. Traffic splitting is managed by a
+		// client-side load balancer reading targetCapacity from RayService status. Skip the
+		// Gateway and HTTPRoute readiness checks and proceed directly to target_capacity logic.
+		logger.Info("SkipGateway is enabled: skipping Gateway and HTTPRoute readiness checks in checkIfNeedTargetCapacityUpdate.")
+	} else {
+		// Validate Gateway and HTTPRoute objects are ready
+		gatewayInstance := &gwv1.Gateway{}
+		if err := r.Get(ctx, common.RayServiceGatewayNamespacedName(rayServiceInstance), gatewayInstance); err != nil {
+			return false, fmt.Sprintf("Failed to retrieve Gateway for RayService: %v", err)
+		}
+		if !utils.IsGatewayReady(gatewayInstance) {
+			return false, "Gateway for RayService NewClusterWithIncrementalUpgrade is not ready."
+		}
+		logger.Info("Gateway for RayService NewClusterWithIncrementalUpgrade is ready.")
 
-	httpRouteInstance := &gwv1.HTTPRoute{}
-	if err := r.Get(ctx, common.RayServiceHTTPRouteNamespacedName(rayServiceInstance), httpRouteInstance); err != nil {
-		return false, fmt.Sprintf("Failed to retrieve HTTPRoute for RayService: %v", err)
-	}
-	if !utils.IsHTTPRouteReady(gatewayInstance, httpRouteInstance) {
-		return false, "HTTPRoute for RayService NewClusterWithIncrementalUpgrade is not ready."
+		httpRouteInstance := &gwv1.HTTPRoute{}
+		if err := r.Get(ctx, common.RayServiceHTTPRouteNamespacedName(rayServiceInstance), httpRouteInstance); err != nil {
+			return false, fmt.Sprintf("Failed to retrieve HTTPRoute for RayService: %v", err)
+		}
+		if !utils.IsHTTPRouteReady(gatewayInstance, httpRouteInstance) {
+			return false, "HTTPRoute for RayService NewClusterWithIncrementalUpgrade is not ready."
+		}
+		logger.Info("HTTPRoute for RayService NewClusterWithIncrementalUpgrade is ready.")
 	}
 
 	// Retrieve the current observed NewClusterWithIncrementalUpgrade Status fields for each RayService.