Repository: skypilot-org/skypilot
Pull request: UX: hint for unmanaged spot; spot queue logging.
<!-- Describe the changes in this PR -->

Fixes #1553.

- Hint for unmanaged spot
- Some logging changes in `sky spot queue [-s]`


<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

`spot queue` logging:

<img width="679" alt="Screen Shot 2023-01-13 at 09 22 22" src="https://user-images.githubusercontent.com/592670/212381792-41e58bed-a97a-46a1-8369-b1300a424341.png">

Checked that `sky launch --use-spot` (unmanaged) prints:

<img width="1132" alt="Screen Shot 2023-01-13 at 09 23 14" src="https://user-images.githubusercontent.com/592670/212381778-969a3f02-f011-4a31-a9d5-591e77529d39.png">


Checked that `sky spot launch` then `sky logs <controller>` (managed) does not print the above.



Problem statement:
UX: if unmanaged spot is used, print a hint that only managed spot automatically recovers
If we detected unmanaged spot is used, we should print a hint a la
- Unmanaged spot cluster is being created, which does not automatically recover from preemptions. 
- To have auto recovery, use managed spot jobs instead: CLI `sky spot launch`, Python `sky.spot_launch()`.

Hints from issue discussion:

Commit message: Pylint

Patch to review:
diff --git a/sky/cli.py b/sky/cli.py
index e4c9eff8f46..5cb7478004e 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -2995,26 +2995,43 @@ def spot_queue(all: bool, refresh: bool, skip_finished: bool):
       watch -n60 sky spot queue
     """
     click.secho('Fetching managed spot job statuses...', fg='yellow')
+    no_jobs_found_str = '  No jobs found.'
     try:
         job_table = core.spot_queue(refresh=refresh,
                                     skip_finished=skip_finished)
     except exceptions.ClusterNotUpError:
+        # TODO(mehul): respect skip_finished here?
         cache = spot_lib.load_job_table_cache()
         if cache is not None:
             readable_time = log_utils.readable_time_duration(cache[0])
+            if not cache[1].strip():
+                table_str = no_jobs_found_str
+                cached_hint = ''
+            else:
+                table_str = cache[1]
+                # Show a helpful hint at the end, if cached table is non-empty.
+                cached_hint = (
+                    '\n\nNOTE: displaying the cached job status table '
+                    f'[last updated: {readable_time}]')
             table_message = (
                 f'\n{colorama.Fore.YELLOW}Cached job status table '
                 f'[last updated: {readable_time}]:{colorama.Style.RESET_ALL}\n'
-                f'{cache[1]}\n')
+                f'{table_str}{cached_hint}')
         else:
             table_message = 'No cached job status table found.'
         click.echo(table_message)
         return
     job_table = spot_lib.format_job_table(job_table, all)
 
-    # Dump cache
     spot_lib.dump_job_table_cache(job_table)
-    click.echo(f'Managed spot jobs:\n{job_table}')
+
+    if not skip_finished:
+        in_progress_only_hint = ''
+    else:
+        in_progress_only_hint = ' (showing in-progress jobs only)'
+    if not job_table:
+        job_table = no_jobs_found_str
+    click.echo(f'Managed spot jobs{in_progress_only_hint}:\n{job_table}')
 
 
 _add_command_alias_to_group(spot, spot_queue, 'status', hidden=True)
diff --git a/sky/execution.py b/sky/execution.py
index 77554786f03..038764309b8 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -119,6 +119,9 @@ def _execute(
     detach_run: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
     no_setup: bool = False,
+    # Internal only:
+    # pylint: disable=invalid-name
+    _is_launched_by_spot_controller: bool = False,
 ) -> None:
     """Execute a entrypoint.
 
@@ -167,6 +170,16 @@ def _execute(
             raise ValueError(
                 'Spot recovery is specified in the task. To launch the '
                 'managed spot job, please use: sky spot launch')
+    if task.use_spot and not _is_launched_by_spot_controller:
+        yellow = colorama.Fore.YELLOW
+        bold = colorama.Style.BRIGHT
+        reset = colorama.Style.RESET_ALL
+        logger.info(
+            f'{yellow}Launching an unmanaged spot task, which does not '
+            f'automatically recover from preemptions.{reset}\n{yellow}To get '
+            'automatic recovery, use managed spot instead: '
+            f'{reset}{bold}sky spot launch{reset} {yellow}or{reset} '
+            f'{bold}sky.spot_launch(){reset}.')
 
     cluster_exists = False
     if cluster_name is not None:
@@ -299,6 +312,9 @@ def launch(
     detach_setup: bool = False,
     detach_run: bool = False,
     no_setup: bool = False,
+    # Internal only:
+    # pylint: disable=invalid-name
+    _is_launched_by_spot_controller: bool = False,
 ) -> None:
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Launch a task.
@@ -373,6 +389,7 @@ def launch(
         detach_run=detach_run,
         idle_minutes_to_autostop=idle_minutes_to_autostop,
         no_setup=no_setup,
+        _is_launched_by_spot_controller=_is_launched_by_spot_controller,
     )
 
 
diff --git a/sky/spot/recovery_strategy.py b/sky/spot/recovery_strategy.py
index 12035d35ba2..14a7d21e153 100644
--- a/sky/spot/recovery_strategy.py
+++ b/sky/spot/recovery_strategy.py
@@ -154,7 +154,8 @@ def _launch(self, max_retry=3, raise_on_failure=True) -> Optional[float]:
                 usage_lib.messages.usage.set_internal()
                 sky.launch(self.dag,
                            cluster_name=self.cluster_name,
-                           detach_run=True)
+                           detach_run=True,
+                           _is_launched_by_spot_controller=True)
                 logger.info('Spot cluster launched.')
             except exceptions.InvalidClusterNameError as e:
                 # The cluster name is too long.
diff --git a/sky/task.py b/sky/task.py
index 2466bd46fc2..613f23edb14 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -381,6 +381,10 @@ def set_envs(
     def need_spot_recovery(self) -> bool:
         return any(r.spot_recovery is not None for r in self.resources)
 
+    @property
+    def use_spot(self) -> bool:
+        return any(r.use_spot for r in self.resources)
+
     @num_nodes.setter
     def num_nodes(self, num_nodes: Optional[int]) -> None:
         if num_nodes is None:


Code review comments:
diff_hunk: @@ -167,6 +170,16 @@ def _execute(
             raise ValueError(
                 'Spot recovery is specified in the task. To launch the '
                 'managed spot job, please use: sky spot launch')
+    if task.use_spot and not _is_launched_by_spot_controller:
+        yellow = colorama.Fore.YELLOW
+        bold = colorama.Style.BRIGHT
+        reset = colorama.Style.RESET_ALL
+        logger.info(
+            f'{yellow}Launching an unmanaged spot task, which does not '
+            f'automatically recover from preemptions.{reset}\n{yellow}To get '
+            'automatic recovery, use managed spot instead: '
+            f'{reset}{bold}sky spot launch{reset} {yellow}or{reset} '
+            f'{bold}sky.spot_launch(){reset}.')
path: sky/execution.py
text: @user1:
Instead of adding the `_is_launched_by_spot_controller`, we can check whether `task.spot_task` is `None`. If it is not `None`, the task is a controller task and vice versa.

@user1:
Actually, how about we place the logging in the `backend.provision`?
Reason:
1. We may only want to print the message once, when it is provisioning. In the `backend.provision`, we will have the information, including the actually launched resources and whether the cluster is firstly launched. The current logging will be printed even when user `sky exec mycluster task.yaml`
2. The spot property sounds more tied with the backend, and it would be better to handle it in the backend?

@author:
It looks like `task.spot_task` is not serialized into / deserialized from a yaml file, which `controller.py` depends on. So when the controller process for a spot job launches and runs `sky.launch(actual_task)`, `actual_task.spot_task` doesn't exist.

On moving it into `backend.provision()`: it needs a `_is_launched_by_spot_controller` to be plumbed through from this function as well.

Given the above, how about we keep the logging in this func? We can add an additional check `Stage.PROVISION in stages` to print it for `sky launch` only.

@user1:
Ahh, that makes sense. I misunderstood the purpose of this. Having the check under the `PROVISION` stage should be fine, though it may still log out the information when the user `sky launch -c existing-cluster` again.

Could not think of a better way at the moment, but I wouldn't say I like the internal argument in the API. It might be better to make it a global config or have a wrapper inner function `_launch` which hash the argument, but not for the `launch` function?

@author:
We could change this 

https://github.com/skypilot-org/skypilot/blob/master/sky/spot/recovery_strategy.py#L155

into a `execution.internal_launch_from_spot_controller(...)`? That said, it may bloat the interface since the new function will have only one argument difference from the original. 

Or, we can use a `**kwargs` to pass in this bool.

Examples with internal args aren't uncommon:
- https://docs.ray.io/en/latest/ray-core/package-ref.html#placement-group
- https://docs.ray.io/en/latest/ray-core/package-ref.html#ray-shutdown
- https://docs.ray.io/en/latest/ray-core/package-ref.html#ray-put

@user1:
Good point! Bloating the interface seems worse to me. Let's keep the internal argument then. : )

@author:
Moved logging, and a second `launch` would not trigger the logging again.