import asyncio import sys from copy import deepcopy from typing import Dict, Optional import pytest from pydantic import BaseModel import ray from ray import serve from ray._common.test_utils import SignalActor, wait_for_condition from ray.exceptions import RayActorError from ray.serve import Application from ray.serve._private.client import ServeControllerClient from ray.serve._private.common import ( DeploymentID, DeploymentStatus, DeploymentStatusTrigger, ReplicaState, TargetCapacityDirection, ) from ray.serve._private.constants import SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE from ray.serve.config import AutoscalingConfig from ray.serve.context import _get_global_client from ray.serve.schema import ( ApplicationStatus, ServeApplicationSchema, ServeDeploySchema, ) INGRESS_DEPLOYMENT_NAME = "ingress" INGRESS_DEPLOYMENT_NUM_REPLICAS = 6 DOWNSTREAM_DEPLOYMENT_NAME = "downstream" DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS = 2 SCALE_TO_ZERO_DEPLOYMENT_NAME = "zero" SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS = 4 START_AT_10_DEPLOYMENT_NAME = "start_at_ten" START_AT_10_DEPLOYMENT_INITIAL_REPLICAS = 10 START_AT_10_DEPLOYMENT_MIN_REPLICAS = 0 START_AT_10_DEPLOYMENT_MAX_REPLICAS = 20 @pytest.fixture def shutdown_ray_and_serve(): serve.shutdown() if ray.is_initialized(): # wait_for_processes=True blocks until the raylet/GCS/etc. subprocesses # have fully exited, so the next test's serve.start() (which calls # ray.init()) doesn't race a still-terminating raylet. ray.shutdown(wait_for_processes=True) yield serve.shutdown() if ray.is_initialized(): ray.shutdown(wait_for_processes=True) @serve.deployment(ray_actor_options={"num_cpus": 0}) class DummyDeployment: def __init__(self, *args): pass def __call__(self, *args): return "hi" test_app = DummyDeployment.options( name=INGRESS_DEPLOYMENT_NAME, num_replicas=INGRESS_DEPLOYMENT_NUM_REPLICAS, ).bind( DummyDeployment.options( name=DOWNSTREAM_DEPLOYMENT_NAME, num_replicas=DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS, ).bind() ) def check_expected_num_replicas( deployment_to_num_replicas: Dict[str, int], app_name: str = SERVE_DEFAULT_APP_NAME ) -> bool: """Checks that the expected number of replicas for each deployment is running. Looks for the deployments in the app named app_name. """ status = serve.status() assert app_name in status.applications application = status.applications[app_name] assert application.status == ApplicationStatus.RUNNING for name, num_replicas in deployment_to_num_replicas.items(): assert name in application.deployments assert ( sum(application.deployments[name].replica_states.values()) == num_replicas ) return True @pytest.fixture def client() -> ServeControllerClient: serve.start() yield _get_global_client() def test_incremental_scale_up(shutdown_ray_and_serve, client: ServeControllerClient): config = ServeDeploySchema( applications=[ ServeApplicationSchema( import_path="ray.serve.tests.test_target_capacity:test_app" ) ] ) # Initially deploy at target_capacity 0, should have no replicas. config.target_capacity = 0.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 0.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: 0, DOWNSTREAM_DEPLOYMENT_NAME: 0, }, timeout=30, ) # Initially deploy at target_capacity 1, should have 1 replica of each. config.target_capacity = 1.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 1.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: 1, DOWNSTREAM_DEPLOYMENT_NAME: 1, }, timeout=30, ) # Increase target_capacity to 50, ingress deployment should scale up. config.target_capacity = 50.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 50.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2, DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2, }, timeout=30, ) # Increase target_capacity to 100, both should fully scale up. config.target_capacity = 100.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 100.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS, DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS, }, timeout=30, ) # Finish rollout (remove target_capacity), should have no effect. config.target_capacity = None client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity is None) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS, DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS, }, timeout=30, ) def test_incremental_scale_down(shutdown_ray_and_serve, client: ServeControllerClient): config = ServeDeploySchema( applications=[ ServeApplicationSchema( import_path="ray.serve.tests.test_target_capacity:test_app" ) ] ) # Initially deploy with no target_capacity (full scale). client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity is None) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS, DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS, }, timeout=30, ) # Decrease target_capacity to 50, both deployments should scale down. config.target_capacity = 50.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 50.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2, DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2, }, timeout=30, ) # Decrease target_capacity to 1, both should fully scale down. config.target_capacity = 1.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 1.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: 1, DOWNSTREAM_DEPLOYMENT_NAME: 1, }, timeout=30, ) # Decrease target_capacity to 0, both should fully scale down to zero. config.target_capacity = 0.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 0.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: 0, DOWNSTREAM_DEPLOYMENT_NAME: 0, }, timeout=30, ) def test_controller_recover_target_capacity( shutdown_ray_and_serve, client: ServeControllerClient ): config = ServeDeploySchema( target_capacity=100, applications=[ ServeApplicationSchema( import_path="ray.serve.tests.test_target_capacity:test_app" ) ], ) client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 100.0) assert ( ray.get(client._controller._get_target_capacity_direction.remote()) == TargetCapacityDirection.UP ) # Scale down to target_capacity 50, both deployments should be at half scale. config.target_capacity = 50.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 50.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2, DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2, }, timeout=30, ) assert ( ray.get(client._controller._get_target_capacity_direction.remote()) == TargetCapacityDirection.DOWN ) ray.kill(client._controller, no_restart=False) # Verify that the target_capacity is recovered after the controller comes back. wait_for_condition(lambda: serve.status().target_capacity == 50.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS / 2, DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS / 2, }, timeout=30, ) assert ( ray.get(client._controller._get_target_capacity_direction.remote()) == TargetCapacityDirection.DOWN ) @serve.deployment( ray_actor_options={"num_cpus": 0}, autoscaling_config={ "min_replicas": 0, "max_replicas": SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS, "target_ongoing_requests": 1, "upscale_delay_s": 1, "downscale_delay_s": 1, "upscaling_factor": 4, "downscaling_factor": 4, "metrics_interval_s": 1, # The default look_back_period_s is 30, which means the test assertions will be # slow to respond to changes in metrics. Setting it to 2 makes the test assertions # more responsive to changes in metrics, hence reducing flakiness. "look_back_period_s": 2, }, max_ongoing_requests=2, graceful_shutdown_timeout_s=0, ) class ScaleToZeroDeployment: async def __call__(self, *args): await asyncio.sleep(10000) scale_to_zero_app = ScaleToZeroDeployment.options( name=SCALE_TO_ZERO_DEPLOYMENT_NAME, ).bind() def test_autoscaling_scale_to_zero( shutdown_ray_and_serve, client: ServeControllerClient ): config = ServeDeploySchema( applications=[ ServeApplicationSchema( import_path="ray.serve.tests.test_target_capacity:scale_to_zero_app" ) ] ) # Initially deploy at target_capacity 0, should have 0 replicas. config.target_capacity = 0.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 0.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ SCALE_TO_ZERO_DEPLOYMENT_NAME: 0, }, ) # Increase to target_capacity 1, should remain at 0 replicas initially. config.target_capacity = 1.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 1.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ SCALE_TO_ZERO_DEPLOYMENT_NAME: 0, }, ) # Send a bunch of requests. Autoscaler will want to scale it up to max replicas, # but it should stay at one due to the target_capacity. handle = serve.get_app_handle(SERVE_DEFAULT_APP_NAME) responses = [ handle.remote() for _ in range(5 * SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS) ] wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ SCALE_TO_ZERO_DEPLOYMENT_NAME: 1, }, timeout=30, ) # Increase to target_capacity 100, should scale all the way up. config.target_capacity = 100.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 100.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ SCALE_TO_ZERO_DEPLOYMENT_NAME: SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS, }, timeout=30, ) # Decrease to target_capacity 50, should scale down. config.target_capacity = 50.0 client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == 50.0) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ SCALE_TO_ZERO_DEPLOYMENT_NAME: SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS / 2, }, timeout=30, ) # Cancel all of the requests, should scale down to zero. [r.cancel() for r in responses] wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ SCALE_TO_ZERO_DEPLOYMENT_NAME: 0, }, timeout=30, ) @serve.deployment(ray_actor_options={"num_cpus": 0}) class ControlledLifecycleDeployment: async def __init__(self): self.lifecycle_signal = ray.get_actor( name="lifecycle_signal", namespace=SERVE_NAMESPACE ) await self.lifecycle_signal.wait.remote() try: self.request_signal = ray.get_actor( name="request_signal", namespace=SERVE_NAMESPACE ) except ValueError: # No request signal actor was launched. self.request_signal = None async def __call__(self) -> str: if self.request_signal is not None: try: await self.request_signal.wait.remote() except RayActorError: print("Request signal actor already dead. Running request.") return "Hello world!" async def __del__(self): # Use try-except here to avoid cluttering logs if the SignalActor # gets torn down before the replicas during clean up. try: await self.lifecycle_signal.wait.remote() except RayActorError: print("Lifecycle signal actor already dead. Replica exiting.") class ControllerAppConfig(BaseModel): num_replicas: int def create_controlled_app(config: ControllerAppConfig) -> Application: num_replicas = config.num_replicas return ControlledLifecycleDeployment.options( name="controlled", num_replicas=num_replicas, ).bind() class AutoscalingControllerAppConfig(BaseModel): min_replicas: int initial_replicas: Optional[int] = None max_replicas: int def create_autoscaling_controlled_app( config: AutoscalingControllerAppConfig, ) -> Application: min_replicas = config.min_replicas initial_replicas = config.initial_replicas max_replicas = config.max_replicas return ControlledLifecycleDeployment.options( name="controlled", autoscaling_config=AutoscalingConfig( min_replicas=min_replicas, initial_replicas=initial_replicas, max_replicas=max_replicas, target_ongoing_requests=1, metrics_interval_s=0.1, look_back_period_s=0.2, upscale_delay_s=0.1, downscale_delay_s=0.1, ), graceful_shutdown_timeout_s=0, ).bind() class TestTargetCapacityUpdateAndServeStatus: def check_num_replicas( self, expected_num_replicas: int, app_name: str, deployment_name: str, replica_state: ReplicaState = ReplicaState.RUNNING, controller_handle: Optional[ray.actor.ActorHandle] = None, ) -> bool: """Checks that the number of replicas are as expected. Args: expected_num_replicas: the expected number of replicas. app_name: the deployment's application name. deployment_name: the deployment's name. replica_state: only replicas in this state are counted. controller_handle: this is an optional argument. If provided, the controller handle is used to get the current autoscaling metrics and print them if the assertion fails. Returns: True when the replica count matches (raises ``AssertionError`` otherwise). """ deployment = serve.status().applications[app_name].deployments[deployment_name] num_running_replicas = deployment.replica_states.get(replica_state, 0) if controller_handle is None: assert num_running_replicas == expected_num_replicas, f"{deployment}" else: deployment_id = DeploymentID(name=deployment_name, app_name=app_name) autoscaling_metrics = ray.get( controller_handle._get_metrics_for_deployment_for_testing.remote( deployment_id ) ) assert num_running_replicas == expected_num_replicas, ( f"Status: {deployment}" f"\nAutoscaling metrics: {autoscaling_metrics}" ) return True def apply_config_and_check_status( self, client: ServeControllerClient, target_capacity: Optional[float], config: ServeDeploySchema, app_name: str, deployment_name: str, expected_app_status: Optional[ApplicationStatus] = None, expected_deployment_status: Optional[DeploymentStatus] = None, expected_deployment_status_trigger: Optional[DeploymentStatusTrigger] = None, timeout=10, ): """Applies config with specified target_capacity.""" config = deepcopy(config) config.target_capacity = target_capacity client.deploy_apps(config) def check(): status = serve.status() assert status.target_capacity == target_capacity if expected_app_status is not None: assert status.applications[app_name].status == expected_app_status dep_status = status.applications[app_name].deployments[deployment_name] if expected_deployment_status is not None: assert dep_status.status == expected_deployment_status if expected_deployment_status_trigger is not None: assert dep_status.status_trigger == expected_deployment_status_trigger return True wait_for_condition(check, timeout=timeout) def unblock_replica_creation_and_deletion(self, lifecycle_signal, app_name: str): """Unblocks creating and deleting ControlledLifecycleDeployment replicas. These replicas can't initialize or be deleted until the "lifecycle_signal" actor runs send, so this method runs send, waits until the replicas start or stop running, and then resets the signal. """ def check_running(): app_status_data = serve.status().applications[app_name] app_status = app_status_data.status assert app_status == ApplicationStatus.RUNNING, f"{app_status_data}" return True ray.get(lifecycle_signal.send.remote()) wait_for_condition(check_running, timeout=30, retry_interval_ms=500) ray.get(lifecycle_signal.send.remote(clear=True)) def test_static_num_replicas_target_capacity_update( self, shutdown_ray_and_serve, client: ServeControllerClient ): """Check how Serve's status updates when target_capacity changes.""" app_name = "controlled_app" deployment_name = "controlled" num_replicas = 20 signal = SignalActor.options( name="lifecycle_signal", namespace=SERVE_NAMESPACE ).remote() config = ServeDeploySchema( applications=[ ServeApplicationSchema( name=app_name, import_path=( "ray.serve.tests.test_target_capacity:create_controlled_app" ), args={"num_replicas": num_replicas}, ) ] ) # Initially deploy at target_capacity 0, and check status. self.apply_config_and_check_status( client, target_capacity=0.0, config=config, app_name=app_name, deployment_name=deployment_name, expected_app_status=ApplicationStatus.RUNNING, expected_deployment_status=DeploymentStatus.HEALTHY, expected_deployment_status_trigger=( DeploymentStatusTrigger.CONFIG_UPDATE_COMPLETED ), ) self.check_num_replicas(0, app_name, deployment_name) # Increase the target_capacity, and check again. self.apply_config_and_check_status( client, target_capacity=50.0, config=config, app_name=app_name, deployment_name=deployment_name, expected_app_status=ApplicationStatus.DEPLOYING, expected_deployment_status=DeploymentStatus.UPSCALING, expected_deployment_status_trigger=( DeploymentStatusTrigger.CONFIG_UPDATE_STARTED ), ) self.check_num_replicas(0, app_name, deployment_name) self.unblock_replica_creation_and_deletion(signal, app_name) self.check_num_replicas(int(0.5 * num_replicas), app_name, deployment_name) # Decrease the target_capacity, and check again. self.apply_config_and_check_status( client, target_capacity=10.0, config=config, app_name=app_name, deployment_name=deployment_name, expected_app_status=ApplicationStatus.DEPLOYING, expected_deployment_status=DeploymentStatus.DOWNSCALING, expected_deployment_status_trigger=( DeploymentStatusTrigger.CONFIG_UPDATE_STARTED ), ) # DeploymentStateManager marks replicas as STOPPING once the deployment # starts downscaling. wait_for_condition( self.check_num_replicas, expected_num_replicas=int(0.1 * num_replicas), app_name=app_name, deployment_name=deployment_name, replica_state=ReplicaState.RUNNING, timeout=20, ) self.check_num_replicas( int(0.4 * num_replicas), app_name, deployment_name, replica_state=ReplicaState.STOPPING, ) self.unblock_replica_creation_and_deletion(signal, app_name) self.check_num_replicas(int(0.1 * num_replicas), app_name, deployment_name) self.check_num_replicas( 0, app_name, deployment_name, replica_state=ReplicaState.STOPPING ) # Send a signal so all replicas shut down quickly when the test finishes. ray.get(signal.send.remote()) @pytest.mark.skipif(sys.platform == "win32", reason="Autoscaling flaky on Windows.") def test_autoscaling_target_capacity_update( self, shutdown_ray_and_serve, client: ServeControllerClient ): """Check Serve's status when target_capacity changes while autoscaling.""" # TODO(landscapepainter): This test fails locally due to the stall for replica initialization # during upscaling and delayed response from serve.status(). It does not fail from # buildkite, but need to investigate why it fails locally. app_name = "controlled_app" deployment_name = "controlled" min_replicas = 10 initial_replicas = 12 max_replicas = 20 lifecycle_signal = SignalActor.options( name="lifecycle_signal", namespace=SERVE_NAMESPACE ).remote() request_signal = SignalActor.options( name="request_signal", namespace=SERVE_NAMESPACE ).remote() config = ServeDeploySchema( applications=[ ServeApplicationSchema( name=app_name, import_path=( "ray.serve.tests.test_target_capacity:" "create_autoscaling_controlled_app" ), args=dict( min_replicas=min_replicas, initial_replicas=initial_replicas, max_replicas=max_replicas, ), ) ] ) # Initially deploy at target_capacity 0, and check status. self.apply_config_and_check_status( client, target_capacity=0.0, config=config, app_name=app_name, deployment_name=deployment_name, timeout=20, expected_app_status=ApplicationStatus.RUNNING, expected_deployment_status=DeploymentStatus.HEALTHY, expected_deployment_status_trigger=( DeploymentStatusTrigger.CONFIG_UPDATE_COMPLETED ), ) self.check_num_replicas(0, app_name, deployment_name) # Increase the target_capacity, and check again. self.apply_config_and_check_status( client, target_capacity=50.0, config=config, app_name=app_name, deployment_name=deployment_name, expected_app_status=ApplicationStatus.DEPLOYING, expected_deployment_status=DeploymentStatus.UPSCALING, expected_deployment_status_trigger=( DeploymentStatusTrigger.CONFIG_UPDATE_STARTED ), ) self.check_num_replicas(0, app_name, deployment_name) self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name) self.check_num_replicas(int(0.5 * initial_replicas), app_name, deployment_name) # Send requests and check that the application scales up. requests = [] handle = serve.get_app_handle(app_name) for _ in range(max_replicas): requests.append(handle.remote()) ray.get(lifecycle_signal.send.remote()) wait_for_condition( self.check_num_replicas, expected_num_replicas=int(0.5 * max_replicas), app_name=app_name, deployment_name=deployment_name, timeout=30, ) # Clear requests and check that application scales down. ray.get(request_signal.send.remote()) results = [request.result() for request in requests] assert results == ["Hello world!"] * (max_replicas) wait_for_condition( self.check_num_replicas, expected_num_replicas=int(0.5 * initial_replicas), app_name=app_name, deployment_name=deployment_name, replica_state=ReplicaState.RUNNING, timeout=20, retry_interval_ms=1000, ) wait_for_condition( self.check_num_replicas, expected_num_replicas=0, app_name=app_name, deployment_name=deployment_name, replica_state=ReplicaState.STOPPING, timeout=20, retry_interval_ms=1000, ) ray.get(lifecycle_signal.send.remote(clear=True)) ray.get(request_signal.send.remote(clear=True)) # Decrease the target_capacity, and check that min_replicas is used # to create the lower bound. self.apply_config_and_check_status( client, target_capacity=10.0, config=config, app_name=app_name, deployment_name=deployment_name, expected_app_status=ApplicationStatus.DEPLOYING, expected_deployment_status=DeploymentStatus.DOWNSCALING, expected_deployment_status_trigger=( DeploymentStatusTrigger.CONFIG_UPDATE_STARTED ), ) self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name) self.check_num_replicas( int(0.1 * min_replicas), app_name, deployment_name, replica_state=ReplicaState.RUNNING, ) self.check_num_replicas( 0, app_name, deployment_name, replica_state=ReplicaState.STOPPING ) # Check that target_capacity * max_replicas is still the upper bound. requests = [] handle = serve.get_app_handle(app_name) for _ in range(max_replicas): requests.append(handle.remote()) ray.get(lifecycle_signal.send.remote()) wait_for_condition( self.check_num_replicas, expected_num_replicas=int(0.1 * max_replicas), app_name=app_name, deployment_name=deployment_name, controller_handle=client._controller, timeout=25, retry_interval_ms=2000, ) # Clear requests and check that application scales down to # target_capacity * min_replicas. ray.get(request_signal.send.remote()) results = [request.result() for request in requests] assert results == ["Hello world!"] * (max_replicas) wait_for_condition( self.check_num_replicas, expected_num_replicas=int(0.1 * min_replicas), app_name=app_name, deployment_name=deployment_name, controller_handle=client._controller, timeout=25, retry_interval_ms=2000, ) ray.get(lifecycle_signal.send.remote(clear=True)) ray.get(request_signal.send.remote(clear=True)) # Scaling up to 100% target_capacity should make Serve use # initial_replicas as lower bound. self.apply_config_and_check_status( client, target_capacity=100.0, config=config, app_name=app_name, deployment_name=deployment_name, expected_app_status=ApplicationStatus.DEPLOYING, expected_deployment_status=DeploymentStatus.UPSCALING, expected_deployment_status_trigger=( DeploymentStatusTrigger.CONFIG_UPDATE_STARTED ), ) self.check_num_replicas(int(0.1 * min_replicas), app_name, deployment_name) self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name) self.check_num_replicas(initial_replicas, app_name, deployment_name) # Unsetting target_capacity should make Serve use # min_replicas as lower bound. The current number of # replicas is already a valid number, so the application stays # RUNNING. self.apply_config_and_check_status( client, target_capacity=None, config=config, app_name=app_name, deployment_name=deployment_name, expected_app_status=ApplicationStatus.RUNNING, expected_deployment_status=DeploymentStatus.DOWNSCALING, expected_deployment_status_trigger=(DeploymentStatusTrigger.AUTOSCALING), ) self.unblock_replica_creation_and_deletion(lifecycle_signal, app_name) wait_for_condition( self.check_num_replicas, expected_num_replicas=min_replicas, app_name=app_name, deployment_name=deployment_name, ) # Send a signal so all replicas shut down quickly when the test finishes. ray.get(request_signal.send.remote()) ray.get(lifecycle_signal.send.remote()) @serve.deployment( ray_actor_options={"num_cpus": 0}, max_ongoing_requests=2, graceful_shutdown_timeout_s=0, ) class HangDeployment: async def __call__(self, *args): await asyncio.sleep(10000) def create_hang_app(config: Dict) -> Application: name: str = config["name"] min_replicas: int = config["min_replicas"] initial_replicas: int = config["initial_replicas"] max_replicas: int = config["max_replicas"] return HangDeployment.options( name=name, autoscaling_config={ "min_replicas": min_replicas, "initial_replicas": initial_replicas, "max_replicas": max_replicas, "target_ongoing_requests": 1, "metrics_interval_s": 0.01, "downscale_delay_s": 0.01, }, graceful_shutdown_timeout_s=0.01, ).bind() class TestInitialReplicasHandling: def deploy_config_and_wait_for_target_capacity( self, client: ServeControllerClient, config: ServeDeploySchema, target_capacity: float, ): config.target_capacity = target_capacity client.deploy_apps(config) wait_for_condition(lambda: serve.status().target_capacity == target_capacity) def test_initial_replicas_scales_down( self, shutdown_ray_and_serve, client: ServeControllerClient ): # TODO(landscapepainter): This test fails locally due to the stall for replica initialization # during upscaling and delayed response from serve.status(). It does not fail from # buildkite, but need to investigate why it fails locally. deployment_name = "start_at_ten" min_replicas = 5 initial_replicas = 10 config = ServeDeploySchema( applications=[ ServeApplicationSchema( import_path="ray.serve.tests.test_target_capacity:create_hang_app", args={ "name": deployment_name, "min_replicas": min_replicas, "initial_replicas": initial_replicas, "max_replicas": 15, }, ), ], ) client.deploy_apps(config) wait_for_condition(lambda: len(serve.status().applications) == 1) assert serve.status().target_capacity is None # Wait for initial deployment to reach RUNNING before starting the # target_capacity loop. Without this, the first loop iteration must # handle both initial replica startup AND target_capacity convergence # within its timeout, which is marginal under CI resource contention. wait_for_condition( lambda: ( SERVE_DEFAULT_APP_NAME in serve.status().applications and serve.status().applications[SERVE_DEFAULT_APP_NAME].status == ApplicationStatus.RUNNING ), timeout=30, ) # Kick off downscaling pattern. test_target_capacities = [100, 60, 40, 0] expected_num_replicas = [ int(min_replicas * capacity / 100) for capacity in test_target_capacities ] for target_capacity, num_replicas in zip( test_target_capacities, expected_num_replicas ): print(f"target_capacity: {target_capacity}, num_replicas: {num_replicas}") config.target_capacity = target_capacity client.deploy_apps(config) wait_for_condition( lambda: serve.status().target_capacity == target_capacity ) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={deployment_name: num_replicas}, timeout=30, ) def test_initial_replicas_scales_up_and_down( self, shutdown_ray_and_serve, client: ServeControllerClient ): # TODO(landscapepainter): This test fails locally due to the stall for replica initialization # during upscaling and delayed response from serve.status(). It does not fail from # buildkite, but need to investigate why it fails locally. deployment_name = "start_at_five" min_replicas = 0 initial_replicas = 5 config = ServeDeploySchema( target_capacity=0, applications=[ ServeApplicationSchema( import_path="ray.serve.tests.test_target_capacity:create_hang_app", args={ "name": deployment_name, "min_replicas": min_replicas, "initial_replicas": initial_replicas, "max_replicas": 2 * initial_replicas, }, ), ], ) test_target_capacities = [0, 1, 20, 60, 40, 30, 100, None] expected_num_replicas = [ 0, 1, 1, 3, min_replicas, min_replicas, initial_replicas, min_replicas, ] for target_capacity, num_replicas in zip( test_target_capacities, expected_num_replicas ): print(f"target_capacity: {target_capacity}, num_replicas: {num_replicas}") config.target_capacity = target_capacity client.deploy_apps(config) wait_for_condition( lambda: serve.status().target_capacity == target_capacity ) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={deployment_name: num_replicas}, timeout=30, ) def test_initial_replicas_zero( self, shutdown_ray_and_serve, client: ServeControllerClient ): deployment_name = "start_at_zero" min_replicas = 0 initial_replicas = 0 config = ServeDeploySchema( target_capacity=0, applications=[ ServeApplicationSchema( import_path="ray.serve.tests.test_target_capacity:create_hang_app", args={ "name": deployment_name, "min_replicas": min_replicas, "initial_replicas": initial_replicas, "max_replicas": 20, }, ), ], ) test_target_capacities = [0, 1, 20, 60, 40, 30, 100, None] expected_num_replicas = [0] * len(test_target_capacities) for target_capacity, num_replicas in zip( test_target_capacities, expected_num_replicas ): config.target_capacity = target_capacity client.deploy_apps(config) wait_for_condition( lambda: serve.status().target_capacity == target_capacity ) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={deployment_name: num_replicas}, ) def test_initial_replicas_new_configs( self, shutdown_ray_and_serve, client: ServeControllerClient ): # TODO(landscapepainter): This test fails locally due to the stall for replica initialization # during upscaling and delayed response from serve.status(). It does not fail from # buildkite, but need to investigate why it fails locally. deployment_name = "start_at_ten" min_replicas = 0 initial_replicas = 10 config_target_capacity = 40 config = ServeDeploySchema( target_capacity=config_target_capacity, applications=[ ServeApplicationSchema( name="app1", import_path="ray.serve.tests.test_target_capacity:create_hang_app", args={ "name": deployment_name, "min_replicas": min_replicas, "initial_replicas": initial_replicas, "max_replicas": initial_replicas * 2, }, ), ], ) # When deploying first config, initial_replicas * target_capacity # should be treated as floor client.deploy_apps(config) wait_for_condition( lambda: serve.status().target_capacity == config_target_capacity ) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ deployment_name: int(initial_replicas * config_target_capacity / 100) }, app_name="app1", timeout=30, ) # When deploying a new config, initial_replicas * target_capacity # should be treated as floor for all deployments, even if # target_capacity is lower new_deployment_name = f"new_{deployment_name}" new_config_target_capacity = config_target_capacity / 2 new_config = deepcopy(config) new_config.applications.append( ServeApplicationSchema( name="app2", import_path="ray.serve.tests.test_target_capacity:create_hang_app", args={ "name": new_deployment_name, "min_replicas": min_replicas, "initial_replicas": initial_replicas, "max_replicas": initial_replicas * 2, }, ), ) new_config.target_capacity = new_config_target_capacity client.deploy_apps(new_config) wait_for_condition( lambda: serve.status().target_capacity == new_config_target_capacity ) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ deployment_name: int( initial_replicas * new_config_target_capacity / 100 ) }, app_name="app1", timeout=30, ) wait_for_condition( check_expected_num_replicas, deployment_to_num_replicas={ new_deployment_name: int( initial_replicas * new_config_target_capacity / 100 ) }, app_name="app2", timeout=30, ) if __name__ == "__main__": sys.exit(pytest.main(["-v", "-s", __file__]))