diff --git a/ray-operator/apis/ray/v1/rayservice_types.go b/ray-operator/apis/ray/v1/rayservice_types.go index 83b9775f377..6dbc3daab6f 100644 --- a/ray-operator/apis/ray/v1/rayservice_types.go +++ b/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 { 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..e64bf2c3937 100644 --- a/ray-operator/config/crd/bases/ray.io_rayservices.yaml +++ b/ray-operator/config/crd/bases/ray.io_rayservices.yaml @@ -8593,13 +8593,11 @@ spec: default: 100 format: int32 type: integer + skipGateway: + type: boolean stepSizePercent: format: int32 type: integer - required: - - gatewayClassName - - 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 8d05334b7de..c6b74836d21 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. @@ -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,12 +379,37 @@ 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.") @@ -381,7 +417,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.") @@ -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,6 +1433,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 +1441,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/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)) + }) +} 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/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 && diff --git a/ray-operator/controllers/ray/utils/validation.go b/ray-operator/controllers/ray/utils/validation.go index 652ff0e14b6..07ada3adb39 100644 --- a/ray-operator/controllers/ray/utils/validation.go +++ b/ray-operator/controllers/ray/utils/validation.go @@ -403,16 +403,24 @@ 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 !skipGateway { + if options.StepSizePercent == nil || *options.StepSizePercent < 0 || *options.StepSizePercent > 100 { + return fmt.Errorf("stepSizePercent must be between 0 and 100") + } - if options.IntervalSeconds == nil || *options.IntervalSeconds <= 0 { - return fmt.Errorf("intervalSeconds must be greater than 0") + 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") + } } - if options.GatewayClassName == "" { - return fmt.Errorf("gatewayClassName is required for NewClusterWithIncrementalUpgrade") + if options.UpgradeTimeoutSeconds != nil && *options.UpgradeTimeoutSeconds < 0 { + return fmt.Errorf("upgradeTimeoutSeconds must be greater than or equal to 0") } return nil diff --git a/ray-operator/controllers/ray/utils/validation_test.go b/ray-operator/controllers/ray/utils/validation_test.go index 014d0917d68..4ac6c7dfa7f 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 with autoscaler", + maxSurgePercent: ptr.To(int32(20)), + skipGateway: true, + enableAutoscaling: true, + expectError: false, + }, + { + name: "skipGateway still requires autoscaler for targetCapacity to scale workers", + maxSurgePercent: ptr.To(int32(20)), + skipGateway: true, + enableAutoscaling: false, + expectError: true, + }, + { + name: "skipGateway does not require stepSizePercent, intervalSeconds, or gatewayClassName", + maxSurgePercent: ptr.To(int32(50)), + skipGateway: true, + enableAutoscaling: true, + 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 {