[Storage]Mount cached mode

sky/backends/cloud_vm_ray_backend.py

@@ -58,6 +58,7 @@
 from sky.utils import command_runner
 from sky.utils import common_utils
 from sky.utils import controller_utils
+from sky.utils import env_options
 from sky.utils import log_utils
 from sky.utils import resources_utils
 from sky.utils import rich_utils
@@ -4543,7 +4544,7 @@ def _execute_storage_mounts(
         # Handle cases where `storage_mounts` is None. This occurs when users
         # initiate a 'sky start' command from a Skypilot version that predates
         # the introduction of the `storage_mounts_metadata` feature.
-        if not storage_mounts:
+        if storage_mounts is None:
             return
 
         # Process only mount mode objects here. COPY mode objects have been
@@ -4552,10 +4553,12 @@ def _execute_storage_mounts(
         storage_mounts = {
             path: storage_mount
             for path, storage_mount in storage_mounts.items()
-            if storage_mount.mode == storage_lib.StorageMode.MOUNT
+            if (storage_mount.mode == storage_lib.StorageMode.MOUNT or
+                storage_mount.mode == storage_lib.StorageMode.MOUNT_CACHED)
         }
 
-        # Handle cases when there aren't any Storages with MOUNT mode.
+        # Handle cases when there aren't any Storages with either MOUNT or
+        # MOUNT_CACHED mode.
         if not storage_mounts:
             return
         start = time.time()
@@ -4581,7 +4584,13 @@ def _execute_storage_mounts(
                         'successfully without mounting the bucket.')
             # Get the first store and use it to mount
             store = list(storage_obj.stores.values())[0]
-            mount_cmd = store.mount_command(dst)
+            if storage_obj.mode == storage_lib.StorageMode.MOUNT:
+                mount_cmd = store.mount_command(dst)
+                action_message = 'Mounting'
+            else:
+                assert storage_obj.mode == storage_lib.StorageMode.MOUNT_CACHED
+                mount_cmd = store.mount_cached_command(dst)
+                action_message = 'Mounting cached mode'
             src_print = (storage_obj.source
                          if storage_obj.source else storage_obj.name)
             if isinstance(src_print, list):
@@ -4593,7 +4602,7 @@ def _execute_storage_mounts(
                     target=dst,
                     cmd=mount_cmd,
                     run_rsync=False,
-                    action_message='Mounting',
+                    action_message=action_message,
                     log_path=log_path,
                     # Need to source bashrc, as the cloud specific CLI or SDK
                     # may require PATH in bashrc.
@@ -4611,12 +4620,23 @@ def _execute_storage_mounts(
                                  f' to an empty or non-existent path.')
                     raise RuntimeError(error_msg) from None
                 else:
-                    # Strip the command (a big heredoc) from the exception
-                    raise exceptions.CommandError(
-                        e.returncode,
-                        command='to mount',
-                        error_msg=e.error_msg,
-                        detailed_reason=e.detailed_reason) from None
+                    # By default, raising an error caused from mounting_utils
+                    # shows a big heredoc as part of it. Here, we want to
+                    # conditionally show the heredoc only if SKYPILOT_DEBUG
+                    # is set
+                    if env_options.Options.SHOW_DEBUG_INFO.get():
+                        raise exceptions.CommandError(
+                            e.returncode,
+                            command='to mount',
+                            error_msg=e.error_msg,
+                            detailed_reason=e.detailed_reason)
+                    else:
+                        # Strip the command (a big heredoc) from the exception
+                        raise exceptions.CommandError(
+                            e.returncode,
+                            command='to mount',
+                            error_msg=e.error_msg,
+                            detailed_reason=e.detailed_reason) from None
 
         end = time.time()
         logger.debug(f'Storage mount sync took {end - start} seconds.')

sky/cloud_stores.py

@@ -438,19 +438,18 @@ def is_directory(self, url: str) -> bool:
     def _get_rclone_sync_command(self, source: str, destination: str):
         bucket_name, data_path, bucket_region = data_utils.split_cos_path(
             source)
-        bucket_rclone_profile = Rclone.generate_rclone_bucket_profile_name(
-            bucket_name, Rclone.RcloneClouds.IBM)
-        data_path_in_bucket = bucket_name + data_path
-        rclone_config_data = Rclone.get_rclone_config(bucket_name,
-                                                      Rclone.RcloneClouds.IBM,
-                                                      bucket_region)
+        rclone_profile_name = data_utils.Rclone.RcloneStores.IBM.get_profile_name(
+            bucket_name)
+        data_path_in_bucket = f'{bucket_name}{data_path}'
+        rclone_config = data_utils.Rclone.RcloneStores.IBM.get_config(
+            rclone_profile_name=rclone_profile_name, region=bucket_region)
         # configure_rclone stores bucket profile in remote cluster's rclone.conf
         configure_rclone = (
-            f' mkdir -p ~/.config/rclone/ &&'
-            f' echo "{rclone_config_data}">> {Rclone.RCLONE_CONFIG_PATH}')
+            f' mkdir -p {constants.RCLONE_CONFIG_DIR} &&'
+            f' echo "{rclone_config}">> {constants.RCLONE_CONFIG_PATH}')
         download_via_rclone = (
             'rclone copy '
-            f'{bucket_rclone_profile}:{data_path_in_bucket} {destination}')
+            f'{rclone_profile_name}:{data_path_in_bucket} {destination}')
 
         all_commands = list(self._GET_RCLONE)
         all_commands.append(configure_rclone)