Repository: skypilot-org/skypilot

Issue:
sky spot timestamp error when the shell prints additional warning
Hi Sky team,

There is a problem related to #1181 . The error 
```
bash: warning: setlocale: LC_ALL: cannot change locale (en_US.UTF-8)
```
leads to the failure of get_job_status in recovery_strategy.py. This made the controller kill the normal running jobs and restart the workers. See the full error message below.
```
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:28 spot_utils.py:58] === Checking the job status... ===
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:31 spot_utils.py:64] Job status: JobStatus.RUNNING
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:31 spot_utils.py:67] ==================================
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:33 recovery_strategy.py:234] Unexpected Exception: could not convert string to float: 'bash: warning: setlocale: LC_ALL: cannot change locale (en_US.UTF-8)\n1665684801.9676614\n'
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:33 recovery_strategy.py:234] Failed to get the job start timestamp. Retrying.
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:40 spot_utils.py:58] === Checking the job status... ===
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:41 spot_utils.py:64] Job status: JobStatus.RUNNING
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:41 spot_utils.py:67] ==================================
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:43 recovery_strategy.py:234] Unexpected Exception: could not convert string to float: 'bash: warning: setlocale: LC_ALL: cannot change locale (en_US.UTF-8)\n1665684801.9676614\n'
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:43 recovery_strategy.py:234] Failed to get the job start timestamp. Retrying.
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:49 spot_utils.py:58] === Checking the job status... ===
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:50 spot_utils.py:64] Job status: JobStatus.RUNNING
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:50 spot_utils.py:67] ==================================
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:52 recovery_strategy.py:234] Unexpected Exception: could not convert string to float: 'bash: warning: setlocale: LC_ALL: cannot change locale (en_US.UTF-8)\n1665684801.9676614\n'
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:52 recovery_strategy.py:234] Failed to get the job start timestamp. Retrying.
(worker-221013-test-merge-0 pid=801) I 10-13 18:14:52 recovery_strategy.py:184] Failed to get the job status, due to unexpected job submission error.
(worker-221013-test-merge-0 pid=801) I 10-13 18:15:28 recovery_strategy.py:253] Retrying to launch the spot cluster in 192.4 seconds.
(worker-221013-test-merge-0 pid=801) I 10-13 18:18:41 cloud_vm_ray_backend.py:2705] Creating a new cluster: "worker-221013-test-merge-0-1" [1x GCP(n2d-standard-64[Spot])].
(worker-221013-test-merge-0 pid=801) I 10-13 18:18:41 cloud_vm_ray_backend.py:2705] Tip: to reuse an existing cluster, specify --cluster (-c). Run `sky status` to see existing clusters.
(worker-221013-test-merge-0 pid=801) I 10-13 18:18:41 cloud_vm_ray_backend.py:903] To view detailed progress: tail -n100 -f /home/jzhou_salk_edu/sky_logs/sky-2022-10-13-18-18-41-040313/provision.log
(worker-221013-test-merge-0 pid=801) I 10-13 18:18:42 authentication.py:241] OS Login is enabled for GCP project prod-635e. Running additional authentication steps.
```




Discussion:
Michaelvll: Hi @zhoujt1994, this PR should fix the issue you mentioned in #1243. If you get time, would you like to try it out, by switching the branch in your folder where you download the skypilot codebase, using the following commands:
```
git pull
git checkout fix-spot-messages
pip install -e ".[gcp]"
```
 Then please try the following command to see if it works.
```
sky spot launch -n test-message --cloud gcp 'echo hi'
```

Patch:
diff --git a/docs/source/examples/syncing-code-artifacts.rst b/docs/source/examples/syncing-code-artifacts.rst
index 46d2acb4..e7634872 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
 --------------------------------------
diff --git a/docs/source/reference/storage.rst b/docs/source/reference/storage.rst
index e7808cba..8ec6992f 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 35fac797..b6d04806 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 72504041..b3e27955 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.')
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index aef17542..ddd8ca95 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 991853b3..1784d986 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 00874079..f4207cde 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/utils/common_utils.py b/sky/utils/common_utils.py
index 94fcce2e..93e3b23e 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,18 +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.
     """
-    matched = _PAYLOAD_PATTERN.match(payload_str)
-    if matched is None:
+    matched = _PAYLOAD_PATTERN.findall(payload_str)
+    if not matched:
         raise ValueError(f'Invalid payload string: \n{payload_str}')
-    payload_str = matched.group(1)
+    payload_str = matched[0]
     payload = json.loads(payload_str)
     return payload


Test patch:
diff --git a/tests/test_storage.py b/tests/test_storage.py
index b3866359..b579e6d5 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"""