Repository: ray-project/ray
File: python/ray/data/_internal/cluster_autoscaler/default_cluster_autoscaler_v2.py (Python)

Code before:
    DEFAULT_CLUSTER_UTIL_AVG_WINDOW_S: int = env_integer(
        "RAY_DATA_CLUSTER_UTIL_AVG_WINDOW_S",
        10,
    )
    # Default number of nodes to add per node type.
    DEFAULT_CLUSTER_SCALING_UP_DELTA: int = env_integer(
        "RAY_DATA_CLUSTER_SCALING_UP_DELTA",
        1,
    )

    # Min number of seconds between two autoscaling requests.
    MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS: int = env_integer(
        "RAY_DATA_MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS",
        10,
    )
    # The time in seconds after which an autoscaling request will expire.
    AUTOSCALING_REQUEST_EXPIRE_TIME_S: int = env_integer(
        "RAY_DATA_AUTOSCALING_REQUEST_EXPIRE_TIME_S",
        180,
    )

    def __init__(
        self,
        resource_manager: "ResourceManager",
        execution_id: str,
        resource_limits: ExecutionResources = ExecutionResources.for_limits(),
        resource_utilization_calculator: Optional[ResourceUtilizationGauge] = None,
        cluster_scaling_up_util_threshold: float = DEFAULT_CLUSTER_SCALING_UP_UTIL_THRESHOLD,  # noqa: E501
        cluster_scaling_up_delta: float = DEFAULT_CLUSTER_SCALING_UP_DELTA,
        cluster_util_avg_window_s: float = DEFAULT_CLUSTER_UTIL_AVG_WINDOW_S,
        min_gap_between_autoscaling_requests_s: float = MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS,  # noqa: E501
        autoscaling_coordinator: Optional[AutoscalingCoordinator] = None,
        get_node_counts: Callable[[], Dict[_NodeResourceSpec, int]] = (
            _get_node_resource_spec_and_count
        ),
    ):
        assert cluster_scaling_up_delta > 0
        assert cluster_util_avg_window_s > 0
        assert min_gap_between_autoscaling_requests_s >= 0

        if resource_utilization_calculator is None:
            resource_utilization_calculator = RollingLogicalUtilizationGauge(
                resource_manager, cluster_util_avg_window_s=cluster_util_avg_window_s
            )

        if autoscaling_coordinator is None:
            autoscaling_coordinator = DefaultAutoscalingCoordinator()

        self._resource_limits = resource_limits
        self._resource_utilization_calculator = resource_utilization_calculator
        # Threshold of cluster utilization to trigger scaling up.

Review comment (suggestion):
Let's use `inf` method from Execution Options so it'll be more understandable.

Code after:
    DEFAULT_CLUSTER_UTIL_AVG_WINDOW_S: int = env_integer(
        "RAY_DATA_CLUSTER_UTIL_AVG_WINDOW_S",
        10,
    )
    # Default number of nodes to add per node type.
    DEFAULT_CLUSTER_SCALING_UP_DELTA: int = env_integer(
        "RAY_DATA_CLUSTER_SCALING_UP_DELTA",
        1,
    )

    # Min number of seconds between two autoscaling requests.
    MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS: int = env_integer(
        "RAY_DATA_MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS",
        10,
    )
    # The time in seconds after which an autoscaling request will expire.
    AUTOSCALING_REQUEST_EXPIRE_TIME_S: int = env_integer(
        "RAY_DATA_AUTOSCALING_REQUEST_EXPIRE_TIME_S",
        180,
    )
    # Whether to disable INFO-level logs.
    RAY_DATA_DISABLE_AUTOSCALER_LOGGING = env_bool(
        "RAY_DATA_DISABLE_AUTOSCALER_LOGGING", False
    )

    def __init__(
        self,
        resource_manager: "ResourceManager",
        execution_id: str,
        resource_limits: ExecutionResources = ExecutionResources.inf(),
        resource_utilization_calculator: Optional[ResourceUtilizationGauge] = None,
        cluster_scaling_up_util_threshold: float = DEFAULT_CLUSTER_SCALING_UP_UTIL_THRESHOLD,  # noqa: E501
        cluster_scaling_up_delta: float = DEFAULT_CLUSTER_SCALING_UP_DELTA,
        cluster_util_avg_window_s: float = DEFAULT_CLUSTER_UTIL_AVG_WINDOW_S,
        min_gap_between_autoscaling_requests_s: float = MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS,  # noqa: E501
        autoscaling_coordinator: Optional[AutoscalingCoordinator] = None,
        get_node_counts: Callable[[], Dict[_NodeResourceSpec, int]] = (
            _get_node_resource_spec_and_count
        ),
    ):
        assert cluster_scaling_up_delta > 0
        assert cluster_util_avg_window_s > 0
        assert min_gap_between_autoscaling_requests_s >= 0

        if resource_utilization_calculator is None:
            resource_utilization_calculator = RollingLogicalUtilizationGauge(
                resource_manager, cluster_util_avg_window_s=cluster_util_avg_window_s
            )

        if autoscaling_coordinator is None:
            autoscaling_coordinator = DefaultAutoscalingCoordinator()