Repository: skypilot-org/skypilot
Pull request #1239 (2022-10-12 20:46:42)

[Storage] Refactor storage and fix data transfer service
Adopt changes for storage from #1152.

Tested:
- [x] `tests/run_smoke_tests.sh TestStorageWithCredentials`
- [x] The following command
```
gsutil mb gs://sky-imagenet-bucket-gcp
python - <<EOF
from sky.data import data_transfer
data_transfer.s3_to_gcs('sky-imagenet-bucket', 'sky-imagenet-bucket-gcp') 
EOF
```


Patch:
diff --git a/docs/source/examples/syncing-code-artifacts.rst b/docs/source/examples/syncing-code-artifacts.rst
index 635c03661..e76348728 100644
--- a/docs/source/examples/syncing-code-artifacts.rst
+++ b/docs/source/examples/syncing-code-artifacts.rst
@@ -64,6 +64,7 @@ scripts, access checkpoints, etc.).
     $ # cluster1 will get the updated code.
     $ sky exec cluster1 task.yaml
 
+.. _file-mounts-example:
 
 Uploading files outside of workdir
 --------------------------------------
@@ -93,6 +94,9 @@ Example file mounts:
 
 For more details, see `this example <https://github.com/skypilot-org/skypilot/blob/master/examples/using_file_mounts.yaml>`_ and :ref:`YAML Configuration <yaml-spec>`.
 
+If you have edited the ``file_mounts`` section and would like to have it reflected on an existing cluster without rerunning the ``setup`` commands,
+pass the ``--no-setup`` flag to ``sky launch``. For example, ``sky launch --no-setup -c <cluster_name> <task.yaml>``.
+
 .. note::
 
     Items listed in a :code:`.gitignore` file (or a ``.git/info/exclude`` file)
diff --git a/docs/source/reference/faq.rst b/docs/source/reference/faq.rst
index 8a122518d..ff2ff8971 100644
--- a/docs/source/reference/faq.rst
+++ b/docs/source/reference/faq.rst
@@ -56,10 +56,32 @@ To get around this, mount the files to a different path, then symlink to them.
     ln -s /tmp/tmp.txt ~/code-repo/
 
 
+
+How to make SkyPilot clusters use my Weights & Biases credentials?
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Install the wandb library on your laptop and login to your account via ``wandb login``.
+Then, add the following lines in your task yaml file:
+
+.. code-block:: yaml
+
+  file_mounts:
+    ~/.netrc: ~/.netrc
+
+How to update an existing cluster's ``file_mounts`` without rerunning ``setup``?
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+If you have edited the ``file_mounts`` section (e.g., by adding some files) and would like to have it reflected on an existing cluster, running ``sky launch -c <cluster> ..`` would work, but it would rerun the ``setup`` commands.
+
+To avoid rerunning the ``setup`` commands, pass the ``--no-setup`` flag to ``sky launch``.
+
+
+
 (Advanced) How to edit or update the regions or pricing information used by SkyPilot?
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-SkyPilot stores regions and pricing information for different cloud resource types in CSV files known as "service catalogs".
+SkyPilot stores regions and pricing information for different cloud resource types in CSV files known as
+`"service catalogs" <https://github.com/skypilot-org/skypilot-catalog>`_.
 These catalogs are cached in the ``~/.sky/catalogs/<schema-version>/`` directory.
 Check out your schema version by running the following command:
 
@@ -71,15 +93,3 @@ You can customize the catalog files to your needs.
 For example, if you have access to special regions of GCP, add the data to ``~/.sky/catalogs/<schema-version>/gcp.csv``.
 Also, you can update the catalog for a specific cloud by deleting the CSV file (e.g., ``rm ~/.sky/catalogs/<schema-version>/gcp.csv``).
 SkyPilot will automatically download the latest catalog in the next run.
-
-
-How to make SkyPilot clusters use my Weights & Biases credentials?
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-Install the wandb library on your laptop and login to your account via ``wandb login``.
-Then, add the following lines in your task yaml file:
-
-.. code-block:: yaml
-
-  file_mounts:
-    ~/.netrc: ~/.netrc
diff --git a/docs/source/reference/storage.rst b/docs/source/reference/storage.rst
index e7808cbae..8ec6992f7 100644
--- a/docs/source/reference/storage.rst
+++ b/docs/source/reference/storage.rst
@@ -4,7 +4,8 @@ SkyPilot Storage
 =================
 
 A SkyPilot Storage object represents an abstract data store containing large data
-files required by the task. Compared to file_mounts, storage is faster and
+files required by the task. Think of it as a bucket of files that can be attached
+to your task. Compared to file_mounts, storage is faster and
 can persist across runs, requiring fewer uploads from your local machine.
 Behind the scenes, storage automatically uploads all data in the source
 to a backing object store in a particular cloud (S3/GCS/Azure Blob).
@@ -175,6 +176,11 @@ and storage mounting:
     For mounts not using SkyPilot Storage (e.g., those using rsync) the symbolic links are directly copied, not their target data.
     The targets must be separately mounted or else the symlinks may break.
 
+.. note::
+    Storage only supports uploading directories (i.e., :code:`source` cannot be a file).
+    To upload a single file to a bucket, please put in a directory and specify the directory as the source.
+    To directly copy a file to a VM, please use regular :ref:`file mounts <file-mounts-example>`.
+
 Creating a shared file system
 -----------------------------
 
diff --git a/examples/storage_demo.yaml b/examples/storage_demo.yaml
index 35fac797e..b6d048064 100644
--- a/examples/storage_demo.yaml
+++ b/examples/storage_demo.yaml
@@ -94,7 +94,7 @@ file_mounts:
   # When the VM is initialized, the contents of the bucket are copied to
   # /datasets-storage. If the bucket already exists, it is fetched and re-used.
   /datasets-storage:
-    name: sky-dataset-romil # Make sure this name is unique or you own this bucket
+    name: sky-dataset-romilzz # Make sure this name is unique or you own this bucket
     source: ~/datasets
     store: s3 # Could be either of [s3, gcs]; default: None
     persistent: True  # Defaults to True, can be set to false.
@@ -110,7 +110,7 @@ file_mounts:
   # other storage mounts using the same bucket with MOUNT mode. Note that the
   # source is synced with the remote bucket everytime this task is run.
   /dataset-storage-mount:
-    name: sky-dataset-romil
+    name: sky-dataset-romilzz
     source: ~/datasets
     mode: MOUNT
 
@@ -137,7 +137,7 @@ file_mounts:
   # this approach can also be used to create a shared filesystem across workers.
   # See examples/storage/pingpong.yaml for an example.
   /outputs-mount:
-    name: romil-output-bucket
+    name: romil-output-bucketzz
     mode: MOUNT
 
 run: |
diff --git a/examples/using_file_mounts.yaml b/examples/using_file_mounts.yaml
index 75f1ab3af..328e0b398 100644
--- a/examples/using_file_mounts.yaml
+++ b/examples/using_file_mounts.yaml
@@ -30,6 +30,7 @@ file_mounts:
 
   ~/tmpfile: ~/tmpfile
   ~/a/b/c/tmpfile: ~/tmpfile
+  ~/tmpfile2: ~/tmp-workdir/tmp file
 
   # Intermediate dirs are created and chown-ed to $USER if they don't already
   # exist.  Thus they can be used too, e.g., 'mkdir -p /data/logs'.  See 'run'.
diff --git a/sky/backends/backend.py b/sky/backends/backend.py
index 4dcff6e5d..272ffb7e3 100644
--- a/sky/backends/backend.py
+++ b/sky/backends/backend.py
@@ -44,7 +44,7 @@ class Backend:
                   retry_until_up: bool = False) -> ResourceHandle:
         if cluster_name is None:
             cluster_name = sky.backends.backend_utils.generate_cluster_name()
-        usage_lib.messages.usage.update_cluster_name(cluster_name)
+        usage_lib.record_cluster_name_for_current_operation(cluster_name)
         usage_lib.messages.usage.update_actual_task(task)
         return self._provision(task, to_provision, dryrun, stream_logs,
                                cluster_name, retry_until_up)
@@ -76,7 +76,8 @@ class Backend:
     @usage_lib.messages.usage.update_runtime('execute')
     def execute(self, handle: ResourceHandle, task: 'task_lib.Task',
                 detach_run: bool) -> None:
-        usage_lib.messages.usage.update_cluster_name(handle.get_cluster_name())
+        usage_lib.record_cluster_name_for_current_operation(
+            handle.get_cluster_name())
         usage_lib.messages.usage.update_actual_task(task)
         return self._execute(handle, task, detach_run)
 
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 51acef19e..2f0739321 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -31,6 +31,7 @@ from ray.autoscaler._private import commands as ray_commands
 from ray.autoscaler._private import util as ray_util
 import rich.console as rich_console
 import rich.progress as rich_progress
+import yaml
 
 import sky
 from sky import authentication as auth
@@ -109,12 +110,33 @@ CLUSTER_STATUS_LOCK_TIMEOUT_SECONDS = 20
 # Remote dir that holds our runtime files.
 _REMOTE_RUNTIME_FILES_DIR = '~/.sky/.runtime_files'
 
+# Include the fields that will be used for generating tags that  distinguishes the
+# cluster in ray, to avoid the stopped cluster being discarded due to updates in
+# the yaml template.
+# Some notes on the fields:
+# - 'provider' fields will be used for bootstrapping and insert more new items in
+#   'node_config'.
+# - keeping the auth is not enough becuase the content of the key file will be used
+#   for calculating the hash.
+# TODO(zhwu): Keep in sync with the fields used in https://github.com/ray-project/ray/blob/e4ce38d001dbbe09cd21c497fedd03d692b2be3e/python/ray/autoscaler/_private/commands.py#L687-L701
+_RAY_YAML_KEYS_TO_RESTORE_FOR_BACK_COMPATIBILITY = {
+    'cluster_name', 'provider', 'auth', 'node_config'
+}
+
 
 def is_ip(s: str) -> bool:
     """Returns whether this string matches IP_ADDR_REGEX."""
     return len(re.findall(IP_ADDR_REGEX, s)) == 1
 
 
+def _get_yaml_path_from_cluster_name(cluster_name: str,
+                                     prefix: str = SKY_USER_FILE_PATH) -> str:
+    output_path = pathlib.Path(
+        prefix).expanduser().resolve() / f'{cluster_name}.yml'
+    os.makedirs(output_path.parents[0], exist_ok=True)
+    return str(output_path)
+
+
 def fill_template(template_name: str,
                   variables: Dict,
                   output_path: Optional[str] = None,
@@ -129,10 +151,8 @@ def fill_template(template_name: str,
     if output_path is None:
         assert ('cluster_name' in variables), ('cluster_name is required.')
         cluster_name = variables.get('cluster_name')
-        output_path = pathlib.Path(
-            output_prefix).expanduser() / f'{cluster_name}.yml'
-        os.makedirs(output_path.parents[0], exist_ok=True)
-        output_path = str(output_path)
+        output_path = _get_yaml_path_from_cluster_name(cluster_name,
+                                                       output_prefix)
     output_path = os.path.abspath(output_path)
 
     # Add yaml file path to the template variables.
@@ -222,8 +242,11 @@ def _optimize_file_mounts(yaml_path: str) -> None:
 
     # (For local) Move all runtime files, including the just-written yaml, to
     # local_runtime_files_dir/.
-    all_local_sources = ' '.join(
-        local_src for local_src in file_mounts.values())
+    all_local_sources = ''
+    for local_src in file_mounts.values():
+        full_local_src = str(pathlib.Path(local_src).expanduser())
+        # Add quotes for paths containing spaces.
+        all_local_sources += f'{full_local_src!r} '
     # Takes 10-20 ms on laptop incl. 3 clouds' credentials.
     subprocess.run(f'cp -r {all_local_sources} {local_runtime_files_dir}/',
                    shell=True,
@@ -244,7 +267,7 @@ def path_size_megabytes(path: str) -> int:
         subprocess.check_output(
             f'rsync {command_runner.RSYNC_DISPLAY_OPTION} '
             f'{command_runner.RSYNC_FILTER_OPTION} '
-            f'{git_exclude_filter} --dry-run {path}',
+            f'{git_exclude_filter} --dry-run {path!r}',
             shell=True).splitlines()[-1])
     total_bytes = rsync_output.split(' ')[3].replace(',', '')
     return int(total_bytes) // 10**6
@@ -655,18 +678,46 @@ class SSHConfigHelper(object):
                 break
 
 
+def _replace_yaml_dicts(new_yaml: str, old_yaml: str,
+                        key_names: Set[str]) -> str:
+    """Replaces 'new' with 'old' for all keys in key_names.
+
+    The replacement will be applied recursively and only for the blocks
+    with the key in key_names, and have the same ancestors in both 'new'
+    and 'old' YAML tree.
+    """
+
+    def _restore_block(new_block: Dict[str, Any], old_block: Dict[str, Any]):
+        for key, value in new_block.items():
+            if key in key_names:
+                if key in old_block:
+                    new_block[key] = old_block[key]
+                else:
+                    del new_block[key]
+            elif isinstance(value, dict):
+                if key in old_block:
+                    _restore_block(value, old_block[key])
+
+    new_config = yaml.safe_load(new_yaml)
+    old_config = yaml.safe_load(old_yaml)
+    _restore_block(new_config, old_config)
+    return common_utils.dump_yaml_str(new_config)
+
+
 # TODO: too many things happening here - leaky abstraction. Refactor.
 @timeline.event
-def write_cluster_config(to_provision: 'resources.Resources',
-                         num_nodes: int,
-                         cluster_config_template: str,
-                         cluster_name: str,
-                         local_wheel_path: pathlib.Path,
-                         wheel_hash: str,
-                         region: Optional[clouds.Region] = None,
-                         zones: Optional[List[clouds.Zone]] = None,
-                         auth_config: Optional[Dict[str, str]] = None,
-                         dryrun: bool = False) -> Dict[str, str]:
+def write_cluster_config(
+        to_provision: 'resources.Resources',
+        num_nodes: int,
+        cluster_config_template: str,
+        cluster_name: str,
+        local_wheel_path: pathlib.Path,
+        wheel_hash: str,
+        region: Optional[clouds.Region] = None,
+        zones: Optional[List[clouds.Zone]] = None,
+        auth_config: Optional[Dict[str, str]] = None,
+        dryrun: bool = False,
+        keep_launch_fields_in_existing_config: bool = True) -> Dict[str, str]:
     """Fills in cluster configuration templates and writes them out.
 
     Returns: {provisioner: path to yaml, the provisioning spec}.
@@ -700,6 +751,12 @@ def write_cluster_config(to_provision: 'resources.Resources',
         auth_config = onprem_utils.get_local_auth_config(cluster_name)
     region_name = resources_vars.get('region')
 
+    yaml_path = _get_yaml_path_from_cluster_name(cluster_name)
+    old_yaml_content = None
+    if os.path.exists(yaml_path) and keep_launch_fields_in_existing_config:
+        with open(yaml_path, 'r') as f:
+            old_yaml_content = f.read()
+
     yaml_path = fill_template(
         cluster_config_template,
         dict(
@@ -750,6 +807,16 @@ def write_cluster_config(to_provision: 'resources.Resources',
         # been fully tested yet.
         _optimize_file_mounts(yaml_path)
 
+    # Restore the old yaml content for backward compatibility.
+    if old_yaml_content is not None:
+        with open(yaml_path, 'r') as f:
+            new_yaml_content = f.read()
+        restored_yaml_content = _replace_yaml_dicts(
+            new_yaml_content, old_yaml_content,
+            _RAY_YAML_KEYS_TO_RESTORE_FOR_BACK_COMPATIBILITY)
+        with open(yaml_path, 'w') as f:
+            f.write(restored_yaml_content)
+
     usage_lib.messages.usage.update_ray_yaml(yaml_path)
     # For TPU nodes. TPU VMs do not need TPU_NAME.
     if (resources_vars.get('tpu_type') is not None and
@@ -966,12 +1033,6 @@ def parallel_data_transfer_to_nodes(
     style = colorama.Style
 
     origin_source = source
-    if run_rsync:
-        # Do this for local src paths, not for cloud store URIs
-        # (otherwise we have '<abs path to cwd>/gs://.../object/').
-        full_src = os.path.abspath(os.path.expanduser(origin_source))
-        if not os.path.islink(full_src) and not os.path.isfile(full_src):
-            source = os.path.join(full_src, '')
 
     def _sync_node(runner: 'command_runner.SSHCommandRunner') -> None:
         if cmd is not None:
@@ -1041,8 +1102,9 @@ def query_head_ip_with_retries(cluster_yaml: str, max_attempts: int = 1) -> str:
     backoff = common_utils.Backoff(initial_backoff=5, max_backoff_factor=5)
     for i in range(max_attempts):
         try:
+            full_cluster_yaml = str(pathlib.Path(cluster_yaml).expanduser())
             out = subprocess_utils.run(
-                f'ray get-head-ip {cluster_yaml}',
+                f'ray get-head-ip {full_cluster_yaml!r}',
                 stdout=subprocess.PIPE,
                 stderr=subprocess.DEVNULL).stdout.decode().strip()
             head_ip = re.findall(IP_ADDR_REGEX, out)
@@ -1096,8 +1158,9 @@ def get_node_ips(cluster_yaml: str,
 
         for retry_cnt in range(worker_ip_max_attempts):
             try:
+                full_cluster_yaml = str(pathlib.Path(cluster_yaml).expanduser())
                 proc = subprocess_utils.run(
-                    f'ray get-worker-ips {cluster_yaml}',
+                    f'ray get-worker-ips {full_cluster_yaml!r}',
                     stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE)
                 out = proc.stdout.decode()
@@ -1587,9 +1650,11 @@ def _update_cluster_status_no_lock(
         try:
             backend = backends.CloudVmRayBackend()
             backend.set_autostop(handle, -1, stream_logs=False)
-        except (Exception, SystemExit):  # pylint: disable=broad-except
-            logger.debug('Failed to reset autostop.')
-        global_user_state.set_cluster_autostop_value(handle.cluster_name, -1)
+        except (Exception, SystemExit) as e:  # pylint: disable=broad-except
+            logger.debug(f'Failed to reset autostop. Due to {type(e)}: {e}')
+        global_user_state.set_cluster_autostop_value(handle.cluster_name,
+                                                     -1,
+                                                     to_down=False)
 
         # If the user starts part of a STOPPED cluster, we still need a status to
         # represent the abnormal status. For spot cluster, it can also represent
@@ -1724,13 +1789,9 @@ def get_clusters(
         f'[bold cyan]Refreshing status for {len(records)} cluster{plural}[/]',
         total=len(records))
 
-    terminated_clusters = []
-
     def _refresh_cluster(cluster_name):
         record = _update_cluster_status(cluster_name,
                                         acquire_per_cluster_status_lock=True)
-        if record is None:
-            terminated_clusters.append(cluster_name)
         progress.update(task, advance=1)
         return record
 
@@ -1738,14 +1799,31 @@ def get_clusters(
     with progress:
         updated_records = subprocess_utils.run_in_parallel(
             _refresh_cluster, cluster_names)
-    if terminated_clusters:
-        plural = 's were' if len(terminated_clusters) > 1 else ' was'
-        cluster_str = ', '.join(repr(name) for name in terminated_clusters)
-        yellow = colorama.Fore.YELLOW
-        reset = colorama.Style.RESET_ALL
-        logger.warning(f'{yellow}The following cluster{plural} terminated on '
-                       'the cloud and removed from the cluster table: '
-                       f'{cluster_str}{reset}')
+
+    # Show information for removed clusters.
+    autodown_clusters, remaining_clusters = [], []
+    for i, record in enumerate(records):
+        if updated_records[i] is None:
+            if record['to_down']:
+                autodown_clusters.append(cluster_names[i])
+            else:
+                remaining_clusters.append(cluster_names[i])
+
+    yellow = colorama.Fore.YELLOW
+    bright = colorama.Style.BRIGHT
+    reset = colorama.Style.RESET_ALL
+    if autodown_clusters:
+        plural = 's' if len(autodown_clusters) > 1 else ''
+        cluster_str = ', '.join(autodown_clusters)
+        logger.info(f'Autodowned cluster{plural}: '
+                    f'{bright}{cluster_str}{reset}')
+    if remaining_clusters:
+        plural = 's' if len(remaining_clusters) > 1 else ''
+        cluster_str = ', '.join(name for name in remaining_clusters)
+        logger.warning(f'{yellow}Cluster{plural} terminated on '
+                       f'the cloud: {reset}{bright}{cluster_str}{reset}')
+
+    # Filter out removed clusters.
     updated_records = [
         record for record in updated_records if record is not None
     ]
@@ -1911,7 +1989,7 @@ def kill_children_processes():
 # Handle ctrl-c
 def interrupt_handler(signum, frame):
     del signum, frame
-    logger.warning(f'{colorama.Fore.LIGHTBLACK_EX}The job will keep '
+    logger.warning(f'{colorama.Style.DIM}The job will keep '
                    f'running after Ctrl-C.{colorama.Style.RESET_ALL}')
     kill_children_processes()
     with ux_utils.print_exception_no_traceback():
@@ -1921,7 +1999,7 @@ def interrupt_handler(signum, frame):
 # Handle ctrl-z
 def stop_handler(signum, frame):
     del signum, frame
-    logger.warning(f'{colorama.Fore.LIGHTBLACK_EX}The job will keep '
+    logger.warning(f'{colorama.Style.DIM}The job will keep '
                    f'running after Ctrl-Z.{colorama.Style.RESET_ALL}')
     kill_children_processes()
     with ux_utils.print_exception_no_traceback():
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 404305119..fc22bdf91 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -924,7 +924,9 @@ class RetryingVmProvisioner(object):
                 self._wheel_hash,
                 region=region,
                 zones=zones,
-                dryrun=dryrun)
+                dryrun=dryrun,
+                keep_launch_fields_in_existing_config=prev_cluster_status
+                is not None)
             if dryrun:
                 return
             cluster_config_file = config_dict['ray']
@@ -1116,6 +1118,7 @@ class RetryingVmProvisioner(object):
             # different order from directly running in the console. The
             # `--log-style` and `--log-color` flags do not work. To reproduce,
             # `ray up --log-style pretty --log-color true | tee tmp.out`.
+
             returncode, stdout, stderr = log_lib.run_with_log(
                 # NOTE: --no-restart solves the following bug.  Without it, if
                 # 'ray up' (sky launch) twice on a cluster with >1 node, the
@@ -1133,7 +1136,15 @@ class RetryingVmProvisioner(object):
                 line_processor=log_utils.RayUpLineProcessor(),
                 # Reduce BOTO_MAX_RETRIES from 12 to 5 to avoid long hanging
                 # time during 'ray up' if insufficient capacity occurs.
-                env=dict(os.environ, BOTO_MAX_RETRIES='5'),
+                env=dict(
+                    os.environ,
+                    BOTO_MAX_RETRIES='5',
+                    # Use environment variables to disable the ray usage stats
+                    # (to avoid the 10 second wait for usage collection
+                    # confirmation), as the ray version on the user's machine
+                    # may be lower version that does not support the
+                    # `--disable-usage-stats` flag.
+                    RAY_USAGE_STATS_ENABLED='0'),
                 require_outputs=True,
                 # Disable stdin to avoid ray outputs mess up the terminal with
                 # misaligned output when multithreading/multiprocessing are used
@@ -1333,10 +1344,16 @@ class RetryingVmProvisioner(object):
                 'of the local cluster. Check if ray[default]==1.13.0 '
                 'is installed or running correctly.')
         backend.run_on_head(handle, 'ray stop', use_cached_head_ip=False)
+
         log_lib.run_with_log(
             ['ray', 'up', '-y', '--restart-only', handle.cluster_yaml],
             log_abs_path,
             stream_logs=False,
+            # Use environment variables to disable the ray usage collection
+            # (avoid the 10 second wait for usage collection confirmation),
+            # as the ray version on the user's machine may be lower version
+            # that does not support the `--disable-usage-stats` flag.
+            env=dict(os.environ, RAY_USAGE_STATS_ENABLED='0'),
             # Disable stdin to avoid ray outputs mess up the terminal with
             # misaligned output when multithreading/multiprocessing is used.
             # Refer to: https://github.com/ray-project/ray/blob/d462172be7c5779abf37609aed08af112a533e1e/python/ray/autoscaler/_private/subprocess_output_util.py#L264 # pylint: disable=line-too-long
@@ -1868,8 +1885,6 @@ class CloudVmRayBackend(backends.Backend):
         else:
             assert os.path.isdir(
                 full_workdir), f'{full_workdir} should be a directory.'
-            # FIXME(zongheng): audit; why not give users control to add '/'?
-            workdir = os.path.join(workdir, '')  # Adds trailing / if needed.
 
         # Raise warning if directory is too large
         dir_size = backend_utils.path_size_megabytes(full_workdir)
@@ -2608,10 +2623,11 @@ class CloudVmRayBackend(backends.Backend):
     def set_autostop(self,
                      handle: ResourceHandle,
                      idle_minutes_to_autostop: Optional[int],
+                     down: bool = False,
                      stream_logs: bool = True) -> None:
         if idle_minutes_to_autostop is not None:
             code = autostop_lib.AutostopCodeGen.set_autostop(
-                idle_minutes_to_autostop, self.NAME)
+                idle_minutes_to_autostop, self.NAME, down)
             returncode, _, stderr = self.run_on_head(handle,
                                                      code,
                                                      require_outputs=True,
@@ -2622,7 +2638,7 @@ class CloudVmRayBackend(backends.Backend):
                                                stderr=stderr,
                                                stream_logs=stream_logs)
             global_user_state.set_cluster_autostop_value(
-                handle.cluster_name, idle_minutes_to_autostop)
+                handle.cluster_name, idle_minutes_to_autostop, down)
 
     # TODO(zhwu): Refactor this to a CommandRunner class, so different backends
     # can support its own command runner.
diff --git a/sky/cli.py b/sky/cli.py
index 3be2ae374..40870c373 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -604,6 +604,7 @@ def _launch_with_confirm(
     detach_run: bool,
     no_confirm: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
+    down: bool = False,  # pylint: disable=redefined-outer-name
     retry_until_up: bool = False,
     no_setup: bool = False,
     node_type: Optional[str] = None,
@@ -654,6 +655,7 @@ def _launch_with_confirm(
             detach_run=detach_run,
             backend=backend,
             idle_minutes_to_autostop=idle_minutes_to_autostop,
+            down=down,
             retry_until_up=retry_until_up,
             no_setup=no_setup,
         )
@@ -991,7 +993,19 @@ def cli():
           'job queue. '
           'Setting this flag is equivalent to '
           'running ``sky launch -d ...`` and then ``sky autostop -i <minutes>``'
-          '. If not set, the cluster will not be auto-stopped.'))
+          '. If not set, the cluster will not be autostopped.'))
+@click.option(
+    '--down',
+    default=False,
+    is_flag=True,
+    required=False,
+    help=
+    ('Tear down the cluster after all jobs finish (successfully or '
+     'abnormally). If --idle-minutes-to-autostop is also set, the cluster will '
+     'be torn down after the specified idle time. '
+     'Note that if errors occur during provisioning/data syncing/setting up, '
+     'the cluster will not be torn down for debugging purposes.'),
+)
 @click.option(
     '--retry-until-up',
     '-r',
@@ -1033,6 +1047,7 @@ def launch(
     env: List[Dict[str, str]],
     disk_size: Optional[int],
     idle_minutes_to_autostop: Optional[int],
+    down: bool,  # pylint: disable=redefined-outer-name
     retry_until_up: bool,
     yes: bool,
     no_setup: bool,
@@ -1083,6 +1098,7 @@ def launch(
         detach_run=detach_run,
         no_confirm=yes,
         idle_minutes_to_autostop=idle_minutes_to_autostop,
+        down=down,
         retry_until_up=retry_until_up,
         no_setup=no_setup,
         is_local_cloud=onprem_utils.check_if_local_cloud(cluster))
@@ -1247,6 +1263,10 @@ def status(all: bool, refresh: bool):  # pylint: disable=redefined-builtin
     - STOPPED: The cluster is stopped and the storage is persisted. Use
       ``sky start`` to restart the cluster.
 
+    The autostop column indicates how long the cluster will be autostopped
+    after minutes of idling (no jobs running). If the time is followed by
+    '(down)', e.g. '1m (down)', the cluster will be autodowned, rather than
+    autostopped.
     """
     cluster_records = core.status(all=all, refresh=refresh)
     local_clusters = onprem_utils.check_and_get_local_clusters(
@@ -1391,7 +1411,7 @@ def logs(
         job_status_str = job_status.value if job_status is not None else 'None'
         click.echo(f'Job {job_id}: {job_status_str}')
         if job_status == job_lib.JobStatus.SUCCEEDED:
-            sys.exit(0)
+            return
         else:
             if job_status is None:
                 id_str = '' if job_id is None else f'{job_id} '
@@ -1431,7 +1451,7 @@ def cancel(cluster: str, all: bool, jobs: List[int]):  # pylint: disable=redefin
               '-a',
               default=None,
               is_flag=True,
-              help='Tear down all existing clusters.')
+              help='Stop all existing clusters.')
 @click.option('--yes',
               '-y',
               is_flag=True,
@@ -1472,10 +1492,10 @@ def stop(
       sky stop -a
 
     """
-    _terminate_or_stop_clusters(clusters,
-                                apply_to_all=all,
-                                terminate=False,
-                                no_confirm=yes)
+    _down_or_stop_clusters(clusters,
+                           apply_to_all=all,
+                           down=False,
+                           no_confirm=yes)
 
 
 @cli.command(cls=_DocumentedCodeCommand)
@@ -1493,12 +1513,21 @@ def stop(
               type=int,
               default=None,
               required=False,
-              help='Set the idle minutes before auto-stopping the cluster.')
-@click.option('--cancel',
-              default=False,
-              is_flag=True,
-              required=False,
-              help='Cancel the auto-stopping.')
+              help='Set the idle minutes before autostopping the cluster.')
+@click.option(
+    '--cancel',
+    default=False,
+    is_flag=True,
+    required=False,
+    help='Cancel the currently active auto{stop,down} setting for the '
+    'cluster.')
+@click.option(
+    '--down',
+    default=False,
+    is_flag=True,
+    required=False,
+    help='Use autodown (tear down the cluster; non-restartable), instead '
+    'of autostop (restartable).')
 @click.option('--yes',
               '-y',
               is_flag=True,
@@ -1511,15 +1540,21 @@ def autostop(
     all: Optional[bool],  # pylint: disable=redefined-builtin
     idle_minutes: Optional[int],
     cancel: bool,  # pylint: disable=redefined-outer-name
+    down: bool,  # pylint: disable=redefined-outer-name
     yes: bool,
 ):
-    """Schedule or cancel auto-stopping for cluster(s).
+    """Schedule or cancel an autostop or autodown for cluster(s).
 
     CLUSTERS are the name (or glob pattern) of the clusters to stop.  If both
     CLUSTERS and ``--all`` are supplied, the latter takes precedence.
 
+    If --down is passed, autodown (tear down the cluster; non-restartable) is
+    used, rather than autostop (restartable).
+
     ``--idle-minutes`` is the number of minutes of idleness (no pending/running
     jobs) after which the cluster will be stopped automatically.
+    Scheduling autostop twice on the same cluster will overwrite the previous
+    autostop schedule.
 
     ``--cancel`` will cancel the autostopping. If the cluster was not scheduled
     autostop, this will do nothing to autostop.
@@ -1527,6 +1562,9 @@ def autostop(
     If ``--idle-minutes`` and ``--cancel`` are not specified, default to 5
     minutes.
 
+    When multiple configurations are specified for the same cluster, e.g. using
+    ``sky autostop`` or ``sky launch -i``, the last one takes precedence.
+
     Examples:
 
     .. code-block:: bash
@@ -1546,11 +1584,11 @@ def autostop(
         idle_minutes = -1
     elif idle_minutes is None:
         idle_minutes = 5
-    _terminate_or_stop_clusters(clusters,
-                                apply_to_all=all,
-                                terminate=False,
-                                no_confirm=yes,
-                                idle_minutes_to_autostop=idle_minutes)
+    _down_or_stop_clusters(clusters,
+                           apply_to_all=all,
+                           down=down,
+                           no_confirm=yes,
+                           idle_minutes_to_autostop=idle_minutes)
 
 
 @cli.command(cls=_DocumentedCodeCommand)
@@ -1583,7 +1621,17 @@ def autostop(
           'job queue. '
           'Setting this flag is equivalent to '
           'running ``sky launch -d ...`` and then ``sky autostop -i <minutes>``'
-          '. If not set, the cluster will not be auto-stopped.'))
+          '. If not set, the cluster will not be autostopped.'))
+@click.option(
+    '--down',
+    default=False,
+    is_flag=True,
+    required=False,
+    help=
+    ('Tear down the cluster after all jobs finish (successfully or '
+     'abnormally). If --idle-minutes-to-autostop is also set, the cluster will '
+     'be torn down after the specified idle time.'),
+)
 @click.option(
     '--retry-until-up',
     '-r',
@@ -1594,8 +1642,13 @@ def autostop(
           'if we fail to start the cluster due to unavailability errors.'))
 @usage_lib.entrypoint
 # pylint: disable=redefined-builtin
-def start(clusters: Tuple[str], all: bool, yes: bool,
-          idle_minutes_to_autostop: int, retry_until_up: bool):
+def start(
+        clusters: Tuple[str],
+        all: bool,
+        yes: bool,
+        idle_minutes_to_autostop: Optional[int],
+        down: bool,  # pylint: disable=redefined-outer-name
+        retry_until_up: bool):
     """Restart cluster(s).
 
     If a cluster is previously stopped (status is STOPPED) or failed in
@@ -1623,6 +1676,9 @@ def start(clusters: Tuple[str], all: bool, yes: bool,
       sky start -a
 
     """
+    if down and idle_minutes_to_autostop is None:
+        raise click.UsageError(
+            '--idle-minutes-to-autostop must be set if --down is set.')
     to_start = []
 
     if not clusters and not all:
@@ -1711,7 +1767,10 @@ def start(clusters: Tuple[str], all: bool, yes: bool,
 
     for name in to_start:
         try:
-            core.start(name, idle_minutes_to_autostop, retry_until_up)
+            core.start(name,
+                       idle_minutes_to_autostop,
+                       retry_until_up,
+                       down=down)
         except exceptions.NotSupportedError as e:
             click.echo(str(e))
         click.secho(f'Cluster {name} started.', fg='green')
@@ -1752,7 +1811,7 @@ def down(
     CLUSTER is the name of the cluster (or glob pattern) to tear down.  If both
     CLUSTER and ``--all`` are supplied, the latter takes precedence.
 
-    Terminating a cluster will delete all associated resources (all billing
+    Tearing down a cluster will delete all associated resources (all billing
     stops), and any data on the attached disks will be lost. For local clusters,
     `sky down` does not terminate the local cluster, but instead removes the
     cluster from `sky status` and terminates the calling user's running jobs.
@@ -1776,38 +1835,40 @@ def down(
       sky down -a
 
     """
-    _terminate_or_stop_clusters(clusters,
-                                apply_to_all=all,
-                                terminate=True,
-                                no_confirm=yes,
-                                purge=purge)
+    _down_or_stop_clusters(clusters,
+                           apply_to_all=all,
+                           down=True,
+                           no_confirm=yes,
+                           purge=purge)
 
 
-def _terminate_or_stop_clusters(
+def _down_or_stop_clusters(
         names: Tuple[str],
         apply_to_all: Optional[bool],
-        terminate: bool,
+        down: bool,  # pylint: disable=redefined-outer-name
         no_confirm: bool,
         purge: bool = False,
         idle_minutes_to_autostop: Optional[int] = None) -> None:
-    """Terminates or (auto-)stops a cluster (or all clusters).
+    """Tears down or (auto-)stops a cluster (or all clusters).
 
     Reserved clusters (spot controller) can only be terminated if the cluster
     name is explicitly and uniquely specified (not via glob) and purge is set
     to True.
     """
-    assert idle_minutes_to_autostop is None or not terminate, (
-        idle_minutes_to_autostop, terminate)
-    command = 'down' if terminate else 'stop'
+    command = 'down' if down else 'stop'
     if not names and apply_to_all is None:
         raise click.UsageError(
             f'sky {command} requires either a cluster name (see `sky status`) '
             'or --all.')
 
-    operation = 'Terminating' if terminate else 'Stopping'
+    operation = 'Terminating' if down else 'Stopping'
     if idle_minutes_to_autostop is not None:
-        verb = 'Scheduling' if idle_minutes_to_autostop >= 0 else 'Cancelling'
-        operation = f'{verb} auto-stop on'
+        is_cancel = idle_minutes_to_autostop < 0
+        verb = 'Cancelling' if is_cancel else 'Scheduling'
+        option_str = 'down' if down else 'stop'
+        if is_cancel:
+            option_str = '{stop,down}'
+        operation = f'{verb} auto{option_str} on'
 
     if len(names) > 0:
         reserved_clusters = [
@@ -1819,7 +1880,7 @@ def _terminate_or_stop_clusters(
             name for name in _get_glob_clusters(names)
             if name not in backend_utils.SKY_RESERVED_CLUSTER_NAMES
         ]
-        if not terminate:
+        if not down:
             local_clusters = onprem_utils.check_and_get_local_clusters()
             # Local clusters are allowed to `sky down`, but not
             # `sky start/stop`. `sky down` unregisters the local cluster
@@ -1836,7 +1897,7 @@ def _terminate_or_stop_clusters(
             if not purge:
                 msg = (f'{operation} reserved cluster(s) '
                        f'{reserved_clusters_str} is not supported.')
-                if terminate:
+                if down:
                     msg += (
                         '\nPlease specify --purge (-p) to force-terminate the '
                         'reserved cluster(s).')
@@ -1874,7 +1935,7 @@ def _terminate_or_stop_clusters(
             # should've been printed by _get_glob_clusters() above.
             continue
         clusters.append(name)
-    usage_lib.messages.usage.update_cluster_name(clusters)
+    usage_lib.record_cluster_name_for_current_operation(clusters)
 
     if not clusters:
         click.echo('\nCluster(s) not found (tip: see `sky status`).')
@@ -1898,11 +1959,11 @@ def _terminate_or_stop_clusters(
         f'[bold cyan]{operation} {len(clusters)} cluster{plural}[/]',
         total=len(clusters))
 
-    def _terminate_or_stop(name: str):
+    def _down_or_stop(name: str):
         success_progress = False
         if idle_minutes_to_autostop is not None:
             try:
-                core.autostop(name, idle_minutes_to_autostop)
+                core.autostop(name, idle_minutes_to_autostop, down)
             except (exceptions.NotSupportedError,
                     exceptions.ClusterNotUpError) as e:
                 message = str(e)
@@ -1911,16 +1972,20 @@ def _terminate_or_stop_clusters(
                 message = (f'{colorama.Fore.GREEN}{operation} '
                            f'cluster {name!r}...done{colorama.Style.RESET_ALL}')
                 if idle_minutes_to_autostop >= 0:
+                    option_str = 'down' if down else 'stop'
+                    passive_str = 'downed' if down else 'stopped'
+                    plural = 's' if idle_minutes_to_autostop != 1 else ''
                     message += (
-                        f'\n  The cluster will be stopped after '
-                        f'{idle_minutes_to_autostop} minutes of idleness.'
-                        '\n  To cancel the autostop, run: '
+                        f'\n  The cluster will be auto{passive_str} after '
+                        f'{idle_minutes_to_autostop} minute{plural} of '
+                        'idleness.'
+                        f'\n  To cancel the auto{option_str}, run: '
                         f'{colorama.Style.BRIGHT}'
                         f'sky autostop {name} --cancel'
                         f'{colorama.Style.RESET_ALL}')
         else:
             try:
-                if terminate:
+                if down:
                     core.down(name, purge=purge)
                 else:
                     core.stop(name, purge=purge)
@@ -1935,7 +2000,7 @@ def _terminate_or_stop_clusters(
                 message = (
                     f'{colorama.Fore.GREEN}{operation} cluster {name}...done.'
                     f'{colorama.Style.RESET_ALL}')
-                if not terminate:
+                if not down:
                     message += ('\n  To restart the cluster, run: '
                                 f'{colorama.Style.BRIGHT}sky start {name}'
                                 f'{colorama.Style.RESET_ALL}')
@@ -1947,7 +2012,7 @@ def _terminate_or_stop_clusters(
         progress.start()
 
     with progress:
-        subprocess_utils.run_in_parallel(_terminate_or_stop, clusters)
+        subprocess_utils.run_in_parallel(_down_or_stop, clusters)
         progress.live.transient = False
         # Make sure the progress bar not mess up the terminal.
         progress.refresh()
@@ -2398,7 +2463,7 @@ def admin_deploy(clusterspec_yaml: str):
     if not isinstance(ips, list):
         ips = [ips]
     local_cluster_name = yaml_config['cluster']['name']
-    usage_lib.messages.usage.update_cluster_name(local_cluster_name)
+    usage_lib.record_cluster_name_for_current_operation(local_cluster_name)
     usage_lib.messages.usage.update_cluster_resources(
         len(ips), sky.Resources(sky.Local()))
 
@@ -2787,7 +2852,7 @@ def bench():
     help=('Automatically stop the cluster after this many minutes '
           'of idleness after setup/file_mounts. This is equivalent to '
           'running `sky launch -d ...` and then `sky autostop -i <minutes>`. '
-          'If not set, the cluster will not be auto-stopped.'))
+          'If not set, the cluster will not be autostopped.'))
 @click.option('--yes',
               '-y',
               is_flag=True,
@@ -3180,7 +3245,7 @@ def benchmark_down(
     clusters_to_exclude: List[str],
     yes: bool,
 ) -> None:
-    """Terminate all clusters belonging to a benchmark."""
+    """Tear down all clusters belonging to a benchmark."""
     record = benchmark_state.get_benchmark_from_name(benchmark)
     if record is None:
         raise click.BadParameter(f'Benchmark {benchmark} does not exist.')
@@ -3194,10 +3259,10 @@ def benchmark_down(
             continue
         to_stop.append(cluster)
 
-    _terminate_or_stop_clusters(to_stop,
-                                apply_to_all=False,
-                                terminate=True,
-                                no_confirm=yes)
+    _down_or_stop_clusters(to_stop,
+                           apply_to_all=False,
+                           down=True,
+                           no_confirm=yes)
 
 
 @bench.command('delete', cls=_DocumentedCodeCommand)
@@ -3206,7 +3271,7 @@ def benchmark_down(
               '-a',
               default=None,
               is_flag=True,
-              help='Tear down all existing clusters.')
+              help='Delete all benchmark reports from the history.')
 @click.option('--yes',
               '-y',
               is_flag=True,
diff --git a/sky/core.py b/sky/core.py
index 87d2a9846..15d20fbaf 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -15,6 +15,7 @@ from sky.backends import backend_utils
 from sky.backends import onprem_utils
 from sky.skylet import constants
 from sky.skylet import job_lib
+from sky.usage import usage_lib
 from sky.utils import tpu_utils
 from sky.utils import ux_utils
 from sky.utils import subprocess_utils
@@ -25,8 +26,10 @@ logger = sky_logging.init_logger(__name__)
 # = Cluster Management =
 # ======================
 
-
 # pylint: disable=redefined-builtin
+
+
+@usage_lib.entrypoint
 def status(all: bool, refresh: bool) -> List[Dict[str, Any]]:
     """Get the cluster status in dict.
 
@@ -42,6 +45,8 @@ def status(all: bool, refresh: bool) -> List[Dict[str, Any]]:
                 'last_use': (int) timestamp of last use,
                 'status': (sky.ClusterStatus) cluster status,
                 'autostop': (int) idle time before autostop,
+                'to_down': (bool) whether autodown is used instead of
+                    autostop,
                 'metadata': (dict) metadata of the cluster,
             }
         ]
@@ -51,12 +56,17 @@ def status(all: bool, refresh: bool) -> List[Dict[str, Any]]:
     return cluster_records
 
 
-def _start(cluster_name: str,
-           idle_minutes_to_autostop: Optional[int] = None,
-           retry_until_up: bool = False) -> backends.Backend.ResourceHandle:
+def _start(
+        cluster_name: str,
+        idle_minutes_to_autostop: Optional[int] = None,
+        retry_until_up: bool = False,
+        down: bool = False,  # pylint: disable=redefined-outer-name
+) -> backends.Backend.ResourceHandle:
 
     cluster_status, handle = backend_utils.refresh_cluster_status_handle(
         cluster_name)
+    if handle is None:
+        raise ValueError(f'Cluster {cluster_name!r} does not exist.')
     if cluster_status == global_user_state.ClusterStatus.UP:
         print(f'Cluster {cluster_name!r} is already up.')
         return
@@ -69,6 +79,11 @@ def _start(cluster_name: str,
         raise exceptions.NotSupportedError(
             f'Starting cluster {cluster_name!r} with backend {backend.NAME} '
             'is not supported.')
+
+    # NOTE: if spot_status() calls _start() and hits here, that entrypoint
+    # would have a cluster name (the controller) filled in.
+    usage_lib.record_cluster_name_for_current_operation(cluster_name)
+
     with dag.Dag():
         dummy_task = task.Task().set_resources(handle.launched_resources)
         dummy_task.num_nodes = handle.launched_nodes
@@ -79,25 +94,31 @@ def _start(cluster_name: str,
                                cluster_name=cluster_name,
                                retry_until_up=retry_until_up)
     if idle_minutes_to_autostop is not None:
-        backend.set_autostop(handle, idle_minutes_to_autostop)
+        backend.set_autostop(handle, idle_minutes_to_autostop, down=down)
     return handle
 
 
-def start(cluster_name: str,
-          idle_minutes_to_autostop: Optional[int] = None,
-          retry_until_up: bool = False):
+@usage_lib.entrypoint
+def start(
+        cluster_name: str,
+        idle_minutes_to_autostop: Optional[int] = None,
+        retry_until_up: bool = False,
+        down: bool = False,  # pylint: disable=redefined-outer-name
+):
     """Start the cluster.
 
     Please refer to the sky.cli.start for the document.
 
     Raises:
+        ValueError: cluster does not exist.
         sky.exceptions.NotSupportedError: the cluster is not supported.
     """
-    _start(cluster_name, idle_minutes_to_autostop, retry_until_up)
+    _start(cluster_name, idle_minutes_to_autostop, retry_until_up, down)
 
 
+@usage_lib.entrypoint
 def stop(cluster_name: str, purge: bool = False):
-    """Stop the cluster
+    """Stop the cluster.
 
     Please refer to the sky.cli.stop for the document.
 
@@ -132,11 +153,13 @@ def stop(cluster_name: str, purge: bool = False):
             'disks will be lost.\n'
             '  To terminate the cluster instead, run: '
             f'{colorama.Style.BRIGHT}sky down {cluster_name}')
+    usage_lib.record_cluster_name_for_current_operation(cluster_name)
     backend.teardown(handle, terminate=False, purge=purge)
 
 
+@usage_lib.entrypoint
 def down(cluster_name: str, purge: bool = False):
-    """Down the cluster
+    """Down the cluster.
 
     Please refer to the sky.cli.down for the document.
 
@@ -152,11 +175,17 @@ def down(cluster_name: str, purge: bool = False):
     if handle is None:
         raise ValueError(f'Cluster {cluster_name!r} does not exist.')
 
+    usage_lib.record_cluster_name_for_current_operation(cluster_name)
     backend = backend_utils.get_backend_from_handle(handle)
     backend.teardown(handle, terminate=True, purge=purge)
 
 
-def autostop(cluster_name: str, idle_minutes_to_autostop: int):
+@usage_lib.entrypoint
+def autostop(
+        cluster_name: str,
+        idle_minutes_to_autostop: int,
+        down: bool = False,  # pylint: disable=redefined-outer-name
+):
     """Set the autostop time of the cluster.
 
     Please refer to the sky.cli.autostop for the document.
@@ -169,8 +198,12 @@ def autostop(cluster_name: str, idle_minutes_to_autostop: int):
         sky.exceptions.NotSupportedError: the cluster is not supported.
         sky.exceptions.ClusterNotUpError: the cluster is not UP.
     """
-    verb = 'Scheduling' if idle_minutes_to_autostop >= 0 else 'Cancelling'
-    operation = f'{verb} auto-stop on'
+    is_cancel = idle_minutes_to_autostop < 0
+    verb = 'Cancelling' if is_cancel else 'Scheduling'
+    option_str = 'down' if down else 'stop'
+    if is_cancel:
+        option_str = '{stop,down}'
+    operation = f'{verb} auto{option_str} on'
     if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
         raise exceptions.NotSupportedError(
             f'{operation} sky reserved cluster {cluster_name!r} '
@@ -192,7 +225,7 @@ def autostop(cluster_name: str, idle_minutes_to_autostop: int):
             raise exceptions.NotSupportedError(
                 f'{colorama.Fore.YELLOW}{operation} cluster '
                 f'{cluster_name!r}... skipped{colorama.Style.RESET_ALL}'
-                '\n  Auto-stopping is only supported by backend: '
+                f'\n  auto{option_str} is only supported by backend: '
                 f'{backends.CloudVmRayBackend.NAME}')
     if cluster_status != global_user_state.ClusterStatus.UP:
         with ux_utils.print_exception_no_traceback():
@@ -200,9 +233,10 @@ def autostop(cluster_name: str, idle_minutes_to_autostop: int):
                 f'{colorama.Fore.YELLOW}{operation} cluster '
                 f'{cluster_name!r} (status: {cluster_status.value})... skipped'
                 f'{colorama.Style.RESET_ALL}'
-                '\n  Auto-stop can only be set/unset for '
+                f'\n  auto{option_str} can only be set/unset for '
                 f'{global_user_state.ClusterStatus.UP.value} clusters.')
-    backend.set_autostop(handle, idle_minutes_to_autostop)
+    usage_lib.record_cluster_name_for_current_operation(cluster_name)
+    backend.set_autostop(handle, idle_minutes_to_autostop, down)
 
 
 # ==================
@@ -243,6 +277,7 @@ def _check_cluster_available(cluster_name: str,
     return handle
 
 
+@usage_lib.entrypoint
 def queue(cluster_name: str,
           skip_finished: bool = False,
           all_users: bool = False) -> List[dict]:
@@ -291,6 +326,7 @@ def queue(cluster_name: str,
     return jobs
 
 
+@usage_lib.entrypoint
 # pylint: disable=redefined-builtin
 def cancel(cluster_name: str,
            all: bool = False,
@@ -331,6 +367,7 @@ def cancel(cluster_name: str,
     backend.cancel_jobs(handle, job_ids)
 
 
+@usage_lib.entrypoint
 def tail_logs(cluster_name: str, job_id: Optional[str], follow: bool = True):
     """Tail the logs of a job.
 
@@ -352,9 +389,11 @@ def tail_logs(cluster_name: str, job_id: Optional[str], follow: bool = True):
           f'Tailing logs of {job_str} on cluster {cluster_name!r}...'
           f'{colorama.Style.RESET_ALL}')
 
+    usage_lib.record_cluster_name_for_current_operation(cluster_name)
     backend.tail_logs(handle, job_id, follow=follow)
 
 
+@usage_lib.entrypoint
 def download_logs(
         cluster_name: str,
         job_ids: Optional[List[str]],
@@ -374,6 +413,7 @@ def download_logs(
     if job_ids is not None and len(job_ids) == 0:
         return []
 
+    usage_lib.record_cluster_name_for_current_operation(cluster_name)
     print(f'{colorama.Fore.YELLOW}'
           'Syncing down logs to local...'
           f'{colorama.Style.RESET_ALL}')
@@ -381,6 +421,7 @@ def download_logs(
     return local_log_dirs
 
 
+@usage_lib.entrypoint
 def job_status(
         cluster_name: str,
         job_ids: Optional[List[str]],
@@ -407,6 +448,7 @@ def job_status(
           'Getting job status...'
           f'{colorama.Style.RESET_ALL}')
 
+    usage_lib.record_cluster_name_for_current_operation(cluster_name)
     statuses = backend.get_job_status(handle, job_ids, stream_logs=stream_logs)
     return statuses
 
@@ -436,6 +478,7 @@ def _is_spot_controller_up(
     return controller_status, handle
 
 
+@usage_lib.entrypoint
 def spot_status(refresh: bool) -> List[Dict[str, Any]]:
     """Get statuses of managed spot jobs.
 
@@ -503,6 +546,7 @@ def spot_status(refresh: bool) -> List[Dict[str, Any]]:
     return jobs
 
 
+@usage_lib.entrypoint
 # pylint: disable=redefined-builtin
 def spot_cancel(name: Optional[str] = None,
                 job_ids: Optional[Tuple[int]] = None,
@@ -556,6 +600,7 @@ def spot_cancel(name: Optional[str] = None,
                 'Please specify the job ID instead of the job name.')
 
 
+@usage_lib.entrypoint
 def spot_tail_logs(name: Optional[str], job_id: Optional[int], follow: bool):
     """Tail logs of managed spot jobs.
 
@@ -586,6 +631,7 @@ def spot_tail_logs(name: Optional[str], job_id: Optional[int], follow: bool):
 # ======================
 # = Storage Management =
 # ======================
+@usage_lib.entrypoint
 def storage_ls() -> List[Dict[str, Any]]:
     """Get the storages.
 
@@ -606,7 +652,8 @@ def storage_ls() -> List[Dict[str, Any]]:
     return storages
 
 
-def storage_delete(name: Optional[str] = None):
+@usage_lib.entrypoint
+def storage_delete(name: str):
     """Delete a storage.
 
     Raises:
diff --git a/sky/data/data_transfer.py b/sky/data/data_transfer.py
index a40c78fab..4c934b96f 100644
--- a/sky/data/data_transfer.py
+++ b/sky/data/data_transfer.py
@@ -15,24 +15,29 @@ TODO:
 - All combinations of Azure Transfer
 - GCS -> S3
 """
-from datetime import datetime
 import json
 import subprocess
-from typing import Any
+import time
+
+import colorama
 
 from sky import clouds
 from sky import sky_logging
 from sky.adaptors import aws, gcp
+from sky.backends import backend_utils
+from sky.utils import ux_utils
 
 logger = sky_logging.init_logger(__name__)
 
-S3Store = Any
-GcsStore = Any
+MAX_POLLS = 120000
+POLL_INTERVAL = 1
 
 
 def s3_to_gcs(s3_bucket_name: str, gs_bucket_name: str) -> None:
     """Creates a one-time transfer from Amazon S3 to Google Cloud Storage.
+
     Can be viewed from: https://console.cloud.google.com/transfer/cloud
+    it will block until the transfer is complete.
 
     Args:
       s3_bucket_name: str; Name of the Amazon S3 Bucket
@@ -56,24 +61,11 @@ def s3_to_gcs(s3_bucket_name: str, gs_bucket_name: str) -> None:
     _add_bucket_iam_member(gs_bucket_name, 'roles/storage.admin',
                            'serviceAccount:' + storage_account['accountEmail'])
 
-    starttime = datetime.utcnow()
     transfer_job = {
         'description': f'Transferring data from S3 Bucket \
         {s3_bucket_name} to GCS Bucket {gs_bucket_name}',
         'status': 'ENABLED',
         'projectId': project_id,
-        'schedule': {
-            'scheduleStartDate': {
-                'day': starttime.day,
-                'month': starttime.month,
-                'year': starttime.year,
-            },
-            'scheduleEndDate': {
-                'day': starttime.day,
-                'month': starttime.month,
-                'year': starttime.year,
-            },
-        },
         'transferSpec': {
             'awsS3DataSource': {
                 'bucketName': s3_bucket_name,
@@ -88,8 +80,40 @@ def s3_to_gcs(s3_bucket_name: str, gs_bucket_name: str) -> None:
         }
     }
 
-    result = storagetransfer.transferJobs().create(body=transfer_job).execute()
-    logger.info(f'AWS -> GCS Transfer Job: {json.dumps(result, indent=4)}')
+    response = storagetransfer.transferJobs().create(
+        body=transfer_job).execute()
+    operation = storagetransfer.transferJobs().run(jobName=response['name'],
+                                                   body={
+                                                       'projectId': project_id
+                                                   }).execute()
+
+    logger.info(f'{colorama.Fore.GREEN}Transfer job scheduled: '
+                f'{colorama.Style.RESET_ALL}'
+                f's3://{s3_bucket_name} -> gs://{gs_bucket_name} ')
+    logger.debug(json.dumps(operation, indent=4))
+    logger.info('Waiting for the transfer to finish')
+    start = time.time()
+    with backend_utils.safe_console_status('Transferring'):
+        for _ in range(MAX_POLLS):
+            result = (storagetransfer.transferOperations().get(
+                name=operation['name']).execute())
+            if 'error' in result:
+                with ux_utils.print_exception_no_traceback():
+                    raise RuntimeError(result['error'])
+
+            if 'done' in result and result['done']:
+                break
+            time.sleep(POLL_INTERVAL)
+        else:
+            # If we get here, we timed out.
+            logger.info(
+                f'Transfer timed out after {(time.time() - start) / 3600:.2f} '
+                'hours. Please check the status of the transfer job in the GCP '
+                'Storage Transfer Service console at '
+                'https://cloud.google.com/storage-transfer-service')
+            return
+    logger.info(
+        f'Transfer finished in {(time.time() - start) / 60:.2f} minutes.')
 
 
 def gcs_to_s3(gs_bucket_name: str, s3_bucket_name: str) -> None:
@@ -114,4 +138,4 @@ def _add_bucket_iam_member(bucket_name: str, role: str, member: str) -> None:
 
     bucket.set_iam_policy(policy)
 
-    logger.info(f'Added {member} with role {role} to {bucket_name}.')
+    logger.debug(f'Added {member} with role {role} to {bucket_name}.')
diff --git a/sky/data/storage.py b/sky/data/storage.py
index c7cf64aaf..5a4009b2a 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -40,10 +40,32 @@ _BUCKET_FAIL_TO_CONNECT_MESSAGE = (
 
 
 class StoreType(enum.Enum):
+    """Enum for the different types of stores."""
     S3 = 'S3'
     GCS = 'GCS'
     AZURE = 'AZURE'
 
+    @classmethod
+    def from_cloud(cls, cloud: clouds.Cloud) -> 'StoreType':
+        if isinstance(cloud, clouds.AWS):
+            return StoreType.S3
+        elif isinstance(cloud, clouds.GCP):
+            return StoreType.GCS
+        elif isinstance(cloud, clouds.Azure):
+            return StoreType.AZURE
+
+        raise ValueError(f'Unsupported cloud for StoreType: {cloud}')
+
+    @classmethod
+    def from_store(cls, store: 'AbstractStore') -> 'StoreType':
+        if isinstance(store, S3Store):
+            return StoreType.S3
+        elif isinstance(store, GcsStore):
+            return StoreType.GCS
+        else:
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(f'Unknown store type: {store}')
+
 
 class StorageMode(enum.Enum):
     MOUNT = 'MOUNT'
@@ -76,16 +98,6 @@ def get_store_prefix(storetype: StoreType) -> str:
             raise ValueError(f'Unknown store type: {storetype}')
 
 
-def _get_storetype_from_store(store: 'Storage') -> StoreType:
-    if isinstance(store, S3Store):
-        return StoreType.S3
-    elif isinstance(store, GcsStore):
-        return StoreType.GCS
-    else:
-        with ux_utils.print_exception_no_traceback():
-            raise ValueError(f'Unknown store type: {store}')
-
-
 class AbstractStore:
     """AbstractStore abstracts away the different storage types exposed by
     different clouds.
@@ -312,11 +324,11 @@ class Storage(object):
                     f'\n\tstores={self.sky_stores})')
 
         def add_store(self, store: AbstractStore) -> None:
-            storetype = _get_storetype_from_store(store)
+            storetype = StoreType.from_store(store)
             self.sky_stores[storetype] = store.get_metadata()
 
         def remove_store(self, store: AbstractStore) -> None:
-            storetype = _get_storetype_from_store(store)
+            storetype = StoreType.from_store(store)
             if storetype in self.sky_stores:
                 del self.sky_stores[storetype]
 
@@ -459,15 +471,26 @@ class Storage(object):
                         '(try "/mydir: /mydir" or "/myfile: /myfile"). '
                         f'Found source={source}')
             # Local path, check if it exists
-            source = os.path.abspath(os.path.expanduser(source))
+            full_src = os.path.abspath(os.path.expanduser(source))
             # Only check if local source exists if it is synced to the bucket
-            if not os.path.exists(source) and sync_on_reconstruction:
+            if not os.path.exists(full_src) and sync_on_reconstruction:
                 with ux_utils.print_exception_no_traceback():
                     raise exceptions.StorageSourceError(
                         'Local source path does not'
                         f' exist: {source}')
+            # Check if source is a file - throw error if it is
+            if os.path.isfile(full_src):
+                with ux_utils.print_exception_no_traceback():
+                    raise exceptions.StorageSourceError(
+                        'Storage source path cannot be a file - only '
+                        'directories are supported as a source. '
+                        'To upload a single file, either\n'
+                        '1) Use regular file mounts by writing '
+                        f'<destination_path>: {source} in the file_mounts '
+                        'section of your YAML, or\n2) Place the file in a '
+                        'directory and specify the directory as the source.')
             # Raise warning if user's path is a symlink
-            elif os.path.islink(source):
+            elif os.path.islink(full_src):
                 logger.warning(f'Source path {source} is a symlink. '
                                'Referenced contents are uploaded, matching '
                                'the default behavior for S3 and GCS syncing.')
@@ -582,18 +605,18 @@ class Storage(object):
         except exceptions.StorageBucketCreateError:
             # Creation failed, so this must be sky managed store. Add failure
             # to state.
-            logger.error(f'Sky could not create {store_type} store '
+            logger.error(f'Could not create {store_type} store '
                          f'with name {self.name}.')
             global_user_state.set_storage_status(self.name,
                                                  StorageStatus.INIT_FAILED)
             raise
         except exceptions.StorageBucketGetError:
             # Bucket get failed, so this is not sky managed. Do not update state
-            logger.error(f'Sky could not get {store_type} store '
+            logger.error(f'Could not get {store_type} store '
                          f'with name {self.name}.')
             raise
         except exceptions.StorageInitError:
-            logger.error(f'Sky could not initialize {store_type} store with '
+            logger.error(f'Could not initialize {store_type} store with '
                          f'name {self.name}. General initialization error.')
             raise
 
@@ -607,7 +630,7 @@ class Storage(object):
 
     def _add_store(self, store: AbstractStore, is_reconstructed: bool = False):
         # Adds a store object to the storage
-        store_type = _get_storetype_from_store(store)
+        store_type = StoreType.from_store(store)
         self.stores[store_type] = store
         # If store initialized and is sky managed, add to state
         if store.is_sky_managed:
@@ -669,7 +692,7 @@ class Storage(object):
         try:
             store.upload()
         except exceptions.StorageUploadError:
-            logger.error(f'Sky could not upload {self.source} to store '
+            logger.error(f'Could not upload {self.source} to store '
                          f'name {store.name}.')
             if store.is_sky_managed:
                 global_user_state.set_storage_status(
@@ -829,7 +852,7 @@ class S3Store(AbstractStore):
                         f'{source} s3://{self.name}/')
         with backend_utils.safe_console_status(
                 f'[bold cyan]Syncing '
-                f'[green]{self.source} to s3://{self.name}/'):
+                f'[green]{self.source}[/] to [green]s3://{self.name}/[/]'):
             # TODO(zhwu): Use log_lib.run_with_log() and redirect the output
             # to a log file.
             with subprocess.Popen(sync_command.split(' '),
@@ -846,7 +869,7 @@ class S3Store(AbstractStore):
                         process.kill()
                         with ux_utils.print_exception_no_traceback():
                             raise PermissionError(
-                                'Sky Storage failed to upload files to '
+                                'Failed to upload files to '
                                 'the S3 bucket. The bucket does not have '
                                 'write permissions. It is possible that '
                                 'the bucket is public.')
@@ -1079,7 +1102,7 @@ class GcsStore(AbstractStore):
         sync_command = f'gsutil -m rsync -d -r {source} gs://{self.name}/'
         with backend_utils.safe_console_status(
                 f'[bold cyan]Syncing '
-                f'[green]{self.source} to gs://{self.name}/'):
+                f'[green]{self.source}[/] to [green]gs://{self.name}/[/]'):
             with subprocess.Popen(sync_command.split(' '),
                                   stderr=subprocess.PIPE,
                                   stdout=subprocess.DEVNULL) as process:
@@ -1094,7 +1117,7 @@ class GcsStore(AbstractStore):
                         process.kill()
                         with ux_utils.print_exception_no_traceback():
                             raise PermissionError(
-                                'Sky Storage failed to upload files to '
+                                'Failed to upload files to '
                                 'GCS. The bucket does not have '
                                 'write permissions. It is possible that '
                                 'the bucket is public.')
diff --git a/sky/execution.py b/sky/execution.py
index da60737fe..95db48a98 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -43,7 +43,6 @@ from sky.utils import ux_utils
 logger = sky_logging.init_logger(__name__)
 
 OptimizeTarget = optimizer.OptimizeTarget
-_MAX_SPOT_JOB_LENGTH = 10
 
 # Message thrown when APIs sky.{exec,launch,spot_launch}() received a string
 # instead of a Dag.  CLI (cli.py) is implemented by us so should not trigger
@@ -93,13 +92,13 @@ class Stage(enum.Enum):
     SETUP = enum.auto()
     PRE_EXEC = enum.auto()
     EXEC = enum.auto()
-    TEARDOWN = enum.auto()
+    DOWN = enum.auto()
 
 
 def _execute(
     dag: sky.Dag,
     dryrun: bool = False,
-    teardown: bool = False,
+    down: bool = False,
     stream_logs: bool = True,
     handle: Any = None,
     backend: Optional[backends.Backend] = None,
@@ -120,8 +119,11 @@ def _execute(
       dag: sky.Dag.
       dryrun: bool; if True, only print the provision info (e.g., cluster
         yaml).
-      teardown: bool; whether to teardown the launched resources after
-        execution.
+      down: bool; whether to tear down the launched resources after all jobs
+        finish (successfully or abnormally). If idle_minutes_to_autostop is
+        also set, the cluster will be torn down after the specified idle time.
+        Note that if errors occur during provisioning/data syncing/setting up,
+        the cluster will not be torn down for debugging purposes.
       stream_logs: bool; whether to stream all tasks' outputs to the client.
       handle: Any; if provided, execution will use an existing backend cluster
         handle instead of provisioning a new one.
@@ -137,7 +139,7 @@ def _execute(
       cluster_name: Name of the cluster to create/reuse.  If None,
         auto-generate a name.
       detach_run: bool; whether to detach the process after the job submitted.
-      autostop_idle_minutes: int; if provided, the cluster will be set to
+      idle_minutes_to_autostop: int; if provided, the cluster will be set to
         autostop after this many minutes of idleness.
       no_setup: bool; whether to skip setup commands or not when (re-)launching.
     """
@@ -157,16 +159,36 @@ def _execute(
             cluster_name)
         cluster_exists = existing_handle is not None
 
+    stages = stages if stages is not None else list(Stage)
+
     backend = backend if backend is not None else backends.CloudVmRayBackend()
-    if not isinstance(backend, backends.CloudVmRayBackend
-                     ) and idle_minutes_to_autostop is not None:
+    if isinstance(backend, backends.CloudVmRayBackend):
+        if down and idle_minutes_to_autostop is None:
+            # Use auto{stop,down} to terminate the cluster after the task is
+            # done.
+            idle_minutes_to_autostop = 0
+        if idle_minutes_to_autostop is not None:
+            if idle_minutes_to_autostop == 0:
+                # idle_minutes_to_autostop=0 can cause the following problem:
+                # After we set the autostop in the PRE_EXEC stage with -i 0,
+                # it could be possible that the cluster immediately found
+                # itself have no task running and start the auto{stop,down}
+                # process, before the task is submitted in the EXEC stage.
+                verb = 'torn down' if down else 'stopped'
+                logger.info(f'{colorama.Style.DIM}The cluster will '
+                            f'be {verb} after 1 minutes of idleness '
+                            '(after all jobs finish).'
+                            f'{colorama.Style.RESET_ALL}')
+                idle_minutes_to_autostop = 1
+            stages.remove(Stage.DOWN)
+    elif idle_minutes_to_autostop is not None:
         # TODO(zhwu): Autostop is not supported for non-CloudVmRayBackend.
         with ux_utils.print_exception_no_traceback():
             raise ValueError(
                 f'Backend {backend.NAME} does not support autostop, please try '
                 f'{backends.CloudVmRayBackend.NAME}')
 
-    if not cluster_exists and (stages is None or Stage.OPTIMIZE in stages):
+    if not cluster_exists and Stage.OPTIMIZE in stages:
         if task.best_resources is None:
             # TODO: fix this for the situation where number of requested
             # accelerators is not an integer.
@@ -186,7 +208,7 @@ def _execute(
         task.sync_storage_mounts()
 
     try:
-        if stages is None or Stage.PROVISION in stages:
+        if Stage.PROVISION in stages:
             if handle is None:
                 handle = backend.provision(task,
                                            task.best_resources,
@@ -199,33 +221,35 @@ def _execute(
             logger.info('Dry run finished.')
             return
 
-        if stages is None or Stage.SYNC_WORKDIR in stages:
+        if Stage.SYNC_WORKDIR in stages:
             if task.workdir is not None:
                 backend.sync_workdir(handle, task.workdir)
 
-        if stages is None or Stage.SYNC_FILE_MOUNTS in stages:
+        if Stage.SYNC_FILE_MOUNTS in stages:
             backend.sync_file_mounts(handle, task.file_mounts,
                                      task.storage_mounts)
 
         if no_setup:
             logger.info('Setup commands skipped.')
-        elif stages is None or Stage.SETUP in stages:
+        elif Stage.SETUP in stages:
             backend.setup(handle, task)
 
-        if stages is None or Stage.PRE_EXEC in stages:
+        if Stage.PRE_EXEC in stages:
             if idle_minutes_to_autostop is not None:
-                backend.set_autostop(handle, idle_minutes_to_autostop)
+                backend.set_autostop(handle,
+                                     idle_minutes_to_autostop,
+                                     down=down)
 
-        if stages is None or Stage.EXEC in stages:
+        if Stage.EXEC in stages:
             try:
                 global_user_state.update_last_use(handle.get_cluster_name())
                 backend.execute(handle, task, detach_run)
             finally:
                 # Enables post_execute() to be run after KeyboardInterrupt.
-                backend.post_execute(handle, teardown)
+                backend.post_execute(handle, down)
 
-        if stages is None or Stage.TEARDOWN in stages:
-            if teardown:
+        if Stage.DOWN in stages:
+            if down and idle_minutes_to_autostop is None:
                 backend.teardown_ephemeral_storage(task)
                 backend.teardown(handle, terminate=True)
     finally:
@@ -253,7 +277,7 @@ def launch(
     retry_until_up: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
     dryrun: bool = False,
-    teardown: bool = False,
+    down: bool = False,
     stream_logs: bool = True,
     backend: Optional[backends.Backend] = None,
     optimize_target: OptimizeTarget = OptimizeTarget.COST,
@@ -268,8 +292,27 @@ def launch(
             auto-generate a name.
         retry_until_up: whether to retry launching the cluster until it is
             up.
-        idle_minutes_to_autostop: if provided, the cluster will be auto-stop
-            after this many minutes of idleness.
+        idle_minutes_to_autostop: automatically stop the cluster after this
+            many minute of idleness, i.e., no running or pending jobs in the
+            cluster's job queue. Idleness starts counting after
+            setup/file_mounts are done; the clock gets reset whenever there
+            are running/pending jobs in the job queue. Setting this flag is
+            equivalent to running ``sky.launch(..., detach_run=True, ...)``
+            and then ``sky.autostop(idle_minutes=<minutes>)``. If not set,
+            the cluster will not be autostopped.
+        down: Tear down the cluster after all jobs finish (successfully or
+            abnormally). If --idle-minutes-to-autostop is also set, the
+            cluster will be torn down after the specified idle time.
+            Note that if errors occur during provisioning/data syncing/setting
+            up, the cluster will not be torn down for debugging purposes.
+        dryrun: if True, do not actually launch the cluster.
+        stream_logs: if True, show the logs in the terminal.
+        backend: backend to use.  If None, use the default backend
+            (CloudVMRayBackend).
+        optimize_target: target to optimize for. Choices: OptimizeTarget.COST,
+            OptimizeTarget.TIME.
+        detach_run: If True, run setup first (blocking), then detach from the
+            job's execution.
         no_setup: if true, the cluster will not re-run setup instructions
 
     Examples:
@@ -290,7 +333,7 @@ def launch(
     _execute(
         dag=dag,
         dryrun=dryrun,
-        teardown=teardown,
+        down=down,
         stream_logs=stream_logs,
         handle=None,
         backend=backend,
@@ -308,7 +351,7 @@ def exec(  # pylint: disable=redefined-builtin
     dag: sky.Dag,
     cluster_name: str,
     dryrun: bool = False,
-    teardown: bool = False,
+    down: bool = False,
     stream_logs: bool = True,
     backend: Optional[backends.Backend] = None,
     optimize_target: OptimizeTarget = OptimizeTarget.COST,
@@ -327,7 +370,7 @@ def exec(  # pylint: disable=redefined-builtin
                              'Use `sky status` to check the status.')
     _execute(dag=dag,
              dryrun=dryrun,
-             teardown=teardown,
+             down=down,
              stream_logs=stream_logs,
              handle=handle,
              backend=backend,
@@ -340,6 +383,7 @@ def exec(  # pylint: disable=redefined-builtin
              detach_run=detach_run)
 
 
+@usage_lib.entrypoint
 def spot_launch(
     dag: sky.Dag,
     name: Optional[str] = None,
@@ -385,7 +429,7 @@ def spot_launch(
               f'{colorama.Style.RESET_ALL}')
         return
 
-    task = _translate_local_file_mounts(task)
+    task = _maybe_translate_local_file_mounts_and_sync_up(task)
 
     with tempfile.NamedTemporaryFile(prefix=f'spot-task-{name}-',
                                      mode='w') as f:
@@ -425,7 +469,17 @@ def spot_launch(
         )
 
 
-def _translate_local_file_mounts(task: task_lib.Task) -> task_lib.Task:
+def _maybe_translate_local_file_mounts_and_sync_up(
+        task: task_lib.Task) -> task_lib.Task:
+    """Translates local->VM mounts into Storage->VM, then syncs up any Storage.
+
+    Eagerly syncing up local->Storage ensures Storage->VM would work at task
+    launch time.
+
+    If there are no local source paths to be translated, this function would
+    still sync up any storage mounts with local source paths (which do not
+    undergo translation).
+    """
     # ================================================================
     # Translate the workdir and local file mounts to cloud file mounts.
     # ================================================================
@@ -434,10 +488,17 @@ def _translate_local_file_mounts(task: task_lib.Task) -> task_lib.Task:
     original_file_mounts = task.file_mounts if task.file_mounts else {}
     original_storage_mounts = task.storage_mounts if task.storage_mounts else {}
 
+    copy_mounts = task.get_local_to_remote_file_mounts()
+    if copy_mounts is None:
+        copy_mounts = {}
+
+    has_local_source_paths = (task.workdir is not None) or copy_mounts
+    if has_local_source_paths:
+        logger.info(
+            f'{colorama.Fore.YELLOW}Translating file_mounts with local '
+            f'source paths to SkyPilot Storage...{colorama.Style.RESET_ALL}')
+
     # Step 1: Translate the workdir to SkyPilot storage.
-    logger.info(
-        f'{colorama.Fore.YELLOW}Translating file_mounts with local '
-        f'source paths to SkyPilot Storage...{colorama.Style.RESET_ALL}')
     new_storage_mounts = dict()
     if task.workdir is not None:
         bucket_name = spot.constants.SPOT_WORKDIR_BUCKET_NAME.format(
@@ -467,9 +528,6 @@ def _translate_local_file_mounts(task: task_lib.Task) -> task_lib.Task:
     # TODO(zhwu): Optimize this by:
     # 1. Use the same bucket for all the mounts.
     # 2. When the src is the same, use the same bucket.
-    copy_mounts = task.get_local_to_remote_file_mounts()
-    if copy_mounts is None:
-        copy_mounts = {}
     copy_mounts_with_file_in_src = dict()
     for i, (dst, src) in enumerate(copy_mounts.items()):
         task.file_mounts.pop(dst)
@@ -492,9 +550,6 @@ def _translate_local_file_mounts(task: task_lib.Task) -> task_lib.Task:
 
     # Step 3: Translate local file mounts with file in src to SkyPilot storage.
     # Hard link the files in src to a temporary directory, and upload folder.
-    original_storage_mounts = {}
-    if task.storage_mounts is not None:
-        original_storage_mounts = task.storage_mounts
     local_fm_path = os.path.join(
         tempfile.gettempdir(),
         spot.constants.SPOT_FM_LOCAL_TMP_DIR.format(id=run_id))
@@ -533,8 +588,11 @@ def _translate_local_file_mounts(task: task_lib.Task) -> task_lib.Task:
     # Upload the local source to a bucket. The task will not be executed
     # locally, so we need to upload the files/folders to the bucket manually
     # here before sending the task to the remote spot controller.
-    logger.info(f'{colorama.Fore.YELLOW}Uploading sources to cloud storage.'
-                f'{colorama.Style.RESET_ALL} See sky storage ls')
+    if task.storage_mounts:
+        # There may be existing (non-translated) storage mounts, so log this
+        # whenever task.storage_mounts is non-empty.
+        logger.info(f'{colorama.Fore.YELLOW}Uploading sources to cloud storage.'
+                    f'{colorama.Style.RESET_ALL} See: sky storage ls')
     task.sync_storage_mounts()
 
     # Step 5: Add the file download into the file mounts, such as
diff --git a/sky/global_user_state.py b/sky/global_user_state.py
index 4a03be70b..ae8717420 100644
--- a/sky/global_user_state.py
+++ b/sky/global_user_state.py
@@ -61,6 +61,8 @@ def create_table(cursor, conn):
     db_utils.add_column_to_table(cursor, conn, 'clusters', 'metadata',
                                  'TEXT DEFAULT "{}"')
 
+    db_utils.add_column_to_table(cursor, conn, 'clusters', 'to_down',
+                                 'INTEGER DEFAULT 0')
     conn.commit()
 
 
@@ -114,7 +116,7 @@ def add_or_update_cluster(cluster_name: str,
     status = ClusterStatus.UP if ready else ClusterStatus.INIT
     _DB.cursor.execute(
         'INSERT or REPLACE INTO clusters'
-        '(name, launched_at, handle, last_use, status, autostop) '
+        '(name, launched_at, handle, last_use, status, autostop, to_down) '
         'VALUES ('
         # name
         '?, '
@@ -132,7 +134,11 @@ def add_or_update_cluster(cluster_name: str,
         # Keep the old autostop value if it exists, otherwise set it to
         # default -1.
         'COALESCE('
-        '(SELECT autostop FROM clusters WHERE name=? AND status!=?), -1)'
+        '(SELECT autostop FROM clusters WHERE name=? AND status!=?), -1), '
+        # Keep the old to_down value if it exists, otherwise set it to
+        # default 0.
+        'COALESCE('
+        '(SELECT to_down FROM clusters WHERE name=? AND status!=?), 0)'
         ')',
         (
             # name
@@ -150,6 +156,8 @@ def add_or_update_cluster(cluster_name: str,
             # autostop
             cluster_name,
             ClusterStatus.STOPPED.value,
+            cluster_name,
+            ClusterStatus.STOPPED.value,
         ))
     _DB.conn.commit()
 
@@ -211,11 +219,14 @@ def set_cluster_status(cluster_name: str, status: ClusterStatus) -> None:
         raise ValueError(f'Cluster {cluster_name} not found.')
 
 
-def set_cluster_autostop_value(cluster_name: str, idle_minutes: int) -> None:
-    _DB.cursor.execute('UPDATE clusters SET autostop=(?) WHERE name=(?)', (
-        idle_minutes,
-        cluster_name,
-    ))
+def set_cluster_autostop_value(cluster_name: str, idle_minutes: int,
+                               to_down: bool) -> None:
+    _DB.cursor.execute(
+        'UPDATE clusters SET autostop=(?), to_down=(?) WHERE name=(?)', (
+            idle_minutes,
+            int(to_down),
+            cluster_name,
+        ))
     count = _DB.cursor.rowcount
     _DB.conn.commit()
     assert count <= 1, count
@@ -248,7 +259,8 @@ def get_cluster_from_name(
         cluster_name: Optional[str]) -> Optional[Dict[str, Any]]:
     rows = _DB.cursor.execute('SELECT * FROM clusters WHERE name=(?)',
                               (cluster_name,))
-    for name, launched_at, handle, last_use, status, autostop, metadata in rows:
+    for (name, launched_at, handle, last_use, status, autostop, metadata,
+         to_down) in rows:
         record = {
             'name': name,
             'launched_at': launched_at,
@@ -256,6 +268,7 @@ def get_cluster_from_name(
             'last_use': last_use,
             'status': ClusterStatus[status],
             'autostop': autostop,
+            'to_down': bool(to_down),
             'metadata': json.loads(metadata),
         }
         return record
@@ -265,7 +278,8 @@ def get_clusters() -> List[Dict[str, Any]]:
     rows = _DB.cursor.execute(
         'select * from clusters order by launched_at desc')
     records = []
-    for name, launched_at, handle, last_use, status, autostop, metadata in rows:
+    for (name, launched_at, handle, last_use, status, autostop, metadata,
+         to_down) in rows:
         # TODO: use namedtuple instead of dict
         record = {
             'name': name,
@@ -274,6 +288,7 @@ def get_clusters() -> List[Dict[str, Any]]:
             'last_use': last_use,
             'status': ClusterStatus[status],
             'autostop': autostop,
+            'to_down': bool(to_down),
             'metadata': json.loads(metadata),
         }
         records.append(record)
diff --git a/sky/skylet/autostop_lib.py b/sky/skylet/autostop_lib.py
index 73c74f3f5..1e6c95f70 100644
--- a/sky/skylet/autostop_lib.py
+++ b/sky/skylet/autostop_lib.py
@@ -10,14 +10,23 @@ AUTOSTOP_CONFIG_KEY = 'autostop_config'
 
 
 class AutostopConfig:
+    """Autostop configuration."""
 
-    def __init__(self, autostop_idle_minutes: int, boot_time: int,
-                 backend: Optional[str]):
+    def __init__(self,
+                 autostop_idle_minutes: int,
+                 boot_time: int,
+                 backend: Optional[str],
+                 down: bool = False):
         assert autostop_idle_minutes < 0 or backend is not None, (
             autostop_idle_minutes, backend)
         self.autostop_idle_minutes = autostop_idle_minutes
         self.boot_time = boot_time
         self.backend = backend
+        self.down = down
+
+    def __set_state__(self, state: dict):
+        state.setdefault('down', False)
+        self.__dict__.update(state)
 
 
 def get_autostop_config() -> Optional[AutostopConfig]:
@@ -27,9 +36,9 @@ def get_autostop_config() -> Optional[AutostopConfig]:
     return pickle.loads(config_str)
 
 
-def set_autostop(idle_minutes: int, backend: Optional[str]) -> None:
+def set_autostop(idle_minutes: int, backend: Optional[str], down: bool) -> None:
     boot_time = psutil.boot_time()
-    autostop_config = AutostopConfig(idle_minutes, boot_time, backend)
+    autostop_config = AutostopConfig(idle_minutes, boot_time, backend, down)
     configs.set_config(AUTOSTOP_CONFIG_KEY, pickle.dumps(autostop_config))
 
 
@@ -43,9 +52,10 @@ class AutostopCodeGen:
     _PREFIX = ['from sky.skylet import autostop_lib']
 
     @classmethod
-    def set_autostop(cls, idle_minutes: int, backend: str) -> str:
+    def set_autostop(cls, idle_minutes: int, backend: str, down: bool) -> str:
         code = [
-            f'autostop_lib.set_autostop({idle_minutes}, {backend!r})',
+            f'autostop_lib.set_autostop({idle_minutes}, {backend!r},'
+            f' {down})',
         ]
         return cls._build(code)
 
diff --git a/sky/skylet/events.py b/sky/skylet/events.py
index 7f5fbaba1..4d736fee9 100644
--- a/sky/skylet/events.py
+++ b/sky/skylet/events.py
@@ -80,7 +80,6 @@ class AutostopEvent(SkyletEvent):
     """Skylet event for autostop."""
     EVENT_INTERVAL_SECONDS = 60
 
-    _NUM_WORKER_PATTERN = re.compile(r'((?:min|max))_workers: (\d+)')
     _UPSCALING_PATTERN = re.compile(r'upscaling_speed: (\d+)')
     _CATCH_NODES = re.compile(r'cache_stopped_nodes: (.*)')
 
@@ -119,13 +118,16 @@ class AutostopEvent(SkyletEvent):
     def _stop_cluster(self, autostop_config):
         if (autostop_config.backend ==
                 cloud_vm_ray_backend.CloudVmRayBackend.NAME):
-            self._replace_yaml_for_stopping(self.ray_yaml_path)
+            self._replace_yaml_for_stopping(self.ray_yaml_path,
+                                            autostop_config.down)
             # `ray up` is required to reset the upscaling speed and min/max
             # workers. Otherwise, `ray down --workers-only` will continuously
             # scale down and up.
-            subprocess.run(
-                ['ray', 'up', '-y', '--restart-only', self.ray_yaml_path],
-                check=True)
+            subprocess.run([
+                'ray', 'up', '-y', '--restart-only', '--disable-usage-stats',
+                self.ray_yaml_path
+            ],
+                           check=True)
             # Stop the workers first to avoid orphan workers.
             subprocess.run(
                 ['ray', 'down', '-y', '--workers-only', self.ray_yaml_path],
@@ -135,17 +137,20 @@ class AutostopEvent(SkyletEvent):
         else:
             raise NotImplementedError
 
-    def _replace_yaml_for_stopping(self, yaml_path: str):
+    def _replace_yaml_for_stopping(self, yaml_path: str, down: bool):
         with open(yaml_path, 'r') as f:
             yaml_str = f.read()
-        # Update the number of workers to 0.
-        yaml_str = self._NUM_WORKER_PATTERN.sub(r'\g<1>_workers: 0', yaml_str)
         yaml_str = self._UPSCALING_PATTERN.sub(r'upscaling_speed: 0', yaml_str)
-        yaml_str = self._CATCH_NODES.sub(r'cache_stopped_nodes: true', yaml_str)
+        if down:
+            yaml_str = self._CATCH_NODES.sub(r'cache_stopped_nodes: false',
+                                             yaml_str)
+        else:
+            yaml_str = self._CATCH_NODES.sub(r'cache_stopped_nodes: true',
+                                             yaml_str)
         config = yaml.safe_load(yaml_str)
         # Set the private key with the existed key on the remote instance.
         config['auth']['ssh_private_key'] = '~/ray_bootstrap_key.pem'
         # Empty the file_mounts.
         config['file_mounts'] = dict()
         common_utils.dump_yaml(yaml_path, config)
-        logger.debug('Replaced worker num and upscaling speed to 0.')
+        logger.debug('Replaced upscaling speed to 0.')
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index aef17542d..ddd8ca957 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -248,12 +248,12 @@ def get_latest_job_id() -> Optional[int]:
         return job_id
 
 
-def get_job_time(job_id: int, is_end: bool) -> Optional[int]:
+def get_job_time_payload(job_id: int, is_end: bool) -> Optional[int]:
     field = 'end_at' if is_end else 'start_at'
     rows = _CURSOR.execute(f'SELECT {field} FROM jobs WHERE job_id=(?)',
                            (job_id,))
     for (timestamp,) in rows:
-        return timestamp
+        return common_utils.encode_payload(timestamp)
 
 
 def _get_records_from_rows(rows) -> List[Dict[str, Any]]:
@@ -632,13 +632,13 @@ class JobLibCodeGen:
         return cls._build(code)
 
     @classmethod
-    def get_job_time(cls,
-                     job_id: Optional[int] = None,
-                     is_end: bool = False) -> str:
+    def get_job_time_payload(cls,
+                             job_id: Optional[int] = None,
+                             is_end: bool = False) -> str:
         code = [
             f'job_id = {job_id} if {job_id} is not None '
             'else job_lib.get_latest_job_id()',
-            f'job_time = job_lib.get_job_time(job_id, {is_end})',
+            f'job_time = job_lib.get_job_time_payload(job_id, {is_end})',
             'print(job_time, flush=True)',
         ]
         return cls._build(code)
diff --git a/sky/spot/spot_utils.py b/sky/spot/spot_utils.py
index 991853b38..1784d986a 100644
--- a/sky/spot/spot_utils.py
+++ b/sky/spot/spot_utils.py
@@ -114,7 +114,8 @@ def update_spot_job_status(job_id: Optional[int] = None):
 def get_job_timestamp(backend: 'backends.CloudVmRayBackend', cluster_name: str,
                       get_end_time: bool) -> float:
     """Get the started/ended time of the job."""
-    code = job_lib.JobLibCodeGen.get_job_time(job_id=None, is_end=get_end_time)
+    code = job_lib.JobLibCodeGen.get_job_time_payload(job_id=None,
+                                                      is_end=get_end_time)
     handle = global_user_state.get_handle_from_cluster_name(cluster_name)
     returncode, stdout, stderr = backend.run_on_head(handle,
                                                      code,
@@ -123,6 +124,7 @@ def get_job_timestamp(backend: 'backends.CloudVmRayBackend', cluster_name: str,
     subprocess_utils.handle_returncode(returncode, code,
                                        'Failed to get job time.',
                                        stdout + stderr)
+    stdout = common_utils.decode_payload(stdout)
     return float(stdout)
 
 
diff --git a/sky/task.py b/sky/task.py
index 00874079b..f4207cded 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -10,6 +10,7 @@ import yaml
 import sky
 from sky import check
 from sky import clouds
+from sky import exceptions
 from sky import global_user_state
 from sky.backends import backend_utils
 from sky.data import storage as storage_lib
@@ -260,7 +261,13 @@ class Task:
             mount_path = storage[0]
             assert mount_path, \
                 'Storage mount path cannot be empty.'
-            storage_obj = storage_lib.Storage.from_yaml_config(storage[1])
+            try:
+                storage_obj = storage_lib.Storage.from_yaml_config(storage[1])
+            except exceptions.StorageSourceError as e:
+                # Patch the error message to include the mount path, if included
+                e.args = (e.args[0].replace('<destination_path>',
+                                            mount_path),) + e.args[1:]
+                raise e
             task_storage_mounts[mount_path] = storage_obj
         task.set_storage_mounts(task_storage_mounts)
 
diff --git a/sky/templates/aws-ray.yml.j2 b/sky/templates/aws-ray.yml.j2
index da3fac071..5732d61dc 100644
--- a/sky/templates/aws-ray.yml.j2
+++ b/sky/templates/aws-ray.yml.j2
@@ -26,10 +26,15 @@ available_node_types:
     node_config:
       InstanceType: {{instance_type}}
       ImageId: {{image_id}}  # Deep Learning AMI (Ubuntu 18.04); see aws.py.
+      # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
       BlockDeviceMappings:
         - DeviceName: /dev/sda1
           Ebs:
             VolumeSize: {{disk_size}}
+            # use default Iops for gp3
+            VolumeType: gp3
+            Iops: 3000
+            Throughput: 125 
       {% if use_spot %}
       InstanceMarketOptions:
           MarketType: spot
@@ -49,6 +54,9 @@ available_node_types:
         - DeviceName: /dev/sda1
           Ebs:
             VolumeSize: {{disk_size}}
+            VolumeType: gp3
+            Iops: 3000
+            Throughput: 125 
       {% if use_spot %}
       InstanceMarketOptions:
           MarketType: spot
@@ -121,13 +129,13 @@ head_start_ray_commands:
   # Line "which prlimit ..": increase the limit of the number of open files for the raylet process, as the `ulimit` may not take effect at this point, because it requires
   # all the sessions to be reloaded. This is a workaround.
   - ((ps aux | grep "-m sky.skylet.skylet" | grep -q python3) || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
-    ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}};
-    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id; done;
+    ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 
 {%- if num_nodes > 1 %}
 worker_start_ray_commands:
-  - ray stop; ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}};
-    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id; done;
+  - ray stop; ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 {%- else %}
 worker_start_ray_commands: []
 {%- endif %}
diff --git a/sky/templates/azure-ray.yml.j2 b/sky/templates/azure-ray.yml.j2
index 0e136f058..88f442a4d 100644
--- a/sky/templates/azure-ray.yml.j2
+++ b/sky/templates/azure-ray.yml.j2
@@ -125,13 +125,13 @@ head_start_ray_commands:
   # Start skylet daemon. (Should not place it in the head_setup_commands, otherwise it will run before skypilot is installed.)
   # NOTE: --disable-usage-stats in `ray start` saves 10 seconds of idle wait.
   - ((ps aux | grep "-m sky.skylet.skylet" | grep -q python3) || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
-    ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}};
-    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id; done;
+    ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 
 {%- if num_nodes > 1 %}
 worker_start_ray_commands:
-  - ray stop; ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}};
-    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id; done;
+  - ray stop; ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 {%- else %}
 worker_start_ray_commands: []
 {%- endif %}
diff --git a/sky/templates/gcp-ray.yml.j2 b/sky/templates/gcp-ray.yml.j2
index 1994094a6..892b4986b 100644
--- a/sky/templates/gcp-ray.yml.j2
+++ b/sky/templates/gcp-ray.yml.j2
@@ -164,15 +164,15 @@ head_start_ray_commands:
   # all the sessions to be reloaded. This is a workaround.
   - ((ps aux | grep "-m sky.skylet.skylet" | grep -q python3) || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
     export SKY_NUM_GPUS=0 && which nvidia-smi > /dev/null && SKY_NUM_GPUS=$(nvidia-smi --query-gpu=index,name --format=csv,noheader | wc -l);
-    ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} --num-gpus=$SKY_NUM_GPUS;
-    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id; done;
+    ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} --num-gpus=$SKY_NUM_GPUS || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 
 # Worker commands are needed for TPU VM Pods
 {%- if num_nodes > 1 or tpu_vm %}
 worker_start_ray_commands:
   - SKY_NUM_GPUS=0 && which nvidia-smi > /dev/null && SKY_NUM_GPUS=$(nvidia-smi --query-gpu=index,name --format=csv,noheader | wc -l);
-    ray stop; ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}} --num-gpus=$SKY_NUM_GPUS;
-    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id; done;
+    ray stop; ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}} --num-gpus=$SKY_NUM_GPUS || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 {%- else %}
 worker_start_ray_commands: []
 {%- endif %}
diff --git a/sky/usage/usage_lib.py b/sky/usage/usage_lib.py
index 493226a17..403e6c984 100644
--- a/sky/usage/usage_lib.py
+++ b/sky/usage/usage_lib.py
@@ -430,3 +430,24 @@ def entrypoint(name_or_fn: str, fallback: bool = False):
     return common_utils.make_decorator(entrypoint_context,
                                        name_or_fn,
                                        fallback=fallback)
+
+
+# Convenience methods below.
+
+
+def record_cluster_name_for_current_operation(
+        cluster_name: Union[List[str], str]) -> None:
+    """Records cluster name(s) for the current operation.
+
+    Usage:
+
+       def op():  # CLI or programmatic API
+
+           ...validate errors...
+
+           usage_lib.record_cluster_name_for_current_operation(
+              <actual clusters being operated on>)
+
+           do_actual_op()
+    """
+    messages.usage.update_cluster_name(cluster_name)
diff --git a/sky/utils/cli_utils/status_utils.py b/sky/utils/cli_utils/status_utils.py
index 94528e37c..6931ee875 100644
--- a/sky/utils/cli_utils/status_utils.py
+++ b/sky/utils/cli_utils/status_utils.py
@@ -34,7 +34,7 @@ class StatusColumn:
 
 def show_status_table(cluster_records: List[Dict[str, Any]], show_all: bool):
     """Compute cluster table values and display."""
-    # TODO(zhwu): Update the information for auto-stop clusters.
+    # TODO(zhwu): Update the information for autostop clusters.
 
     status_columns = [
         StatusColumn('NAME', _get_name),
@@ -72,8 +72,8 @@ def show_status_table(cluster_records: List[Dict[str, Any]], show_all: bool):
         if pending_autostop:
             click.echo(
                 '\n'
-                f'You have {pending_autostop} clusters with autostop scheduled.'
-                ' Refresh statuses with: `sky status --refresh`.')
+                f'You have {pending_autostop} clusters with auto{{stop,down}} '
+                'scheduled. Refresh statuses with: `sky status --refresh`.')
     else:
         click.echo('No existing clusters.')
 
@@ -208,10 +208,17 @@ def _get_zone(cluster_status):
 
 
 def _get_autostop(cluster_status):
-    autostop_str = '-'
+    autostop_str = ''
+    separtion = ''
     if cluster_status['autostop'] >= 0:
         # TODO(zhwu): check the status of the autostop cluster.
-        autostop_str = str(cluster_status['autostop']) + ' min'
+        autostop_str = str(cluster_status['autostop']) + 'm'
+        separtion = ' '
+
+    if cluster_status['to_down']:
+        autostop_str += f'{separtion}(down)'
+    if autostop_str == '':
+        autostop_str = '-'
     return autostop_str
 
 
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index cd018570e..71652ccc9 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -304,18 +304,24 @@ class SSHCommandRunner:
             rsync_command.append(
                 RSYNC_EXCLUDE_OPTION.format(str(resolved_source / GIT_EXCLUDE)))
 
+        # rsync doesn't support '~' in a quoted target path. need to expand it.
+        full_source_str = str(resolved_source)
+        if resolved_source.is_dir():
+            full_source_str = os.path.join(full_source_str, '')
+
         ssh_options = ' '.join(
             ssh_options_list(self.ssh_private_key, self.ssh_control_name))
         rsync_command.append(f'-e "ssh {ssh_options}"')
+        # To support spaces in the path, we need to quote source and target.
         if up:
             rsync_command.extend([
-                source,
-                f'{self.ssh_user}@{self.ip}:{target}',
+                f'{full_source_str!r}',
+                f'{self.ssh_user}@{self.ip}:{target!r}',
             ])
         else:
             rsync_command.extend([
-                f'{self.ssh_user}@{self.ip}:{source}',
-                target,
+                f'{self.ssh_user}@{self.ip}:{full_source_str!r}',
+                f'{target!r}',
             ])
         command = ' '.join(rsync_command)
 
diff --git a/sky/utils/common_utils.py b/sky/utils/common_utils.py
index 6e5f77986..93e3b23e2 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -12,7 +12,7 @@ import socket
 import sys
 import time
 import uuid
-from typing import Dict, List, Union
+from typing import Any
 import yaml
 
 from sky import sky_logging
@@ -223,11 +223,17 @@ def retry(method, max_retries=3, initial_backoff=1):
     return method_with_retries
 
 
-def encode_payload(payload: Union[List, Dict]) -> str:
+def encode_payload(payload: Any) -> str:
     """Encode a payload to make it more robust for parsing.
 
+    The make the message transfer more robust to any additional
+    strings added to the message during transfering.
+
+    An example message that is polluted by the system warning:
+    "LC_ALL: cannot change locale (en_US.UTF-8)\n<sky-payload>hello, world</sky-payload>" # pylint: disable=line-too-long
+
     Args:
-        payload: A dict or list to be encoded.
+        payload: A str, dict or list to be encoded.
 
     Returns:
         A string that is encoded from the payload.
@@ -237,15 +243,18 @@ def encode_payload(payload: Union[List, Dict]) -> str:
     return payload_str
 
 
-def decode_payload(payload_str: str) -> Union[List, Dict]:
+def decode_payload(payload_str: str) -> Any:
     """Decode a payload string.
 
     Args:
         payload_str: A string that is encoded from a payload.
 
     Returns:
-        A dict or list that is decoded from the payload string.
+        A str, dict or list that is decoded from the payload string.
     """
-    payload_str = _PAYLOAD_PATTERN.match(payload_str).group(1)
+    matched = _PAYLOAD_PATTERN.findall(payload_str)
+    if not matched:
+        raise ValueError(f'Invalid payload string: \n{payload_str}')
+    payload_str = matched[0]
     payload = json.loads(payload_str)
     return payload


Test patch:
diff --git a/tests/backward_comaptibility_tests.sh b/tests/backward_comaptibility_tests.sh
index d18b2bc05..fcdd4bd5d 100644
--- a/tests/backward_comaptibility_tests.sh
+++ b/tests/backward_comaptibility_tests.sh
@@ -1,71 +1,78 @@
 #!/bin/bash
-set -ex
+set -ev
 
-CLUSTER_NAME="test-back-compat"
-. $(conda info --base 2> /dev/null)/etc/profile.d/conda.sh
+source ~/.bashrc 
+CLUSTER_NAME="test-back-compat-$USER"
+source $(conda info --base 2> /dev/null)/etc/profile.d/conda.sh
 
 git clone https://github.com/skypilot-org/skypilot.git sky-master || true
 
 
 # Create environment for compatibility tests
-conda create -n sky-back-compat-master -y --clone sky-dev
-conda create -n sky-back-compat-current -y --clone sky-dev
-
-conda activate sky-back-compat-master
-rm -r  ~/.sky/sky_wheels || true
-pip uninstall skypilot -y
-
+mamba > /dev/null 2>&1 || conda install -n base -c conda-forge mamba -y
+mamba init
+source ~/.bashrc
+mamba env list | grep sky-back-compat-master || mamba create -n sky-back-compat-master -y python=3.9
+
+mamba activate sky-back-compat-master
+mamba install -c conda-forge google-cloud-sdk -y
+rm -r  ~/.sky/wheels || true
 cd sky-master
 git pull origin master
 pip install -e ".[all]"
 cd -
 
-conda activate sky-back-compat-current
-rm -r  ~/.sky/sky_wheels || true
+mamba env list | grep sky-back-compat-current || mamba create -n sky-back-compat-current -y --clone sky-back-compat-master
+mamba activate sky-back-compat-current
+rm -r  ~/.sky/wheels || true
 pip uninstall -y skypilot
 pip install -e ".[all]"
 
 
 # exec + launch
-conda activate sky-back-compat-master
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-master
+rm -r  ~/.sky/wheels || true
 which sky
-sky launch --cloud gcp -c ${CLUSTER_NAME} examples/minimal.yaml
+sky launch --cloud gcp -y -c ${CLUSTER_NAME} examples/minimal.yaml
 
-conda activate sky-back-compat-current
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-current
+rm -r  ~/.sky/wheels || true
 sky exec --cloud gcp ${CLUSTER_NAME} examples/minimal.yaml
-sky launch --cloud gcp -c ${CLUSTER_NAME} examples/minimal.yaml
-sky queue
+s=$(sky launch --cloud gcp -d -c ${CLUSTER_NAME} examples/minimal.yaml)
+echo $s
+echo $s | grep "Job ID: 3" || exit 1
+sky queue ${CLUSTER_NAME}
 
 # sky stop + sky start + sky exec
-conda activate sky-back-compat-master
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-master
+rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-2 examples/minimal.yaml
-conda activate sky-back-compat-current
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-current
+rm -r  ~/.sky/wheels || true
 sky stop -y ${CLUSTER_NAME}-2
 sky start -y ${CLUSTER_NAME}-2
-sky exec --cloud gcp ${CLUSTER_NAME}-2 examples/minimal.yaml
+s=$(sky exec --cloud gcp -d ${CLUSTER_NAME}-2 examples/minimal.yaml)
+echo $s
+echo $s | grep "Job ID: 2" || exit 1
 
 # `sky autostop` + `sky status -r`
-conda activate sky-back-compat-master
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-master
+rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-3 examples/minimal.yaml
-conda activate sky-back-compat-current
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-current
+rm -r  ~/.sky/wheels || true
 sky autostop -y -i0 ${CLUSTER_NAME}-3
 sleep 100
-sky status -r | grep ${CLUSTER_NAME}-3 | grep STOPPED
+sky status -r | grep ${CLUSTER_NAME}-3 | grep STOPPED || exit 1
 
 
 # (1 node) sky launch + sky exec + sky queue + sky logs
-conda activate sky-back-compat-master
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-master
+rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-4 examples/minimal.yaml
 sky stop -y ${CLUSTER_NAME}-4
-conda activate sky-back-compat-current
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-current
+rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-4 examples/minimal.yaml
 sky queue ${CLUSTER_NAME}-4
 sky logs ${CLUSTER_NAME}-4 1 --status
@@ -74,12 +81,12 @@ sky logs ${CLUSTER_NAME}-4 1
 sky logs ${CLUSTER_NAME}-4 2
 
 # (1 node) sky start + sky exec + sky queue + sky logs
-conda activate sky-back-compat-master
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-master
+rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-5 examples/minimal.yaml
 sky stop -y ${CLUSTER_NAME}-5
-conda activate sky-back-compat-current
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-current
+rm -r  ~/.sky/wheels || true
 sky start -y ${CLUSTER_NAME}-5
 sky queue ${CLUSTER_NAME}-5
 sky logs ${CLUSTER_NAME}-5 1 --status
@@ -90,12 +97,12 @@ sky logs ${CLUSTER_NAME}-5 2 --status
 sky logs ${CLUSTER_NAME}-5 2
 
 # (2 nodes) sky launch + sky exec + sky queue + sky logs
-conda activate sky-back-compat-master
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-master
+rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-6 examples/multi_hostname.yaml
 sky stop -y ${CLUSTER_NAME}-6
-conda activate sky-back-compat-current
-rm -r  ~/.sky/sky_wheels || true
+mamba activate sky-back-compat-current
+rm -r  ~/.sky/wheels || true
 sky start -y ${CLUSTER_NAME}-6
 sky queue ${CLUSTER_NAME}-6
 sky logs ${CLUSTER_NAME}-6 1 --status
@@ -104,3 +111,5 @@ sky exec --cloud gcp ${CLUSTER_NAME}-6 examples/multi_hostname.yaml
 sky queue ${CLUSTER_NAME}-6
 sky logs ${CLUSTER_NAME}-6 2 --status
 sky logs ${CLUSTER_NAME}-6 2
+
+sky down ${CLUSTER_NAME}* -y
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index ba7f1e376..cd8e45d51 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -243,6 +243,7 @@ def test_file_mounts():
         [
             'touch ~/tmpfile',
             'mkdir -p ~/tmp-workdir',
+            'touch ~/tmp-workdir/tmp\ file',
             'touch ~/tmp-workdir/foo',
             'ln -f -s ~/tmp-workdir/ ~/tmp-workdir/circle-link',
             f'sky launch -y -c {name} examples/using_file_mounts.yaml',
@@ -511,7 +512,7 @@ def test_autostop():
             f'sky launch -y -d -c {name} --num-nodes 2 examples/minimal.yaml',
             f'sky autostop -y {name} -i 1',
             # Ensure autostop is set.
-            f'sky status | grep {name} | grep "1 min"',
+            f'sky status | grep {name} | grep "1m"',
             'sleep 180',
             # Ensure the cluster is STOPPED.
             f'sky status --refresh | grep {name} | grep STOPPED',
@@ -526,6 +527,38 @@ def test_autostop():
     run_one_test(test)
 
 
+# ---------- Testing Autodowning ----------
+def test_autodown():
+    name = _get_cluster_name()
+    test = Test(
+        'autodown',
+        [
+            f'sky launch -y -d -c {name} --num-nodes 2 --cloud gcp examples/minimal.yaml',
+            f'sky autostop -y {name} --down -i 1',
+            # Ensure autostop is set.
+            f'sky status | grep {name} | grep "1m (down)"',
+            'sleep 240',
+            # Ensure the cluster is terminated.
+            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && {{ echo $s | grep {name} | grep "Autodowned cluster\|terminated on the cloud"; }} || {{ echo $s | grep {name} && exit 1 || exit 0; }}',
+            f'sky launch -y -d -c {name} --cloud aws --num-nodes 2 --down examples/minimal.yaml',
+            f'sky status | grep {name} | grep UP',  # Ensure the cluster is UP.
+            f'sky exec {name} --cloud aws examples/minimal.yaml',
+            f'sky status | grep {name} | grep "1m (down)"',
+            'sleep 240',
+            # Ensure the cluster is terminated.
+            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && {{ echo $s | grep {name} | grep "Autodowned cluster\|terminated on the cloud"; }} || {{ echo $s | grep {name} && exit 1 || exit 0; }}',
+            f'sky launch -y -d -c {name} --cloud aws --num-nodes 2 --down examples/minimal.yaml',
+            f'sky autostop -y {name} --cancel',
+            'sleep 240',
+            # Ensure the cluster is still UP.
+            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && echo $s | grep {name} | grep UP',
+        ],
+        f'sky down -y {name}',
+        timeout=20 * 60,
+    )
+    run_one_test(test)
+
+
 def _get_cancel_task_with_cloud(name, cloud, timeout=15 * 60):
     test = Test(
         f'{cloud}-cancel-task',
@@ -730,7 +763,7 @@ def test_inline_spot_env():
         [
             f'sky spot launch -n {name} -y --env TEST_ENV="hello world" -- "([[ ! -z \\"\$TEST_ENV\\" ]] && [[ ! -z \\"\$SKY_NODE_IPS\\" ]] && [[ ! -z \\"\$SKY_NODE_RANK\\" ]]) || exit 1"',
             'sleep 10',
-            f'sky spot status | grep {name} | grep SUCCEEDED',
+            f's=$(sky spot status) && printf "$s" && echo "$s"  | grep {name} | grep SUCCEEDED',
         ],
         f'sky spot cancel -y -n {name}',
     )
diff --git a/tests/test_spot.py b/tests/test_spot.py
index d1bc6a7c5..549411a4d 100644
--- a/tests/test_spot.py
+++ b/tests/test_spot.py
@@ -141,7 +141,7 @@ class TestReservedClustersOperations:
         cli_runner = cli_testing.CliRunner()
         result = cli_runner.invoke(cli.autostop, [spot.SPOT_CONTROLLER_NAME])
         assert result.exit_code == click.UsageError.exit_code
-        assert ('Scheduling auto-stop on reserved cluster(s) '
+        assert ('Scheduling autostop on reserved cluster(s) '
                 f'\'{spot.SPOT_CONTROLLER_NAME}\' is not supported'
                 in result.output)
 
diff --git a/tests/test_storage.py b/tests/test_storage.py
index b38663594..b579e6d54 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -1,6 +1,7 @@
 import time
 
 import pytest
+import tempfile
 
 from sky import exceptions
 from sky.data import storage as storage_lib
@@ -20,6 +21,12 @@ class TestStorageSpecLocalSource:
             storage_lib.Storage(name='test', source='/bin/')
         assert 'Storage source paths cannot end with a slash' in str(e)
 
+    def test_source_single_file(self):
+        with pytest.raises(exceptions.StorageSourceError) as e:
+            with tempfile.NamedTemporaryFile() as f:
+                storage_lib.Storage(name='test', source=f.name)
+        assert 'Storage source path cannot be a file' in str(e)
+
 
 class TestStorageSpecValidation:
     """Storage specification validation tests"""