Repository: skypilot-org/skypilot
Pull request: [Spot] Auto file_mounts translation for spot
Fixes #1073.

This PR makes sure that a normal task YAML can be run with `sky spot launch` without any modification.

~~It is currently blocked by #1069.~~

Tested:
- [x] `sky spot launch -n test-storage exampels/managed_spot_with_storage.yaml`

TODO:
- [x] Add test

Problem statement:
[Spot] Automatically translate the local file mounts and workdir to sky Storage
The friction of changing an original available sky yaml with `workdir` and local file mounts set to the managed spot yaml has been observed multiple times. We may want to have SkyPilot translate those for the users automatically, i.e. translate those into SkyPilot Storage with `persistent: false` and `mode: COPY`

Hints from issue discussion:

Commit message: Add a TODO

Patch to review:
diff --git a/examples/managed_spot_with_storage.yaml b/examples/managed_spot_with_storage.yaml
index d82aff7db4b..c9fe823b19d 100644
--- a/examples/managed_spot_with_storage.yaml
+++ b/examples/managed_spot_with_storage.yaml
@@ -11,8 +11,10 @@ resources:
   use_spot: true
   spot_recovery: failover
 
+workdir: ./examples
+
 file_mounts:
-  ~/sky_workdir:
+  ~/bucket_workdir:
     # Change this to the your own globally unique bucket name.
     name: sky-workdir-zhwu
     source: ./examples
@@ -20,8 +22,33 @@ file_mounts:
     mode: COPY
   /imagenet-image:
     source: s3://sky-imagenet-data
+  
+  # File mounts for folder
+  /tmp/workdir: ~/tmp-workdir
+
+
+  # File mounts for file
+  ~/tmpfile: ~/tmpfile
+  ~/a/b/c/tmpfile: ~/tmpfile
+
+  # 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'.
+  /data/checkpoints/best.pt: ~/tmpfile
+
+  ~/.ssh/id_rsa.pub: ~/.ssh/id_rsa.pub
+
+
 
 run: |
   set -ex
   ls ~/sky_workdir/managed_spot_with_storage.yaml
+  ls ~/bucket_workdir/managed_spot_with_storage.yaml
   ls -l /imagenet-image/datasets
+  
+
+  mkdir -p /data/logs
+  touch /data/logs/test.log
+  touch /data/checkpoints/last.pt
+
+  echo hi >> /tmp/workdir/new_file
+  ls -al /tmp/workdir
diff --git a/sky/authentication.py b/sky/authentication.py
index 10a0877bfbd..8378801f58e 100644
--- a/sky/authentication.py
+++ b/sky/authentication.py
@@ -12,6 +12,7 @@
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.backends import default_backend
 
+from sky import clouds
 from sky import sky_logging
 from sky.adaptors import aws, gcp
 from sky.utils import common_utils
@@ -30,11 +31,6 @@
 MAX_TRIALS = 64
 PRIVATE_SSH_KEY_PATH = '~/.ssh/sky-key'
 
-GCP_CONFIGURE_PATH = '~/.config/gcloud/configurations/config_default'
-# Do not place the backup under the gcloud config directory, as ray
-# autoscaler can overwrite that directory on the remote nodes.
-GCP_CONFIGURE_SKY_BACKUP_PATH = '~/.sky/.sky_gcp_config_default'
-
 
 def generate_rsa_key_pair():
     key = rsa.generate_private_key(backend=default_backend(),
@@ -212,27 +208,13 @@ def setup_gcp_authentication(config):
             'additional authentication steps.')
         # Read the account information from the credential file, since the user
         # should be set according the account, when the oslogin is enabled.
-        config_path = os.path.expanduser(GCP_CONFIGURE_PATH)
+        config_path = os.path.expanduser(clouds.gcp.GCP_CONFIG_PATH)
         sky_backup_config_path = os.path.expanduser(
-            GCP_CONFIGURE_SKY_BACKUP_PATH)
-        if not os.path.exists(sky_backup_config_path):
-            if not os.path.exists(config_path):
-                with ux_utils.print_exception_no_traceback():
-                    raise RuntimeError(
-                        'GCP authentication failed, as the oslogin is enabled '
-                        f'but the file {config_path} is not found.')
-
-            # Create a backup of the config_default file in the same folder (the
-            # folder will be uploaded by `sky launch`), as the original file can
-            # be modified on the remote cluster by ray causing failure of
-            # launching GCP cluster on a remote cluster.
-            subprocess.run(f'cp {config_path} {sky_backup_config_path}',
-                           shell=True,
-                           check=True)
-        new_file_mounts = config.get('file_mounts', {})
-        new_file_mounts[
-            GCP_CONFIGURE_SKY_BACKUP_PATH] = GCP_CONFIGURE_SKY_BACKUP_PATH
-        config['file_mounts'] = new_file_mounts
+            clouds.gcp.GCP_CONFIG_SKY_BACKUP_PATH)
+        assert os.path.exists(sky_backup_config_path), (
+            'GCP credential backup file '
+            f'{sky_backup_config_path!r} does not exist.')
+
         with open(sky_backup_config_path, 'r') as infile:
             for line in infile:
                 if line.startswith('account'):
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 488c2b15bd8..f34c807c852 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -37,12 +37,12 @@
 from sky import backends
 from sky import check as sky_check
 from sky import clouds
-from sky import constants
 from sky import exceptions
 from sky import global_user_state
 from sky import sky_logging
 from sky import spot as spot_lib
 from sky.backends import onprem_utils
+from sky.skylet import constants
 from sky.skylet import log_lib
 from sky.utils import common_utils
 from sky.utils import command_runner
@@ -60,7 +60,6 @@
 console = rich_console.Console()
 
 # NOTE: keep in sync with the cluster template 'file_mounts'.
-SKY_REMOTE_WORKDIR = log_lib.SKY_REMOTE_WORKDIR
 SKY_REMOTE_APP_DIR = '~/.sky/sky_app'
 SKY_RAY_YAML_REMOTE_PATH = '~/.sky/sky_ray.yml'
 IP_ADDR_REGEX = r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'
@@ -1910,7 +1909,8 @@ def interrupt_handler(signum, frame):
     logger.warning(f'{colorama.Fore.LIGHTBLACK_EX}The job will keep '
                    f'running after Ctrl-C.{colorama.Style.RESET_ALL}')
     kill_children_processes()
-    raise KeyboardInterrupt(exceptions.KEYBOARD_INTERRUPT_CODE)
+    with ux_utils.print_exception_no_traceback():
+        raise KeyboardInterrupt(exceptions.KEYBOARD_INTERRUPT_CODE)
 
 
 # Handle ctrl-z
@@ -1919,7 +1919,8 @@ def stop_handler(signum, frame):
     logger.warning(f'{colorama.Fore.LIGHTBLACK_EX}The job will keep '
                    f'running after Ctrl-Z.{colorama.Style.RESET_ALL}')
     kill_children_processes()
-    raise KeyboardInterrupt(exceptions.SIGTSTP_CODE)
+    with ux_utils.print_exception_no_traceback():
+        raise KeyboardInterrupt(exceptions.SIGTSTP_CODE)
 
 
 def validate_schema(obj, schema, err_msg_prefix=''):
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index f00f99a952d..b726e7b7a59 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -22,7 +22,6 @@
 from sky import backends
 from sky import clouds
 from sky import cloud_stores
-from sky import constants
 from sky import exceptions
 from sky import global_user_state
 from sky import resources as resources_lib
@@ -36,6 +35,7 @@
 from sky.backends import onprem_utils
 from sky.backends import wheel_utils
 from sky.skylet import autostop_lib
+from sky.skylet import constants
 from sky.skylet import job_lib
 from sky.skylet import log_lib
 from sky.usage import usage_lib
@@ -54,7 +54,7 @@
 Path = str
 
 SKY_REMOTE_APP_DIR = backend_utils.SKY_REMOTE_APP_DIR
-SKY_REMOTE_WORKDIR = backend_utils.SKY_REMOTE_WORKDIR
+SKY_REMOTE_WORKDIR = constants.SKY_REMOTE_WORKDIR
 
 logger = sky_logging.init_logger(__name__)
 
@@ -176,10 +176,11 @@ def add_prologue(self,
             import ray
             import ray.util as ray_util
 
+            from sky.skylet import constants
             from sky.skylet import job_lib
             from sky.utils import log_utils
 
-            SKY_REMOTE_WORKDIR = {log_lib.SKY_REMOTE_WORKDIR!r}
+            SKY_REMOTE_WORKDIR = {constants.SKY_REMOTE_WORKDIR!r}
             job_lib.set_status({job_id!r}, job_lib.JobStatus.PENDING)
 
             ray.init(address={ray_address!r}, namespace='__sky__{job_id}__', log_to_driver=True)
diff --git a/sky/benchmark/benchmark_utils.py b/sky/benchmark/benchmark_utils.py
index ff67e15196e..969fe1d1ee4 100644
--- a/sky/benchmark/benchmark_utils.py
+++ b/sky/benchmark/benchmark_utils.py
@@ -21,14 +21,14 @@
 
 import sky
 from sky import backends
-from sky import constants
 from sky import data
 from sky import global_user_state
 from sky import sky_logging
 from sky.backends import backend_utils
 from sky.benchmark import benchmark_state
-from sky.skylet import job_lib
+from sky.skylet import constants
 from sky.skylet import log_lib
+from sky.skylet import job_lib
 from sky.utils import log_utils
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
diff --git a/sky/cli.py b/sky/cli.py
index cf10ba07005..4d58515a600 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -47,7 +47,6 @@
 from sky import backends
 from sky import check as sky_check
 from sky import clouds
-from sky import constants
 from sky import exceptions
 from sky import global_user_state
 from sky import sky_logging
@@ -59,6 +58,7 @@
 from sky.benchmark import benchmark_utils
 from sky.clouds import service_catalog
 from sky.data import storage_utils
+from sky.skylet import constants
 from sky.skylet import job_lib
 from sky.utils import log_utils
 from sky.utils import common_utils
diff --git a/sky/cloud_stores.py b/sky/cloud_stores.py
index dc8e695741e..a5320bbf9a2 100644
--- a/sky/cloud_stores.py
+++ b/sky/cloud_stores.py
@@ -10,6 +10,7 @@
 import subprocess
 import urllib.parse
 
+from sky.clouds import gcp
 from sky.data import data_utils
 from sky.adaptors import aws
 
@@ -93,26 +94,19 @@ class GcsCloudStorage(CloudStorage):
     # We use gsutil as a basic implementation.  One pro is that its -m
     # multi-threaded download is nice, which frees us from implementing
     # parellel workers on our end.
-    _GET_GSUTIL = [
-        # Skip if gsutil already exists.
-        'pushd /tmp &>/dev/null',
-        '(test -f ~/google-cloud-sdk/bin/gsutil || (wget --quiet '
-        'https://dl.google.com/dl/cloudsdk/channels/rapid/downloads/'
-        'google-cloud-sdk-367.0.0-linux-x86_64.tar.gz && '
-        'tar xzf google-cloud-sdk-367.0.0-linux-x86_64.tar.gz && '
-        'mv google-cloud-sdk ~/ && '
-        '~/google-cloud-sdk/install.sh -q ))',
-        'popd &>/dev/null',
-    ]
+    # The gsutil command is part of the Google Cloud SDK, and we reuse
+    # the installation logic here.
+    _GET_GSUTIL = gcp.GCLOUD_INSTALLATION_COMMAND
 
-    _GSUTIL = '~/google-cloud-sdk/bin/gsutil'
+    _GSUTIL = ('GOOGLE_APPLICATION_CREDENTIALS='
+               f'{gcp.DEFAULT_GCP_APPLICATION_CREDENTIAL_PATH} gsutil')
 
     def is_directory(self, url: str) -> bool:
         """Returns whether 'url' is a directory.
         In cloud object stores, a "directory" refers to a regular object whose
         name is a prefix of other objects.
         """
-        commands = list(self._GET_GSUTIL)
+        commands = [self._GET_GSUTIL]
         commands.append(f'{self._GSUTIL} ls -d {url}')
         command = ' && '.join(commands)
         p = subprocess.run(command,
@@ -142,14 +136,14 @@ def make_sync_dir_command(self, source: str, destination: str) -> str:
         """Downloads a directory using gsutil."""
         download_via_gsutil = (
             f'{self._GSUTIL} -m rsync -r {source} {destination}')
-        all_commands = list(self._GET_GSUTIL)
+        all_commands = [self._GET_GSUTIL]
         all_commands.append(download_via_gsutil)
         return ' && '.join(all_commands)
 
     def make_sync_file_command(self, source: str, destination: str) -> str:
         """Downloads a file using gsutil."""
         download_via_gsutil = f'{self._GSUTIL} -m cp {source} {destination}'
-        all_commands = list(self._GET_GSUTIL)
+        all_commands = [self._GET_GSUTIL]
         all_commands.append(download_via_gsutil)
         return ' && '.join(all_commands)
 
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index 46ab4e2eb4f..480650347e0 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -18,6 +18,11 @@
     '~/.config/gcloud/'
     'application_default_credentials.json')
 
+GCP_CONFIG_PATH = '~/.config/gcloud/configurations/config_default'
+# Do not place the backup under the gcloud config directory, as ray
+# autoscaler can overwrite that directory on the remote nodes.
+GCP_CONFIG_SKY_BACKUP_PATH = '~/.sky/.sky_gcp_config_default'
+
 # Minimum set of files under ~/.config/gcloud that grant GCP access.
 _CREDENTIAL_FILES = [
     'credentials.db',
@@ -29,6 +34,23 @@
 
 _IMAGE_ID_PREFIX = ('projects/deeplearning-platform-release/global/images/')
 
+_GCLOUD_INSTALLATION_LOG = '~/.sky/logs/gcloud_installation.log'
+# Need to be run with /bin/bash
+# We factor out the installation logic to keep it align in both spot
+# controller and cloud stores.
+GCLOUD_INSTALLATION_COMMAND = f'pushd /tmp &>/dev/null && \
+    gcloud --help > /dev/null 2>&1 || \
+    {{ mkdir -p {os.path.dirname(_GCLOUD_INSTALLATION_LOG)} && \
+    wget --quiet https://dl.google.com/dl/cloudsdk/channels/rapid/downloads/google-cloud-sdk-382.0.0-linux-x86_64.tar.gz > {_GCLOUD_INSTALLATION_LOG} && \
+    tar xzf google-cloud-sdk-382.0.0-linux-x86_64.tar.gz >> {_GCLOUD_INSTALLATION_LOG} && \
+    rm -rf ~/google-cloud-sdk >> {_GCLOUD_INSTALLATION_LOG}  && \
+    mv google-cloud-sdk ~/ && \
+    ~/google-cloud-sdk/install.sh -q >> {_GCLOUD_INSTALLATION_LOG} 2>&1 && \
+    echo "source ~/google-cloud-sdk/path.bash.inc > /dev/null 2>&1" >> ~/.bashrc && \
+    source ~/google-cloud-sdk/path.bash.inc >> {_GCLOUD_INSTALLATION_LOG} 2>&1; }} && \
+    {{ cp {GCP_CONFIG_SKY_BACKUP_PATH} {GCP_CONFIG_PATH} > /dev/null 2>&1 || true; }} && \
+    popd &>/dev/null'
+
 
 def _run_output(cmd):
     proc = subprocess.run(cmd,
@@ -328,13 +350,30 @@ def check_credentials(self) -> Tuple[bool, Optional[str]]:
         return True, None
 
     def get_credential_file_mounts(self) -> Dict[str, str]:
+        # Create a backup of the config_default file, as the original file can
+        # be modified on the remote cluster by ray causing authentication
+        # problems. The backup file will be updated to the remote cluster
+        # whenever the original file is not empty and will be applied
+        # appropriately on the remote cluster when neccessary.
+        if (os.path.exists(os.path.expanduser(GCP_CONFIG_PATH)) and
+                os.path.getsize(os.path.expanduser(GCP_CONFIG_PATH)) > 0):
+            subprocess.run(f'cp {GCP_CONFIG_PATH} {GCP_CONFIG_SKY_BACKUP_PATH}',
+                           shell=True,
+                           check=True)
+        elif not os.path.exists(os.path.expanduser(GCP_CONFIG_SKY_BACKUP_PATH)):
+            raise RuntimeError(
+                'GCP credential file is empty. Please make sure you '
+                'have run: gcloud init')
+
         # Excluding the symlink to the python executable created by the gcp
         # credential, which causes problem for ray up multiple nodes, tracked
         # in #494, #496, #483.
-        return {
+        credentials = {
             f'~/.config/gcloud/{filename}': f'~/.config/gcloud/{filename}'
             for filename in _CREDENTIAL_FILES
         }
+        credentials[GCP_CONFIG_SKY_BACKUP_PATH] = GCP_CONFIG_SKY_BACKUP_PATH
+        return credentials
 
     def instance_type_exists(self, instance_type):
         return service_catalog.instance_type_exists(instance_type, 'gcp')
diff --git a/sky/core.py b/sky/core.py
index 70faee49cd7..0430363ea31 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -6,7 +6,6 @@
 from sky import dag
 from sky import task
 from sky import backends
-from sky import constants
 from sky import data
 from sky import exceptions
 from sky import global_user_state
@@ -14,6 +13,7 @@
 from sky import spot
 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.utils import tpu_utils
 from sky.utils import ux_utils
@@ -465,6 +465,9 @@ def spot_status(refresh: bool) -> List[Dict[str, Any]]:
         stop_msg = 'To view the latest job table: sky spot status --refresh'
     controller_status, handle = _is_spot_controller_up(stop_msg)
 
+    if controller_status is None:
+        return []
+
     if (refresh and controller_status in [
             global_user_state.ClusterStatus.STOPPED,
             global_user_state.ClusterStatus.INIT
diff --git a/sky/data/storage.py b/sky/data/storage.py
index 1dff096fe53..c7cf64aafc8 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -6,6 +6,7 @@
 from typing import Any, Dict, Optional, Tuple, Union
 import urllib.parse
 
+from sky import clouds
 from sky.adaptors import aws
 from sky.adaptors import gcp
 from sky.backends import backend_utils
@@ -24,6 +25,10 @@
 StorageHandle = Any
 StorageStatus = global_user_state.StorageStatus
 
+# Clouds with object storage implemented in this module. Azure Blob
+# Storage isn't supported yet (even though Azure is).
+STORE_ENABLED_CLOUDS = [clouds.AWS(), clouds.GCP()]
+
 # Max number of objects a GCS bucket can be directly deleted with
 _GCS_RM_MAX_OBJS = 256
 
@@ -45,6 +50,32 @@ class StorageMode(enum.Enum):
     COPY = 'COPY'
 
 
+def get_storetype_from_cloud(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):
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Azure Blob Storage is not supported yet.')
+    else:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(f'Unknown cloud type: {cloud}')
+
+
+def get_store_prefix(storetype: StoreType) -> str:
+    if storetype == StoreType.S3:
+        return 's3://'
+    elif storetype == StoreType.GCS:
+        return 'gs://'
+    elif storetype == StoreType.AZURE:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Azure Blob Storage is not supported yet.')
+    else:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(f'Unknown store type: {storetype}')
+
+
 def _get_storetype_from_store(store: 'Storage') -> StoreType:
     if isinstance(store, S3Store):
         return StoreType.S3
@@ -803,15 +834,14 @@ def sync_local_dir(self) -> None:
             # to a log file.
             with subprocess.Popen(sync_command.split(' '),
                                   stderr=subprocess.PIPE,
-                                  stdout=subprocess.PIPE) as process:
+                                  stdout=subprocess.DEVNULL) as process:
                 stderr = []
                 while True:
                     line = process.stderr.readline()
                     if not line:
                         break
                     str_line = line.decode('utf-8')
-                    logger.info(str_line)
-                    stderr.append(line)
+                    stderr.append(str_line)
                     if 'Access Denied' in str_line:
                         process.kill()
                         with ux_utils.print_exception_no_traceback():
@@ -824,8 +854,7 @@ def sync_local_dir(self) -> None:
                 if returncode != 0:
                     stderr = '\n'.join(stderr)
                     with ux_utils.print_exception_no_traceback():
-                        logger.error(
-                            f'{process.stdout.decode("utf-8")}\n{stderr}')
+                        logger.error(stderr)
                         raise exceptions.StorageUploadError(
                             f'Upload to S3 failed for store {self.name} and '
                             f'source {self.source}. Please check the logs.')
@@ -1045,20 +1074,21 @@ def get_handle(self) -> StorageHandle:
     def sync_local_dir(self) -> None:
         """Syncs a local directory to a GCS bucket."""
         source = os.path.abspath(os.path.expanduser(self.source))
+        # TODO(zhwu): Speed up the upload process by providing a list of files
+        # so that we can utilize the parallelism.
         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}/'):
             with subprocess.Popen(sync_command.split(' '),
                                   stderr=subprocess.PIPE,
-                                  stdout=subprocess.PIPE) as process:
+                                  stdout=subprocess.DEVNULL) as process:
                 stderr = []
                 while True:
                     line = process.stderr.readline()
                     if not line:
                         break
                     str_line = line.decode('utf-8')
-                    logger.info(str_line)
                     stderr.append(str_line)
                     if 'AccessDeniedException' in str_line:
                         process.kill()
@@ -1072,8 +1102,7 @@ def sync_local_dir(self) -> None:
                 if returncode != 0:
                     with ux_utils.print_exception_no_traceback():
                         stderr = '\n'.join(stderr)
-                        logger.error(
-                            f'{process.stdout.decode("utf-8")}\n{stderr}')
+                        logger.error(stderr)
                         raise exceptions.StorageUploadError(
                             f'Upload to GCS failed for store {self.name} and '
                             f'source {self.source}. Please check logs.')
diff --git a/sky/execution.py b/sky/execution.py
index da9d67ca4e5..df0ebb219d0 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -12,7 +12,9 @@
 
   - ray exec + each task's commands
 """
+import copy
 import enum
+import getpass
 import tempfile
 import os
 import time
@@ -27,10 +29,13 @@
 from sky import optimizer
 from sky import sky_logging
 from sky import spot
+from sky import task as task_lib
 from sky.backends import backend_utils
+from sky.clouds import gcp
 from sky.data import data_utils
-from sky.data import storage
+from sky.data import storage as storage_lib
 from sky.usage import usage_lib
+from sky.skylet import constants
 from sky.utils import common_utils
 from sky.utils import env_options, timeline
 from sky.utils import subprocess_utils
@@ -177,7 +182,7 @@ def _execute(
 
     if task.storage_mounts is not None:
         # Optimizer should eventually choose where to store bucket
-        task.add_storage_mounts()
+        task.sync_storage_mounts()
 
     try:
         if stages is None or Stage.PROVISION in stages:
@@ -362,11 +367,8 @@ def spot_launch(
 
     change_default_value = dict()
     if not resources.use_spot_specified:
-        logger.info('Field use_spot not specified; defaulting to True.')
         change_default_value['use_spot'] = True
     if resources.spot_recovery is None:
-        logger.info('No spot recovery strategy specified; defaulting to '
-                    f'{spot.SPOT_DEFAULT_STRATEGY}.')
         change_default_value['spot_recovery'] = spot.SPOT_DEFAULT_STRATEGY
 
     new_resources = resources.copy(**change_default_value)
@@ -378,54 +380,7 @@ def spot_launch(
               f'{colorama.Style.RESET_ALL}')
         return
 
-    # TODO(zhwu): Refactor the Task (as Resources), so that we can enforce the
-    # following validations.
-    # Check the file mounts in the task.
-    # Disallow all local file mounts (copy mounts).
-    if task.workdir is not None:
-        with ux_utils.print_exception_no_traceback():
-            raise exceptions.NotSupportedError(
-                'Workdir is currently not allowed for managed spot jobs.\n'
-                'Hint: use Storage to auto-upload the workdir to a cloud '
-                'bucket.')
-    copy_mounts = task.get_local_to_remote_file_mounts()
-    if copy_mounts:
-        local_sources = '\t' + '\n\t'.join(
-            src for _, src in copy_mounts.items())
-        with ux_utils.print_exception_no_traceback():
-            raise exceptions.NotSupportedError(
-                'Local file mounts are currently not allowed for managed spot '
-                f'jobs.\nFound local source paths:\n{local_sources}\nHint: use '
-                'Storage to auto-upload local files to a cloud bucket.')
-
-    # Copy the local source to a bucket. The task will not be executed locally,
-    # so we need to copy the files to the bucket manually here before sending to
-    # the remote spot controller.
-    with backend_utils.suppress_output():
-        task.add_storage_mounts()
-
-    # Replace the source field that is local path in all storage_mounts with
-    # bucket URI and remove the name field.
-    for storage_obj in task.storage_mounts.values():
-        if (storage_obj.source is not None and
-                not data_utils.is_cloud_store_url(storage_obj.source)):
-            # Need to replace the local path with bucket URI, and remove the
-            # name field, so that the storage mount can work on the spot
-            # controller.
-            store_types = list(storage_obj.stores.keys())
-            assert len(store_types) == 1, (
-                'We only support one store type for now.', storage_obj.stores)
-            store_type = store_types[0]
-            if store_type == storage.StoreType.S3:
-                storage_obj.source = f's3://{storage_obj.name}'
-            elif store_type == storage.StoreType.GCS:
-                storage_obj.source = f'gs://{storage_obj.name}'
-            else:
-                with ux_utils.print_exception_no_traceback():
-                    raise exceptions.NotSupportedError(
-                        f'Unsupported store type: {store_type}')
-            storage_obj.name = None
-            storage_obj.force_delete = True
+    task = _translate_local_file_mounts(task)
 
     with tempfile.NamedTemporaryFile(prefix=f'spot-task-{name}-',
                                      mode='w') as f:
@@ -439,6 +394,7 @@ def spot_launch(
                 'user_yaml_path': f.name,
                 'spot_controller': controller_name,
                 'cluster_name': name,
+                'gcloud_installation_commands': gcp.GCLOUD_INSTALLATION_COMMAND,
                 'is_dev': env_options.Options.IS_DEVELOPER.get(),
                 'disable_logging': env_options.Options.DISABLE_LOGGING.get(),
                 'logging_user_hash': common_utils.get_user_hash(),
@@ -446,7 +402,7 @@ def spot_launch(
             },
             output_prefix=spot.SPOT_CONTROLLER_YAML_PREFIX)
         with sky.Dag() as spot_dag:
-            controller_task = sky.Task.from_yaml(yaml_path)
+            controller_task = task_lib.Task.from_yaml(yaml_path)
             controller_task.spot_task = task
             assert len(controller_task.resources) == 1
         print(f'{colorama.Fore.YELLOW}'
@@ -462,3 +418,151 @@ def spot_launch(
             SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP,
             retry_until_up=True,
         )
+
+
+def _translate_local_file_mounts(task: task_lib.Task) -> task_lib.Task:
+    # ================================================================
+    # Translate the workdir and local file mounts to cloud file mounts.
+    # ================================================================
+    task = copy.deepcopy(task)
+    run_id = common_utils.get_run_id()[:8]
+    original_file_mounts = task.file_mounts if task.file_mounts else {}
+    original_storage_mounts = task.storage_mounts if task.storage_mounts else {}
+
+    # 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(
+            username=getpass.getuser(), id=run_id)
+        workdir = task.workdir
+        task.workdir = None
+        if (constants.SKY_REMOTE_WORKDIR in original_file_mounts or
+                constants.SKY_REMOTE_WORKDIR in original_storage_mounts):
+            raise ValueError(
+                f'Cannot mount {constants.SKY_REMOTE_WORKDIR} as both the '
+                'workdir and file_mounts contains it as the target.')
+        new_storage_mounts[
+            constants.
+            SKY_REMOTE_WORKDIR] = storage_lib.Storage.from_yaml_config({
+                'name': bucket_name,
+                'source': workdir,
+                'persistent': False,
+                'mode': 'COPY',
+            })
+        # Check of the existence of the workdir in file_mounts is done in
+        # the task construction.
+        logger.info(f'Workdir {workdir!r} will be synced to cloud storage '
+                    f'{bucket_name!r}.')
+
+    # Step 2: Translate the local file mounts with folder in src to SkyPilot
+    # storage.
+    # 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()
+    copy_mounts_with_file_in_src = dict()
+    for i, (dst, src) in enumerate(copy_mounts.items()):
+        task.file_mounts.pop(dst)
+        if os.path.isfile(os.path.abspath(os.path.expanduser(src))):
+            copy_mounts_with_file_in_src[dst] = src
+            continue
+        bucket_name = spot.constants.SPOT_FM_BUCKET_NAME.format(
+            username=getpass.getuser(),
+            id=f'{run_id}-{i}',
+        )
+        new_storage_mounts[dst] = storage_lib.Storage.from_yaml_config({
+            'name': bucket_name,
+            'source': src,
+            'persistent': False,
+            'mode': 'COPY',
+        })
+        logger.info(
+            f'Folder in local file mount {src!r} will be synced to SkyPilot '
+            f'storage {bucket_name}.')
+
+    # 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))
+    os.makedirs(local_fm_path, exist_ok=True)
+    file_bucket_name = spot.constants.SPOT_FM_FILE_ONLY_BUCKET_NAME.format(
+        username=getpass.getuser(), id=run_id)
+    if copy_mounts_with_file_in_src:
+        src_to_file_id = dict()
+        for i, src in enumerate(set(copy_mounts_with_file_in_src.values())):
+            src_to_file_id[src] = i
+            os.link(os.path.abspath(os.path.expanduser(src)),
+                    os.path.join(local_fm_path, f'file-{i}'))
+
+        new_storage_mounts[
+            spot.constants.
+            SPOT_FM_REMOTE_TMP_DIR] = storage_lib.Storage.from_yaml_config({
+                'name': file_bucket_name,
+                'source': local_fm_path,
+                'persistent': False,
+                'mode': 'MOUNT',
+            })
+        if spot.constants.SPOT_FM_REMOTE_TMP_DIR in original_storage_mounts:
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(
+                    'Failed to translate file mounts, due to the default '
+                    f'destination {spot.constants.SPOT_FM_REMOTE_TMP_DIR} '
+                    'being taken.')
+        sources = list(src_to_file_id.keys())
+        sources_str = '\n\t'.join(sources)
+        logger.info('Source files in file_mounts will be synced to '
+                    f'cloud storage {bucket_name}:'
+                    f'\n\t{sources_str}')
+    task.update_storage_mounts(new_storage_mounts)
+
+    # Step 4: Upload storage from sources
+    # 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')
+    task.sync_storage_mounts()
+
+    # Step 5: Add the file download into the file mounts, such as
+    #  /original-dst: s3://spot-fm-file-only-bucket-name/file-0
+    new_file_mounts = dict()
+    for dst, src in copy_mounts_with_file_in_src.items():
+        storage = task.storage_mounts[spot.constants.SPOT_FM_REMOTE_TMP_DIR]
+        store_type = list(storage.stores.keys())[0]
+        store_prefix = storage_lib.get_store_prefix(store_type)
+        bucket_url = store_prefix + file_bucket_name
+        file_id = src_to_file_id[src]
+        new_file_mounts[dst] = bucket_url + f'/file-{file_id}'
+    task.update_file_mounts(new_file_mounts)
+
+    # Step 6: Replace the source field that is local path in all storage_mounts
+    # with bucket URI and remove the name field.
+    for storage_obj in task.storage_mounts.values():
+        if (storage_obj.source is not None and
+                not data_utils.is_cloud_store_url(storage_obj.source)):
+            # Need to replace the local path with bucket URI, and remove the
+            # name field, so that the storage mount can work on the spot
+            # controller.
+            store_types = list(storage_obj.stores.keys())
+            assert len(store_types) == 1, (
+                'We only support one store type for now.', storage_obj.stores)
+            store_type = store_types[0]
+            if store_type == storage_lib.StoreType.S3:
+                storage_obj.source = f's3://{storage_obj.name}'
+            elif store_type == storage_lib.StoreType.GCS:
+                storage_obj.source = f'gs://{storage_obj.name}'
+            else:
+                with ux_utils.print_exception_no_traceback():
+                    raise exceptions.NotSupportedError(
+                        f'Unsupported store type: {store_type}')
+            storage_obj.name = None
+            storage_obj.force_delete = True
+
+    return task
diff --git a/sky/constants.py b/sky/skylet/constants.py
similarity index 90%
rename from sky/constants.py
rename to sky/skylet/constants.py
index bafba180aa1..c0d98ecef2b 100644
--- a/sky/constants.py
+++ b/sky/skylet/constants.py
@@ -1,6 +1,7 @@
 """Constants for SkyPilot."""
 
 SKY_LOGS_DIRECTORY = '~/sky_logs'
+SKY_REMOTE_WORKDIR = '~/sky_workdir'
 SKY_REMOTE_RAY_VERSION = '1.13.0'
 
 # TODO(mluo): Make explicit `sky launch -c <name> ''` optional.
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index 1672c22ec22..0819285bba4 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -11,8 +11,8 @@
 
 import filelock
 
-from sky import constants
 from sky import sky_logging
+from sky.skylet import constants
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
 from sky.utils import db_utils
diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py
index f97295f81c3..641f3a72f6c 100644
--- a/sky/skylet/log_lib.py
+++ b/sky/skylet/log_lib.py
@@ -15,10 +15,10 @@
 import colorama
 
 from sky import sky_logging
+from sky.skylet import constants
 from sky.skylet import job_lib
 from sky.utils import log_utils
 
-SKY_REMOTE_WORKDIR = '~/sky_workdir'
 _SKY_LOG_WAITING_GAP_SECONDS = 1
 _SKY_LOG_WAITING_MAX_RETRY = 5
 _SKY_LOG_TAILING_GAP_SECONDS = 0.2
@@ -248,7 +248,7 @@ def make_task_bash_script(codegen: str,
             set -a
             . $(conda info --base 2> /dev/null)/etc/profile.d/conda.sh > /dev/null 2>&1 || true
             set +a
-            cd {SKY_REMOTE_WORKDIR}"""),
+            cd {constants.SKY_REMOTE_WORKDIR}"""),
     ]
     if env_vars is not None:
         for k, v in env_vars.items():
diff --git a/sky/skylet/ray_patches/__init__.py b/sky/skylet/ray_patches/__init__.py
index ce2e29d742c..8ab9b73e203 100644
--- a/sky/skylet/ray_patches/__init__.py
+++ b/sky/skylet/ray_patches/__init__.py
@@ -27,7 +27,7 @@
 import os
 import subprocess
 
-from sky import constants
+from sky.skylet import constants
 
 
 def _to_absolute(pwd_file):
diff --git a/sky/spot/__init__.py b/sky/spot/__init__.py
index a0544d3de9a..da769c145e7 100644
--- a/sky/spot/__init__.py
+++ b/sky/spot/__init__.py
@@ -19,7 +19,6 @@
 
 pathlib.Path(SPOT_TASK_YAML_PREFIX).expanduser().parent.mkdir(parents=True,
                                                               exist_ok=True)
-
 __all__ = [
     'SpotController',
     'SPOT_STRATEGIES',
diff --git a/sky/spot/constants.py b/sky/spot/constants.py
index e4b47536857..9accab396dc 100644
--- a/sky/spot/constants.py
+++ b/sky/spot/constants.py
@@ -6,3 +6,9 @@
 SPOT_CONTROLLER_YAML_PREFIX = '~/.sky/spot_controller'
 
 SPOT_TASK_YAML_PREFIX = '~/.sky/spot_tasks'
+
+SPOT_WORKDIR_BUCKET_NAME = 'skypilot-workdir-{username}-{id}'
+SPOT_FM_BUCKET_NAME = 'skypilot-filemounts-folder-{username}-{id}'
+SPOT_FM_FILE_ONLY_BUCKET_NAME = 'skypilot-filemounts-files-{username}-{id}'
+SPOT_FM_LOCAL_TMP_DIR = 'skypilot-filemounts-files-{id}'
+SPOT_FM_REMOTE_TMP_DIR = '/tmp/sky-spot-filemounts-files'
diff --git a/sky/spot/spot_utils.py b/sky/spot/spot_utils.py
index d67fb8b677b..37ab0cbed9a 100644
--- a/sky/spot/spot_utils.py
+++ b/sky/spot/spot_utils.py
@@ -321,7 +321,7 @@ def load_spot_job_queue(payload: str) -> List[Dict[str, Any]]:
     return jobs
 
 
-def format_job_table(jobs: Dict[str, Any], show_all: bool) -> str:
+def format_job_table(jobs: List[Dict[str, Any]], show_all: bool) -> str:
     """Show all spot jobs."""
     columns = [
         'ID', 'NAME', 'RESOURCES', 'SUBMITTED', 'TOT. DURATION', 'JOB DURATION',
diff --git a/sky/task.py b/sky/task.py
index 98de6ec8e37..3bda4a89f42 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -8,11 +8,13 @@
 import yaml
 
 import sky
+from sky import check
 from sky import clouds
+from sky import global_user_state
 from sky.backends import backend_utils
 from sky.data import storage as storage_lib
-from sky.data import data_transfer as data_transfer_lib
 from sky.data import data_utils
+from sky.skylet import constants
 from sky.utils import schemas
 from sky.utils import ux_utils
 
@@ -464,25 +466,77 @@ def set_storage_mounts(
             self.storage_mounts = None
             return self
         for target, _ in storage_mounts.items():
+            if (target == constants.SKY_REMOTE_WORKDIR and
+                    self.workdir is not None):
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        f'Cannot use {constants.SKY_REMOTE_WORKDIR!r} as a '
+                        'destination path of a file mount, as it will be used '
+                        'by the workdir. If uploading a file/folder to the '
+                        'workdir is needed, please specify the full path to '
+                        'the file/folder.')
+
             if data_utils.is_cloud_store_url(target):
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         'Storage mount destination path cannot be cloud storage'
                     )
         # Storage source validation is done in Storage object
-
         self.storage_mounts = storage_mounts
         return self
 
-    def add_storage_mounts(self) -> None:
-        """Adds storage mounts to the Task."""
-        # TODO(romilb): The optimizer should look at the source and destination
-        #  to figure out the right stores to use. For now, we hardcode
-        #  storage_plans to AWS as the optimal plan.
+    def update_storage_mounts(self, storage_mounts: Dict[str,
+                                                         storage_lib.Storage]):
+        """Updates the storage mounts for this Task"""
+        if not storage_mounts:
+            return self
+        task_storage_mounts = self.storage_mounts if self.storage_mounts else {}
+        task_storage_mounts.update(storage_mounts)
+        return self.set_storage_mounts(task_storage_mounts)
+
+    def get_preferred_store_type(self) -> storage_lib.StoreType:
+        # TODO(zhwu, romilb): The optimizer should look at the source and
+        #  destination to figure out the right stores to use. For now, we
+        #  use a heuristic solution to find the store type by the following
+        #  order:
+        #  1. cloud decided in best_resources.
+        #  2. cloud specified in the task resources.
+        #  3. the first enabled cloud.
+        # This should be refactored and moved to the optimizer.
+        assert len(self.resources) == 1, self.resources
+        storage_cloud = None
+        if self.best_resources is not None:
+            storage_cloud = self.best_resources.cloud
+        else:
+            resources = list(self.resources)[0]
+            storage_cloud = resources.cloud
+            if storage_cloud is None:
+                # Get the first enabled cloud.
+                enabled_clouds = global_user_state.get_enabled_clouds()
+                if len(enabled_clouds) == 0:
+                    check.check(quiet=True)
+                    enabled_clouds = global_user_state.get_enabled_clouds()
+                if len(enabled_clouds) == 0:
+                    raise ValueError('No enabled clouds.')
+
+                for cloud in storage_lib.STORE_ENABLED_CLOUDS:
+                    for enabled_cloud in enabled_clouds:
+                        if cloud.is_same_cloud(enabled_cloud):
+                            storage_cloud = cloud
+                            break
+        if storage_cloud is None:
+            raise ValueError('No available cloud to mount storage.')
+        store_type = storage_lib.get_storetype_from_cloud(storage_cloud)
+        return store_type
+
+    def sync_storage_mounts(self) -> None:
+        """Syncs storage mounts: sync files/dirs to cloud storage."""
+
         for storage in self.storage_mounts.values():
             if len(storage.stores) == 0:
-                self.storage_plans[storage] = storage_lib.StoreType.S3
-                storage.add_store(storage_lib.StoreType.S3)
+                store_type = self.get_preferred_store_type()
+                self.storage_plans[storage] = store_type
+                storage.add_store(store_type)
             else:
                 # We will download the first store that is added to remote.
                 self.storage_plans[storage] = list(storage.stores.keys())[0]
@@ -503,12 +557,6 @@ def add_storage_mounts(self) -> None:
                         mnt_path: blob_path,
                     })
                 elif store_type is storage_lib.StoreType.GCS:
-                    # Remember to run `gcloud auth application-default login`
-                    self.setup = (
-                        '([[ -z $GOOGLE_APPLICATION_CREDENTIALS ]] && '
-                        'echo GOOGLE_APPLICATION_CREDENTIALS='
-                        f'{data_transfer_lib.DEFAULT_GCS_CREDENTIALS_PATH} '
-                        f'>> ~/.bashrc || true); {self.setup or "true"}')
                     if storage.source.startswith('gs://'):
                         blob_path = storage.source
                     else:
@@ -570,6 +618,17 @@ def set_file_mounts(self, file_mounts: Optional[Dict[str, str]]) -> None:
                             f'File mount source {source!r} does not exist '
                             'locally. To fix: check if it exists, and correct '
                             'the path.')
+            # TODO(zhwu): /home/username/sky_workdir as the target path need
+            # to be filtered out as well.
+            if (target == constants.SKY_REMOTE_WORKDIR and
+                    self.workdir is not None):
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        f'Cannot use {constants.SKY_REMOTE_WORKDIR!r} as a '
+                        'destination path of a file mount, as it will be used '
+                        'by the workdir. If uploading a file/folder to the '
+                        'workdir is needed, please specify the full path to '
+                        'the file/folder.')
 
         self.file_mounts = file_mounts
         return self
diff --git a/sky/templates/spot-controller.yaml.j2 b/sky/templates/spot-controller.yaml.j2
index 00e6e4878df..d56a5ce14b5 100644
--- a/sky/templates/spot-controller.yaml.j2
+++ b/sky/templates/spot-controller.yaml.j2
@@ -25,13 +25,7 @@ setup: |
   # pip list | grep azure-cli > /dev/null 2>&1 || \
   #  pip3 install azure-cli==2.31.0 azure-core
 
-  gcloud --help > /dev/null 2>&1 || \
-  (wget --quiet https://dl.google.com/dl/cloudsdk/channels/rapid/downloads/google-cloud-sdk-382.0.0-linux-x86_64.tar.gz && \
-    tar xzf google-cloud-sdk-382.0.0-linux-x86_64.tar.gz && \
-    mv google-cloud-sdk ~/ && \
-    ~/google-cloud-sdk/install.sh -q > /dev/null 2>&1 && \
-    source ~/google-cloud-sdk/path.bash.inc > /dev/null 2>&1 && \
-    echo 'source ~/google-cloud-sdk/path.bash.inc > /dev/null 2>&1' >> ~/.bashrc)
+  {{gcloud_installation_commands}}
 
 run: |
   python -u -m sky.spot.controller \
diff --git a/sky/usage/usage_lib.py b/sky/usage/usage_lib.py
index 4c084b5e03d..493226a176e 100644
--- a/sky/usage/usage_lib.py
+++ b/sky/usage/usage_lib.py
@@ -11,7 +11,6 @@
 import traceback
 import typing
 from typing import Any, Dict, List, Optional, Union
-import uuid
 
 import requests
 
@@ -28,16 +27,6 @@
 
 logger = sky_logging.init_logger(__name__)
 
-_run_id = None
-
-
-def _get_logging_run_id():
-    """Returns a unique run id for this logging."""
-    global _run_id
-    if _run_id is None:
-        _run_id = str(uuid.uuid4())
-    return _run_id
-
 
 def _get_current_timestamp_ns() -> int:
     return int(datetime.datetime.now(datetime.timezone.utc).timestamp() * 1e9)
@@ -88,7 +77,7 @@ def __init__(self) -> None:
         super().__init__(constants.USAGE_MESSAGE_SCHEMA_VERSION)
         # Message identifier.
         self.user: str = _get_user_hash()
-        self.run_id: str = _get_logging_run_id()
+        self.run_id: str = common_utils.get_run_id()
         self.sky_version: str = sky.__version__
 
         # Entry
diff --git a/sky/utils/common_utils.py b/sky/utils/common_utils.py
index 0e46f119139..6e5f7798681 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -11,6 +11,7 @@
 import socket
 import sys
 import time
+import uuid
 from typing import Dict, List, Union
 import yaml
 
@@ -23,6 +24,21 @@
 
 logger = sky_logging.init_logger(__name__)
 
+_run_id = None
+
+
+def get_run_id():
+    """Returns a unique run id for each 'run'.
+
+    A run is defined as the lifetime of a process that has imported `sky`
+    and has called its CLI or programmatic APIs. For example, two successive
+    `sky launch` are two runs.
+    """
+    global _run_id
+    if _run_id is None:
+        _run_id = str(uuid.uuid4())
+    return _run_id
+
 
 def get_user_hash():
     """Returns a unique user-machine specific hash as a user id."""
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 1ea5ff76d39..eff7edf3770 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -2,7 +2,6 @@
 import textwrap
 
 from click import testing as cli_testing
-import pytest
 
 import sky
 from sky import clouds
@@ -57,6 +56,10 @@ def test_accelerator_mismatch(monkeypatch):
         lambda: enabled_clouds,
     )
     monkeypatch.setattr('sky.check.check', lambda *_args, **_kwargs: None)
+    config_file_backup = tempfile.NamedTemporaryFile(
+        prefix='tmp_backup_config_default', delete=False)
+    monkeypatch.setattr('sky.clouds.gcp.GCP_CONFIG_SKY_BACKUP_PATH',
+                        config_file_backup.name)
 
     spec = textwrap.dedent("""\
         resources:
diff --git a/tests/test_onprem.py b/tests/test_onprem.py
index 472aefc1f1f..d0d74ab922e 100644
--- a/tests/test_onprem.py
+++ b/tests/test_onprem.py
@@ -11,7 +11,6 @@
 import yaml
 
 from sky import cli
-from sky import exceptions
 from sky.utils import command_runner
 from sky.utils import subprocess_utils
 
diff --git a/tests/test_optimizer_dryruns.py b/tests/test_optimizer_dryruns.py
index edd7e88e186..079d28232a0 100644
--- a/tests/test_optimizer_dryruns.py
+++ b/tests/test_optimizer_dryruns.py
@@ -38,6 +38,12 @@ def _make_resources(
         lambda: enabled_clouds,
     )
     monkeypatch.setattr('sky.check.check', lambda *_args, **_kwargs: None)
+
+    config_file_backup = tempfile.NamedTemporaryFile(
+        prefix='tmp_backup_config_default', delete=False)
+    monkeypatch.setattr('sky.clouds.gcp.GCP_CONFIG_SKY_BACKUP_PATH',
+                        config_file_backup.name)
+
     # Should create Resources here, since it uses the enabled clouds.
     return sky.Resources(*resources_args, **resources_kwargs)
 
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 28feb6b734c..2b30040d5d4 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -282,7 +282,7 @@ def test_job_queue():
     test = Test(
         'job_queue',
         [
-            f'sky launch -y -c {name} examples/job_queue/cluster.yaml',
+            f'sky launch --cloud gcp -y -c {name} examples/job_queue/cluster.yaml',
             f'sky exec {name} -d examples/job_queue/job.yaml',
             f'sky exec {name} -d examples/job_queue/job.yaml',
             f'sky exec {name} -d examples/job_queue/job.yaml',
@@ -299,7 +299,7 @@ def test_n_node_job_queue():
     test = Test(
         'job_queue_multinode',
         [
-            f'sky launch -y -c {name} examples/job_queue/cluster_multinode.yaml',
+            f'sky launch --cloud gcp -y -c {name} examples/job_queue/cluster_multinode.yaml',
             f'sky exec {name} -d examples/job_queue/job_multinode.yaml',
             f'sky exec {name} -d examples/job_queue/job_multinode.yaml',
             f'sky exec {name} -d examples/job_queue/job_multinode.yaml',


Code review comments:
diff_hunk: @@ -282,7 +282,7 @@ def test_job_queue():
     test = Test(
         'job_queue',
         [
-            f'sky launch -y -c {name} examples/job_queue/cluster.yaml',
+            f'sky launch --cloud gcp -y -c {name} examples/job_queue/cluster.yaml',
path: tests/test_smoke.py
text: @user1:
Shall we leave these out?

@author:
Done. Thanks!