Repository: skypilot-org/skypilot
Pull request: Make `sky start` on the spot controller aware of autostop.
Fixes #1447.

If controller is to be started:
 - disallow: -i, --down; disallow being specified with other clusters

Tested:

CLI
```
# Expectedly fail:
sky start sky-cpunode-zongheng sky-spot-controller-8a3968f2
sky start sky-spot-controller-8a3968f2 -i1
sky start sky-spot-controller-8a3968f2 -i1 --down
sky start sky-spot-controller-8a3968f2 --down

# Starting a stopped controller: OK. Autostop reset to 10m.
sky start sky-spot-controller-8a3968f2  

# Starting an UP controller: OK. Autostop reset to 10m.
sky start sky-spot-controller-8a3968f2 --force 
```
Python API
- [x] repeated the above and observed appropriate ValueError
- [x] manually called `core.start()` on a stopped controller; verfied autostop is set to 10m

Smoke
- [x] `bash tests/run_smoke_tests.sh test_spot`


Problem statement:
[spot] Interaction between `sky start [--force]` and spot controller's autostop
User reported that:

>Is the spot controller autostopping actually working? When checking sky status, the comment says will be autostopped if idle for 10min  but the AUTOSTOP column shows nothing.

---

After digging into potential issues, we aren't sure why this happened. The below *could* be one reason:

We should either 
- disallow `sky start [--force]` on a (stopped) spot controller (which resets its autostop setting to 0); or
- ensure  `sky start [--force]` on a (stopped) spot controller properly restores its autostop setting

cc @Michaelvll

Hints from issue discussion:
We should probably do the second option, to support cases like: a controller gets auto-stopped -> user wants to run `sky spot logs X`. Relatedly - this will print:

https://github.com/skypilot-org/skypilot/blob/master/sky/core.py#L709-L712

which asks users to manually run `sky start` with an updated `-i` number of minutes.
Good catch! I have experienced this before as well. It makes sense to support the second option by automatically set the autostop when starting the spot controller.

Commit message: pytest

Patch to review:
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index ce053b1a5a3..85a362f880d 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -1956,7 +1956,7 @@ def check_cluster_name_not_reserved(
     return None.
     """
     if cluster_name in SKY_RESERVED_CLUSTER_NAMES:
-        msg = (f'Cluster {cluster_name!r} is reserved for '
+        msg = (f'Cluster {cluster_name!r} is reserved for the '
                f'{SKY_RESERVED_CLUSTER_NAMES[cluster_name].lower()}.')
         if operation_str is not None:
             msg += f' {operation_str} is not allowed.'
diff --git a/sky/cli.py b/sky/cli.py
index 011518d8f41..287b5a7f7f2 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -1174,7 +1174,7 @@ def launch(
     and they undergo job queue scheduling.
     """
     backend_utils.check_cluster_name_not_reserved(
-        cluster, operation_str='Launching task on it')
+        cluster, operation_str='Launching tasks on it')
     if backend_name is None:
         backend_name = backends.CloudVmRayBackend.NAME
 
@@ -1917,6 +1917,32 @@ def start(
     if not to_start:
         return
 
+    # Checks for reserved clusters (spot controller).
+    reserved, non_reserved = [], []
+    for name in to_start:
+        if name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+            reserved.append(name)
+        else:
+            non_reserved.append(name)
+    if reserved and non_reserved:
+        assert len(reserved) == 1, reserved
+        # Keep this behavior the same as _down_or_stop_clusters().
+        raise click.UsageError(
+            'Starting the spot controller with other cluster(s) '
+            'is currently not supported.\n'
+            'Please start the former independently.')
+    if reserved:
+        assert len(reserved) == 1, reserved
+        bold = backend_utils.BOLD
+        reset_bold = backend_utils.RESET_BOLD
+        if idle_minutes_to_autostop is not None:
+            raise click.UsageError(
+                'Autostop options are currently not allowed when starting the '
+                'spot controller. Use the default autostop settings by directly'
+                f' callling: {bold}sky start {reserved[0]}{reset_bold}')
+        idle_minutes_to_autostop = (
+            spot_lib.SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP)
+
     if not yes:
         cluster_str = 'clusters' if len(to_start) > 1 else 'cluster'
         cluster_list = ', '.join(to_start)
@@ -2107,13 +2133,13 @@ def _down_or_stop_clusters(
                 names_str = ', '.join(map(repr, names))
                 raise click.UsageError(
                     f'{operation} reserved cluster(s) '
-                    f'{reserved_clusters_str} with multiple other cluster(s) '
-                    f'{names_str} is not supported.\n'
+                    f'{reserved_clusters_str} with other cluster(s) '
+                    f'{names_str} is currently not supported.\n'
                     f'Please omit the reserved cluster(s) {reserved_clusters}.')
             if not down:
                 raise click.UsageError(
                     f'{operation} reserved cluster(s) '
-                    f'{reserved_clusters_str} is not supported.')
+                    f'{reserved_clusters_str} is currently not supported.')
             else:
                 # TODO(zhwu): We can only have one reserved cluster (spot
                 # controller).
diff --git a/sky/core.py b/sky/core.py
index 74a2a8a653d..b03011614c5 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -89,6 +89,20 @@ def _start(
             f'Starting cluster {cluster_name!r} with backend {backend.NAME} '
             'is not supported.')
 
+    if cluster_name == spot.SPOT_CONTROLLER_NAME:
+        if down:
+            raise ValueError('Using autodown (rather than autostop) is not '
+                             'supported for the spot controller. Pass '
+                             '`down=False` or omit it instead.')
+        if (idle_minutes_to_autostop is not None and idle_minutes_to_autostop !=
+                spot.SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP):
+            raise ValueError(
+                'Passing a custom autostop setting is currently not '
+                'supported when starting the spot controller. To '
+                'fix: omit the `idle_minutes_to_autostop` argument to use the '
+                'default autostop settings.')
+        idle_minutes_to_autostop = spot.SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP
+
     # NOTE: if spot_queue() calls _start() and hits here, that entrypoint
     # would have a cluster name (the controller) filled in.
     usage_lib.record_cluster_name_for_current_operation(cluster_name)
@@ -149,7 +163,10 @@ def start(
 
     Raises:
         ValueError: the specified cluster does not exist; or if ``down`` is set
-          to True but ``idle_minutes_to_autostop`` is None.
+          to True but ``idle_minutes_to_autostop`` is None; or if the specified
+          cluster is the managed spot controller, and either
+          ``idle_minutes_to_autostop`` is set to a non-default value or
+          ``down`` is True (omit them to use the default autostop settings).
         sky.exceptions.NotSupportedError: if the cluster to restart was
           launched using a non-default backend that does not support this
           operation.
@@ -709,7 +726,7 @@ def spot_tail_logs(name: Optional[str], job_id: Optional[int],
     # TODO(zhwu): Automatically restart the spot controller
     controller_status, handle = _is_spot_controller_up(
         'Please restart the spot controller with '
-        f'`sky start {spot.SPOT_CONTROLLER_NAME} -i 5`.')
+        f'`sky start {spot.SPOT_CONTROLLER_NAME}`.')
     if handle is None or handle.head_ip is None:
         msg = 'All jobs finished.'
         if controller_status == global_user_state.ClusterStatus.INIT:
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 4257833c381..eca2f7a0038 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -758,6 +758,8 @@ def test_use_spot():
 def test_spot():
     """Test the spot yaml."""
     name = _get_cluster_name()
+    cancel_command = (
+        f'sky spot cancel -y -n {name}-1; sky spot cancel -y -n {name}-2')
     test = Test(
         'managed-spot',
         [
@@ -770,8 +772,15 @@ def test_spot():
             'sleep 200',
             f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name}-1 | head -n1 | grep CANCELLED',
             f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name}-2 | head -n1 | grep "RUNNING\|SUCCEEDED"',
+            # Test autostop. This assumes no regular spot jobs are running.
+            cancel_command,
+            'sleep 720',  # Sleep for a bit more than the default 10m.
+            'sky status --refresh | grep sky-spot-controller- | grep STOPPED',
+            'sky start "sky-spot-controller-*" -y',
+            # Ensures it's up and the autostop setting is restored.
+            'sky status | grep sky-spot-controller- | grep UP | grep 10m',
         ],
-        f'sky spot cancel -y -n {name}-1; sky spot cancel -y -n {name}-2',
+        cancel_command,
     )
     run_one_test(test)
 
diff --git a/tests/test_spot.py b/tests/test_spot.py
index 30f0e0e92a0..5d920913c36 100644
--- a/tests/test_spot.py
+++ b/tests/test_spot.py
@@ -141,7 +141,7 @@ def test_stop_spot_controller(self, _mock_cluster_state):
         assert result.exit_code == click.UsageError.exit_code
         assert (
             f'Stopping reserved cluster(s) \'{spot.SPOT_CONTROLLER_NAME}\' is '
-            'not supported' in result.output)
+            'currently not supported' in result.output)
 
         result = cli_runner.invoke(cli.stop, ['sky-spot-con*'])
         assert not result.exception
@@ -157,7 +157,7 @@ def test_autostop_spot_controller(self, _mock_cluster_state):
         result = cli_runner.invoke(cli.autostop, [spot.SPOT_CONTROLLER_NAME])
         assert result.exit_code == click.UsageError.exit_code
         assert ('Scheduling autostop on reserved cluster(s) '
-                f'\'{spot.SPOT_CONTROLLER_NAME}\' is not supported'
+                f'\'{spot.SPOT_CONTROLLER_NAME}\' is currently not supported'
                 in result.output)
 
         result = cli_runner.invoke(cli.autostop, ['sky-spot-con*'])


Code review comments:
diff_hunk: @@ -89,6 +89,20 @@ def _start(
             f'Starting cluster {cluster_name!r} with backend {backend.NAME} '
             'is not supported.')
 
+    if cluster_name == spot.SPOT_CONTROLLER_NAME:
+        if down:
+            raise ValueError('Using autodown (rather than autostop) is not '
+                             'supported for the spot controller. Pass '
+                             '`down=False` or omit it instead.')
+        if (idle_minutes_to_autostop is not None and idle_minutes_to_autostop !=
path: sky/core.py
text: @user1:
The second part of the if statement `idle_minutes_to_autostop != spot.SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP`, can we just completely disallow the user to specify idle stop time even if it is the right time?

@author:
Good catch, done.

diff_hunk: @@ -149,7 +163,10 @@ def start(
 
     Raises:
         ValueError: the specified cluster does not exist; or if ``down`` is set
-          to True but ``idle_minutes_to_autostop`` is None.
+          to True but ``idle_minutes_to_autostop`` is None; or if the specified
path: sky/core.py
text: @user1:
The decision boundrary is getting rather complex here. Possible to simplify?

@author:
Yes, agreed it is complex. We could show "argument values are invalid" instead, but I slightly prefer being more precise in this case.

@user1:
In that case, maybe split this into bullet points, i.e.

1) specified cluster does not exists

2) `idle_minutes_to_autostop` is None and `down`=True

etc.

@author:
Good call, done.