Decouple ray up and user's file_mounts / setup (#407)

* wip: Add setup in provision pipeline

* Fix gcp/azure

* remove useless variables

* minor fix

* Add some TODOs

* Fix comments

* Fix comments

* Fix gcp/azure initialization_commands

* Remove setup from template

* Fix setup directory

* Change rsync back to -Pavz

* Remove unused argument

* fix file_mount/dir_mount

Author: Michaelvll

examples/horovod_distributed_tf_app.py

@@ -1,4 +1,4 @@
-# TODO (zhwu): Fix this for #272.
+# TODO(zhwu): Fix this for #272.
 import json
 from typing import Dict, List
 

sky/backends/backend.py

@@ -42,6 +42,9 @@ def sync_file_mounts(
     ) -> None:
         raise NotImplementedError
 
+    def setup(self, handle: ResourceHandle, task: Task) -> None:
+        raise NotImplementedError
+
     def add_storage_objects(self, task: Task) -> None:
         raise NotImplementedError
 

sky/backends/backend_utils.py

@@ -6,13 +6,11 @@
 import pathlib
 import shlex
 import subprocess
-import tempfile
 import textwrap
 import time
 from typing import Dict, List, Optional, Tuple, Union
 import uuid
 import yaml
-import zlib
 
 import jinja2
 
@@ -22,7 +20,6 @@
 from sky import clouds
 from sky import sky_logging
 from sky import resources
-from sky import task as task_lib
 from sky.backends import wheel_utils
 from sky.adaptors import azure
 from sky.skylet import log_lib
@@ -90,12 +87,7 @@ def wrap_file_mount(cls, path: str) -> str:
         return os.path.join(_SKY_REMOTE_FILE_MOUNTS_DIR, path.lstrip('/'))
 
     @classmethod
-    def make_safe_symlink_command(
-            cls,
-            *,
-            source: str,
-            target: str,
-            download_target_commands: Optional[List[str]] = None) -> str:
+    def make_safe_symlink_command(cls, *, source: str, target: str) -> str:
         """Returns a command that safely symlinks 'source' to 'target'.
 
         All intermediate directories of 'source' will be owned by $USER,
@@ -147,8 +139,6 @@ def make_safe_symlink_command(
             f'(echo "!!! Failed mounting because path exists ({source})"; '
             'exit 1)))',
         ]
-        if download_target_commands is not None:
-            commands += download_target_commands
         commands += [
             # Link.
             f'sudo ln -s {target} {source}',
@@ -311,8 +301,7 @@ def remove_cluster(cls, ip: str, auth_config: Dict[str, str]):
 
 
 # TODO: too many things happening here - leaky abstraction. Refactor.
-def write_cluster_config(task: task_lib.Task,
-                         to_provision: Resources,
+def write_cluster_config(to_provision: Resources,
                          num_nodes: int,
                          cluster_config_template: str,
                          cluster_name: str,
@@ -373,48 +362,6 @@ def write_cluster_config(task: task_lib.Task,
 
     assert cluster_name is not None
 
-    setup_sh_path = None
-    if task.setup is not None:
-        codegen = textwrap.dedent(f"""#!/bin/bash
-            . $(conda info --base)/etc/profile.d/conda.sh
-            {task.setup}
-        """)
-        # Use a stable path, /<tempdir>/sky_setup_<checksum>.sh, because
-        # rerunning the same task without any changes to the content of the
-        # setup command should skip the setup step.  Using NamedTemporaryFile()
-        # would generate a random path every time, hence re-triggering setup.
-        checksum = zlib.crc32(codegen.encode())
-        tempdir = tempfile.gettempdir()
-        # TODO: file lock on this path, in case tasks have the same setup cmd.
-        with open(os.path.join(tempdir, f'sky_setup_{checksum}.sh'), 'w') as f:
-            f.write(codegen)
-        setup_sh_path = f.name
-
-    # File mounts handling for remote paths possibly without write access:
-    #  (1) in 'file_mounts' sections, add <prefix> to these target paths.
-    #  (2) then, create symlinks from '/.../file' to '<prefix>/.../file'.
-    # We need to do these since as of Ray 1.8, this is not supported natively
-    # (Docker works though, of course):
-    #  https://github.com/ray-project/ray/pull/9332
-    #  https://github.com/ray-project/ray/issues/9326
-    mounts = task.get_local_to_remote_file_mounts()
-    wrapped_file_mounts = {}
-    initialization_commands = []
-    if mounts is not None:
-        for remote, local in mounts.items():
-            if not os.path.isabs(remote) and not remote.startswith('~/'):
-                remote = f'~/{remote}'
-            if remote.startswith('~/') or remote.startswith('/tmp/'):
-                # Skip as these should be writable locations.
-                wrapped_file_mounts[remote] = local
-                continue
-            assert os.path.isabs(remote), (remote, local)
-            wrapped_remote = FileMountHelper.wrap_file_mount(remote)
-            wrapped_file_mounts[wrapped_remote] = local
-            command = FileMountHelper.make_safe_symlink_command(
-                source=remote, target=wrapped_remote)
-            initialization_commands.append(command)
-
     # TODO(suquark): once we have sky on PYPI, we should directly install sky
     # from PYPI
     local_wheel_path = wheel_utils.build_sky_wheel()
@@ -424,14 +371,8 @@ def write_cluster_config(task: task_lib.Task,
             resources_vars,
             **{
                 'cluster_name': cluster_name,
-                'setup_sh_path': setup_sh_path,
-                'workdir': task.workdir,
-                'docker_image': task.docker_image,
                 'num_nodes': num_nodes,
                 'disk_size': to_provision.disk_size,
-                # File mounts handling.
-                'file_mounts': wrapped_file_mounts,
-                'initialization_commands': initialization_commands or None,
                 # Region/zones.
                 'region': region,
                 'zones': ','.join(zones),