wehub-resource-sync
1576 lines
58 KiB
Python
1576 lines
58 KiB
Python
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__]))
|