import math import sys from unittest.mock import MagicMock, patch import pytest from ray.serve._private.autoscaling_state import DeploymentAutoscalingState from ray.serve._private.common import DeploymentID, ReplicaID, TimeStampedValue from ray.serve._private.constants import ( CONTROL_LOOP_INTERVAL_S, SERVE_AUTOSCALING_DECISION_COUNTERS_KEY, SERVE_AUTOSCALING_DECISION_TIMESTAMP_KEY, ) from ray.serve._private.gang_scheduling_autoscaling_policy import ( GangSchedulingAutoscalingPolicy, ) from ray.serve._private.metrics_utils import ( aggregate_timeseries, merge_instantaneous_total, time_weighted_average, ) from ray.serve.autoscaling_policy import ( _apply_app_level_autoscaling_config, _apply_autoscaling_config, _apply_delay_logic, _apply_scaling_factors, replica_queue_length_autoscaling_policy, ) from ray.serve.config import ( AggregationFunction, AutoscalingConfig, AutoscalingContext, GangSchedulingConfig, ) wrapped_replica_queue_length_autoscaling_policy = _apply_autoscaling_config( replica_queue_length_autoscaling_policy ) def create_context_with_overrides( base_ctx: AutoscalingContext, **kwargs ) -> AutoscalingContext: """Helper to create a new AutoscalingContext with specified attributes overridden. Args: base_ctx: The base AutoscalingContext to copy values from. **kwargs: Attributes to override in the new context. Returns: A new AutoscalingContext with overridden values. """ # Get all constructor parameters with defaults from base context params = { "config": base_ctx.config, "deployment_id": base_ctx.deployment_id, "deployment_name": base_ctx.deployment_name, "app_name": base_ctx.app_name, "current_num_replicas": base_ctx.current_num_replicas, "target_num_replicas": base_ctx.target_num_replicas, "running_replicas": base_ctx.running_replicas, "total_num_requests": base_ctx.total_num_requests, "total_queued_requests": base_ctx.total_queued_requests, "aggregated_metrics": base_ctx.aggregated_metrics, "raw_metrics": base_ctx.raw_metrics, "capacity_adjusted_min_replicas": base_ctx.capacity_adjusted_min_replicas, "capacity_adjusted_max_replicas": base_ctx.capacity_adjusted_max_replicas, "policy_state": base_ctx.policy_state, "last_scale_up_time": base_ctx.last_scale_up_time, "last_scale_down_time": base_ctx.last_scale_down_time, "current_time": base_ctx.current_time, "total_pending_async_requests": base_ctx.total_pending_async_requests, } # Override with provided kwargs params.update(kwargs) return AutoscalingContext(**params) def test_exclude_early_partial_period_in_timeseries_aggregation(): """Test that time-weighted average excludes the early partial period when series have misaligned start times. When merging multiple timeseries (e.g., from different replicas), series that start late are implicitly 0 before their first data point. This undercounts the total and biases the mean downward. The fix excludes the partial period by starting the averaging window when all series have contributed at least one point. """ # Two replicas with misaligned starts: r1 starts at 0.2, r2 starts at 0.1 # From 0.1 to 0.2, only r2 contributes (total=3); from 0.2 onward both (total=8) base_time = 1000.0 # Use epoch-like timestamps series1 = [ TimeStampedValue(base_time + 0.2, 5.0), TimeStampedValue(base_time + 0.8, 7.0), TimeStampedValue(base_time + 1.5, 6.0), ] series2 = [ TimeStampedValue(base_time + 0.1, 3.0), TimeStampedValue(base_time + 0.9, 4.0), TimeStampedValue(base_time + 1.4, 8.0), ] merged = merge_instantaneous_total([series1, series2]) # Merged: (0.1, 3), (0.2, 8), (0.8, 10), (0.9, 11), (1.4, 15), (1.5, 14) last_window_s = 0.5 # Extend window to base_time + 2.0 # Without aligned window: includes partial period [0.1, 0.2) where total=3 avg_without_aligned = time_weighted_average( merged, window_start=None, window_end=None, last_window_s=last_window_s, ) # With aligned window (start at 0.2 when all series have reported): excludes partial aligned_start = base_time + 0.2 avg_with_aligned = time_weighted_average( merged, window_start=aligned_start, window_end=None, last_window_s=last_window_s, ) # The aligned average should be higher because we excluded the period # [0.1, 0.2) where the total was underestimated (3 instead of 8) assert avg_with_aligned > avg_without_aligned, ( f"Aligned avg ({avg_with_aligned}) should exceed unaligned ({avg_without_aligned}) " "since we exclude the partial period with underestimated total" ) # Verify aggregate_timeseries with window_start produces the aligned result result = aggregate_timeseries( merged, AggregationFunction.MEAN, last_window_s=last_window_s, window_start=aligned_start, ) assert result == avg_with_aligned def _run_upscale_downscale_flow( policy, config: AutoscalingConfig, ctx: AutoscalingContext, overload_requests: int, start_replicas: int, upscale_target: int, ): """ This runs the upscale and downscale flow to test the delays during upscale and downscale. This can be used by both the default autoscaling policy and custom autoscaling policy with default parameters to verify scaling is properly enabled. The downscale flow is from upscale_target upto zero. Uses a mocked clock so that each policy call advances wall-clock time by CONTROL_LOOP_INTERVAL_S, matching what the real control loop would do. Wait counts use math.ceil(delay_s * ticks_per_second): int(delay / interval) undercounts when float division yields 5.999... for 0.6/0.1, but wall-clock delay logic requires elapsed >= delay_s. """ ticks_per_second = round(1.0 / CONTROL_LOOP_INTERVAL_S) upscale_wait_periods = math.ceil(config.upscale_delay_s * ticks_per_second) downscale_wait_periods = math.ceil(config.downscale_delay_s * ticks_per_second) if config.downscale_to_zero_delay_s: downscale_to_zero_wait_periods = math.ceil( config.downscale_to_zero_delay_s * ticks_per_second ) else: downscale_to_zero_wait_periods = math.ceil( config.downscale_delay_s * ticks_per_second ) # Initialize local policy_state from the base context, so both default and # decorated policies can persist their internal state policy_state = ctx.policy_state or {} # Track a fake clock that advances by CONTROL_LOOP_INTERVAL_S per call, # simulating real control-loop timing for the wall-clock delay logic. # Uses integer division (tick / ticks_per_second) instead of multiplication # (tick * 0.1) to avoid IEEE 754 double-rounding: a*0.1 rounds 0.1 first then # multiplies, whereas a/10 is a single correctly-rounded operation. fake_tick = [0] def _advance_time(): current = fake_tick[0] / ticks_per_second fake_tick[0] += 1 return current with patch("ray.serve.autoscaling_policy.time") as mock_time: mock_time.time = _advance_time # We should scale up only after enough consecutive scale-up decisions. for i in range(upscale_wait_periods): ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=start_replicas, target_num_replicas=start_replicas, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == start_replicas, i ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=start_replicas, target_num_replicas=start_replicas, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == upscale_target no_requests = 0 # We should scale down only after enough consecutive scale-down decisions. for i in range(downscale_wait_periods): ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=upscale_target, target_num_replicas=upscale_target, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == upscale_target, i ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=upscale_target, target_num_replicas=upscale_target, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 1 # We should scale down to zero only after enough consecutive # downscale-to-zero decisions. for i in range(downscale_to_zero_wait_periods): ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 1, i ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 0 # Get some scale-up decisions, but not enough to trigger a scale up. for i in range(int(upscale_wait_periods / 2)): ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 1, i # Interrupt with a scale-down decision. ctx = create_context_with_overrides( ctx, total_num_requests=0, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) _, policy_state = policy(ctx=ctx) # The counter should be reset, so it should require # `upscale_wait_periods` more periods before we actually scale up. for i in range(upscale_wait_periods): ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 1, i ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == upscale_target # Get some scale-down decisions, but not enough to trigger a scale # down. for i in range(int(downscale_wait_periods / 2)): ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=upscale_target, target_num_replicas=upscale_target, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == upscale_target, i # Interrupt with a scale-up decision. ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=upscale_target, target_num_replicas=upscale_target, policy_state=policy_state, ) _, policy_state = policy(ctx=ctx) # The counter should be reset so it should require # `downscale_wait_periods` more periods before we actually scale down. for i in range(downscale_wait_periods): ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=upscale_target, target_num_replicas=upscale_target, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == upscale_target, i # First scale down to 1 replica ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=upscale_target, target_num_replicas=upscale_target, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 1 # Scale down to 0, but not enough to trigger a complete scale down # to zero. for i in range(int(downscale_to_zero_wait_periods / 2)): ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 1, i # Interrupt with a scale-up decision. ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) _, policy_state = policy(ctx=ctx) # The counter should be reset so it should require # `downscale_to_zero_wait_periods` more periods before we actually # scale down. for i in range(downscale_to_zero_wait_periods): ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 1, i ctx = create_context_with_overrides( ctx, total_num_requests=no_requests, current_num_replicas=1, target_num_replicas=1, policy_state=policy_state, ) new_num_replicas, policy_state = policy(ctx=ctx) assert new_num_replicas == 0 class TestReplicaQueueLengthPolicy: @pytest.mark.parametrize( "use_upscale_smoothing_factor,use_upscaling_factor", [(True, True), (True, False), (False, True)], ) def test_scaling_factor_scale_up_from_0_replicas( self, use_upscale_smoothing_factor, use_upscaling_factor ): """Test that the scaling factor is respected when scaling up from 0 replicas. """ min_replicas = 0 max_replicas = 2 config = AutoscalingConfig( min_replicas=min_replicas, max_replicas=max_replicas, upscale_smoothing_factor=10 if use_upscale_smoothing_factor else None, upscaling_factor=10 if use_upscaling_factor else None, ) ctx = AutoscalingContext( target_num_replicas=0, total_num_requests=1, current_num_replicas=0, config=config, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state={}, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) # 1 * 10 assert new_num_replicas == 10 if use_upscale_smoothing_factor: config.upscale_smoothing_factor = 0.5 if use_upscaling_factor: config.upscaling_factor = 0.5 new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) # math.ceil(1 * 0.5) assert new_num_replicas == 1 @pytest.mark.parametrize( "use_downscale_smoothing_factor,use_downscaling_factor", [(True, True), (True, False), (False, True)], ) def test_scaling_factor_scale_down_to_0_replicas( self, use_downscale_smoothing_factor, use_downscaling_factor ): """Test that a deployment scales down to 0 for non-default smoothing factors.""" # With smoothing factor > 1, the desired number of replicas should # immediately drop to 0 (while respecting upscale and downscale delay) min_replicas = 0 max_replicas = 5 policy_state = {} config = AutoscalingConfig( min_replicas=min_replicas, max_replicas=max_replicas, downscale_smoothing_factor=10 if use_downscale_smoothing_factor else None, downscaling_factor=10 if use_downscaling_factor else None, upscale_delay_s=0, downscale_delay_s=0, ) ctx = AutoscalingContext( config=config, total_num_requests=0, current_num_replicas=5, target_num_replicas=5, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state=policy_state, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) # Downscaling to 0 first stops at 1 assert new_num_replicas == 1 # Need to trigger this the second time to go to zero ctx.target_num_replicas = 1 ctx.current_num_replicas = 1 new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert new_num_replicas == 0 # With smoothing factor < 1, the desired number of replicas shouldn't # get stuck at a positive number, and instead should eventually drop # to zero if use_downscale_smoothing_factor: config.downscale_smoothing_factor = 0.2 if use_downscaling_factor: config.downscaling_factor = 0.2 # policy_manager = AutoscalingPolicyManager(config) num_replicas = 5 for _ in range(5): ctx = create_context_with_overrides( ctx, total_num_requests=0, current_num_replicas=num_replicas, target_num_replicas=num_replicas, ) num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert num_replicas == 0 @pytest.mark.parametrize("downscale_to_zero_delay_s", [None, 300]) def test_upscale_downscale_delay(self, downscale_to_zero_delay_s): """Unit test for upscale_delay_s, downscale_delay_s and downscale_to_zero_delay_s""" min_replicas = 0 max_replicas = 2 policy_state = {} config = AutoscalingConfig( min_replicas=min_replicas, max_replicas=max_replicas, target_ongoing_requests=1, upscale_delay_s=30.0, downscale_delay_s=600.0, downscale_to_zero_delay_s=downscale_to_zero_delay_s, ) # Use overload_requests that naturally produce the desired upscale_target # without depending on bounds clamping (bounds are applied at a higher # level in get_decision_num_replicas, not in the policy wrapper). # desired = start_replicas * (overload_requests / target_ongoing_requests) # = 1 * (2 / 1) = 2 overload_requests = 2 ctx = AutoscalingContext( config=config, total_num_requests=1, current_num_replicas=0, target_num_replicas=0, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state=policy_state, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) # Scale up when there are 0 replicas and current_handle_queued_queries > 0 new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert new_num_replicas == 1 # Run the basic upscale/downscale flow _run_upscale_downscale_flow( wrapped_replica_queue_length_autoscaling_policy, config, ctx, overload_requests=overload_requests, start_replicas=1, upscale_target=2, ) def test_replicas_delayed_startup(self): """Unit test simulating replicas taking time to start up.""" min_replicas = 1 max_replicas = 200 policy_state = {} config = { "min_replicas": min_replicas, "max_replicas": max_replicas, "upscale_delay_s": 0, "downscale_delay_s": 100000, "target_ongoing_requests": 1, } config = AutoscalingConfig(**config) ctx = AutoscalingContext( config=config, target_num_replicas=1, total_num_requests=100, current_num_replicas=1, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state=policy_state, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) # new_num_replicas = policy_manager.get_decision_num_replicas(1, 100, 1) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert new_num_replicas == 100 # New target is 100, but no new replicas finished spinning up during this # timestep. ctx = create_context_with_overrides( ctx, total_num_requests=100, current_num_replicas=1, target_num_replicas=100, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert new_num_replicas == 100 # Two new replicas spun up during this timestep. ctx = create_context_with_overrides( ctx, total_num_requests=123, current_num_replicas=3, target_num_replicas=100, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert new_num_replicas == 123 # A lot of queries got drained and a lot of replicas started up, but # new_num_replicas should not decrease, because of the downscale delay. ctx = create_context_with_overrides( ctx, total_num_requests=10, current_num_replicas=4, target_num_replicas=123, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert new_num_replicas == 123 @pytest.mark.parametrize("delay_s", [30.0, 0.0]) def test_fluctuating_ongoing_requests(self, delay_s): """ Simulates a workload that switches between too many and too few ongoing requests. """ min_replicas = 1 max_replicas = 10 policy_state = {} config = { "min_replicas": min_replicas, "max_replicas": max_replicas, "upscale_delay_s": delay_s, "downscale_delay_s": delay_s, "target_ongoing_requests": 50, } config = AutoscalingConfig(**config) if delay_s > 0: wait_periods = int(delay_s / CONTROL_LOOP_INTERVAL_S) assert wait_periods > 1 underload_requests, overload_requests = 2 * 20, 100 trials = 1000 ctx = AutoscalingContext( config=config, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state=policy_state, target_num_replicas=None, total_num_requests=None, current_num_replicas=None, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) new_num_replicas = None for trial in range(trials): if trial % 2 == 0: ctx = create_context_with_overrides( ctx, total_num_requests=overload_requests, current_num_replicas=1, target_num_replicas=1, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy( ctx=ctx ) if delay_s > 0: assert new_num_replicas == 1, trial else: assert new_num_replicas == 2, trial else: ctx = create_context_with_overrides( ctx, total_num_requests=underload_requests, current_num_replicas=2, target_num_replicas=2, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy( ctx=ctx ) if delay_s > 0: assert new_num_replicas == 2, trial else: assert new_num_replicas == 1, trial @pytest.mark.parametrize("ongoing_requests", [20, 100, 10]) def test_single_replica_receives_all_requests(self, ongoing_requests): target_requests = 5 min_replicas = 1 max_replicas = 50 policy_state = {} config = AutoscalingConfig( min_replicas=min_replicas, max_replicas=max_replicas, target_ongoing_requests=target_requests, upscale_delay_s=0.0, downscale_delay_s=0.0, ) ctx = AutoscalingContext( config=config, total_num_requests=ongoing_requests, current_num_replicas=4, target_num_replicas=4, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state=policy_state, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) new_num_replicas, _ = wrapped_replica_queue_length_autoscaling_policy(ctx=ctx) assert new_num_replicas == ongoing_requests / target_requests def test_callable_and_direct_values(self): config = AutoscalingConfig(min_replicas=1, max_replicas=10) deployment_id = DeploymentID(name="test", app_name="test_app") replica_id = ReplicaID(unique_id="r1", deployment_id=deployment_id) # Test callables with lazy evaluation and caching call_counts = {"requests": 0, "queued": 0, "agg": 0, "raw": 0} ctx = AutoscalingContext( config=config, deployment_id=None, deployment_name="test", app_name=None, current_num_replicas=5, target_num_replicas=5, running_replicas=[], total_num_requests=lambda: ( call_counts.update({"requests": call_counts["requests"] + 1}), 42.0, )[1], total_queued_requests=lambda: ( call_counts.update({"queued": call_counts["queued"] + 1}), 10.0, )[1], aggregated_metrics=lambda: ( call_counts.update({"agg": call_counts["agg"] + 1}), {"m": {replica_id: 5.0}}, )[1], raw_metrics=lambda: ( call_counts.update({"raw": call_counts["raw"] + 1}), {"m": {replica_id: [TimeStampedValue(1.0, 5.0)]}}, )[1], capacity_adjusted_min_replicas=1, capacity_adjusted_max_replicas=10, policy_state={}, last_scale_up_time=None, last_scale_down_time=None, current_time=None, total_pending_async_requests=0, ) # Callables not executed until accessed assert all(c == 0 for c in call_counts.values()) # First access executes callables assert ctx.total_num_requests == 42.0 assert ctx.total_queued_requests == 10.0 assert ctx.aggregated_metrics == {"m": {replica_id: 5.0}} assert ctx.raw_metrics["m"][replica_id][0].value == 5.0 assert all(c == 1 for c in call_counts.values()) # Second access uses cached values _ = ctx.total_num_requests _ = ctx.total_queued_requests _ = ctx.aggregated_metrics _ = ctx.raw_metrics assert all(c == 1 for c in call_counts.values()) # Test direct values (non-callable) ctx2 = AutoscalingContext( config=config, deployment_id=None, deployment_name="test", app_name=None, current_num_replicas=5, target_num_replicas=5, running_replicas=[], total_num_requests=100.0, total_queued_requests=20.0, aggregated_metrics={"m2": {replica_id: 15.0}}, raw_metrics={"m2": {replica_id: [TimeStampedValue(2.0, 25.0)]}}, capacity_adjusted_min_replicas=1, capacity_adjusted_max_replicas=10, policy_state={}, last_scale_up_time=None, last_scale_down_time=None, current_time=None, total_pending_async_requests=0, ) assert ctx2.total_num_requests == 100.0 assert ctx2.total_queued_requests == 20.0 assert ctx2.aggregated_metrics == {"m2": {replica_id: 15.0}} assert ctx2.raw_metrics["m2"][replica_id][0].value == 25.0 class TestAutoscalingConfigParameters: def test_apply_scaling_factors_upscale(self): config = AutoscalingConfig( min_replicas=1, max_replicas=30, upscaling_factor=0.5 ) desired_num_replicas = 20 current_num_replicas = 10 result = _apply_scaling_factors( desired_num_replicas, current_num_replicas, config ) # Expected: 10 + 0.5 * (20 - 10) = 15 assert result == 15 def test_apply_scaling_factors_downscale(self): config = AutoscalingConfig( min_replicas=1, max_replicas=30, downscaling_factor=0.5 ) desired_num_replicas = 5 current_num_replicas = 20 result = _apply_scaling_factors( desired_num_replicas, current_num_replicas, config ) # Expected: 20 - 0.5 * (20 - 5) = ceil(12.5) = 13 assert result == 13 def test_apply_scaling_factors_stuck_downscale(self): config = AutoscalingConfig( min_replicas=1, max_replicas=30, downscaling_factor=0.5 ) desired_num_replicas = 9 current_num_replicas = 10 result = _apply_scaling_factors( desired_num_replicas, current_num_replicas, config ) # Expected: 10 - 0.5 * (10 - 9) = 9.5 = ceil(9.5) = 10. The logic then adjusts it to 9. assert result == 9 def test_upscale_delay_respected_with_slow_loop_steps(self): """Regression test: upscale_delay_s must be respected even when each control loop step takes longer than CONTROL_LOOP_INTERVAL_S. Previously, the code counted consecutive iterations and compared to int(delay_s / CONTROL_LOOP_INTERVAL_S), assuming each iteration is exactly 0.1s. With 600ms/iteration and upscale_delay_s=100s that caused a ~6× overshoot (~600s actual vs 100s configured). The fix uses wall-clock timestamps, so the upscale fires after exactly upscale_delay_s of real elapsed time regardless of iteration duration. This test simulates slow iterations via _now and verifies that: - upscale fires after ~upscale_delay_s of simulated time (not 6× later) - the elapsed simulated time is within one iteration of upscale_delay_s """ upscale_delay_s = 100.0 # Simulate a loaded cluster: 500ms loop step + 100ms sleep = 600ms/iter. loop_step_duration_s = 0.5 actual_iteration_s = loop_step_duration_s + CONTROL_LOOP_INTERVAL_S config = AutoscalingConfig( min_replicas=1, max_replicas=10, upscale_delay_s=upscale_delay_s, ) ctx = AutoscalingContext( target_num_replicas=1, current_num_replicas=1, config=config, capacity_adjusted_min_replicas=1, capacity_adjusted_max_replicas=10, policy_state={}, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_num_requests=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) # Simulate wall-clock time advancing by actual_iteration_s per call. start_time = 0.0 simulated_now = start_time fire_time = None iterations = 0 decision = ctx.target_num_replicas while decision == ctx.target_num_replicas: decision, ctx.policy_state = _apply_delay_logic( desired_num_replicas=5, curr_target_num_replicas=ctx.target_num_replicas, config=ctx.config, policy_state=ctx.policy_state, _now=simulated_now, ) if decision != ctx.target_num_replicas: fire_time = simulated_now simulated_now += actual_iteration_s iterations += 1 actual_elapsed_s = fire_time - start_time # With wall-clock tracking, actual elapsed time must be close to # upscale_delay_s — within one iteration at most. assert actual_elapsed_s <= upscale_delay_s + actual_iteration_s, ( f"Upscale fired too late: elapsed={actual_elapsed_s:.1f}s " f"vs configured={upscale_delay_s}s" ) assert actual_elapsed_s >= upscale_delay_s, ( f"Upscale fired too early: elapsed={actual_elapsed_s:.1f}s " f"vs configured={upscale_delay_s}s" ) # Overshoot must be at most one iteration (not 6× as in the old code). overshoot_factor = actual_elapsed_s / upscale_delay_s assert ( overshoot_factor < 1.1 ), f"Expected ≤1.1× overshoot with wall-clock fix, got {overshoot_factor:.2f}×" def test_apply_delay_logic_upscale(self): """Test upscale delay uses wall-clock time.""" config = AutoscalingConfig( min_replicas=1, max_replicas=10, upscale_delay_s=0.3, ) ctx = AutoscalingContext( target_num_replicas=1, current_num_replicas=1, config=config, capacity_adjusted_min_replicas=1, capacity_adjusted_max_replicas=10, policy_state={}, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_num_requests=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) base_time = 1000.0 # Simulate calls before the delay elapses — should not scale up. for i in range(3): decision, ctx.policy_state = _apply_delay_logic( desired_num_replicas=5, curr_target_num_replicas=ctx.target_num_replicas, config=ctx.config, policy_state=ctx.policy_state, _now=base_time + i * 0.05, ) assert decision == 1, f"Should not scale up on iteration {i}" # Simulate a call after the delay has elapsed (use +0.31 to avoid float precision # issues: 1000.3 - 1000.0 evaluates to 0.2999... in IEEE 754). decision, _ = _apply_delay_logic( desired_num_replicas=5, curr_target_num_replicas=ctx.target_num_replicas, config=ctx.config, policy_state=ctx.policy_state, _now=base_time + 0.31, ) assert decision == 5 def test_apply_delay_logic_downscale(self): """Test downscale delay uses wall-clock time.""" config = AutoscalingConfig( min_replicas=0, max_replicas=10, downscale_to_zero_delay_s=0.4, downscale_delay_s=0.3, ) ctx = AutoscalingContext( target_num_replicas=4, current_num_replicas=4, config=config, capacity_adjusted_min_replicas=0, capacity_adjusted_max_replicas=10, policy_state={}, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_num_requests=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) base_time = 1000.0 # Downscale from 4->1: calls before downscale_delay_s (0.3s) elapses for i in range(3): decision_num_replicas, ctx.policy_state = _apply_delay_logic( desired_num_replicas=0, curr_target_num_replicas=ctx.target_num_replicas, config=ctx.config, policy_state=ctx.policy_state, _now=base_time + i * 0.05, ) assert decision_num_replicas == 4, f"Should not scale down on iteration {i}" # Call after 0.3s has elapsed — should downscale 4->1 (use +0.31 to avoid # float precision issues: 1000.3 - 1000.0 evaluates to 0.2999... in IEEE 754). decision_num_replicas, ctx.policy_state = _apply_delay_logic( desired_num_replicas=0, curr_target_num_replicas=ctx.target_num_replicas, config=ctx.config, policy_state=ctx.policy_state, _now=base_time + 0.31, ) assert decision_num_replicas == 1 ctx.target_num_replicas = decision_num_replicas # Downscale from 1->0: calls before downscale_to_zero_delay_s (0.4s) elapses base_time2 = base_time + 0.5 for i in range(3): decision_num_replicas, ctx.policy_state = _apply_delay_logic( desired_num_replicas=0, curr_target_num_replicas=ctx.target_num_replicas, config=ctx.config, policy_state=ctx.policy_state, _now=base_time2 + i * 0.1, ) assert ( decision_num_replicas == 1 ), f"Should not scale down from 1 to 0 on tick {i}" # Call after 0.4s has elapsed — should downscale 1->0 (use +0.41 to avoid # float precision issues with large base timestamps). decision_num_replicas, _ = _apply_delay_logic( desired_num_replicas=0, curr_target_num_replicas=ctx.target_num_replicas, config=ctx.config, policy_state=ctx.policy_state, _now=base_time2 + 0.41, ) assert decision_num_replicas == 0 @_apply_autoscaling_config def simple_custom_policy(ctx: AutoscalingContext): """ Custom policy to check default parameters are applied """ if ctx.total_num_requests > 0: desired_num_replicas = 3 else: desired_num_replicas = 0 return desired_num_replicas, {} @_apply_app_level_autoscaling_config def simple_app_level_policy(ctxs): """App-level policy that always requests scaling up to 5 replicas.""" return {deployment_id: 5 for deployment_id in ctxs.keys()}, {} class TestCustomPolicyWithDefaultParameters: @pytest.mark.parametrize("downscale_to_zero_delay_s", [None, 300]) def test_upscale_downscale_delay(self, downscale_to_zero_delay_s): """Unit test for upscale_delay_s, downscale_delay_s and downscale_to_zero_delay_s""" min_replicas = 0 max_replicas = 4 policy_state = {} config = AutoscalingConfig( min_replicas=min_replicas, max_replicas=max_replicas, target_ongoing_requests=1, upscale_delay_s=20.0, downscale_delay_s=400.0, downscale_to_zero_delay_s=downscale_to_zero_delay_s, ) ctx = AutoscalingContext( config=config, total_num_requests=1, current_num_replicas=0, target_num_replicas=0, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state=policy_state, deployment_id=None, deployment_name=None, app_name=None, running_replicas=None, current_time=None, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) # Scale up when there are 0 replicas and current_handle_queued_queries > 0 new_num_replicas, _ = simple_custom_policy(ctx=ctx) assert new_num_replicas == 1 _run_upscale_downscale_flow( simple_custom_policy, config, ctx, overload_requests=70, start_replicas=1, upscale_target=3, ) @pytest.mark.parametrize( "current_replicas, total_requests, upscale_factor, downscale_factor, expected_replicas", [ # Upscale cases # current=1, desired_raw=3 # factor=0.5 → ceil(1 + 0.5*(3-1)) = 2 (1, 100, 0.5, None, 2), # factor=1.0 → ceil(1 + 1.0*(3-1)) = 3 (1, 100, 1.0, None, 3), # Downscale cases # current=3, desired_raw=0 # factor=0.5 → ceil(3 + 0.5*(0-3)) = ceil(1.5) = 2 (3, 0, None, 0.5, 2), # factor=1.0 → max(ceil(3 + 1.0*(0-3)),1) = 1 (3, 0, None, 1.0, 1), ], ) def test_apply_scaling_factors( self, current_replicas, total_requests, upscale_factor, downscale_factor, expected_replicas, ): """ The test checks if the scaling factors are applied """ config = AutoscalingConfig( min_replicas=0, max_replicas=10, upscaling_factor=upscale_factor, downscaling_factor=downscale_factor, upscale_delay_s=0.0, downscale_delay_s=0.0, downscale_to_zero_delay_s=0.0, ) ctx = AutoscalingContext( config=config, deployment_id=None, deployment_name="test", app_name=None, current_num_replicas=current_replicas, target_num_replicas=current_replicas, running_replicas=None, total_num_requests=0, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, capacity_adjusted_min_replicas=config.min_replicas, capacity_adjusted_max_replicas=config.max_replicas, policy_state={}, last_scale_up_time=None, last_scale_down_time=None, current_time=None, total_pending_async_requests=0, ) ctx = create_context_with_overrides(ctx, total_num_requests=total_requests) num_replicas, _ = simple_custom_policy(ctx) assert num_replicas == expected_replicas class TestAppLevelPolicyWithDefaultParameters: def test_cold_start_fast_path(self): """App-level decorator should cold-start immediately (0 -> 1) even with delays.""" config = AutoscalingConfig( min_replicas=0, max_replicas=10, target_ongoing_requests=10, upscale_delay_s=20.0, downscale_delay_s=200.0, ) d1 = DeploymentID(name="d1", app_name="app") d2 = DeploymentID(name="d2", app_name="app") contexts = { d1: AutoscalingContext( config=config, deployment_id=d1, deployment_name="d1", app_name="app", current_num_replicas=0, target_num_replicas=0, running_replicas=[], total_num_requests=1, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, capacity_adjusted_min_replicas=0, capacity_adjusted_max_replicas=10, policy_state={}, last_scale_up_time=None, last_scale_down_time=None, current_time=None, total_pending_async_requests=0, ), d2: AutoscalingContext( config=config, deployment_id=d2, deployment_name="d2", app_name="app", current_num_replicas=0, target_num_replicas=0, running_replicas=[], total_num_requests=1, total_queued_requests=None, aggregated_metrics=None, raw_metrics=None, capacity_adjusted_min_replicas=0, capacity_adjusted_max_replicas=10, policy_state={}, last_scale_up_time=None, last_scale_down_time=None, current_time=None, total_pending_async_requests=0, ), } decisions, _ = simple_app_level_policy(contexts) assert decisions[d1] == 1 assert decisions[d2] == 1 class TestGangSchedulingAutoscalingPolicy: """Tests for GangSchedulingAutoscalingPolicy which aligns replica counts to gang size multiples.""" def _make_policy(self, gang_size, inner_result): def base_scaling_policy(ctx): return inner_result, {} return GangSchedulingAutoscalingPolicy(base_scaling_policy, gang_size) def _make_ctx(self, current_num_replicas): return AutoscalingContext( deployment_id=DeploymentID(name="test", app_name="app"), deployment_name="test", app_name=None, current_num_replicas=current_num_replicas, target_num_replicas=0, running_replicas=[], total_num_requests=0, total_queued_requests=0, aggregated_metrics=None, raw_metrics=None, capacity_adjusted_min_replicas=0, capacity_adjusted_max_replicas=0, policy_state={}, last_scale_up_time=None, last_scale_down_time=None, current_time=None, config=None, total_pending_async_requests=0, ) @pytest.mark.parametrize( "raw_autoscaling_decision,expected", [ (5, 8), # ceil(5/4)*4 = 8 (7, 8), # ceil(7/4)*4 = 8 (8, 8), # already aligned (9, 12), # ceil(9/4)*4 = 12 ], ) def test_rounds_up(self, raw_autoscaling_decision, expected): """Always rounds up, independent of current replica count.""" policy = self._make_policy(gang_size=4, inner_result=raw_autoscaling_decision) for current in [4, 12]: ctx = self._make_ctx(current_num_replicas=current) assert policy(ctx)[0] == expected def test_zero_replicas_unchanged(self): ctx = self._make_ctx(current_num_replicas=4) policy = self._make_policy(gang_size=4, inner_result=0) assert policy(ctx)[0] == 0 def test_gang_size_one_no_op(self): ctx = self._make_ctx(current_num_replicas=3) policy = self._make_policy(gang_size=1, inner_result=5) assert policy(ctx)[0] == 5 def _create_state(self, gang_size, min_replicas, max_replicas, running=0): """Create a DeploymentAutoscalingState with gang config registered.""" dep_id = DeploymentID(name="test", app_name="app") state = DeploymentAutoscalingState(dep_id) info = MagicMock() info.deployment_config.autoscaling_config = AutoscalingConfig( min_replicas=min_replicas, max_replicas=max_replicas, ) info.deployment_config.gang_scheduling_config = GangSchedulingConfig( gang_size=gang_size, ) info.target_capacity = None info.target_capacity_direction = None info.config_changed.return_value = False state.register(info, curr_target_num_replicas=min_replicas) state._running_replicas = list(range(running)) return state def test_integration_with_deployment_state(self): """Test that gang autoscaling policy is auto-injected when registering with gang config.""" state = self._create_state(gang_size=4, min_replicas=4, max_replicas=16) assert isinstance(state._policy, GangSchedulingAutoscalingPolicy) assert state._policy._gang_size == 4 def test_integration_no_gang(self): """Without gang config, the policy is not wrapped.""" dep_id = DeploymentID(name="test", app_name="app") state = DeploymentAutoscalingState(dep_id) info = MagicMock() info.deployment_config.autoscaling_config = AutoscalingConfig( min_replicas=1, max_replicas=10, ) info.deployment_config.gang_scheduling_config = None info.target_capacity = None info.target_capacity_direction = None info.config_changed.return_value = False state.register(info, curr_target_num_replicas=1) assert not isinstance(state._policy, GangSchedulingAutoscalingPolicy) def test_integration_scale_up_respects_max(self): """Gang alignment rounds up, but apply_bounds clips to max.""" state = self._create_state( gang_size=4, min_replicas=4, max_replicas=8, running=4 ) # ceil(9/4)*4 = 12, clipped to max_replicas=8 state._policy = GangSchedulingAutoscalingPolicy( lambda ctx: (9, {}), gang_size=4 ) decision = state.get_decision_num_replicas(curr_target_num_replicas=4) assert decision == 8 def test_integration_scale_down_respects_min(self): """Gang alignment rounds up, but apply_bounds clips to min.""" state = self._create_state( gang_size=4, min_replicas=4, max_replicas=16, running=8 ) # ceil(1/4)*4 = 4, clipped to min_replicas=4 state._policy = GangSchedulingAutoscalingPolicy( lambda ctx: (1, {}), gang_size=4 ) decision = state.get_decision_num_replicas(curr_target_num_replicas=8) assert decision == 4 class TestWarmupScalingFeedbackLoop: """Regression tests for a feedback loop that causes deployments to scale to max_replicas during warmup when upscaling_factor > 1. When current_num_replicas == 0 (replicas scheduled but not yet RUNNING) and there is no traffic, the policy should hold the target steady. """ @pytest.mark.parametrize("upscaling_factor", [0.5, 1.0, 1.5, 2.0, 10.0]) def test_no_runaway_scaling_during_warmup(self, upscaling_factor): min_replicas = 2 max_replicas = 10 config = AutoscalingConfig( min_replicas=min_replicas, max_replicas=max_replicas, target_ongoing_requests=0.2, upscaling_factor=upscaling_factor, upscale_delay_s=5.0, downscale_delay_s=30.0, ) target = min_replicas policy_state = {} for tick in range(100): ctx = AutoscalingContext( config=config, current_num_replicas=0, target_num_replicas=target, total_num_requests=0, total_queued_requests=0, capacity_adjusted_min_replicas=min_replicas, capacity_adjusted_max_replicas=max_replicas, policy_state=policy_state, deployment_id=None, deployment_name=None, app_name=None, running_replicas=[], current_time=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) new_target, policy_state = wrapped_replica_queue_length_autoscaling_policy( ctx=ctx ) assert new_target == target, ( f"Tick {tick}: target changed from {target} to {new_target} " f"with upscaling_factor={upscaling_factor}, " f"current_num_replicas=0, total_num_requests=0. " f"This indicates a feedback loop bug." ) target = new_target assert target == min_replicas class TestAppLevelPolicyStateIsolation: """ Test that no internal state cross contamination when a user policy returns the same dict object for multiple deployments. """ def _make_context(self, dep_id, policy_state=None): config = AutoscalingConfig( min_replicas=1, max_replicas=5, upscale_delay_s=100, downscale_delay_s=100, ) return AutoscalingContext( config=config, current_num_replicas=2, target_num_replicas=2, total_num_requests=0, total_queued_requests=0, capacity_adjusted_min_replicas=config.min_replicas, capacity_adjusted_max_replicas=config.max_replicas, policy_state=policy_state or {}, deployment_id=dep_id, deployment_name=dep_id.name, app_name=dep_id.app_name, running_replicas=[], current_time=None, aggregated_metrics=None, raw_metrics=None, last_scale_up_time=None, last_scale_down_time=None, total_pending_async_requests=0, ) def test_shared_user_state_does_not_contaminate_internal_state(self): d1 = DeploymentID("d1", "app") d2 = DeploymentID("d2", "app") fake_now = 1000.0 d1_internal_state = { SERVE_AUTOSCALING_DECISION_COUNTERS_KEY: 3, SERVE_AUTOSCALING_DECISION_TIMESTAMP_KEY: fake_now, } d2_internal_state = { SERVE_AUTOSCALING_DECISION_COUNTERS_KEY: 0, SERVE_AUTOSCALING_DECISION_TIMESTAMP_KEY: None, } shared_state = {"counter": 5} def fake_policy(contexts): return {d1: 3, d2: 3}, {d1: shared_state, d2: shared_state} wrapped = _apply_app_level_autoscaling_config(fake_policy) contexts = { d1: self._make_context(d1, policy_state=d1_internal_state), d2: self._make_context(d2, policy_state=d2_internal_state), } with patch("ray.serve.autoscaling_policy.time") as mock_time: mock_time.time = lambda: fake_now _, final_state = wrapped(contexts) # d1 had counter=3, timestamp=fake_now. Delay logic sees scale-up # (desired=3 > target=2), counter was positive so it increments to 4. # Delay hasn't elapsed (0s < 100s) so no reset. assert final_state[d1][SERVE_AUTOSCALING_DECISION_COUNTERS_KEY] == 4 assert final_state[d1][SERVE_AUTOSCALING_DECISION_TIMESTAMP_KEY] == fake_now # user state remains intact assert final_state[d1]["counter"] == 5 # d2 had counter=0, timestamp=None. Delay logic sees scale-up, # increments counter to 1, sets timestamp to fake_now. assert final_state[d2][SERVE_AUTOSCALING_DECISION_COUNTERS_KEY] == 1 assert final_state[d2][SERVE_AUTOSCALING_DECISION_TIMESTAMP_KEY] == fake_now # user state remains intact assert final_state[d2]["counter"] == 5 if __name__ == "__main__": sys.exit(pytest.main(["-v", "-s", __file__]))