[runtime_env] Support specifying the runtime_resources directory for testing (#20257)

python/ray/_private/parameter.py

@@ -97,6 +97,8 @@ class RayParams:
             used by the raylet process.
         temp_dir (str): If provided, it will specify the root temporary
             directory for the Ray process.
+        runtime_env_dir_name (str): If provided, specifies the directory that
+            will be created in the session dir to hold runtime_env files.
         include_log_monitor (bool): If True, then start a log monitor to
             monitor the log files for all processes on this node and push their
             contents to Redis.
@@ -159,6 +161,7 @@ class RayParams:
                  plasma_store_socket_name=None,
                  raylet_socket_name=None,
                  temp_dir=None,
+                 runtime_env_dir_name=None,
                  include_log_monitor=None,
                  autoscaling_config=None,
                  start_initial_python_workers_for_first_job=False,
@@ -207,6 +210,8 @@ class RayParams:
         self.plasma_store_socket_name = plasma_store_socket_name
         self.raylet_socket_name = raylet_socket_name
         self.temp_dir = temp_dir
+        self.runtime_env_dir_name = (
+            runtime_env_dir_name or ray_constants.DEFAULT_RUNTIME_ENV_DIR_NAME)
         self.include_log_monitor = include_log_monitor
         self.autoscaling_config = autoscaling_config
         self.metrics_agent_port = metrics_agent_port

python/ray/node.py

@@ -300,8 +300,8 @@ class Node:
         old_logs_dir = os.path.join(self._logs_dir, "old")
         try_to_create_directory(old_logs_dir)
         # Create a directory to be used for runtime environment.
-        self._runtime_env_dir = os.path.join(self._session_dir,
-                                             "runtime_resources")
+        self._runtime_env_dir = os.path.join(
+            self._session_dir, self._ray_params.runtime_env_dir_name)
         try_to_create_directory(self._runtime_env_dir)
 
     def get_resource_spec(self):

python/ray/ray_constants.py

@@ -271,6 +271,9 @@ HEALTHCHECK_EXPIRATION_S = os.environ.get("RAY_HEALTHCHECK_EXPIRATION_S", 10)
 # src/ray/common/constants.h.
 SETUP_WORKER_FILENAME = "setup_worker.py"
 
+# Directory name where runtime_env resources will be created & cached.
+DEFAULT_RUNTIME_ENV_DIR_NAME = "runtime_resources"
+
 # Used to separate lines when formatting the call stack where an ObjectRef was
 # created.
 CALL_STACK_LINE_DELIMITER = " | "

python/ray/tests/test_runtime_env_working_dir.py

@@ -308,13 +308,11 @@ def test_s3_uri(start_cluster, option, per_task_actor):
     "source", [S3_PACKAGE_URI, lazy_fixture("tmp_working_dir")])
 def test_multi_node(start_cluster, option: str, source: str):
     """Tests that the working_dir is propagated across multi-node clusters."""
-    # TODO(architkulkarni): Currently all nodes in cluster_utils share the same
-    # session directory, which isn't the case for real world clusters. Once
-    # this is fixed, we should test GC with NUM_NODES > 1 here.
-    NUM_NODES = 1
+    NUM_NODES = 2
     cluster, address = start_cluster
-    for _ in range(NUM_NODES - 1):  # Head node already added.
-        cluster.add_node(num_cpus=1)
+    for i in range(NUM_NODES - 1):  # Head node already added.
+        cluster.add_node(
+            num_cpus=1, runtime_env_dir_name=f"node_{i}_runtime_resources")
 
     if option == "working_dir":
         ray.init(address, runtime_env={"working_dir": source})