Repository: skypilot-org/skypilot
Pull request #1231 (2022-10-12 00:31:46)

[Storage] Disallow single files for upload to Storage
Mitigates #1226 by raising an exception if files are uploaded to sky storage.

```
(base) romilb@romilbx1yoga:$ sky launch storage_test.yaml 

Task from YAML spec: storage_test.yaml
sky.exceptions.StorageSourceError: Storage source path cannot be a file - only directories are supported as a source. To upload a single file, either
1) Use regular file mounts by writing ~/tmpfile: ~/tmpfile in the file_mounts section of your YAML, or
2) Place the file in a directory and specify the directory as the source.
```

Tested:
- [x] Example from #1226 
- [x] `sky launch examples/storage_demo.yaml`
- [x]  `pytest test_storage.py::TestStorageSpecLocalSource`
- [x] Docs rendered locally  


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/sky/data/storage.py b/sky/data/storage.py
index c7cf64aaf..b3e279559 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -459,15 +459,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 +593,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
 
@@ -669,7 +680,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 +840,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 +857,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 +1090,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 +1105,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..038555bc5 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -385,7 +385,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 +425,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 +444,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 +484,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 +506,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 +544,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/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..58355fa21 100644
--- a/sky/templates/aws-ray.yml.j2
+++ b/sky/templates/aws-ray.yml.j2
@@ -121,13 +121,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/utils/common_utils.py b/sky/utils/common_utils.py
index 6e5f77986..94fcce2e9 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -246,6 +246,9 @@ def decode_payload(payload_str: str) -> Union[List, Dict]:
     Returns:
         A dict or list that is decoded from the payload string.
     """
-    payload_str = _PAYLOAD_PATTERN.match(payload_str).group(1)
+    matched = _PAYLOAD_PATTERN.match(payload_str)
+    if matched is None:
+        raise ValueError(f'Invalid payload string: \n{payload_str}')
+    payload_str = matched.group(1)
     payload = json.loads(payload_str)
     return payload


Test patch:
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"""