[Data] Dump verbose ResourceManager telemetry into ray-data.log (ray-project#58261)

> Thank you for contributing to Ray! 🚀
> Please review the [Ray Contribution
Guide](https://docs.ray.io/en/master/ray-contribute/getting-involved.html)
before opening a pull request.

> ⚠️ Remove these instructions before submitting your PR.

> 💡 Tip: Mark as draft if you want early feedback, or ready for review
when it's complete.

## Description

This change make RD by dump verbose telemetry for `ResourceManager` into
the `ray-data.log` by default.

## Related issues
> Link related issues: "Fixes ray-project#1234", "Closes ray-project#1234", or "Related to
ray-project#1234".

## Additional information
> Optional: Add implementation details, API changes, usage examples,
screenshots, etc.

---------

Signed-off-by: Alexey Kudinkin <ak@anyscale.com>

Author: alexeykudinkin

python/ray/data/_internal/execution/resource_manager.py

@@ -1,12 +1,11 @@
 import logging
 import math
-import os
 import time
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from typing import TYPE_CHECKING, Callable, Dict, Iterable, List, Optional
 
-from ray._private.ray_constants import env_float
+from ray._private.ray_constants import env_bool, env_float
 from ray.data._internal.execution.interfaces.execution_options import (
     ExecutionOptions,
     ExecutionResources,
@@ -33,7 +32,12 @@
 
 
 logger = logging.getLogger(__name__)
-DEBUG_RESOURCE_MANAGER = os.environ.get("RAY_DATA_DEBUG_RESOURCE_MANAGER", "0") == "1"
+
+
+LOG_DEBUG_TELEMETRY_FOR_RESOURCE_MANAGER_OVERRIDE: Optional[bool] = env_bool(
+    "RAY_DATA_DEBUG_RESOURCE_MANAGER", None
+)
+
 
 # These are physical operators that must receive all inputs before they start
 # processing data.
@@ -83,8 +87,6 @@ def __init__(
         # the operator, including the external output buffer in OpState, and the
         # input buffers of the downstream operators.
         self._mem_op_outputs: Dict[PhysicalOperator, int] = defaultdict(int)
-        # Whether to print debug information.
-        self._debug = DEBUG_RESOURCE_MANAGER
 
         self._op_resource_allocator: Optional["OpResourceAllocator"] = None
 
@@ -266,7 +268,7 @@ def get_op_usage(self, op: PhysicalOperator) -> ExecutionResources:
         """Return the resource usage of the given operator at the current time."""
         return self._op_usages[op]
 
-    def get_op_usage_str(self, op: PhysicalOperator) -> str:
+    def get_op_usage_str(self, op: PhysicalOperator, *, verbose: bool) -> str:
         """Return a human-readable string representation of the resource usage of
         the given operator."""
         usage_str = f"{self._op_running_usages[op].cpu:.1f} CPU"
@@ -276,7 +278,11 @@ def get_op_usage_str(self, op: PhysicalOperator) -> str:
             f", {self._op_running_usages[op].object_store_memory_str()} object store"
         )
 
-        if self._debug:
+        # NOTE: Config can override requested verbosity level
+        if LOG_DEBUG_TELEMETRY_FOR_RESOURCE_MANAGER_OVERRIDE is not None:
+            verbose = LOG_DEBUG_TELEMETRY_FOR_RESOURCE_MANAGER_OVERRIDE
+
+        if verbose:
             usage_str += (
                 f" (in={memory_string(self._mem_op_internal[op])},"
                 f"out={memory_string(self._mem_op_outputs[op])})"

python/ray/data/_internal/execution/streaming_executor.py

@@ -734,7 +734,7 @@ def _debug_dump_topology(topology: Topology, resource_manager: ResourceManager)
     for i, (op, state) in enumerate(topology.items()):
         state.update_display_metrics(resource_manager)
         logger.debug(
-            f"{i}: {state.summary_str(resource_manager)}, "
+            f"{i}: {state.summary_str(resource_manager, verbose=True)}, "
             f"Blocks Outputted: {state.num_completed_tasks}/{op.num_outputs_total()}, "
             f"Metrics: {state.op_display_metrics.display_str()}"
         )

python/ray/data/_internal/execution/streaming_executor_state.py

@@ -387,7 +387,9 @@ def refresh_progress_bar(self, resource_manager: ResourceManager) -> None:
             self.progress_bar.set_description(self.summary_str(resource_manager))
             self.progress_bar.refresh()
 
-    def summary_str(self, resource_manager: ResourceManager) -> str:
+    def summary_str(
+        self, resource_manager: ResourceManager, verbose: bool = False
+    ) -> str:
         # Active tasks
         active = self.op.num_active_tasks()
         desc = f"- {self.op.name}: Tasks: {active}"
@@ -409,7 +411,7 @@ def summary_str(self, resource_manager: ResourceManager) -> str:
 
         # Queued blocks
         desc += f"; Queued blocks: {self.total_enqueued_input_blocks()} ({memory_string(self.total_enqueued_input_blocks_bytes())})"
-        desc += f"; Resources: {resource_manager.get_op_usage_str(self.op)}"
+        desc += f"; Resources: {resource_manager.get_op_usage_str(self.op, verbose=verbose)}"
 
         # Any additional operator specific information.
         suffix = self.op.progress_str()