[Data] Info log cluster scale up decisions (ray-project#60357)

> I am thinking there is very limited visibility into the autoscaling
decisions, currently have to look through DEBUG logs FWICT. Adding some
visibility in terms of metrics and events would be nice, and promoting
key action logs to INFO would be my high-level suggestion

This PR logs a message to STDOUT whenever the autoscaler decides to
scale up the cluster.

---------

Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu>

Author: bveeramani

python/ray/data/_internal/cluster_autoscaler/__init__.py

@@ -40,8 +40,8 @@ def create_cluster_autoscaler(
     if selected_autoscaler == ClusterAutoscalerVersion.V2:
         return DefaultClusterAutoscalerV2(
             resource_manager,
-            resource_limits=resource_limits,
             execution_id=execution_id,
+            resource_limits=resource_limits,
         )
 
     elif selected_autoscaler == ClusterAutoscalerVersion.V1:

python/ray/data/_internal/cluster_autoscaler/default_cluster_autoscaler_v2.py

@@ -1,10 +1,9 @@
 import logging
 import math
 import time
-from collections import defaultdict
+from collections import Counter, defaultdict
 from dataclasses import dataclass
-from logging import getLogger
-from typing import TYPE_CHECKING, Callable, Dict, Optional
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
 
 import ray
 from .base_autoscaling_coordinator import AutoscalingCoordinator
@@ -16,14 +15,15 @@
     RollingLogicalUtilizationGauge,
 )
 from .util import cap_resource_request_to_limits
-from ray._private.ray_constants import env_float, env_integer
+from ray._private.ray_constants import env_bool, env_float, env_integer
 from ray.data._internal.cluster_autoscaler import ClusterAutoscaler
 from ray.data._internal.execution.interfaces.execution_options import ExecutionResources
+from ray.data._internal.execution.util import memory_string
 
 if TYPE_CHECKING:
     from ray.data._internal.execution.resource_manager import ResourceManager
 
-logger = getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 
 @dataclass(frozen=True)
@@ -40,13 +40,28 @@ def __post_init__(self):
         assert isinstance(self.mem, int)
         assert self.mem >= 0
 
+    def __str__(self):
+        return (
+            "{"
+            + f"CPU: {self.cpu}, GPU: {self.gpu}, memory: {memory_string(self.mem)}"
+            + "}"
+        )
+
     @classmethod
     def of(cls, *, cpu=0, gpu=0, mem=0):
         cpu = math.floor(cpu)
         gpu = math.floor(gpu)
         mem = math.floor(mem)
         return cls(cpu=cpu, gpu=gpu, mem=mem)
 
+    @classmethod
+    def from_bundle(cls, bundle: Dict[str, Any]) -> "_NodeResourceSpec":
+        return _NodeResourceSpec.of(
+            cpu=bundle.get("CPU", 0),
+            gpu=bundle.get("GPU", 0),
+            mem=bundle.get("memory", 0),
+        )
+
     def to_bundle(self):
         return {"CPU": self.cpu, "GPU": self.gpu, "memory": self.mem}
 
@@ -60,10 +75,9 @@ def _get_node_resource_spec_and_count() -> Dict[_NodeResourceSpec, int]:
         for node_group_config in cluster_config.node_group_configs:
             if not node_group_config.resources or node_group_config.max_count == 0:
                 continue
-            node_resource_spec = _NodeResourceSpec.of(
-                cpu=node_group_config.resources.get("CPU", 0),
-                gpu=node_group_config.resources.get("GPU", 0),
-                mem=node_group_config.resources.get("memory", 0),
+
+            node_resource_spec = _NodeResourceSpec.from_bundle(
+                node_group_config.resources
             )
             nodes_resource_spec_count[node_resource_spec] = 0
 
@@ -75,9 +89,7 @@ def _get_node_resource_spec_and_count() -> Dict[_NodeResourceSpec, int]:
     ]
 
     for r in node_resources:
-        node_resource_spec = _NodeResourceSpec.of(
-            cpu=r.get("CPU", 0), gpu=r.get("GPU", 0), mem=r.get("memory", 0)
-        )
+        node_resource_spec = _NodeResourceSpec.from_bundle(r)
         nodes_resource_spec_count[node_resource_spec] += 1
 
     return nodes_resource_spec_count
@@ -128,12 +140,16 @@ class DefaultClusterAutoscalerV2(ClusterAutoscaler):
         "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",
-        resource_limits: ExecutionResources,
         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,
@@ -160,7 +176,7 @@ def __init__(
         self._resource_utilization_calculator = resource_utilization_calculator
         # Threshold of cluster utilization to trigger scaling up.
         self._cluster_scaling_up_util_threshold = cluster_scaling_up_util_threshold
-        self._cluster_scaling_up_delta = cluster_scaling_up_delta
+        self._cluster_scaling_up_delta = int(math.ceil(cluster_scaling_up_delta))
         self._min_gap_between_autoscaling_requests_s = (
             min_gap_between_autoscaling_requests_s
         )
@@ -204,30 +220,18 @@ def try_trigger_scaling(self):
 
         # We separate active bundles (existing nodes) from pending bundles (scale-up delta)
         # to ensure existing nodes' resources are never crowded out by scale-up requests.
-        active_bundles = []
-        pending_bundles = []
-        debug_msg = ""
-        if logger.isEnabledFor(logging.DEBUG):
-            debug_msg = (
-                "Scaling up cluster. Current utilization: "
-                f"CPU={util.cpu:.2f}, GPU={util.gpu:.2f}, object_store_memory={util.object_store_memory:.2f}."
-                " Requesting resources:"
-            )
         # TODO(hchen): We scale up all nodes by the same delta for now.
         # We may want to distinguish different node types based on their individual
         # utilization.
+        active_bundles = []
+        pending_bundles = []
         node_resource_spec_count = self._get_node_counts()
         for node_resource_spec, count in node_resource_spec_count.items():
             bundle = node_resource_spec.to_bundle()
             # Bundles for existing nodes -> active (must include)
             active_bundles.extend([bundle] * count)
             # Bundles for scale-up delta -> pending (best-effort)
-            delta_count = int(math.ceil(self._cluster_scaling_up_delta))
-            pending_bundles.extend([bundle] * delta_count)
-            if logger.isEnabledFor(logging.DEBUG):
-                num_to_request = count + delta_count
-                debug_msg += f" [{bundle}: {count} -> {num_to_request}]"
-        logger.debug(debug_msg)
+            pending_bundles.extend([bundle] * self._cluster_scaling_up_delta)
 
         # Cap the resource request to respect user-configured limits.
         # Active bundles (existing nodes) are always included; pending bundles
@@ -236,8 +240,42 @@ def try_trigger_scaling(self):
             active_bundles, pending_bundles, self._resource_limits
         )
 
+        if resource_request != active_bundles:
+            self._log_resource_request(util, active_bundles, resource_request)
+
         self._send_resource_request(resource_request)
 
+    def _log_resource_request(
+        self,
+        current_utilization: ExecutionResources,
+        active_bundles: List[Dict[str, float]],
+        resource_request: List[Dict[str, float]],
+    ) -> None:
+        message = (
+            "The utilization of one or more logical resource is higher than the "
+            f"specified threshold of {self._cluster_scaling_up_util_threshold:.0%}: "
+            f"CPU={current_utilization.cpu:.0%}, GPU={current_utilization.gpu:.0%}, "
+            f"object_store_memory={current_utilization.object_store_memory:.0%}. "
+            f"Requesting {self._cluster_scaling_up_delta} node(s) of each shape:"
+        )
+
+        current_node_counts = Counter(
+            [_NodeResourceSpec.from_bundle(bundle) for bundle in active_bundles]
+        )
+        requested_node_counts = Counter(
+            [_NodeResourceSpec.from_bundle(bundle) for bundle in resource_request]
+        )
+        for node_spec, requested_count in requested_node_counts.items():
+            current_count = current_node_counts.get(node_spec, 0)
+            message += f" [{node_spec}: {current_count} -> {requested_count}]"
+
+        if self.RAY_DATA_DISABLE_AUTOSCALER_LOGGING:
+            level = logging.DEBUG
+        else:
+            level = logging.INFO
+
+        logger.log(level, message)
+
     def _send_resource_request(self, resource_request):
         # Make autoscaler resource request.
         self._autoscaling_coordinator.request_resources(

python/ray/data/tests/test_default_cluster_autoscaler_v2.py

@@ -1,3 +1,4 @@
+import logging
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -566,6 +567,35 @@ def get_node_counts():
             f"Got total memory={resources_allocated.memory}"
         )
 
+    def test_try_scale_up_logs_info_message(self, propagate_logs, caplog):
+        fake_coordinator = FakeAutoscalingCoordinator()
+        node_spec = _NodeResourceSpec.of(cpu=1, gpu=0, mem=8 * 1024**3)
+        utilization = ExecutionResources(cpu=1, gpu=1, object_store_memory=1)
+        autoscaler = DefaultClusterAutoscalerV2(
+            resource_manager=MagicMock(),
+            execution_id="test_execution_id",
+            resource_utilization_calculator=StubUtilizationGauge(utilization),
+            min_gap_between_autoscaling_requests_s=0,
+            autoscaling_coordinator=fake_coordinator,
+            get_node_counts=lambda: {node_spec: 1},
+        )
+
+        with caplog.at_level(logging.INFO):
+            autoscaler.try_trigger_scaling()
+
+        expected_message = (
+            "The utilization of one or more logical resource is higher than the "
+            "specified threshold of 75%: CPU=100%, GPU=100%, object_store_memory=100%. "
+            "Requesting 1 node(s) of each shape: "
+            "[{CPU: 1, GPU: 0, memory: 8.0GiB}: 1 -> 2]"
+        )
+        log_messages = [record.message for record in caplog.records]
+        assert expected_message in log_messages, (
+            f"Expected log message not found.\n"
+            f"Expected: {expected_message}\n"
+            f"Actual logs: {log_messages}"
+        )
+
 
 if __name__ == "__main__":
     import sys