From 6cbc10a0aa775c9489e693fa4ba983cd7879f767 Mon Sep 17 00:00:00 2001 From: Savita Manghnani Date: Wed, 3 Jun 2026 15:58:09 -0400 Subject: [PATCH 1/7] [RayService] Add skipGateway option for client-side load balancer support Introduces `skipGateway` field in `ClusterUpgradeOptions` for the `NewClusterWithIncrementalUpgrade` strategy. When enabled: - Gateway and HTTPRoute objects are not created or reconciled - Gateway/HTTPRoute readiness checks are skipped in the target_capacity advancement gate (checkIfNeedTargetCapacityUpdate) - TrafficRoutedPercent advances via the existing time-step logic (IntervalSeconds + StepSizePercent) instead of being read from HTTPRoute backend ref weights This allows deployments using client-side load balancers to use incremental upgrade without requiring a Gateway API implementation. The client-side LB is responsible for reading targetCapacity from RayService status and splitting traffic accordingly. GatewayClassName is now optional when skipGateway is true. Co-Authored-By: Claude Sonnet 4.6 --- ray-operator/apis/ray/v1/rayservice_types.go | 11 ++- .../apis/ray/v1/zz_generated.deepcopy.go | 5 ++ .../config/crd/bases/ray.io_rayservices.yaml | 3 +- .../controllers/ray/rayservice_controller.go | 89 +++++++++++++------ ray-operator/controllers/ray/utils/util.go | 11 +++ 5 files changed, 88 insertions(+), 31 deletions(-) diff --git a/ray-operator/apis/ray/v1/rayservice_types.go b/ray-operator/apis/ray/v1/rayservice_types.go index 83b9775f377..62566a72037 100644 --- a/ray-operator/apis/ray/v1/rayservice_types.go +++ b/ray-operator/apis/ray/v1/rayservice_types.go @@ -71,7 +71,16 @@ type ClusterUpgradeOptions struct { // The interval in seconds between transferring StepSize traffic from the old to new RayCluster. IntervalSeconds *int32 `json:"intervalSeconds"` // 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. + // TrafficRoutedPercent advances on a time-step basis using IntervalSeconds and StepSizePercent. + // GatewayClassName is not required when SkipGateway is true. + // +optional + SkipGateway *bool `json:"skipGateway,omitempty"` } type RayServiceUpgradeStrategy struct { diff --git a/ray-operator/apis/ray/v1/zz_generated.deepcopy.go b/ray-operator/apis/ray/v1/zz_generated.deepcopy.go index cd710592d98..49c21f144f9 100644 --- a/ray-operator/apis/ray/v1/zz_generated.deepcopy.go +++ b/ray-operator/apis/ray/v1/zz_generated.deepcopy.go @@ -136,6 +136,11 @@ func (in *ClusterUpgradeOptions) DeepCopyInto(out *ClusterUpgradeOptions) { *out = new(int32) **out = **in } + if in.SkipGateway != nil { + in, out := &in.SkipGateway, &out.SkipGateway + *out = new(bool) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterUpgradeOptions. diff --git a/ray-operator/config/crd/bases/ray.io_rayservices.yaml b/ray-operator/config/crd/bases/ray.io_rayservices.yaml index baec5cfe1f3..ea0f0c3c1c1 100644 --- a/ray-operator/config/crd/bases/ray.io_rayservices.yaml +++ b/ray-operator/config/crd/bases/ray.io_rayservices.yaml @@ -8593,11 +8593,12 @@ spec: default: 100 format: int32 type: integer + skipGateway: + type: boolean stepSizePercent: format: int32 type: integer required: - - gatewayClassName - intervalSeconds - stepSizePercent type: object diff --git a/ray-operator/controllers/ray/rayservice_controller.go b/ray-operator/controllers/ray/rayservice_controller.go index 8d05334b7de..d90b2b5e4ea 100644 --- a/ray-operator/controllers/ray/rayservice_controller.go +++ b/ray-operator/controllers/ray/rayservice_controller.go @@ -200,17 +200,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 +253,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 +344,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,12 +372,29 @@ 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()} + var activeWeight, pendingWeight int32 + if utils.IsSkipGateway(&rayServiceInstance.Spec) { + // No Gateway: advance TrafficRoutedPercent using the time-step calculation so + // the target_capacity advancement gate in reconcileServeTargetCapacity has a + // value to compare against. The client-side load balancer is responsible for + // reading targetCapacity from RayService status to split traffic accordingly. + logger.Info("SkipGateway is enabled: computing TrafficRoutedPercent via time-step instead of HTTPRoute weights.") + var err error + activeWeight, pendingWeight, err = r.calculateTrafficRoutedPercent(ctx, rayServiceInstance, isPendingClusterReady) + if err != nil { + logger.Info("Failed to calculate TrafficRoutedPercent for skipGateway mode.") + } else { + logger.Info("Calculated TrafficRoutedPercent via time-step (skipGateway)", "activeWeight", activeWeight, "pendingWeight", pendingWeight) + } + } 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.") @@ -381,7 +402,7 @@ func (r *RayServiceReconciler) calculateStatus( } 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.") @@ -1286,6 +1307,7 @@ 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 @@ -1293,21 +1315,30 @@ func (r *RayServiceReconciler) checkIfNeedTargetCapacityUpdate(ctx context.Conte 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. diff --git a/ray-operator/controllers/ray/utils/util.go b/ray-operator/controllers/ray/utils/util.go index 70b1dadf3e1..becbb441a93 100644 --- a/ray-operator/controllers/ray/utils/util.go +++ b/ray-operator/controllers/ray/utils/util.go @@ -820,6 +820,17 @@ func GetRayServiceClusterUpgradeOptions(spec *rayv1.RayServiceSpec) *rayv1.Clust return nil } +// IsSkipGateway returns true when the RayService is configured to skip Gateway and HTTPRoute +// reconciliation during a NewClusterWithIncrementalUpgrade. When true, traffic splitting is +// managed externally by a client-side load balancer reading targetCapacity from RayService status. +func IsSkipGateway(spec *rayv1.RayServiceSpec) bool { + options := GetRayServiceClusterUpgradeOptions(spec) + if options == nil { + return false + } + return ptr.Deref(options.SkipGateway, false) +} + // IsIncrementalUpgradeComplete checks if the conditions for completing an incremental upgrade are met. func IsIncrementalUpgradeComplete(rayServiceInstance *rayv1.RayService, pendingCluster *rayv1.RayCluster) bool { return pendingCluster != nil && From d85d6397b514adb18baadc2ea50497a5902fd30c Mon Sep 17 00:00:00 2001 From: Savita Manghnani Date: Thu, 4 Jun 2026 16:23:04 -0400 Subject: [PATCH 2/7] Replace timer-based traffic step with serve deployment health gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In skipGateway mode, TrafficRoutedPercent previously advanced on a fixed timer (intervalSeconds + stepSizePercent). This caused capacity gaps when pod cold-start exceeded intervalSeconds, routing traffic to pending before replicas were ready. Replace with a serve-deployment health gate: TrafficRoutedPercent is set to match TargetCapacity only when the pending cluster's Serve deployments report HEALTHY, meaning autoscaling has fully settled at the current target_capacity. The existing reconcileServeTargetCapacity gate (pendingTargetCapacity != pendingTrafficRoutedPercent) then naturally advances to the next step — no timer needed. IntervalSeconds and StepSizePercent are made optional (omitempty) since they are not used in skipGateway mode. GatewayClassName was already optional. MaxSurgePercent remains the only functional field for skipGateway: it controls how large each target_capacity step is. Co-Authored-By: Claude Sonnet 4.6 --- ray-operator/apis/ray/v1/rayservice_types.go | 13 +++-- .../config/crd/bases/ray.io_rayservices.yaml | 3 -- .../controllers/ray/rayservice_controller.go | 48 +++++++++++++++---- 3 files changed, 47 insertions(+), 17 deletions(-) diff --git a/ray-operator/apis/ray/v1/rayservice_types.go b/ray-operator/apis/ray/v1/rayservice_types.go index 62566a72037..bd731c2f547 100644 --- a/ray-operator/apis/ray/v1/rayservice_types.go +++ b/ray-operator/apis/ray/v1/rayservice_types.go @@ -67,9 +67,13 @@ 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. // Required when SkipGateway is false or unset. // +optional @@ -77,8 +81,9 @@ type ClusterUpgradeOptions struct { // 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. - // TrafficRoutedPercent advances on a time-step basis using IntervalSeconds and StepSizePercent. - // GatewayClassName is not required when SkipGateway is true. + // 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"` } diff --git a/ray-operator/config/crd/bases/ray.io_rayservices.yaml b/ray-operator/config/crd/bases/ray.io_rayservices.yaml index ea0f0c3c1c1..e64bf2c3937 100644 --- a/ray-operator/config/crd/bases/ray.io_rayservices.yaml +++ b/ray-operator/config/crd/bases/ray.io_rayservices.yaml @@ -8598,9 +8598,6 @@ spec: stepSizePercent: format: int32 type: integer - required: - - intervalSeconds - - stepSizePercent type: object type: type: string diff --git a/ray-operator/controllers/ray/rayservice_controller.go b/ray-operator/controllers/ray/rayservice_controller.go index d90b2b5e4ea..a4d7fbca9e3 100644 --- a/ray-operator/controllers/ray/rayservice_controller.go +++ b/ray-operator/controllers/ray/rayservice_controller.go @@ -375,17 +375,24 @@ func (r *RayServiceReconciler) calculateStatus( now := metav1.Time{Time: time.Now()} var activeWeight, pendingWeight int32 if utils.IsSkipGateway(&rayServiceInstance.Spec) { - // No Gateway: advance TrafficRoutedPercent using the time-step calculation so - // the target_capacity advancement gate in reconcileServeTargetCapacity has a - // value to compare against. The client-side load balancer is responsible for - // reading targetCapacity from RayService status to split traffic accordingly. - logger.Info("SkipGateway is enabled: computing TrafficRoutedPercent via time-step instead of HTTPRoute weights.") - var err error - activeWeight, pendingWeight, err = r.calculateTrafficRoutedPercent(ctx, rayServiceInstance, isPendingClusterReady) - if err != nil { - logger.Info("Failed to calculate TrafficRoutedPercent for skipGateway mode.") + // 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 arePendingServeDeploymentsHealthy(pendingClusterServeApplications) { + pendingWeight = pendingTargetCap + activeWeight = 100 - pendingWeight + logger.Info("SkipGateway: pending Serve deployments HEALTHY, advancing TrafficRoutedPercent to match TargetCapacity.", + "pendingTargetCapacity", pendingTargetCap, "pendingWeight", pendingWeight, "activeWeight", activeWeight) } else { - logger.Info("Calculated TrafficRoutedPercent via time-step (skipGateway)", "activeWeight", activeWeight, "pendingWeight", pendingWeight) + // Gate stays blocked: leave TrafficRoutedPercent at previous value by not updating. + activeWeight = -1 + pendingWeight = -1 + logger.Info("SkipGateway: pending Serve deployments not yet HEALTHY, holding TrafficRoutedPercent until autoscaling settles.", + "pendingTargetCapacity", pendingTargetCap) } } else { // Update TrafficRoutedPercent to each RayService based on current weights from HTTPRoute. @@ -1213,6 +1220,27 @@ 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 +} + 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 == "" { From 3ee51aa4631cb24c51c256c94be792d54a19a620 Mon Sep 17 00:00:00 2001 From: Savita Manghnani Date: Wed, 10 Jun 2026 12:43:38 -0400 Subject: [PATCH 3/7] [RayService] Skip gateway validations when skipGateway is true MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GatewayClassName, StepSizePercent, and IntervalSeconds are not needed when skipGateway is enabled — guard the checks accordingly. Co-Authored-By: Claude Sonnet 4.6 --- .../controllers/ray/utils/validation.go | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/ray-operator/controllers/ray/utils/validation.go b/ray-operator/controllers/ray/utils/validation.go index 652ff0e14b6..fc875a914c6 100644 --- a/ray-operator/controllers/ray/utils/validation.go +++ b/ray-operator/controllers/ray/utils/validation.go @@ -403,16 +403,20 @@ func ValidateClusterUpgradeOptions(rayService *rayv1.RayService) error { return fmt.Errorf("maxSurgePercent must be between 0 and 100") } - if options.StepSizePercent == nil || *options.StepSizePercent < 0 || *options.StepSizePercent > 100 { - return fmt.Errorf("stepSizePercent must be between 0 and 100") - } + skipGateway := options.SkipGateway != nil && *options.SkipGateway - if options.IntervalSeconds == nil || *options.IntervalSeconds <= 0 { - return fmt.Errorf("intervalSeconds must be greater than 0") - } + if !skipGateway { + if options.StepSizePercent == nil || *options.StepSizePercent < 0 || *options.StepSizePercent > 100 { + return fmt.Errorf("stepSizePercent must be between 0 and 100") + } - if options.GatewayClassName == "" { - return fmt.Errorf("gatewayClassName is required for NewClusterWithIncrementalUpgrade") + if options.IntervalSeconds == nil || *options.IntervalSeconds <= 0 { + return fmt.Errorf("intervalSeconds must be greater than 0") + } + + if options.GatewayClassName == "" { + return fmt.Errorf("gatewayClassName is required for NewClusterWithIncrementalUpgrade") + } } return nil From dc97d6216b685a6d4cc31f3599c2ef71d08ea097 Mon Sep 17 00:00:00 2001 From: Savita Manghnani Date: Thu, 11 Jun 2026 16:48:24 -0400 Subject: [PATCH 4/7] skipGateway: capacity-parity check, upgrade timeout, and validation fixes - Replace arePendingServeDeploymentsHealthy with isPendingClusterCapacityReady at the TrafficRoutedPercent gate so that the pending cluster must have enough available worker replicas (ceil(pendingTargetCap/100 * active.availableWorkers)) before traffic routing advances. Includes deadlock-safe bypass when pendingMaxReplicas < activeAvailableWorkers (intentional capacity reduction). - Add UpgradeTimeoutSeconds field to ClusterUpgradeOptions. When set, the operator aborts the upgrade by deleting the pending RayCluster and emitting a Warning event if the pending cluster's age exceeds the timeout. - Add UpgradeTimeout event type constant. - Add validation: upgradeTimeoutSeconds must be >= 0 when set. Co-Authored-By: Claude Sonnet 4.6 --- ray-operator/apis/ray/v1/rayservice_types.go | 6 + .../controllers/ray/rayservice_controller.go | 103 +++++++++++++++++- .../controllers/ray/utils/constant.go | 1 + .../controllers/ray/utils/validation.go | 4 + 4 files changed, 111 insertions(+), 3 deletions(-) diff --git a/ray-operator/apis/ray/v1/rayservice_types.go b/ray-operator/apis/ray/v1/rayservice_types.go index bd731c2f547..6dbc3daab6f 100644 --- a/ray-operator/apis/ray/v1/rayservice_types.go +++ b/ray-operator/apis/ray/v1/rayservice_types.go @@ -86,6 +86,12 @@ type ClusterUpgradeOptions struct { // 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 { diff --git a/ray-operator/controllers/ray/rayservice_controller.go b/ray-operator/controllers/ray/rayservice_controller.go index a4d7fbca9e3..120dcc47311 100644 --- a/ray-operator/controllers/ray/rayservice_controller.go +++ b/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. @@ -382,16 +389,16 @@ func (r *RayServiceReconciler) calculateStatus( // 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 arePendingServeDeploymentsHealthy(pendingClusterServeApplications) { + if isPendingClusterCapacityReady(pendingClusterServeApplications, activeCluster, pendingCluster, pendingTargetCap) { pendingWeight = pendingTargetCap activeWeight = 100 - pendingWeight - logger.Info("SkipGateway: pending Serve deployments HEALTHY, advancing TrafficRoutedPercent to match TargetCapacity.", + logger.Info("SkipGateway: pending cluster capacity ready, advancing TrafficRoutedPercent to match TargetCapacity.", "pendingTargetCapacity", pendingTargetCap, "pendingWeight", pendingWeight, "activeWeight", activeWeight) } else { // Gate stays blocked: leave TrafficRoutedPercent at previous value by not updating. activeWeight = -1 pendingWeight = -1 - logger.Info("SkipGateway: pending Serve deployments not yet HEALTHY, holding TrafficRoutedPercent until autoscaling settles.", + logger.Info("SkipGateway: pending cluster not yet capacity-ready, holding TrafficRoutedPercent until deployments are healthy and replicas are sufficient.", "pendingTargetCapacity", pendingTargetCap) } } else { @@ -1241,6 +1248,96 @@ func arePendingServeDeploymentsHealthy(apps map[string]rayv1.AppStatus) bool { 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 == "" { diff --git a/ray-operator/controllers/ray/utils/constant.go b/ray-operator/controllers/ray/utils/constant.go index 38b94c9b715..9484847b01d 100644 --- a/ray-operator/controllers/ray/utils/constant.go +++ b/ray-operator/controllers/ray/utils/constant.go @@ -371,6 +371,7 @@ const ( FailedToUpdateGateway K8sEventType = "FailedToUpdateGateway" FailedToCreateHTTPRoute K8sEventType = "FailedToCreateHTTPRoute" FailedToUpdateHTTPRoute K8sEventType = "FailedToUpdateHTTPRoute" + UpgradeTimeout K8sEventType = "UpgradeTimeout" // Generic Pod event list DeletedPod K8sEventType = "DeletedPod" diff --git a/ray-operator/controllers/ray/utils/validation.go b/ray-operator/controllers/ray/utils/validation.go index fc875a914c6..07ada3adb39 100644 --- a/ray-operator/controllers/ray/utils/validation.go +++ b/ray-operator/controllers/ray/utils/validation.go @@ -419,6 +419,10 @@ func ValidateClusterUpgradeOptions(rayService *rayv1.RayService) error { } } + if options.UpgradeTimeoutSeconds != nil && *options.UpgradeTimeoutSeconds < 0 { + return fmt.Errorf("upgradeTimeoutSeconds must be greater than or equal to 0") + } + return nil } From 88b0001cf0ccfe1836879d55731fa88738f162e9 Mon Sep 17 00:00:00 2001 From: Savita Manghnani Date: Sat, 13 Jun 2026 05:43:18 -0400 Subject: [PATCH 5/7] skipGateway: skip autoscaler requirement when skipGateway=true autoscaler is required for gateway-based incremental traffic routing but not for skipGateway mode, where traffic routing is handled by a client-side load balancer reading targetCapacity from RayService status. Co-Authored-By: Claude Sonnet 4.6 --- .../controllers/ray/utils/validation.go | 14 ++++++----- .../controllers/ray/utils/validation_test.go | 25 ++++++++++++++++++- 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/ray-operator/controllers/ray/utils/validation.go b/ray-operator/controllers/ray/utils/validation.go index 07ada3adb39..6353c34bc14 100644 --- a/ray-operator/controllers/ray/utils/validation.go +++ b/ray-operator/controllers/ray/utils/validation.go @@ -389,22 +389,24 @@ func ValidateRayServiceSpec(rayService *rayv1.RayService) error { } func ValidateClusterUpgradeOptions(rayService *rayv1.RayService) error { - if !IsAutoscalingEnabled(&rayService.Spec.RayClusterSpec) { - return fmt.Errorf("Ray Autoscaler is required for NewClusterWithIncrementalUpgrade") - } - options := rayService.Spec.UpgradeStrategy.ClusterUpgradeOptions if options == nil { return fmt.Errorf("ClusterUpgradeOptions are required for NewClusterWithIncrementalUpgrade") } + skipGateway := options.SkipGateway != nil && *options.SkipGateway + + // Autoscaler is required for gateway-based incremental upgrade but not for skipGateway mode, + // which uses client-side load balancing driven by targetCapacity in RayService status. + if !skipGateway && !IsAutoscalingEnabled(&rayService.Spec.RayClusterSpec) { + return fmt.Errorf("Ray Autoscaler is required for NewClusterWithIncrementalUpgrade") + } + // MaxSurgePercent defaults to 100% if unset. if *options.MaxSurgePercent < 0 || *options.MaxSurgePercent > 100 { return fmt.Errorf("maxSurgePercent must be between 0 and 100") } - skipGateway := options.SkipGateway != nil && *options.SkipGateway - if !skipGateway { if options.StepSizePercent == nil || *options.StepSizePercent < 0 || *options.StepSizePercent > 100 { return fmt.Errorf("stepSizePercent must be between 0 and 100") diff --git a/ray-operator/controllers/ray/utils/validation_test.go b/ray-operator/controllers/ray/utils/validation_test.go index 014d0917d68..ccad901eaf1 100644 --- a/ray-operator/controllers/ray/utils/validation_test.go +++ b/ray-operator/controllers/ray/utils/validation_test.go @@ -1783,6 +1783,7 @@ func TestValidateClusterUpgradeOptions(t *testing.T) { gatewayClassName string spec rayv1.RayServiceSpec enableAutoscaling bool + skipGateway bool expectError bool }{ { @@ -1842,12 +1843,33 @@ func TestValidateClusterUpgradeOptions(t *testing.T) { enableAutoscaling: true, expectError: true, }, + { + name: "skipGateway valid without autoscaler", + maxSurgePercent: ptr.To(int32(20)), + skipGateway: true, + enableAutoscaling: false, + expectError: false, + }, + { + name: "skipGateway valid with autoscaler", + maxSurgePercent: ptr.To(int32(20)), + skipGateway: true, + enableAutoscaling: true, + expectError: false, + }, + { + name: "skipGateway does not require stepSizePercent or intervalSeconds", + maxSurgePercent: ptr.To(int32(50)), + skipGateway: true, + enableAutoscaling: false, + expectError: false, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { var upgradeStrategy *rayv1.RayServiceUpgradeStrategy - if tt.maxSurgePercent != nil || tt.stepSizePercent != nil || tt.intervalSeconds != nil || tt.gatewayClassName != "" { + if tt.maxSurgePercent != nil || tt.stepSizePercent != nil || tt.intervalSeconds != nil || tt.gatewayClassName != "" || tt.skipGateway { upgradeStrategy = &rayv1.RayServiceUpgradeStrategy{ Type: ptr.To(rayv1.NewClusterWithIncrementalUpgrade), ClusterUpgradeOptions: &rayv1.ClusterUpgradeOptions{ @@ -1855,6 +1877,7 @@ func TestValidateClusterUpgradeOptions(t *testing.T) { StepSizePercent: tt.stepSizePercent, IntervalSeconds: tt.intervalSeconds, GatewayClassName: tt.gatewayClassName, + SkipGateway: ptr.To(tt.skipGateway), }, } } else if tt.expectError { From 951e4fbd50ee51d91e0c3412436520e660dbfbb8 Mon Sep 17 00:00:00 2001 From: Savita Manghnani Date: Sat, 13 Jun 2026 05:48:30 -0400 Subject: [PATCH 6/7] skipGateway: revert autoscaler exemption; keep gateway-specific field waivers autoscaler is required regardless of skipGateway: it drives targetCapacity scaling so the pending cluster ramps up workers proportionally instead of starting at full replicas. Without autoscaler the gradual capacity transfer semantics are broken. skipGateway only waives stepSizePercent, intervalSeconds, and gatewayClassName which are gateway-routing-specific fields not used in the client-side LB path. Co-Authored-By: Claude Sonnet 4.6 --- ray-operator/controllers/ray/utils/validation.go | 14 ++++++-------- .../controllers/ray/utils/validation_test.go | 14 +++++++------- 2 files changed, 13 insertions(+), 15 deletions(-) diff --git a/ray-operator/controllers/ray/utils/validation.go b/ray-operator/controllers/ray/utils/validation.go index 6353c34bc14..07ada3adb39 100644 --- a/ray-operator/controllers/ray/utils/validation.go +++ b/ray-operator/controllers/ray/utils/validation.go @@ -389,24 +389,22 @@ func ValidateRayServiceSpec(rayService *rayv1.RayService) error { } func ValidateClusterUpgradeOptions(rayService *rayv1.RayService) error { + if !IsAutoscalingEnabled(&rayService.Spec.RayClusterSpec) { + return fmt.Errorf("Ray Autoscaler is required for NewClusterWithIncrementalUpgrade") + } + options := rayService.Spec.UpgradeStrategy.ClusterUpgradeOptions if options == nil { return fmt.Errorf("ClusterUpgradeOptions are required for NewClusterWithIncrementalUpgrade") } - skipGateway := options.SkipGateway != nil && *options.SkipGateway - - // Autoscaler is required for gateway-based incremental upgrade but not for skipGateway mode, - // which uses client-side load balancing driven by targetCapacity in RayService status. - if !skipGateway && !IsAutoscalingEnabled(&rayService.Spec.RayClusterSpec) { - return fmt.Errorf("Ray Autoscaler is required for NewClusterWithIncrementalUpgrade") - } - // MaxSurgePercent defaults to 100% if unset. if *options.MaxSurgePercent < 0 || *options.MaxSurgePercent > 100 { return fmt.Errorf("maxSurgePercent must be between 0 and 100") } + skipGateway := options.SkipGateway != nil && *options.SkipGateway + if !skipGateway { if options.StepSizePercent == nil || *options.StepSizePercent < 0 || *options.StepSizePercent > 100 { return fmt.Errorf("stepSizePercent must be between 0 and 100") diff --git a/ray-operator/controllers/ray/utils/validation_test.go b/ray-operator/controllers/ray/utils/validation_test.go index ccad901eaf1..4ac6c7dfa7f 100644 --- a/ray-operator/controllers/ray/utils/validation_test.go +++ b/ray-operator/controllers/ray/utils/validation_test.go @@ -1844,24 +1844,24 @@ func TestValidateClusterUpgradeOptions(t *testing.T) { expectError: true, }, { - name: "skipGateway valid without autoscaler", + name: "skipGateway valid with autoscaler", maxSurgePercent: ptr.To(int32(20)), skipGateway: true, - enableAutoscaling: false, + enableAutoscaling: true, expectError: false, }, { - name: "skipGateway valid with autoscaler", + name: "skipGateway still requires autoscaler for targetCapacity to scale workers", maxSurgePercent: ptr.To(int32(20)), skipGateway: true, - enableAutoscaling: true, - expectError: false, + enableAutoscaling: false, + expectError: true, }, { - name: "skipGateway does not require stepSizePercent or intervalSeconds", + name: "skipGateway does not require stepSizePercent, intervalSeconds, or gatewayClassName", maxSurgePercent: ptr.To(int32(50)), skipGateway: true, - enableAutoscaling: false, + enableAutoscaling: true, expectError: false, }, } From d68528ba5180c3de115d2bbf74de129fd1635288 Mon Sep 17 00:00:00 2001 From: Savita Manghnani Date: Mon, 22 Jun 2026 17:18:19 -0400 Subject: [PATCH 7/7] Fix spurious 'pending cluster not yet capacity-ready' log when no upgrade in progress When upgradeStrategy.type=NewClusterWithIncrementalUpgrade with skipGateway=true, the TC/TRP gate code ran every reconcile regardless of whether a pending cluster existed. With no pending cluster, arePendingServeDeploymentsHealthy(nil) returns false, so the log fired on every reconcile post-upgrade. Fix: guard the SkipGateway TC/TRP block with pendingCluster != nil. Initialize activeWeight/pendingWeight to -1 (the sentinel for "don't update") so the no- pending-cluster case correctly skips the update rather than falling through to the HTTPRoute weight reader. Co-Authored-By: Claude Sonnet 4.6 --- .../controllers/ray/rayservice_controller.go | 39 ++++++------ .../ray/rayservice_controller_unit_test.go | 60 +++++++++++++++++++ 2 files changed, 80 insertions(+), 19 deletions(-) diff --git a/ray-operator/controllers/ray/rayservice_controller.go b/ray-operator/controllers/ray/rayservice_controller.go index 120dcc47311..c6b74836d21 100644 --- a/ray-operator/controllers/ray/rayservice_controller.go +++ b/ray-operator/controllers/ray/rayservice_controller.go @@ -380,27 +380,28 @@ func (r *RayServiceReconciler) calculateStatus( oldPendingPercent := ptr.Deref(rayServiceInstance.Status.PendingServiceStatus.TrafficRoutedPercent, -1) now := metav1.Time{Time: time.Now()} - var activeWeight, pendingWeight int32 + // Default sentinel: -1 means "don't update TrafficRoutedPercent this reconcile". + activeWeight, pendingWeight := int32(-1), int32(-1) if utils.IsSkipGateway(&rayServiceInstance.Spec) { - // 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 { - // Gate stays blocked: leave TrafficRoutedPercent at previous value by not updating. - activeWeight = -1 - pendingWeight = -1 - logger.Info("SkipGateway: pending cluster not yet capacity-ready, holding TrafficRoutedPercent until deployments are healthy and replicas are sufficient.", - "pendingTargetCapacity", pendingTargetCap) + 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) diff --git a/ray-operator/controllers/ray/rayservice_controller_unit_test.go b/ray-operator/controllers/ray/rayservice_controller_unit_test.go index 17f2159019e..25850a20fc4 100644 --- a/ray-operator/controllers/ray/rayservice_controller_unit_test.go +++ b/ray-operator/controllers/ray/rayservice_controller_unit_test.go @@ -2638,3 +2638,63 @@ func TestShouldUpdateCluster_SuspendFlip(t *testing.T) { }) } } + +func makeClusterWithWorkers(available int32, maxReplicas int32) *rayv1.RayCluster { + return &rayv1.RayCluster{ + Spec: rayv1.RayClusterSpec{ + WorkerGroupSpecs: []rayv1.WorkerGroupSpec{ + {MaxReplicas: ptr.To(maxReplicas)}, + }, + }, + Status: rayv1.RayClusterStatus{ + AvailableWorkerReplicas: available, + }, + } +} + +func makeHealthyApps() map[string]rayv1.AppStatus { + return map[string]rayv1.AppStatus{ + "app": { + Status: rayv1.ApplicationStatusEnum.RUNNING, + Deployments: map[string]rayv1.ServeDeploymentStatus{ + "dep": {Status: rayv1.DeploymentStatusEnum.HEALTHY}, + }, + }, + } +} + +func TestIsPendingClusterCapacityReady(t *testing.T) { + t.Run("returns false when no pending apps", func(t *testing.T) { + active := makeClusterWithWorkers(2, 2) + pending := makeClusterWithWorkers(1, 2) + assert.False(t, isPendingClusterCapacityReady(nil, active, pending, 20)) + }) + + t.Run("returns true when both clusters nil", func(t *testing.T) { + assert.True(t, isPendingClusterCapacityReady(makeHealthyApps(), nil, nil, 0)) + }) + + t.Run("returns true when pending maxReplicas < active available (capacity reduction)", func(t *testing.T) { + active := makeClusterWithWorkers(3, 3) + pending := makeClusterWithWorkers(1, 2) // maxReplicas=2 < activeAvailable=3 + assert.True(t, isPendingClusterCapacityReady(makeHealthyApps(), active, pending, 80)) + }) + + t.Run("returns false when pending workers insufficient for TC", func(t *testing.T) { + active := makeClusterWithWorkers(3, 5) + pending := makeClusterWithWorkers(1, 5) // needs ceil(40%*3)=2 workers, has 1 + assert.False(t, isPendingClusterCapacityReady(makeHealthyApps(), active, pending, 40)) + }) + + t.Run("returns true when pending workers sufficient for TC", func(t *testing.T) { + active := makeClusterWithWorkers(3, 5) + pending := makeClusterWithWorkers(2, 5) // needs ceil(40%*3)=2 workers, has 2 + assert.True(t, isPendingClusterCapacityReady(makeHealthyApps(), active, pending, 40)) + }) + + t.Run("TC=0 always passes worker check (no workers needed)", func(t *testing.T) { + active := makeClusterWithWorkers(3, 5) + pending := makeClusterWithWorkers(0, 5) // ceil(0%*3)=0 workers needed + assert.True(t, isPendingClusterCapacityReady(makeHealthyApps(), active, pending, 0)) + }) +}