Make sky start on the spot controller aware of autostop.

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.'

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' calling: {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).

sky/core.py

@@ -89,6 +89,19 @@ 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:
+            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)
@@ -148,8 +161,12 @@ def start(
             Useful for upgrading SkyPilot runtime.
 
     Raises:
-        ValueError: the specified cluster does not exist; or if ``down`` is set
-          to True but ``idle_minutes_to_autostop`` is None.
+        ValueError: argument values are invalid: (1) the specified cluster does
+          not exist; (2) if ``down`` is set to True but
+          ``idle_minutes_to_autostop`` is None; (3) if the specified cluster is
+          the managed spot controller, and either ``idle_minutes_to_autostop``
+          is not None 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.
@@ -612,9 +629,7 @@ def spot_queue(refresh: bool) -> List[Dict[str, Any]]:
               'Restarting controller for latest status...'
               f'{colorama.Style.RESET_ALL}')
 
-        handle = _start(spot.SPOT_CONTROLLER_NAME,
-                        idle_minutes_to_autostop=spot.
-                        SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP)
+        handle = _start(spot.SPOT_CONTROLLER_NAME)
 
     if handle is None or handle.head_ip is None:
         raise exceptions.ClusterNotUpError('Spot controller is not up.')
@@ -709,7 +724,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:

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)
 

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*'])