Repository: skypilot-org/skypilot
Pull request: Add core programmatic API align with CLI
Closes #871 and #980. Also, makes #972 easier.

Our current implementation has many functionalities in the `cli.py`, which makes python API users have to use subprocess to do the cluster, jobs, and storage management, and cannot handle easily handle the return values (needs to be done with string process).

This PR currently just has the skeleton of SDK and APIs. The implementation will be filled later. Please take a look at the API hierarchy and let me know if you find it not appropriate. : )

~~I am not sure whether the `stop`, `start`, `down`, `autostop` should be placed under `sky` or `sky.sdk`. If placed under `sky.sdk`, it is a bit weird to only have `launch` and `exec` under `sky`.~~

I updated the hierarchy of the APIs based on the [suggestion](https://github.com/sky-proj/sky/pull/978#issuecomment-1186033792) of @romilbhardwaj.

### Other changes
1. Added a smoke test for the `sky logs`.

### Hierarchy of  the APIs:
**sky**
  - sky.launch
  - sky.exec
  - sky.stop
  - sky.start
  - sky.down
  - sky.autostop
  - sky.status
  - sky.queue
  - sky.cancel
  - sky.spot_status
  - sky.spot_cancel
  - sky.get_storage
  - sky.delete_storage

### TODOs
- [ ] Add docs for python API (#1046 )

### Tested:
#### CLIs
- [x] `sky launch -c min ./example/minimal.yaml`
- [x] `sky stop min`
- [x] `sky start min` 
- [x] `sky autostop min`
- [x] `sky autostop --cancel min`
- [x] `sky queue`
- [x] `sky spot launch -n test-queue 'sleep 10000; echo hi'`
- [x] `sky spot status`
- [x] `sky spot cancel -n test-queue`
- [x] `sky logs test-multi-echo-zhwu-164b 1 2 3 4 -s`
- [x] `sky logs test-multi-echo-zhwu-164b 3 -s`
- [x] `sky logs test-multi-echo-zhwu-164b 3`
- [x] `sky logs test-multi-echo-zhwu-164b 1 2 3 4` fails
- [x] `tests/run_smoke_test.sh` (Failed TPU VM as #962)


Problem statement:
Move the functionality of cli to sdk
We need to move the functionality of the current CLI to SDK so that we can align the CLI and programmatic APIs. One example of the benefit is we can replace the `subprocess` call of the CLI commands with SDK, e.g. `subprocess.run('sky status')` in the cloud_vm_ray_backend.

[API] `sky.launch()` should not expose `is_spot_controller_task`
(Discovered by @pounde)

This argument is for system-internal usage and thus should be prefixed with `_` (one solution, but other ways of making it not exposed also work).

Hints from issue discussion:
We can have an `execution.launch_impl`(or a better name) method with the same API as the current `sky.launch`, and let all internal call use that method. `sky.launch` and `execution.launch` can be a wrapper of that method without the `is_spot_controller_task` in the param list.

Commit message: Fix test

Patch to review:
diff --git a/sky/__init__.py b/sky/__init__.py
index 64db590a23f..cd9f05482bc 100644
--- a/sky/__init__.py
+++ b/sky/__init__.py
@@ -7,11 +7,15 @@
 from sky import clouds
 from sky.clouds.service_catalog import list_accelerators
 from sky.dag import Dag, DagContext
-from sky.execution import launch, exec  # pylint: disable=redefined-builtin
+from sky.execution import launch, exec, spot_launch  # pylint: disable=redefined-builtin
 from sky.resources import Resources
 from sky.task import Task
 from sky.optimizer import Optimizer, OptimizeTarget
 from sky.data import Storage, StoreType
+from sky.global_user_state import ClusterStatus
+from sky.skylet.job_lib import JobStatus
+from sky.core import (status, start, stop, down, autostop, queue, cancel,
+                      spot_status, spot_cancel, storage_ls, storage_delete)
 
 __root_dir__ = os.path.dirname(os.path.abspath(__file__))
 
@@ -35,10 +39,25 @@
     'Task',
     'backends',
     'benchmark',
-    'launch',
-    'exec',
     'list_accelerators',
     '__root_dir__',
     'Storage',
     'StoreType',
+    'ClusterStatus',
+    'JobStatus',
+    # APIs
+    'launch',
+    'exec',
+    'spot_launch',
+    'status',
+    'start',
+    'stop',
+    'down',
+    'autostop',
+    'queue',
+    'cancel',
+    'spot_status',
+    'spot_cancel',
+    'storage_ls',
+    'storage_delete',
 ]
diff --git a/sky/backends/backend.py b/sky/backends/backend.py
index ebc13e88064..96660bf4f5b 100644
--- a/sky/backends/backend.py
+++ b/sky/backends/backend.py
@@ -94,8 +94,8 @@ def teardown_ephemeral_storage(self, task: 'task_lib.Task') -> None:
     def teardown(self,
                  handle: ResourceHandle,
                  terminate: bool,
-                 purge: bool = False) -> bool:
-        return self._teardown(handle, terminate, purge)
+                 purge: bool = False):
+        self._teardown(handle, terminate, purge)
 
     def register_info(self, **kwargs) -> None:
         """Register backend-specific information."""
@@ -138,5 +138,5 @@ def _teardown_ephemeral_storage(self, task: 'task_lib.Task') -> None:
     def _teardown(self,
                   handle: ResourceHandle,
                   terminate: bool,
-                  purge: bool = False) -> bool:
+                  purge: bool = False):
         raise NotImplementedError
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 05989f33323..fcc82d01b70 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -827,7 +827,7 @@ def ssh_credential_from_yaml(cluster_yaml: str) -> Tuple[str, str, str]:
     return ssh_user, ssh_private_key, ssh_control_name
 
 
-def parallel_data_transfer_to_nodes(
+def parallel_data_transfer_with_nodes(
     runners: List[command_runner.SSHCommandRunner],
     source: str,
     target: str,
@@ -876,9 +876,10 @@ def _sync_node(runner: 'command_runner.SSHCommandRunner') -> None:
         if run_rsync:
             # TODO(zhwu): Optimize for large amount of files.
             # zip / transfer/ unzip
-            runner.rsync_up(
+            runner.rsync(
                 source=source,
                 target=target,
+                up=True,
                 log_path=log_path,
                 stream_logs=stream_logs,
             )
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 9e947ee4c7d..1ec3636be2a 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -15,7 +15,6 @@
 import typing
 from typing import Dict, List, Optional, Tuple, Union
 
-import click
 import colorama
 import filelock
 
@@ -39,10 +38,10 @@
 from sky.skylet import autostop_lib
 from sky.skylet import job_lib
 from sky.skylet import log_lib
-from sky.skylet.utils import log_utils
 from sky.usage import usage_lib
 from sky.utils import common_utils
 from sky.utils import command_runner
+from sky.utils import log_utils
 from sky.utils import subprocess_utils
 from sky.utils import timeline
 from sky.utils import ux_utils
@@ -168,7 +167,7 @@ def add_prologue(self,
             import ray.util as ray_util
 
             from sky.skylet import job_lib
-            from sky.skylet.utils import log_utils
+            from sky.utils import log_utils
 
             SKY_REMOTE_WORKDIR = {log_lib.SKY_REMOTE_WORKDIR!r}
             job_lib.set_status({job_id!r}, job_lib.JobStatus.PENDING)
@@ -1705,9 +1704,10 @@ def _sync_workdir(self, handle: ResourceHandle, workdir: Path) -> None:
             ip_list, *ssh_credentials)
 
         def _sync_workdir_node(runner: command_runner.SSHCommandRunner) -> None:
-            runner.rsync_up(
+            runner.rsync(
                 source=workdir,
                 target=SKY_REMOTE_WORKDIR,
+                up=True,
                 log_path=log_path,
                 stream_logs=False,
             )
@@ -1760,9 +1760,10 @@ def _setup(self, handle: ResourceHandle, task: task_lib.Task) -> None:
                 ip_list, *ssh_credentials)
 
             def _setup_node(runner: command_runner.SSHCommandRunner) -> int:
-                runner.rsync_up(source=setup_sh_path,
-                                target=f'/tmp/{setup_file}',
-                                stream_logs=False)
+                runner.rsync(source=setup_sh_path,
+                             target=f'/tmp/{setup_file}',
+                             up=True,
+                             stream_logs=False)
                 # Need this `-i` option to make sure `source ~/.bashrc` work
                 cmd = f'/bin/bash -i /tmp/{setup_file} 2>&1'
                 returncode = runner.run(
@@ -1807,9 +1808,10 @@ def _exec_code_on_head(
             # We choose to sync code + exec, because the alternative of 'ray
             # submit' may not work as it may use system python (python2) to
             # execute the script.  Happens for AWS.
-            runner.rsync_up(source=fp.name,
-                            target=script_path,
-                            stream_logs=False)
+            runner.rsync(source=fp.name,
+                         target=script_path,
+                         up=True,
+                         stream_logs=False)
         remote_log_dir = self.log_dir
         remote_log_path = os.path.join(remote_log_dir, 'run.log')
 
@@ -1888,9 +1890,10 @@ def _setup_and_create_job_cmd_on_local_head(
             remote_run_file = f'/tmp/{run_file}'
             # We choose to sync code + exec, so that Ray job submission API will
             # work for the multitenant case.
-            runner.rsync_up(source=fp.name,
-                            target=remote_run_file,
-                            stream_logs=False)
+            runner.rsync(source=fp.name,
+                         target=remote_run_file,
+                         up=True,
+                         stream_logs=False)
         runner.run(f'mkdir -p {remote_log_dir}; chmod a+rwx {remote_run_file}',
                    stream_logs=False)
         switch_user_cmd = job_lib.make_job_command_with_user_switching(
@@ -1991,7 +1994,7 @@ def _teardown_ephemeral_storage(self, task: task_lib.Task) -> None:
     def _teardown(self,
                   handle: ResourceHandle,
                   terminate: bool,
-                  purge: bool = False) -> bool:
+                  purge: bool = False):
         cluster_name = handle.cluster_name
         lock_path = os.path.expanduser(
             backend_utils.CLUSTER_STATUS_LOCK_PATH.format(cluster_name))
@@ -2006,19 +2009,20 @@ def _teardown(self,
                 success = self.teardown_no_lock(handle, terminate, purge)
             if success and terminate:
                 os.remove(lock_path)
-            return success
-        except filelock.Timeout:
-            logger.error(f'Cluster {cluster_name} is locked by {lock_path}. '
-                         'Check to see if it is still being launched.')
-        return False
+        except filelock.Timeout as e:
+            raise RuntimeError(
+                f'Cluster {cluster_name!r} is locked by {lock_path}. '
+                'Check to see if it is still being launched.') from e
 
     # --- CloudVMRayBackend Specific APIs ---
 
-    def get_job_status(self,
-                       handle: ResourceHandle,
-                       job_id: Optional[int] = None,
-                       stream_logs: bool = True) -> Optional[job_lib.JobStatus]:
-        code = job_lib.JobLibCodeGen.get_job_status(job_id)
+    def get_job_status(
+        self,
+        handle: ResourceHandle,
+        job_ids: Optional[List[str]] = None,
+        stream_logs: bool = True
+    ) -> Dict[Optional[str], Optional[job_lib.JobStatus]]:
+        code = job_lib.JobLibCodeGen.get_job_status(job_ids)
         # All error messages should have been redirected to stdout.
         returncode, stdout, _ = self.run_on_head(handle,
                                                  code,
@@ -2026,11 +2030,8 @@ def get_job_status(self,
                                                  require_outputs=True)
         subprocess_utils.handle_returncode(returncode, code,
                                            'Failed to get job status.', stdout)
-        result = stdout.strip()
-        if result.endswith(' None'):
-            # "Job 9876 None"
-            return None
-        return job_lib.JobStatus(result.split(' ')[-1])
+        statuses = job_lib.load_statuses_json(stdout)
+        return statuses
 
     def cancel_jobs(self, handle: ResourceHandle, jobs: Optional[List[int]]):
         job_owner = onprem_utils.get_job_owner(handle.cluster_yaml)
@@ -2046,7 +2047,12 @@ def cancel_jobs(self, handle: ResourceHandle, jobs: Optional[List[int]]):
             f'Failed to cancel jobs on cluster {handle.cluster_name}.', stdout)
 
     def sync_down_logs(self, handle: ResourceHandle,
-                       job_ids: Optional[str]) -> None:
+                       job_ids: Optional[str]) -> Dict[str, str]:
+        """Sync down logs for the given job_ids.
+
+        Returns:
+            A dictionary mapping job_id to log path.
+        """
         code = job_lib.JobLibCodeGen.get_log_path_with_globbing(job_ids)
         returncode, log_dirs, stderr = self.run_on_head(handle,
                                                         code,
@@ -2054,18 +2060,18 @@ def sync_down_logs(self, handle: ResourceHandle,
                                                         require_outputs=True)
         subprocess_utils.handle_returncode(returncode, code,
                                            'Failed to sync logs.', stderr)
-        log_dirs = ast.literal_eval(log_dirs)
-        if not log_dirs or len(log_dirs) == 0:
+        log_dirs = json.loads(log_dirs)
+        if not log_dirs:
             logger.info(f'{colorama.Fore.YELLOW}'
                         'No matching log directories found'
                         f'{colorama.Style.RESET_ALL}')
             return
 
-        job_ids = [log_dir[0] for log_dir in log_dirs]
+        job_ids = list(log_dirs.keys())
+        remote_log_dirs = list(log_dirs.values())
         local_log_dirs = [
-            os.path.expanduser(log_dir[1]) for log_dir in log_dirs
+            os.path.expanduser(log_dir) for log_dir in remote_log_dirs
         ]
-        remote_log_dirs = [log_dir[1] for log_dir in log_dirs]
 
         style = colorama.Style
         fore = colorama.Fore
@@ -2073,42 +2079,42 @@ def sync_down_logs(self, handle: ResourceHandle,
             logger.info(f'{fore.CYAN}Job {job_id} logs: {log_dir}'
                         f'{style.RESET_ALL}')
 
-        ips = backend_utils.get_node_ips(handle.cluster_yaml,
-                                         handle.launched_nodes,
-                                         handle=handle)
-
-        def rsync_down(ip: str, local_log_dir: str,
-                       remote_log_dir: str) -> None:
-            from ray.autoscaler import sdk  # pylint: disable=import-outside-toplevel
-            sdk.rsync(
-                handle.cluster_yaml,
-                source=f'{remote_log_dir}/*',
-                target=f'{local_log_dir}',
-                down=True,
-                ip_address=ip,
-                use_internal_ip=False,
-                should_bootstrap=False,
-            )
+        ip_list = backend_utils.get_node_ips(handle.cluster_yaml,
+                                             handle.launched_nodes,
+                                             handle=handle)
+        ssh_credentials = backend_utils.ssh_credential_from_yaml(
+            handle.cluster_yaml)
+        runners = command_runner.SSHCommandRunner.make_runner_list(
+            ip_list, *ssh_credentials)
 
-        # Call the ray sdk to rsync the logs back to local.
-        for job_id, local_log_dir, remote_log_dir in zip(
-                job_ids, local_log_dirs, remote_log_dirs):
-            for i, ip in enumerate(ips):
-                try:
-                    # Disable the output of rsync.
-                    with backend_utils.subpress_output():
-                        rsync_down(ip, local_log_dir, remote_log_dir)
-                    logger.info(
-                        f'{fore.CYAN}Job {job_id} logs: downloaded from '
-                        f'node-{i} ({ip}){style.RESET_ALL}')
-                except click.exceptions.ClickException as e:
+        def _rsync_down(args) -> None:
+            """Rsync down logs from remote nodes.
+
+            Args:
+                args: A tuple of (runner, local_log_dir, remote_log_dir)
+            """
+            (runner, local_log_dir, remote_log_dir) = args
+            try:
+                os.makedirs(local_log_dir, exist_ok=True)
+                runner.rsync(
+                    source=f'{remote_log_dir}/*',
+                    target=local_log_dir,
+                    up=False,
+                    stream_logs=False,
+                )
+            except exceptions.CommandError as e:
+                if e.returncode == exceptions.RSYNC_FILE_NOT_FOUND_CODE:
                     # Raised by rsync_down. Remote log dir may not exist, since
                     # the job can be run on some part of the nodes.
-                    if 'SSH command failed' in str(e):
-                        logger.debug(
-                            f'node-{i} ({ip}) does not have the tasks/*.')
-                    else:
-                        raise e
+                    logger.debug(f'{runner.ip} does not have the tasks/*.')
+                else:
+                    raise
+
+        parallel_args = [[runner, *item]
+                         for item in zip(local_log_dirs, remote_log_dirs)
+                         for runner in runners]
+        subprocess_utils.run_in_parallel(_rsync_down, parallel_args)
+        return log_dirs
 
     def tail_logs(self,
                   handle: ResourceHandle,
@@ -2562,7 +2568,7 @@ def _execute_file_mounts(self, handle: ResourceHandle,
                     mkdir_for_wrapped_dst = f'mkdir -p {wrapped_dst}'
 
                 # TODO(mluo): Fix method so that mkdir and rsync run together
-                backend_utils.parallel_data_transfer_to_nodes(
+                backend_utils.parallel_data_transfer_with_nodes(
                     runners,
                     source=src,
                     target=wrapped_dst,
@@ -2595,7 +2601,7 @@ def _execute_file_mounts(self, handle: ResourceHandle,
             ]
             command = ' && '.join(download_target_commands)
             # dst is only used for message printing.
-            backend_utils.parallel_data_transfer_to_nodes(
+            backend_utils.parallel_data_transfer_with_nodes(
                 runners,
                 source=src,
                 target=dst,
@@ -2667,7 +2673,7 @@ def _execute_storage_mounts(self, handle: ResourceHandle,
             mount_cmd = store.mount_command(dst)
             src_print = (storage_obj.source
                          if storage_obj.source else storage_obj.name)
-            backend_utils.parallel_data_transfer_to_nodes(
+            backend_utils.parallel_data_transfer_with_nodes(
                 runners,
                 source=src_print,
                 target=dst,
diff --git a/sky/backends/local_docker_backend.py b/sky/backends/local_docker_backend.py
index cfb384cd8e8..cb00582830f 100644
--- a/sky/backends/local_docker_backend.py
+++ b/sky/backends/local_docker_backend.py
@@ -293,7 +293,7 @@ def _post_execute(self, handle: ResourceHandle, teardown: bool) -> None:
     def _teardown(self,
                   handle: ResourceHandle,
                   terminate: bool,
-                  purge: bool = False) -> bool:
+                  purge: bool = False):
         """Teardown kills the container."""
         del purge  # Unused.
         if not terminate:
@@ -309,7 +309,6 @@ def _teardown(self,
             container.remove(force=True)
         cluster_name = handle.get_cluster_name()
         global_user_state.remove_cluster(cluster_name, terminate=True)
-        return True
 
     # --- Utilities ---
 
diff --git a/sky/backends/onprem_utils.py b/sky/backends/onprem_utils.py
index 61d4c5b7776..f1454968d02 100644
--- a/sky/backends/onprem_utils.py
+++ b/sky/backends/onprem_utils.py
@@ -204,7 +204,7 @@ def _setup_local_worker(runner: command_runner.SSHCommandRunner):
                     failure_message=f'Failed to run {mkdir_dst} on remote.')
                 if os.path.isdir(src):
                     src = os.path.join(src, '')
-                runner.rsync_up(source=src, target=dst, stream_logs=False)
+                runner.rsync(source=src, target=dst, up=True, stream_logs=False)
 
             setup_cmd = f'/bin/bash -i /tmp/{setup_file} 2>&1'
             rc, stdout, _ = runner.run(setup_cmd,
@@ -336,9 +336,10 @@ def _gather_cluster_accelerators(
             fp.write(code)
             fp.flush()
             runner.run(f'mkdir -p {remote_resource_dir}', stream_logs=False)
-            runner.rsync_up(source=fp.name,
-                            target=_SKY_GET_ACCELERATORS_SCRIPT_PATH,
-                            stream_logs=False)
+            runner.rsync(source=fp.name,
+                         target=_SKY_GET_ACCELERATORS_SCRIPT_PATH,
+                         up=True,
+                         stream_logs=False)
             output = run_command_and_handle_ssh_failure(
                 runner,
                 f'python3 {_SKY_GET_ACCELERATORS_SCRIPT_PATH}',
@@ -454,15 +455,17 @@ def _start_ray_workers(
                 f'Failed to launch ray on worker node {runner.ip}.')
 
             # Connecting ray dashboard with worker node.
-            runner.rsync_up(source=ssh_key,
-                            target=remote_ssh_key,
-                            stream_logs=False)
+            runner.rsync(source=ssh_key,
+                         target=remote_ssh_key,
+                         up=True,
+                         stream_logs=False)
             with tempfile.NamedTemporaryFile('w', prefix='sky_app_') as fp:
                 fp.write(port_cmd)
                 fp.flush()
-                runner.rsync_up(source=fp.name,
-                                target=dashboard_remote_path,
-                                stream_logs=False)
+                runner.rsync(source=fp.name,
+                             target=dashboard_remote_path,
+                             up=True,
+                             stream_logs=False)
             # Kill existing dashboard connection and launch new one
             run_command_and_handle_ssh_failure(
                 runner, f'chmod a+rwx {dashboard_remote_path};'
diff --git a/sky/benchmark/benchmark_utils.py b/sky/benchmark/benchmark_utils.py
index cb5957c9b89..c91a05a58a9 100644
--- a/sky/benchmark/benchmark_utils.py
+++ b/sky/benchmark/benchmark_utils.py
@@ -29,7 +29,7 @@
 from sky.benchmark import benchmark_state
 from sky.skylet import job_lib
 from sky.skylet import log_lib
-from sky.skylet.utils import log_utils
+from sky.utils import log_utils
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -297,8 +297,8 @@ def _update_benchmark_result(benchmark_result: Dict[str, Any]) -> Optional[str]:
             # NOTE: The id of the benchmarking job must be 1.
             # TODO(woosuk): Handle exceptions.
             job_status = backend.get_job_status(handle,
-                                                job_id=1,
-                                                stream_logs=False)
+                                                job_ids=['1'],
+                                                stream_logs=False)['1']
 
     # Update the benchmark status.
     if (cluster_status == global_user_state.ClusterStatus.INIT or
diff --git a/sky/cli.py b/sky/cli.py
index baf31f0d0ec..cecc24f0a18 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -33,7 +33,6 @@
 import os
 import shlex
 import sys
-import tempfile
 import textwrap
 import typing
 from typing import Any, Dict, List, Optional, Tuple
@@ -47,22 +46,22 @@
 from sky import backends
 from sky import check as sky_check
 from sky import clouds
-from sky import data
+from sky import constants
+from sky import exceptions
 from sky import global_user_state
 from sky import sky_logging
 from sky import spot as spot_lib
+from sky import core
 from sky.backends import backend_utils
 from sky.backends import onprem_utils
 from sky.benchmark import benchmark_state
 from sky.benchmark import benchmark_utils
 from sky.clouds import service_catalog
-from sky.data import data_utils
-from sky.data.storage import StoreType
+from sky.data import storage_utils
 from sky.skylet import job_lib
-from sky.skylet.utils import log_utils
+from sky.utils import log_utils
 from sky.utils import common_utils
 from sky.utils import command_runner
-from sky.utils import env_options
 from sky.utils import subprocess_utils
 from sky.utils import timeline
 from sky.utils import ux_utils
@@ -95,11 +94,6 @@
                              use_spot=False),
 }
 
-# TODO(mluo): Make explicit `sky launch -c <name> ''` optional.
-_UNINITIALIZED_CLUSTER_MESSAGE = (
-    'Found uninitialized local cluster {cluster}. Run this '
-    'command to initialize it locally: sky launch -c {cluster} \'\'')
-
 
 def _get_glob_clusters(clusters: List[str]) -> List[str]:
     """Returns a list of clusters that match the glob pattern."""
@@ -109,7 +103,8 @@ def _get_glob_clusters(clusters: List[str]) -> List[str]:
         if len(glob_cluster) == 0:
             if onprem_utils.check_if_local_cloud(cluster):
                 click.echo(
-                    _UNINITIALIZED_CLUSTER_MESSAGE.format(cluster=cluster))
+                    constants.UNINITIALIZED_ONPREM_CLUSTER_MESSAGE.format(
+                        cluster=cluster))
             else:
                 click.echo(f'Cluster {cluster} not found.')
         glob_clusters.extend(glob_cluster)
@@ -252,6 +247,29 @@ def _parse_env_var(env_var: str) -> Tuple[str, str]:
         type=str,
         help=('The region to use. If specified, overrides the '
               '"resources.region" config. Passing "none" resets the config.')),
+    click.option(
+        '--gpus',
+        required=False,
+        type=str,
+        help=
+        ('Type and number of GPUs to use. Example values: '
+         '"V100:8", "V100" (short for a count of 1), or "V100:0.5" '
+         '(fractional counts are supported by the scheduling framework). '
+         'If a new cluster is being launched by this command, this is the '
+         'resources to provision. If an existing cluster is being reused, this'
+         ' is seen as the task demand, which must fit the cluster\'s total '
+         'resources and is used for scheduling the task. '
+         'Overrides the "accelerators" '
+         'config in the YAML if both are supplied. '
+         'Passing "none" resets the config.')),
+    click.option(
+        '--instance-type',
+        required=False,
+        type=str,
+        help=('The instance type to use. If specified, overrides the '
+              '"resources.instance_type" config. Passing "none" resets the '
+              'config.'),
+    ),
     click.option(
         '--num-nodes',
         required=False,
@@ -321,6 +339,7 @@ def _add_options(func):
 def _parse_override_params(cloud: Optional[str] = None,
                            region: Optional[str] = None,
                            gpus: Optional[str] = None,
+                           instance_type: Optional[str] = None,
                            use_spot: Optional[bool] = None,
                            image_id: Optional[str] = None,
                            disk_size: Optional[int] = None) -> Dict[str, Any]:
@@ -341,6 +360,11 @@ def _parse_override_params(cloud: Optional[str] = None,
             override_params['accelerators'] = None
         else:
             override_params['accelerators'] = gpus
+    if instance_type is not None:
+        if instance_type.lower() == 'none':
+            override_params['instance_type'] = None
+        else:
+            override_params['instance_type'] = instance_type
     if use_spot is not None:
         override_params['use_spot'] = use_spot
     if image_id is not None:
@@ -671,6 +695,7 @@ def _make_dag_from_entrypoint_with_overrides(
     cloud: Optional[str] = None,
     region: Optional[str] = None,
     gpus: Optional[str] = None,
+    instance_type: Optional[str] = None,
     num_nodes: Optional[int] = None,
     use_spot: Optional[bool] = None,
     image_id: Optional[str] = None,
@@ -712,6 +737,7 @@ def _make_dag_from_entrypoint_with_overrides(
         override_params = _parse_override_params(cloud=cloud,
                                                  region=region,
                                                  gpus=gpus,
+                                                 instance_type=instance_type,
                                                  use_spot=use_spot,
                                                  image_id=image_id,
                                                  disk_size=disk_size)
@@ -860,11 +886,12 @@ def launch(
     workdir: Optional[str],
     cloud: Optional[str],
     region: Optional[str],
+    gpus: Optional[str],
+    instance_type: Optional[str],
     num_nodes: Optional[int],
     use_spot: Optional[bool],
     image_id: Optional[str],
     env: List[Dict[str, str]],
-    gpus: Optional[str],
     disk_size: Optional[int],
     idle_minutes_to_autostop: Optional[int],
     retry_until_up: bool,
@@ -891,6 +918,7 @@ def launch(
         cloud=cloud,
         region=region,
         gpus=gpus,
+        instance_type=instance_type,
         num_nodes=num_nodes,
         use_spot=use_spot,
         image_id=image_id,
@@ -938,11 +966,12 @@ def exec(
     cloud: Optional[str],
     region: Optional[str],
     workdir: Optional[str],
+    gpus: Optional[str],
+    instance_type: Optional[str],
     num_nodes: Optional[int],
     use_spot: Optional[bool],
     image_id: Optional[str],
     env: List[Dict[str, str]],
-    gpus: Optional[str],
 ):
     """Execute a task or a command on a cluster (skip setup).
 
@@ -1006,7 +1035,7 @@ def exec(
     if handle is None:
         if onprem_utils.check_if_local_cloud(cluster):
             raise click.BadParameter(
-                _UNINITIALIZED_CLUSTER_MESSAGE.format(cluster=cluster))
+                constants.UNINITIALIZED_CLUSTER_MESSAGE.format(cluster=cluster))
         raise click.BadParameter(f'Cluster {cluster!r} not found. '
                                  'Use `sky launch` to provision first.')
     backend = backend_utils.get_backend_from_handle(handle)
@@ -1019,6 +1048,7 @@ def exec(
         cloud=cloud,
         region=region,
         gpus=gpus,
+        instance_type=instance_type,
         use_spot=use_spot,
         image_id=image_id,
         num_nodes=num_nodes,
@@ -1064,8 +1094,9 @@ def status(all: bool, refresh: bool):  # pylint: disable=redefined-builtin
     - STOPPED: The cluster is stopped and the storage is persisted. Use
       ``sky start`` to restart the cluster.
     """
+    cluster_records = core.status(all=all, refresh=refresh)
     local_clusters = onprem_utils.check_and_get_local_clusters()
-    status_utils.show_status_table(all, refresh)
+    status_utils.show_status_table(cluster_records, all)
     status_utils.show_local_status_table(local_clusters)
 
 
@@ -1087,12 +1118,6 @@ def status(all: bool, refresh: bool):  # pylint: disable=redefined-builtin
 def queue(clusters: Tuple[str], skip_finished: bool, all_users: bool):
     """Show the job queue for cluster(s)."""
     click.secho('Fetching and parsing job queue...', fg='yellow')
-    all_jobs = not skip_finished
-    username = getpass.getuser()
-    if all_users:
-        username = None
-    code = job_lib.JobLibCodeGen.show_jobs(username, all_jobs)
-
     show_local_clusters = False
     if clusters:
         clusters = _get_glob_clusters(clusters)
@@ -1101,25 +1126,21 @@ def queue(clusters: Tuple[str], skip_finished: bool, all_users: bool):
         cluster_infos = global_user_state.get_clusters()
         clusters = [c['name'] for c in cluster_infos]
 
-    local_clusters = onprem_utils.check_and_get_local_clusters()
-
     unsupported_clusters = []
     for cluster in clusters:
-        cluster_status, handle = backend_utils.refresh_cluster_status_handle(
-            cluster)
-        backend = backend_utils.get_backend_from_handle(handle)
-        if isinstance(backend, backends.LocalDockerBackend):
-            # LocalDockerBackend does not support job queues
+        try:
+            job_table = core.queue(cluster, skip_finished, all_users)
+        except exceptions.NotSupportedError as e:
             unsupported_clusters.append(cluster)
+            click.echo(str(e))
             continue
-        if cluster_status != global_user_state.ClusterStatus.UP:
-            click.secho(
-                f'Cluster {cluster} is not up (status: {cluster_status.value});'
-                ' skipped.',
-                fg='yellow')
+        except (RuntimeError, exceptions.ClusterNotUpError) as e:
+            click.echo(str(e))
             continue
-        _show_job_queue_on_cluster(cluster, handle, backend, code)
+        job_table = job_lib.format_job_queue(job_table)
+        click.echo(f'{job_table}')
 
+    local_clusters = onprem_utils.check_and_get_local_clusters()
     for local_cluster in local_clusters:
         if local_cluster not in clusters and show_local_clusters:
             click.secho(
@@ -1134,24 +1155,6 @@ def queue(clusters: Tuple[str], skip_finished: bool, all_users: bool):
             fg='yellow')
 
 
-def _show_job_queue_on_cluster(cluster: str, handle: Optional[Any],
-                               backend: 'backend_lib.Backend', code: str):
-    click.echo(f'\nJob queue of cluster {cluster}')
-    if handle.head_ip is None:
-        click.echo(
-            f'Cluster {cluster} has been stopped or not properly set up. '
-            'Please re-launch it with `sky launch` to view the job queue.')
-        return
-
-    returncode, job_table, stderr = backend.run_on_head(handle,
-                                                        code,
-                                                        require_outputs=True)
-    if returncode != 0:
-        click.echo(stderr)
-        click.secho(f'Failed to get job queue on cluster {cluster}.', fg='red')
-    click.echo(f'{job_table}')
-
-
 @cli.command()
 @click.option(
     '--sync-down',
@@ -1167,61 +1170,52 @@ def _show_job_queue_on_cluster(cluster: str, handle: Optional[Any],
     help=('If specified, do not show logs but exit with a status code for the '
           'job\'s status: 0 for succeeded, or 1 for all other statuses.'))
 @click.argument('cluster', required=True, type=str)
-@click.argument('job_id', required=False, type=str)
+@click.argument('job_ids', type=str, nargs=-1)
 # TODO(zhwu): support logs by job name
 @usage_lib.entrypoint
-def logs(cluster: str, job_id: Optional[str], sync_down: bool, status: bool):  # pylint: disable=redefined-outer-name
+def logs(cluster: str, job_ids: Tuple[str], sync_down: bool, status: bool):  # pylint: disable=redefined-outer-name
     """Tail the log of a job.
 
     If JOB_ID is not provided, tails the logs of the last job on the cluster.
     """
-    cluster_name = cluster
-    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
-        cluster_name)
-    if handle is None:
-        if onprem_utils.check_if_local_cloud(cluster):
-            raise click.BadParameter(
-                _UNINITIALIZED_CLUSTER_MESSAGE.format(cluster=cluster_name))
-        raise click.BadParameter(f'Cluster \'{cluster_name}\' not found'
-                                 ' (see `sky status`).')
-    backend = backend_utils.get_backend_from_handle(handle)
-    if isinstance(backend, backends.LocalDockerBackend):
-        raise click.UsageError('Logs are not available with '
-                               'LocalDockerBackend.')
-    if cluster_status != global_user_state.ClusterStatus.UP:
-        click.secho(
-            f'Cluster {cluster_name} is not up '
-            f'(status: {cluster_status.value}).',
-            fg='yellow')
-        return
-
     if sync_down and status:
         raise click.UsageError(
-            'Both --sync_down and --status are specified '
+            'Both sync_down and status are specified '
             '(ambiguous). To fix: specify at most one of them.')
 
+    if len(job_ids) > 1 and not sync_down:
+        raise click.UsageError(
+            f'Cannot stream logs of multiple jobs {job_ids}. '
+            'Set --sync_down to download them.')
+
+    job_ids = None if not job_ids else job_ids
+
     if sync_down:
-        click.secho('Syncing down logs to local...', fg='yellow')
-        backend.sync_down_logs(handle, job_id)
+        core.download_logs(cluster, job_ids)
         return
 
-    if job_id is not None and not job_id.isdigit():
-        click.secho(
-            'Only single job ID supported for streaming or status check, '
-            'consider using --sync_down to download logs for multiple jobs.',
-            fg='yellow')
-        return
-    job_id = int(job_id) if job_id is not None else job_id
+    assert job_ids is None or len(job_ids) <= 1, job_ids
+    job_id = None
+    if job_ids:
+        job_id = job_ids[0]
+        if not job_id.isdigit():
+            raise click.UsageError(f'Invalid job ID {job_id}. '
+                                   'Job ID must be integers.')
     if status:
-        job_status = backend.get_job_status(handle, job_id)
+        job_statuses = core.job_status(cluster, job_ids)
+        job_id = list(job_statuses.keys())[0]
+        job_status = list(job_statuses.values())[0]
+        job_status_str = job_status.value if job_status is not None else 'None'
+        click.echo(f'Job {job_id}: {job_status_str}')
         if job_status == job_lib.JobStatus.SUCCEEDED:
             sys.exit(0)
-        if job_status is None:
-            id_str = '' if job_id is None else str(job_id) + ' '
-            click.secho(f'Job {id_str}not found', fg='red')
-        sys.exit(1)
-    else:
-        backend.tail_logs(handle, job_id)
+        else:
+            if job_status is None:
+                id_str = '' if job_id is None else f'{job_id} '
+                click.secho(f'Job {id_str}not found', fg='red')
+            sys.exit(1)
+
+    core.tail_logs(cluster, job_id)
 
 
 @cli.command()
@@ -1236,45 +1230,10 @@ def logs(cluster: str, job_id: Optional[str], sync_down: bool, status: bool):  #
 @usage_lib.entrypoint
 def cancel(cluster: str, all: bool, jobs: List[int]):  # pylint: disable=redefined-builtin
     """Cancel job(s)."""
-    if len(jobs) == 0 and not all:
-        raise click.UsageError(
-            'sky cancel requires either a job id '
-            f'(see `sky queue {cluster} -s`) or the --all flag.')
-
-    backend_utils.check_cluster_name_not_reserved(
-        cluster, operation_str='Cancelling jobs')
-
-    # Check the status of the cluster.
-    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
-        cluster)
-    if handle is None:
-        if onprem_utils.check_if_local_cloud(cluster):
-            raise click.BadParameter(
-                _UNINITIALIZED_CLUSTER_MESSAGE.format(cluster=cluster))
-        raise click.BadParameter(f'Cluster {cluster!r} not found'
-                                 ' (see `sky status`).')
-    backend = backend_utils.get_backend_from_handle(handle)
-    if not isinstance(backend, backends.CloudVmRayBackend):
-        raise click.UsageError(
-            'Job cancelling is only supported for '
-            f'{backends.CloudVmRayBackend.NAME}, but cluster {cluster!r} '
-            f'is created by {backend.NAME}.')
-    if cluster_status != global_user_state.ClusterStatus.UP:
-        click.secho(
-            f'Cluster {cluster} is not up (status: {cluster_status.value}); '
-            'skipped.',
-            fg='yellow')
-        return
-
-    if all:
-        click.secho(f'Cancelling all jobs on cluster {cluster}...', fg='yellow')
-        jobs = None
-    else:
-        jobs_str = ', '.join(map(str, jobs))
-        click.secho(f'Cancelling jobs ({jobs_str}) on cluster {cluster}...',
-                    fg='yellow')
-
-    backend.cancel_jobs(handle, jobs)
+    try:
+        core.cancel(cluster, all, jobs)
+    except ValueError as e:
+        raise click.UsageError(str(e))
 
 
 @cli.command(cls=_DocumentedCodeCommand)
@@ -1410,6 +1369,20 @@ def autostop(
               default=False,
               required=False,
               help='Skip confirmation prompt.')
+@click.option(
+    '--idle-minutes-to-autostop',
+    '-i',
+    default=None,
+    type=int,
+    required=False,
+    help=('Automatically stop the cluster after this many minutes '
+          'of idleness, i.e., no running or pending jobs in the cluster\'s job '
+          'queue. Idleness starts counting after setup/file_mounts are done; '
+          'the clock gets reset whenever there are running/pending jobs in the '
+          'job queue. '
+          'Setting this flag is equivalent to '
+          'running ``sky launch -d ...`` and then ``sky autostop -i <minutes>``'
+          '. If not set, the cluster will not be auto-stopped.'))
 @click.option(
     '--retry-until-up',
     '-r',
@@ -1419,7 +1392,8 @@ def autostop(
     help=('Retry provisioning infinitely until the cluster is up, '
           'if we fail to start the cluster due to unavailability errors.'))
 @usage_lib.entrypoint
-def start(clusters: Tuple[str], yes: bool, retry_until_up: bool):
+def start(clusters: Tuple[str], yes: bool, idle_minutes_to_autostop: int,
+          retry_until_up: bool):
     """Restart cluster(s).
 
     If a cluster is previously stopped (status is STOPPED) or failed in
@@ -1513,7 +1487,10 @@ def start(clusters: Tuple[str], yes: bool, retry_until_up: bool):
             show_default=True)
 
     for name in to_start:
-        _start_cluster(name, retry_until_up=retry_until_up)
+        try:
+            core.start(name, idle_minutes_to_autostop, retry_until_up)
+        except exceptions.NotSupportedError as e:
+            click.echo(str(e))
         click.secho(f'Cluster {name} started.', fg='green')
 
 
@@ -1607,7 +1584,6 @@ def _terminate_or_stop_clusters(
         operation = f'{verb} auto-stop on'
 
     if len(names) > 0:
-        local_clusters = onprem_utils.check_and_get_local_clusters()
         reserved_clusters = [
             name for name in names
             if name in backend_utils.SKY_RESERVED_CLUSTER_NAMES
@@ -1618,6 +1594,7 @@ def _terminate_or_stop_clusters(
             if name not in backend_utils.SKY_RESERVED_CLUSTER_NAMES
         ]
         if not terminate:
+            local_clusters = onprem_utils.check_and_get_local_clusters()
             # Local clusters are allowed to `sky down`, but not
             # `sky start/stop`. `sky down` unregisters the local cluster
             # from sky.
@@ -1670,9 +1647,8 @@ def _terminate_or_stop_clusters(
             # controller is not in 'sky status'.  Cluster-not-found message
             # should've been printed by _get_glob_clusters() above.
             continue
-        clusters.append({'name': name, 'handle': handle})
-    usage_lib.messages.usage.update_cluster_name(
-        [cluster['name'] for cluster in clusters])
+        clusters.append(name)
+    usage_lib.messages.usage.update_cluster_name(clusters)
 
     if not clusters:
         click.echo('\nCluster(s) not found (tip: see `sky status`).')
@@ -1680,16 +1656,13 @@ def _terminate_or_stop_clusters(
 
     if not no_confirm and len(clusters) > 0:
         cluster_str = 'clusters' if len(clusters) > 1 else 'cluster'
-        cluster_list = ', '.join([r['name'] for r in clusters])
+        cluster_list = ', '.join(clusters)
         click.confirm(
             f'{operation} {len(clusters)} {cluster_str}: '
             f'{cluster_list}. Proceed?',
             default=True,
             abort=True,
             show_default=True)
-        # Add a blank line to separate the confirmation prompt from the
-        # progress bar.
-        click.echo()
 
     plural = 's' if len(clusters) > 1 else ''
     progress = rich_progress.Progress(transient=True,
@@ -1699,56 +1672,40 @@ def _terminate_or_stop_clusters(
         f'[bold cyan]{operation} {len(clusters)} cluster{plural}[/]',
         total=len(clusters))
 
-    def _terminate_or_stop(record):
-        name = record['name']
-        handle = record['handle']
-        backend = backend_utils.get_backend_from_handle(handle)
+    def _terminate_or_stop(name: str):
         success_progress = False
-        if (isinstance(backend, backends.CloudVmRayBackend) and
-                handle.launched_resources.use_spot and not terminate):
-            # Disable spot instances to be stopped.
-            # TODO(suquark): enable GCP+spot to be stopped in the future.
-            message = (
-                f'{colorama.Fore.YELLOW}Stopping cluster {name}... skipped.'
-                f'{colorama.Style.RESET_ALL}\n'
-                '  Stopping spot instances is not supported as the attached '
-                'disks will be lost.\n'
-                '  To terminate the cluster instead, run: '
-                f'{colorama.Style.BRIGHT}sky down {name}'
-                f'{colorama.Style.RESET_ALL}')
-        elif idle_minutes_to_autostop is not None:
-            (cluster_status,
-             handle) = backend_utils.refresh_cluster_status_handle(name)
-            if not isinstance(backend, backends.CloudVmRayBackend):
-                message = (f'{colorama.Fore.YELLOW}{operation} cluster '
-                           f'{name}... skipped{colorama.Style.RESET_ALL}'
-                           '\n  Auto-stopping is only supported by backend: '
-                           f'{backends.CloudVmRayBackend.NAME}')
-            else:
-                if cluster_status != global_user_state.ClusterStatus.UP:
-                    message = (
-                        f'{colorama.Fore.YELLOW}{operation} cluster '
-                        f'{name} (status: {cluster_status.value})... skipped'
-                        f'{colorama.Style.RESET_ALL}'
-                        '\n  Auto-stop can only be set/unset for '
-                        f'{global_user_state.ClusterStatus.UP.value} clusters.')
-                else:
-                    backend.set_autostop(handle, idle_minutes_to_autostop)
-                    message = (
-                        f'{colorama.Fore.GREEN}{operation} '
-                        f'cluster {name}...done{colorama.Style.RESET_ALL}')
-                    if idle_minutes_to_autostop >= 0:
-                        message += (
-                            f'\n  The cluster will be stopped after '
-                            f'{idle_minutes_to_autostop} minutes of idleness.'
-                            '\n  To cancel the autostop, run: '
-                            f'{colorama.Style.BRIGHT}'
-                            f'sky autostop {name} --cancel'
-                            f'{colorama.Style.RESET_ALL}')
-                    success_progress = True
+        if idle_minutes_to_autostop is not None:
+            try:
+                core.autostop(name, idle_minutes_to_autostop)
+            except (exceptions.NotSupportedError,
+                    exceptions.ClusterNotUpError) as e:
+                message = str(e)
+            else:  # no exception raised
+                success_progress = True
+                message = (f'{colorama.Fore.GREEN}{operation} '
+                           f'cluster {name!r}...done{colorama.Style.RESET_ALL}')
+                if idle_minutes_to_autostop >= 0:
+                    message += (
+                        f'\n  The cluster will be stopped after '
+                        f'{idle_minutes_to_autostop} minutes of idleness.'
+                        '\n  To cancel the autostop, run: '
+                        f'{colorama.Style.BRIGHT}'
+                        f'sky autostop {name} --cancel'
+                        f'{colorama.Style.RESET_ALL}')
         else:
-            success = backend.teardown(handle, terminate=terminate, purge=purge)
-            if success:
+            try:
+                if terminate:
+                    core.down(name, purge=purge)
+                else:
+                    core.stop(name, purge=purge)
+            except RuntimeError:
+                message = (
+                    f'{colorama.Fore.RED}{operation} cluster {name}...failed. '
+                    'Please check the logs and try again.'
+                    f'{colorama.Style.RESET_ALL}')
+            except exceptions.NotSupportedError as e:
+                message = str(e)
+            else:  # no exception raised
                 message = (
                     f'{colorama.Fore.GREEN}{operation} cluster {name}...done.'
                     f'{colorama.Style.RESET_ALL}')
@@ -1757,11 +1714,6 @@ def _terminate_or_stop(record):
                                 f'{colorama.Style.BRIGHT}sky start {name}'
                                 f'{colorama.Style.RESET_ALL}')
                 success_progress = True
-            else:
-                message = (
-                    f'{colorama.Fore.RED}{operation} cluster {name}...failed. '
-                    'Please check the logs and try again.'
-                    f'{colorama.Style.RESET_ALL}')
         progress.stop()
         click.echo(message)
         if success_progress:
@@ -2117,45 +2069,21 @@ def storage():
 @usage_lib.entrypoint
 def storage_ls():
     """List storage objects created."""
-    storage_stat = global_user_state.get_storage()
-    storage_table = log_utils.create_table([
-        'NAME',
-        'CREATED',
-        'STORE',
-        'COMMAND',
-        'STATUS',
-    ])
-
-    for row in storage_stat:
-        launched_at = row['launched_at']
-        storage_table.add_row([
-            # NAME
-            row['name'],
-            # LAUNCHED
-            log_utils.readable_time_duration(launched_at),
-            # CLOUDS
-            ', '.join([s.value for s in row['handle'].sky_stores.keys()]),
-            # COMMAND
-            row['last_use'],
-            # STATUS
-            row['status'].value,
-        ])
-    if storage_stat:
-        click.echo(storage_table)
-    else:
-        click.echo('No existing storage.')
+    storages = sky.storage_ls()
+    storage_table = storage_utils.format_storage_table(storages)
+    click.echo(storage_table)
 
 
 @storage.command('delete', cls=_DocumentedCodeCommand)
+@click.argument('names', required=False, type=str, nargs=-1)
 @click.option('--all',
               '-a',
               default=False,
               is_flag=True,
               required=False,
               help='Delete all storage objects.')
-@click.argument('name', required=False, type=str, nargs=-1)
 @usage_lib.entrypoint
-def storage_delete(all: bool, name: str):  # pylint: disable=redefined-builtin
+def storage_delete(names: Tuple[str], all: bool):  # pylint: disable=redefined-builtin
     """Delete storage objects.
 
     Examples:
@@ -2168,29 +2096,14 @@ def storage_delete(all: bool, name: str):  # pylint: disable=redefined-builtin
         # Delete all storage objects.
         sky storage delete -a
     """
+    if sum([len(names) > 0, all]) != 1:
+        raise click.UsageError('Either --all or a name must be specified.')
     if all:
         click.echo('Deleting all storage objects.')
-        storages = global_user_state.get_storage()
-        for row in storages:
-            store_object = data.Storage(name=row['name'],
-                                        source=row['handle'].source,
-                                        sync_on_reconstruction=False)
-            store_object.delete()
-    elif name:
-        for n in name:
-            handle = global_user_state.get_handle_from_storage_name(n)
-            if handle is None:
-                click.echo(f'Storage name {n} not found.')
-            else:
-                click.echo(f'Deleting storage object {n}.')
-                store_object = data.Storage(name=handle.storage_name,
-                                            source=handle.source,
-                                            sync_on_reconstruction=False)
-                store_object.delete()
-    else:
-        raise click.ClickException(
-            'Must pass in \'-a/--all\' or storage names to \'sky '
-            'storage delete\'.')
+        storages = sky.storage_ls()
+        names = [s['name'] for s in storages]
+    for name in names:
+        sky.storage_delete(name)
 
 
 @cli.group(cls=_NaturalOrderGroup)
@@ -2327,6 +2240,7 @@ def spot_launch(
     cloud: Optional[str],
     region: Optional[str],
     gpus: Optional[str],
+    instance_type: Optional[str],
     num_nodes: Optional[int],
     use_spot: Optional[bool],
     image_id: Optional[str],
@@ -2349,6 +2263,7 @@ def spot_launch(
         cloud=cloud,
         region=region,
         gpus=gpus,
+        instance_type=instance_type,
         num_nodes=num_nodes,
         use_spot=use_spot,
         image_id=image_id,
@@ -2362,104 +2277,7 @@ def spot_launch(
         if prompt is not None:
             click.confirm(prompt, default=True, abort=True, show_default=True)
 
-    assert len(dag.tasks) == 1, dag
-    task = dag.tasks[0]
-    assert len(task.resources) == 1, task
-    resources = list(task.resources)[0]
-
-    change_default_value = dict()
-    if not resources.use_spot_specified:
-        logger.info('Field use_spot not specified; defaulting to True.')
-        change_default_value['use_spot'] = True
-    if resources.spot_recovery is None:
-        logger.info('No spot recovery strategy specified; defaulting to '
-                    f'{spot_lib.SPOT_DEFAULT_STRATEGY}.')
-        change_default_value['spot_recovery'] = spot_lib.SPOT_DEFAULT_STRATEGY
-
-    new_resources = resources.copy(**change_default_value)
-    task.set_resources({new_resources})
-
-    if task.run is None:
-        click.secho(
-            'Skipping the managed spot task as the run section is not set.',
-            fg='green')
-        return
-
-    # TODO(zhwu): Refactor the Task (as Resources), so that we can enforce the
-    # following validations.
-    # Check the file mounts in the task.
-    # Disallow all local file mounts (copy mounts).
-    if task.workdir is not None:
-        raise click.UsageError('Workdir is not allowed for managed spot jobs.')
-    copy_mounts = task.get_local_to_remote_file_mounts()
-    if copy_mounts:
-        copy_mounts_str = '\n\t'.join(': '.join(m) for m in copy_mounts)
-        raise click.UsageError(
-            'Local file mounts are not allowed for managed spot jobs, '
-            f'but following are found: {copy_mounts_str}')
-
-    # Copy the local source to a bucket. The task will not be executed locally,
-    # so we need to copy the files to the bucket manually here before sending to
-    # the remote spot controller.
-    task.add_storage_mounts()
-
-    # Replace the source field that is local path in all storage_mounts with
-    # bucket URI and remove the name field.
-    for storage_obj in task.storage_mounts.values():
-        if (storage_obj.source is not None and
-                not data_utils.is_cloud_store_url(storage_obj.source)):
-            # Need to replace the local path with bucket URI, and remove the
-            # name field, so that the storage mount can work on the spot
-            # controller.
-            store_types = list(storage_obj.stores.keys())
-            assert len(store_types) == 1, (
-                'We only support one store type for now.', storage_obj.stores)
-            store_type = store_types[0]
-            if store_type == StoreType.S3:
-                storage_obj.source = f's3://{storage_obj.name}'
-            elif store_type == StoreType.GCS:
-                storage_obj.source = f'gs://{storage_obj.name}'
-            else:
-                with ux_utils.print_exception_no_traceback():
-                    raise ValueError(f'Unsupported store type: {store_type}')
-            storage_obj.name = None
-
-    with tempfile.NamedTemporaryFile(prefix=f'spot-task-{name}-',
-                                     mode='w') as f:
-        task_config = task.to_yaml_config()
-        common_utils.dump_yaml(f.name, task_config)
-
-        controller_name = spot_lib.SPOT_CONTROLLER_NAME
-        yaml_path = backend_utils.fill_template(
-            spot_lib.SPOT_CONTROLLER_TEMPLATE, {
-                'remote_user_yaml_prefix': spot_lib.SPOT_TASK_YAML_PREFIX,
-                'user_yaml_path': f.name,
-                'spot_controller': controller_name,
-                'cluster_name': name,
-                'sky_remote_path': backend_utils.SKY_REMOTE_PATH,
-                'is_dev': env_options.Options.IS_DEVELOPER.get(),
-                'disable_logging': env_options.Options.DISABLE_LOGGING.get(),
-                'logging_user_hash': usage_lib.get_logging_user_hash()
-            },
-            output_prefix=spot_lib.SPOT_CONTROLLER_YAML_PREFIX)
-        with sky.Dag() as dag:
-            controller_task = sky.Task.from_yaml(yaml_path)
-            controller_task.spot_task = task
-            assert len(controller_task.resources) == 1
-        click.secho(
-            f'Launching managed spot job {name} from spot controller...',
-            fg='yellow')
-        click.echo('Launching spot controller...')
-        sky.launch(
-            dag,
-            stream_logs=True,
-            cluster_name=controller_name,
-            detach_run=detach_run,
-            idle_minutes_to_autostop=spot_lib.
-            SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP,
-            is_spot_controller_task=True,
-            retry_until_up=True,
-        )
+    sky.spot_launch(dag, name, detach_run=detach_run)
 
 
 @spot.command('status', cls=_DocumentedCodeCommand)
@@ -2502,44 +2320,25 @@ def spot_status(all: bool, refresh: bool):
     log can be found with ``sky logs sky-spot-controller job_id``.
     """
     click.secho('Fetching managed spot job statuses...', fg='yellow')
-    cache = spot_lib.load_job_table_cache()
-    stop_msg = ''
-    if not refresh:
-        stop_msg = 'To view the latest job table: sky spot status --refresh'
-    controller_status, handle = _is_spot_controller_up(stop_msg)
-
-    if (refresh and controller_status in [
-            global_user_state.ClusterStatus.STOPPED,
-            global_user_state.ClusterStatus.INIT
-    ]):
-        click.secho('Restarting controller for latest status...', fg='yellow')
-        handle = _start_cluster(spot_lib.SPOT_CONTROLLER_NAME,
-                                idle_minutes_to_autostop=spot_lib.
-                                SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP)
-
-    if handle is None:
-        job_table_str = 'No cached job status table found.'
+    try:
+        job_table = core.spot_status(refresh=refresh)
+    except exceptions.ClusterNotUpError:
+        cache = spot_lib.load_job_table_cache()
         if cache is not None:
             readable_time = log_utils.readable_time_duration(cache[0])
-            job_table_str = (
+            job_table_json = (
                 f'\n{colorama.Fore.YELLOW}Cached job status table '
                 f'[last updated: {readable_time}]:{colorama.Style.RESET_ALL}\n'
                 f'{cache[1]}\n')
-        click.echo(job_table_str)
+        else:
+            job_table_json = 'No cached job status table found.'
+        click.echo(job_table_json)
         return
+    job_table = spot_lib.format_job_table(job_table, all)
 
-    backend = backend_utils.get_backend_from_handle(handle)
-    assert isinstance(backend, backends.CloudVmRayBackend)
-
-    code = spot_lib.SpotCodeGen.show_jobs(show_all=all)
-    returncode, job_table_str, stderr = backend.run_on_head(
-        handle, code, require_outputs=True, stream_logs=False)
-    subprocess_utils.handle_returncode(returncode, code,
-                                       'Failed to fetch managed job statuses',
-                                       job_table_str + stderr)
-
-    spot_lib.dump_job_table_cache(job_table_str)
-    click.echo(f'Managed spot jobs:\n{job_table_str}')
+    # Dump cache
+    spot_lib.dump_job_table_cache(job_table)
+    click.echo(f'Managed spot jobs:\n{job_table}')
 
 
 @spot.command('cancel', cls=_DocumentedCodeCommand)
@@ -2604,27 +2403,7 @@ def spot_cancel(name: Optional[str], job_ids: Tuple[int], all: bool, yes: bool):
                       abort=True,
                       show_default=True)
 
-    backend = backend_utils.get_backend_from_handle(handle)
-    assert isinstance(backend, backends.CloudVmRayBackend)
-    codegen = spot_lib.SpotCodeGen
-    if all:
-        code = codegen.cancel_jobs_by_id(None)
-    elif job_ids:
-        code = codegen.cancel_jobs_by_id(job_ids)
-    else:
-        code = codegen.cancel_job_by_name(name)
-    # The stderr is redirected to stdout
-    returncode, stdout, _ = backend.run_on_head(handle,
-                                                code,
-                                                require_outputs=True,
-                                                stream_logs=False)
-    subprocess_utils.handle_returncode(returncode, code,
-                                       'Failed to cancel managed spot job',
-                                       stdout)
-
-    click.echo(stdout)
-    if 'Multiple jobs found with name' in stdout:
-        click.echo('Please specify the job ID instead of the job name.')
+    core.spot_cancel(job_ids=job_ids, name=name, all=all)
 
 
 @spot.command('logs', cls=_DocumentedCodeCommand)
@@ -2637,18 +2416,7 @@ def spot_cancel(name: Optional[str], job_ids: Tuple[int], all: bool, yes: bool):
 @usage_lib.entrypoint
 def spot_logs(name: Optional[str], job_id: Optional[int]):
     """Tail the log of a managed spot job."""
-    # TODO(zhwu): Automatically restart the spot controller
-    _, handle = _is_spot_controller_up(
-        'Please restart the spot controller with '
-        '`sky start sky-spot-controller`.')
-    if handle is None:
-        return
-
-    if name is not None and job_id is not None:
-        click.UsageError('Cannot specify both --name and --job-id.')
-    backend = backend_utils.get_backend_from_handle(handle)
-    # Stream the realtime logs
-    backend.tail_spot_logs(handle, job_id=job_id, job_name=name)
+    core.spot_tail_logs(name=name, job_id=job_id)
 
 
 # ==============================
diff --git a/sky/constants.py b/sky/constants.py
index 0fefb045a44..bafba180aa1 100644
--- a/sky/constants.py
+++ b/sky/constants.py
@@ -2,3 +2,8 @@
 
 SKY_LOGS_DIRECTORY = '~/sky_logs'
 SKY_REMOTE_RAY_VERSION = '1.13.0'
+
+# TODO(mluo): Make explicit `sky launch -c <name> ''` optional.
+UNINITIALIZED_ONPREM_CLUSTER_MESSAGE = (
+    'Found uninitialized local cluster {cluster}. Run this '
+    'command to initialize it locally: sky launch -c {cluster} \'\'')
diff --git a/sky/core.py b/sky/core.py
new file mode 100644
index 00000000000..6cd472eb3f3
--- /dev/null
+++ b/sky/core.py
@@ -0,0 +1,590 @@
+"""SDK functions for cluster/job management."""
+import colorama
+import getpass
+from typing import Any, Dict, List, Optional, Tuple
+
+from sky import dag
+from sky import task
+from sky import backends
+from sky import constants
+from sky import data
+from sky import exceptions
+from sky import global_user_state
+from sky import sky_logging
+from sky import spot
+from sky.backends import backend_utils
+from sky.backends import onprem_utils
+from sky.skylet import job_lib
+from sky.utils import ux_utils
+from sky.utils import subprocess_utils
+
+logger = sky_logging.init_logger(__name__)
+
+# ======================
+# = Cluster Management =
+# ======================
+
+
+# pylint: disable=redefined-builtin
+def status(all: bool, refresh: bool) -> List[Dict[str, Any]]:
+    """Get the cluster status in dict.
+
+    Please refer to the sky.cli.status for the document.
+
+    Returns:
+        List[dict]:
+        [
+            {
+                'name': (str) cluster name,
+                'launched_at': (int) timestamp of launched,
+                'last_use': (int) timestamp of last use,
+                'status': (sky.ClusterStatus) cluster status,
+                'autostop': (int) idle time before autostop,
+                'metadata': (dict) metadata of the cluster,
+            }
+        ]
+    """
+    cluster_records = backend_utils.get_clusters(all, refresh)
+    return cluster_records
+
+
+def _start(cluster_name: str,
+           idle_minutes_to_autostop: Optional[int] = None,
+           retry_until_up: bool = False) -> backends.Backend.ResourceHandle:
+
+    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
+        cluster_name)
+    if cluster_status == global_user_state.ClusterStatus.UP:
+        logger.info(f'Cluster {cluster_name!r} is already up.')
+        return
+    assert cluster_status in (
+        global_user_state.ClusterStatus.INIT,
+        global_user_state.ClusterStatus.STOPPED), cluster_status
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    if not isinstance(backend, backends.CloudVmRayBackend):
+        raise exceptions.NotSupportedError(
+            f'Starting cluster {cluster_name!r} with backend {backend.NAME} '
+            'is not supported.')
+    with dag.Dag():
+        dummy_task = task.Task().set_resources(handle.launched_resources)
+        dummy_task.num_nodes = handle.launched_nodes
+    handle = backend.provision(dummy_task,
+                               to_provision=handle.launched_resources,
+                               dryrun=False,
+                               stream_logs=True,
+                               cluster_name=cluster_name,
+                               retry_until_up=retry_until_up)
+    if idle_minutes_to_autostop is not None:
+        backend.set_autostop(handle, idle_minutes_to_autostop)
+    return handle
+
+
+def start(cluster_name: str,
+          idle_minutes_to_autostop: Optional[int] = None,
+          retry_until_up: bool = False):
+    """Start the cluster.
+
+    Please refer to the sky.cli.start for the document.
+
+    Raises:
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+    """
+    _start(cluster_name, idle_minutes_to_autostop, retry_until_up)
+
+
+def stop(cluster_name: str, purge: bool = False):
+    """Stop the cluster
+
+    Please refer to the sky.cli.stop for the document.
+
+    Raises:
+        ValueError: cluster does not exist.
+        RuntimeError: Fail to stop the cluster.
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+    """
+    if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+        raise exceptions.NotSupportedError(
+            f'Stopping sky reserved cluster {cluster_name!r} '
+            f'is not supported.')
+    handle = global_user_state.get_handle_from_cluster_name(cluster_name)
+    if handle is None:
+        raise ValueError(f'Cluster {cluster_name!r} does not exist.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    if (isinstance(backend, backends.CloudVmRayBackend) and
+            handle.launched_resources.use_spot):
+        # Disable spot instances to be stopped.
+        # TODO(suquark): enable GCP+spot to be stopped in the future.
+        raise exceptions.NotSupportedError(
+            f'{colorama.Fore.YELLOW}Stopping cluster '
+            f'{cluster_name!r}... skipped.{colorama.Style.RESET_ALL}\n'
+            '  Stopping spot instances is not supported as the attached '
+            'disks will be lost.\n'
+            '  To terminate the cluster instead, run: '
+            f'{colorama.Style.BRIGHT}sky down {cluster_name}')
+    backend.teardown(handle, terminate=False, purge=purge)
+
+
+def down(cluster_name: str, purge: bool = False):
+    """Down the cluster
+
+    Please refer to the sky.cli.down for the document.
+
+    Raises:
+        ValueError: cluster does not exist.
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+    """
+    if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+        raise exceptions.NotSupportedError(
+            f'Tearing down sky reserved cluster {cluster_name!r} '
+            f'is not supported.')
+    handle = global_user_state.get_handle_from_cluster_name(cluster_name)
+    if handle is None:
+        raise ValueError(f'Cluster {cluster_name!r} does not exist.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    backend.teardown(handle, terminate=True, purge=purge)
+
+
+def autostop(cluster_name: str, idle_minutes_to_autostop: int):
+    """Set the autostop time of the cluster.
+
+    Please refer to the sky.cli.autostop for the document.
+
+    Args:
+        cluster_name (str): name of the cluster.
+        autostop_time (int): autostop time in minutes.
+    Raises:
+        ValueError: cluster does not exist.
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+        sky.exceptions.ClusterNotUpError: the cluster is not UP.
+    """
+    verb = 'Scheduling' if idle_minutes_to_autostop >= 0 else 'Cancelling'
+    operation = f'{verb} auto-stop on'
+    if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+        raise exceptions.NotSupportedError(
+            f'{operation} sky reserved cluster {cluster_name!r} '
+            f'is not supported.')
+    (cluster_status,
+     handle) = backend_utils.refresh_cluster_status_handle(cluster_name)
+    if handle is None:
+        raise ValueError(f'Cluster {cluster_name!r} does not exist.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    if not isinstance(backend, backends.CloudVmRayBackend):
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'{colorama.Fore.YELLOW}{operation} cluster '
+                f'{cluster_name!r}... skipped{colorama.Style.RESET_ALL}'
+                '\n  Auto-stopping is only supported by backend: '
+                f'{backends.CloudVmRayBackend.NAME}')
+    if cluster_status != global_user_state.ClusterStatus.UP:
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.ClusterNotUpError(
+                f'{colorama.Fore.YELLOW}{operation} cluster '
+                f'{cluster_name!r} (status: {cluster_status.value})... skipped'
+                f'{colorama.Style.RESET_ALL}'
+                '\n  Auto-stop can only be set/unset for '
+                f'{global_user_state.ClusterStatus.UP.value} clusters.')
+    backend.set_autostop(handle, idle_minutes_to_autostop)
+
+
+# ==================
+# = Job Management =
+# ==================
+
+
+def _check_cluster_available(cluster_name: str,
+                             operation: str) -> backends.Backend.ResourceHandle:
+    """Check if the cluster is available."""
+    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
+        cluster_name)
+    backend = backend_utils.get_backend_from_handle(handle)
+    if isinstance(backend, backends.LocalDockerBackend):
+        # LocalDockerBackend does not support job queues
+        raise exceptions.NotSupportedError(
+            f'Cluster {cluster_name} with LocalDockerBackend does '
+            f'not support {operation}.')
+    if cluster_status != global_user_state.ClusterStatus.UP:
+        if onprem_utils.check_if_local_cloud(cluster_name):
+            raise exceptions.ClusterNotUpError(
+                constants.UNINITIALIZED_ONPREM_CLUSTER_MESSAGE.format(
+                    cluster_name))
+        raise exceptions.ClusterNotUpError(
+            f'{colorama.Fore.YELLOW}Cluster {cluster_name!r} is not up '
+            f'(status: {cluster_status.value}); skipped.'
+            f'{colorama.Style.RESET_ALL}')
+
+    if handle.head_ip is None:
+        raise exceptions.ClusterNotUpError(
+            f'Cluster {cluster_name!r} has been stopped or not properly set up.'
+            ' Please re-launch it with `sky start`.')
+    return handle
+
+
+def queue(cluster_name: str,
+          skip_finished: bool = False,
+          all_users: bool = False):
+    """Get the job queue in List[dict].
+
+    Please refer to the sky.cli.queue for the document.
+
+    Returns:
+        List[dict]:
+        [
+            {
+                'job_id': (int) job id,
+                'job_name': (str) job name,
+                'username': (str) username,
+                'submitted_at': (int) timestamp of submitted,
+                'start_at': (int) timestamp of started,
+                'end_at': (int) timestamp of ended,
+                'resources': (str) resources,
+                'status': (job_lib.JobStatus) job status,
+                'log_path': (str) log path,
+            }
+        ]
+    raises:
+        RuntimeError: if failed to get the job queue.
+        sky.exceptions.ClusterNotUpError: the cluster is not up.
+        sky.exceptions.NotSupportedError: the feature is not supported.
+    """
+    all_jobs = not skip_finished
+    username = getpass.getuser()
+    if all_users:
+        username = None
+    code = job_lib.JobLibCodeGen.get_job_queue(username, all_jobs)
+
+    handle = _check_cluster_available(cluster_name, 'getting the job queue')
+    backend = backend_utils.get_backend_from_handle(handle)
+
+    logger.info(f'\nSky Job Queue of Cluster {cluster_name}')
+    returncode, jobs_json, stderr = backend.run_on_head(handle,
+                                                        code,
+                                                        require_outputs=True)
+    if returncode != 0:
+        raise RuntimeError(f'{jobs_json + stderr}\n{colorama.Fore.RED}'
+                           f'Failed to get job queue on cluster {cluster_name}.'
+                           f'{colorama.Style.RESET_ALL}')
+    jobs = job_lib.load_job_queue(jobs_json)
+    return jobs
+
+
+# pylint: disable=redefined-builtin
+def cancel(cluster_name: str,
+           all: bool = False,
+           job_ids: Optional[List[int]] = None):
+    """Cancel jobs.
+
+    Please refer to the sky.cli.cancel for the document.
+
+    Raises:
+        ValueError: arguments are invalid or the cluster is not supported.
+        sky.exceptions.ClusterNotUpError: the cluster is not up.
+        sky.exceptions.NotSupportedError: the feature is not supported.
+    """
+    job_ids = [] if job_ids is None else job_ids
+    if len(job_ids) == 0 and not all:
+        raise ValueError(
+            'sky cancel requires either a job id '
+            f'(see `sky queue {cluster_name} -s`) or the --all flag.')
+
+    backend_utils.check_cluster_name_not_reserved(
+        cluster_name, operation_str='Cancelling jobs')
+
+    # Check the status of the cluster.
+    handle = _check_cluster_available(cluster_name, 'cancelling jobs')
+    backend = backend_utils.get_backend_from_handle(handle)
+
+    if all:
+        logger.info(f'{colorama.Fore.YELLOW}'
+                    f'Cancelling all jobs on cluster {cluster_name!r}...'
+                    f'{colorama.Style.RESET_ALL}')
+        job_ids = None
+    else:
+        jobs_str = ', '.join(map(str, job_ids))
+        logger.info(
+            f'{colorama.Fore.YELLOW}'
+            f'Cancelling jobs ({jobs_str}) on cluster {cluster_name!r}...'
+            f'{colorama.Style.RESET_ALL}')
+
+    backend.cancel_jobs(handle, job_ids)
+
+
+def tail_logs(cluster_name: str, job_id: Optional[str]):
+    """Tail the logs of a job.
+
+    Please refer to the sky.cli.tail_logs for the document.
+
+    Raises:
+        ValueError: arguments are invalid or the cluster is not supported.
+        sky.exceptions.ClusterNotUpError: the cluster is not up.
+        sky.exceptions.NotSupportedError: the feature is not supported.
+    """
+    # Check the status of the cluster.
+    handle = _check_cluster_available(cluster_name, 'tailing logs')
+    backend = backend_utils.get_backend_from_handle(handle)
+
+    job_str = f'job {job_id}'
+    if job_id is None:
+        job_str = 'the last job'
+    logger.info(f'{colorama.Fore.YELLOW}'
+                f'Tailing logs of {job_str} on cluster {cluster_name!r}...'
+                f'{colorama.Style.RESET_ALL}')
+
+    backend.tail_logs(handle, job_id)
+
+
+def download_logs(cluster_name: str,
+                  job_ids: Optional[List[str]]) -> Dict[str, str]:
+    """Download the logs of jobs.
+
+    Args:
+        cluster_name: (str) name of the cluster.
+        job_ids: (List[str]) job ids.
+    Returns:
+        Dict[str, str]: a mapping of job_id to local log path.
+    """
+    # Check the status of the cluster.
+    handle = _check_cluster_available(cluster_name, 'downloading logs')
+    backend = backend_utils.get_backend_from_handle(handle)
+
+    if job_ids is not None and len(job_ids) == 0:
+        return []
+
+    logger.info(f'{colorama.Fore.YELLOW}'
+                'Syncing down logs to local...'
+                f'{colorama.Style.RESET_ALL}')
+    local_log_dirs = backend.sync_down_logs(handle, job_ids)
+    return local_log_dirs
+
+
+def job_status(
+        cluster_name: str,
+        job_ids: Optional[List[str]],
+        stream_logs: bool = False) -> Dict[str, Optional[job_lib.JobStatus]]:
+    """Get the status of jobs.
+
+    Args:
+        cluster_name: (str) name of the cluster.
+        job_ids: (List[str]) job ids. If None, get the status of the last job.
+    Returns:
+        Dict[str, Optional[job_lib.JobStatus]]: A mapping of job_id to job
+        statuses. The status will be None if the job does not exist.
+        If job_ids is None and there is no job on the cluster, it will return
+        {None: None}.
+    """
+    # Check the status of the cluster.
+    handle = _check_cluster_available(cluster_name, 'getting job status')
+    backend = backend_utils.get_backend_from_handle(handle)
+
+    if job_ids is not None and len(job_ids) == 0:
+        return []
+
+    logger.info(f'{colorama.Fore.YELLOW}'
+                'Getting job status...'
+                f'{colorama.Style.RESET_ALL}')
+
+    statuses = backend.get_job_status(handle, job_ids, stream_logs=stream_logs)
+    return statuses
+
+
+# =======================
+# = Spot Job Management =
+# =======================
+
+
+def _is_spot_controller_up(
+    stopped_message: str,
+) -> Tuple[Optional[global_user_state.ClusterStatus],
+           Optional[backends.Backend.ResourceHandle]]:
+    controller_status, handle = backend_utils.refresh_cluster_status_handle(
+        spot.SPOT_CONTROLLER_NAME, force_refresh=True)
+    if controller_status is None:
+        logger.info('No managed spot job has been run.')
+    elif controller_status != global_user_state.ClusterStatus.UP:
+        msg = (f'Spot controller {spot.SPOT_CONTROLLER_NAME} '
+               f'is {controller_status.value}.')
+        if controller_status == global_user_state.ClusterStatus.STOPPED:
+            msg += f'\n{stopped_message}'
+        if controller_status == global_user_state.ClusterStatus.INIT:
+            msg += '\nPlease wait for the controller to be ready.'
+        logger.info(msg)
+        handle = None
+    return controller_status, handle
+
+
+def spot_status(refresh: bool) -> List[Dict[str, Any]]:
+    """Get statuses of managed spot jobs.
+
+    Please refer to the sky.cli.spot_status for the document.
+
+    Returns:
+        [
+            {
+                'job_id': int,
+                'job_name': str,
+                'resources': str,
+                'submitted_at': (float) timestamp of submission,
+                'end_at': (float) timestamp of end,
+                'duration': (float) duration in seconds,
+                'retry_count': int Number of retries,
+                'status': sky.JobStatus status of the job,
+            }
+        ]
+    Raises:
+        sky.exceptions.ClusterNotUpError: the spot controller is not up.
+    """
+
+    stop_msg = ''
+    if not refresh:
+        stop_msg = 'To view the latest job table: sky spot status --refresh'
+    controller_status, handle = _is_spot_controller_up(stop_msg)
+
+    if (refresh and controller_status in [
+            global_user_state.ClusterStatus.STOPPED,
+            global_user_state.ClusterStatus.INIT
+    ]):
+        logger.info(f'{colorama.Fore.YELLOW}'
+                    '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)
+
+    if handle is None or handle.head_ip is None:
+        raise exceptions.ClusterNotUpError('Spot controller is not up.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    assert isinstance(backend, backends.CloudVmRayBackend)
+
+    code = spot.SpotCodeGen.get_job_table()
+    returncode, job_table_json, stderr = backend.run_on_head(
+        handle, code, require_outputs=True, stream_logs=False)
+    try:
+        subprocess_utils.handle_returncode(
+            returncode, code, 'Failed to fetch managed job statuses',
+            job_table_json + stderr)
+    except exceptions.CommandError as e:
+        raise RuntimeError(e.message) from e
+
+    jobs = spot.load_spot_job_queue(job_table_json)
+    return jobs
+
+
+# pylint: disable=redefined-builtin
+def spot_cancel(name: Optional[str] = None,
+                job_ids: Optional[Tuple[int]] = None,
+                all: bool = False):
+    """Cancel managed spot jobs
+
+    Please refer to the sky.cli.spot_cancel for the document.
+
+    Raises:
+        sky.exceptions.ClusterNotUpError: the spot controller is not up.
+        RuntimeError: failed to cancel the job.
+    """
+    job_ids = [] if job_ids is None else job_ids
+    _, handle = _is_spot_controller_up(
+        'All managed spot jobs should have finished.')
+    if handle is None or handle.head_ip is None:
+        raise exceptions.ClusterNotUpError('All jobs finished.')
+
+    job_id_str = ','.join(map(str, job_ids))
+    if sum([len(job_ids) > 0, name is not None, all]) != 1:
+        argument_str = f'job_ids={job_id_str}' if len(job_ids) > 0 else ''
+        argument_str += f' name={name}' if name is not None else ''
+        argument_str += ' all' if all else ''
+        raise ValueError('Can only specify one of JOB_IDS or name or all. '
+                         f'Provided {argument_str!r}.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    assert isinstance(backend, backends.CloudVmRayBackend)
+    if all:
+        code = spot.SpotCodeGen.cancel_jobs_by_id(None)
+    elif job_ids:
+        code = spot.SpotCodeGen.cancel_jobs_by_id(job_ids)
+    else:
+        code = spot.SpotCodeGen.cancel_job_by_name(name)
+    # The stderr is redirected to stdout
+    returncode, stdout, _ = backend.run_on_head(handle,
+                                                code,
+                                                require_outputs=True,
+                                                stream_logs=False)
+    try:
+        subprocess_utils.handle_returncode(returncode, code,
+                                           'Failed to cancel managed spot job',
+                                           stdout)
+    except exceptions.CommandError as e:
+        raise RuntimeError(e.message) from e
+
+    logger.info(stdout)
+    if 'Multiple jobs found with name' in stdout:
+        with ux_utils.print_exception_no_traceback():
+            raise RuntimeError(
+                'Please specify the job ID instead of the job name.')
+
+
+def spot_tail_logs(name: Optional[str], job_id: Optional[int]):
+    """Tail logs of managed spot jobs.
+
+    Please refer to the sky.cli.spot_logs for the document.
+
+    Raises:
+        ValueError: invalid arguments.
+        sky.exceptions.ClusterNotUpError: the spot controller is not up.
+    """
+    # TODO(zhwu): Automatically restart the spot controller
+    _, handle = _is_spot_controller_up(
+        'Please restart the spot controller with '
+        '`sky start sky-spot-controller -i 5`.')
+    if handle is None or handle.head_ip is None:
+        raise exceptions.ClusterNotUpError('All jobs finished.')
+
+    if name is not None and job_id is not None:
+        raise ValueError('Cannot specify both name and job_id.')
+    backend = backend_utils.get_backend_from_handle(handle)
+    # Stream the realtime logs
+    backend.tail_spot_logs(handle, job_id=job_id, job_name=name)
+
+
+# ======================
+# = Storage Management =
+# ======================
+def storage_ls() -> List[Dict[str, Any]]:
+    """Get the storages.
+
+    Returns:
+        [
+            {
+                'name': str,
+                'launched_at': int timestamp of creation,
+                'store': List[sky.StoreType],
+                'last_use': int timestamp of last use,
+                'status': sky.StorageStatus,
+            }
+        ]
+    """
+    storages = global_user_state.get_storage()
+    for storage in storages:
+        storage['store'] = list(storage.pop('handle').sky_stores.keys())
+    return storages
+
+
+def storage_delete(name: Optional[str] = None):
+    """Delete a storage.
+
+    Raises:
+        ValueError: If the storage does not exist.
+    """
+    handle = global_user_state.get_handle_from_storage_name(name)
+    if handle is None:
+        raise ValueError(f'Storage name {name!r} not found.')
+    else:
+        logger.info(f'Deleting storage object {name!r}.')
+        store_object = data.Storage(name=handle.storage_name,
+                                    source=handle.source,
+                                    sync_on_reconstruction=False)
+        store_object.delete()
diff --git a/sky/data/storage_utils.py b/sky/data/storage_utils.py
new file mode 100644
index 00000000000..860c91caa15
--- /dev/null
+++ b/sky/data/storage_utils.py
@@ -0,0 +1,44 @@
+"""Utility functions for the storage module."""
+from typing import Any, Dict, List
+
+from sky import sky_logging
+from sky.utils import log_utils
+
+logger = sky_logging.init_logger(__name__)
+
+
+def format_storage_table(storages: List[Dict[str, Any]]) -> str:
+    """Format the storage table for display.
+
+    Args:
+        storage_table (dict): The storage table.
+
+    Returns:
+        str: The formatted storage table.
+    """
+    storage_table = log_utils.create_table([
+        'NAME',
+        'CREATED',
+        'STORE',
+        'COMMAND',
+        'STATUS',
+    ])
+
+    for row in storages:
+        launched_at = row['launched_at']
+        storage_table.add_row([
+            # NAME
+            row['name'],
+            # LAUNCHED
+            log_utils.readable_time_duration(launched_at),
+            # CLOUDS
+            ', '.join([s.value for s in row['store']]),
+            # COMMAND
+            row['last_use'],
+            # STATUS
+            row['status'].value,
+        ])
+    if storages:
+        return storage_table
+    else:
+        return 'No existing storage.'
diff --git a/sky/exceptions.py b/sky/exceptions.py
index f833e3ac5c6..88fe3249b09 100644
--- a/sky/exceptions.py
+++ b/sky/exceptions.py
@@ -4,6 +4,7 @@
 # Return code for keyboard interruption and SIGTSTP
 KEYBOARD_INTERRUPT_CODE = 130
 SIGTSTP_CODE = 146
+RSYNC_FILE_NOT_FOUND_CODE = 23
 
 
 class ResourcesUnavailableError(Exception):
@@ -37,6 +38,16 @@ def __init__(self, returncode: int, command: str, error_msg: str) -> None:
         super().__init__(message)
 
 
+class ClusterNotUpError(Exception):
+    """Raised when a cluster is not up."""
+    pass
+
+
+class NotSupportedError(Exception):
+    """Raised when a feature is not supported."""
+    pass
+
+
 class StorageError(Exception):
     pass
 
diff --git a/sky/execution.py b/sky/execution.py
index 24868060955..7a80504b5d0 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -13,18 +13,25 @@
   - ray exec + each task's commands
 """
 import enum
+import tempfile
 import os
 import time
 from typing import Any, List, Optional
 
+import colorama
+
 import sky
 from sky import backends
+from sky import exceptions
 from sky import global_user_state
 from sky import optimizer
 from sky import sky_logging
 from sky import spot
 from sky.backends import backend_utils
+from sky.data import data_utils
+from sky.data import storage
 from sky.usage import usage_lib
+from sky.utils import common_utils
 from sky.utils import env_options, timeline
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -32,6 +39,7 @@
 logger = sky_logging.init_logger(__name__)
 
 OptimizeTarget = optimizer.OptimizeTarget
+_MAX_SPOT_JOB_LENGTH = 10
 
 
 class Stage(enum.Enum):
@@ -184,7 +192,8 @@ def _execute(
             # For spot controller task, it requires a while to have the
             # managed spot status shown in the status table.
             time.sleep(0.5)
-            subprocess_utils.run('sky spot status', env=env)
+            subprocess_utils.run(
+                f'sky spot status | head -n {_MAX_SPOT_JOB_LENGTH}')
         else:
             subprocess_utils.run('sky status', env=env)
         print()
@@ -204,11 +213,9 @@ def launch(
     cluster_name: Optional[str] = None,
     detach_run: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
-    is_spot_controller_task: bool = False,
 ) -> None:
-    if not is_spot_controller_task:
-        backend_utils.check_cluster_name_not_reserved(
-            cluster_name, operation_str='sky.launch')
+    backend_utils.check_cluster_name_not_reserved(cluster_name,
+                                                  operation_str='sky.launch')
     _execute(
         dag=dag,
         dryrun=dryrun,
@@ -259,3 +266,128 @@ def exec(  # pylint: disable=redefined-builtin
              ],
              cluster_name=cluster_name,
              detach_run=detach_run)
+
+
+def spot_launch(
+    dag: sky.Dag,
+    name: Optional[str] = None,
+    stream_logs: bool = True,
+    detach_run: bool = False,
+) -> None:
+    """Launch a managed spot job.
+
+    Please refer to the sky.cli.spot_launch for the document.
+
+    Args:
+        dag: DAG to launch.
+        name: Name of the spot job.
+        detach_run: Whether to detach the run.
+
+    Raises:
+        ValueError: cluster does not exist.
+        sky.exceptions.NotSupportedError: the feature is not supported.
+    """
+    if name is None:
+        name = backend_utils.generate_cluster_name()
+
+    assert len(dag.tasks) == 1, ('Only one task is allowed in a spot launch.',
+                                 dag)
+    task = dag.tasks[0]
+    assert len(task.resources) == 1, task
+    resources = list(task.resources)[0]
+
+    change_default_value = dict()
+    if not resources.use_spot_specified:
+        logger.info('Field use_spot not specified; defaulting to True.')
+        change_default_value['use_spot'] = True
+    if resources.spot_recovery is None:
+        logger.info('No spot recovery strategy specified; defaulting to '
+                    f'{spot.SPOT_DEFAULT_STRATEGY}.')
+        change_default_value['spot_recovery'] = spot.SPOT_DEFAULT_STRATEGY
+
+    new_resources = resources.copy(**change_default_value)
+    task.set_resources({new_resources})
+
+    if task.run is None:
+        logger.info(
+            f'{colorama.Fore.GREEN}'
+            'Skipping the managed spot task as the run section is not set.'
+            f'{colorama.Style.RESET_ALL}')
+        return
+
+    # TODO(zhwu): Refactor the Task (as Resources), so that we can enforce the
+    # following validations.
+    # Check the file mounts in the task.
+    # Disallow all local file mounts (copy mounts).
+    if task.workdir is not None:
+        raise exceptions.NotSupportedError(
+            'Workdir is not allowed for managed spot jobs.')
+    copy_mounts = task.get_local_to_remote_file_mounts()
+    if copy_mounts:
+        copy_mounts_str = '\n\t'.join(': '.join(m) for m in copy_mounts)
+        raise exceptions.NotSupportedError(
+            'Local file mounts are not allowed for managed spot jobs, '
+            f'but following are found: {copy_mounts_str}')
+
+    # Copy the local source to a bucket. The task will not be executed locally,
+    # so we need to copy the files to the bucket manually here before sending to
+    # the remote spot controller.
+    task.add_storage_mounts()
+
+    # Replace the source field that is local path in all storage_mounts with
+    # bucket URI and remove the name field.
+    for storage_obj in task.storage_mounts.values():
+        if (storage_obj.source is not None and
+                not data_utils.is_cloud_store_url(storage_obj.source)):
+            # Need to replace the local path with bucket URI, and remove the
+            # name field, so that the storage mount can work on the spot
+            # controller.
+            store_types = list(storage_obj.stores.keys())
+            assert len(store_types) == 1, (
+                'We only support one store type for now.', storage_obj.stores)
+            store_type = store_types[0]
+            if store_type == storage.StoreType.S3:
+                storage_obj.source = f's3://{storage_obj.name}'
+            elif store_type == storage.StoreType.GCS:
+                storage_obj.source = f'gs://{storage_obj.name}'
+            else:
+                with ux_utils.print_exception_no_traceback():
+                    raise exceptions.NotSupportedError(
+                        f'Unsupported store type: {store_type}')
+            storage_obj.name = None
+
+    with tempfile.NamedTemporaryFile(prefix=f'spot-task-{name}-',
+                                     mode='w') as f:
+        task_config = task.to_yaml_config()
+        common_utils.dump_yaml(f.name, task_config)
+
+        controller_name = spot.SPOT_CONTROLLER_NAME
+        yaml_path = backend_utils.fill_template(
+            spot.SPOT_CONTROLLER_TEMPLATE, {
+                'remote_user_yaml_prefix': spot.SPOT_TASK_YAML_PREFIX,
+                'user_yaml_path': f.name,
+                'spot_controller': controller_name,
+                'cluster_name': name,
+                'sky_remote_path': backend_utils.SKY_REMOTE_PATH,
+                'is_dev': env_options.Options.IS_DEVELOPER.get(),
+                'disable_logging': env_options.Options.DISABLE_LOGGING.get(),
+                'logging_user_hash': usage_lib.get_logging_user_hash()
+            },
+            output_prefix=spot.SPOT_CONTROLLER_YAML_PREFIX)
+        with sky.Dag() as spot_dag:
+            controller_task = sky.Task.from_yaml(yaml_path)
+            controller_task.spot_task = task
+            assert len(controller_task.resources) == 1
+        logger.info(f'{colorama.Fore.YELLOW}'
+                    f'Launching managed spot job {name} from spot controller...'
+                    f'{colorama.Style.RESET_ALL}')
+        logger.info('Launching spot controller...')
+        _execute(
+            dag=spot_dag,
+            stream_logs=stream_logs,
+            cluster_name=controller_name,
+            detach_run=detach_run,
+            idle_minutes_to_autostop=spot.
+            SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP,
+            retry_until_up=True,
+        )
diff --git a/sky/global_user_state.py b/sky/global_user_state.py
index 385a1fc0a9d..a1a75f05a6c 100644
--- a/sky/global_user_state.py
+++ b/sky/global_user_state.py
@@ -16,7 +16,7 @@
 from typing import Any, Dict, List, Optional
 
 from sky import clouds
-from sky.skylet.utils import db_utils
+from sky.utils import db_utils
 from sky.utils import common_utils
 
 if typing.TYPE_CHECKING:
diff --git a/sky/optimizer.py b/sky/optimizer.py
index faaaa43afdf..5c2d5c98622 100644
--- a/sky/optimizer.py
+++ b/sky/optimizer.py
@@ -17,7 +17,7 @@
 from sky import task as task_lib
 from sky.utils import env_options
 from sky.utils import ux_utils
-from sky.skylet.utils import log_utils
+from sky.utils import log_utils
 
 if typing.TYPE_CHECKING:
     from sky import dag as dag_lib
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index faec031c55d..f3aee2456da 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -3,6 +3,7 @@
 This is a remote utility module that provides job queue functionality.
 """
 import enum
+import json
 import os
 import pathlib
 import shlex
@@ -14,8 +15,8 @@
 from sky import constants
 from sky import sky_logging
 from sky.utils import subprocess_utils
-from sky.skylet.utils import db_utils
-from sky.skylet.utils import log_utils
+from sky.utils import db_utils
+from sky.utils import log_utils
 
 logger = sky_logging.init_logger(__name__)
 
@@ -218,6 +219,27 @@ def get_status(job_id: int) -> Optional[JobStatus]:
         return get_status_no_lock(job_id)
 
 
+def get_statuses_json(job_ids: List[Optional[int]]) -> str:
+    # Per-job lock is not required here, since the staled job status will not
+    # affect the caller.
+    query_str = ','.join(['?'] * len(job_ids))
+    rows = _CURSOR.execute(
+        f'SELECT job_id, status FROM jobs WHERE job_id IN ({query_str})',
+        job_ids)
+    statuses = {job_id: None for job_id in job_ids}
+    for (job_id, status) in rows:
+        statuses[job_id] = status
+    return json.dumps(statuses)
+
+
+def load_statuses_json(statuses_json: str) -> Dict[int, JobStatus]:
+    statuses = json.loads(statuses_json)
+    for job_id, status in statuses.items():
+        if status is not None:
+            statuses[job_id] = JobStatus[status]
+    return statuses
+
+
 def get_latest_job_id() -> Optional[int]:
     rows = _CURSOR.execute(
         'SELECT job_id FROM jobs ORDER BY job_id DESC LIMIT 1')
@@ -415,12 +437,17 @@ def is_cluster_idle() -> bool:
         return count == 0
 
 
-def _show_job_queue(jobs) -> None:
+def format_job_queue(jobs: List[Dict[str, Any]]):
+    """Format the job queue for display.
+
+    Usage:
+        jobs = get_job_queue()
+        print(format_job_queue(jobs))
+    """
     job_table = log_utils.create_table([
         'ID', 'NAME', 'SUBMITTED', 'STARTED', 'DURATION', 'RESOURCES', 'STATUS',
         'LOG'
     ])
-
     for job in jobs:
         job_table.add_row([
             job['job_id'],
@@ -432,13 +459,13 @@ def _show_job_queue(jobs) -> None:
                                              absolute=True),
             job['resources'],
             job['status'].value,
-            os.path.join(constants.SKY_LOGS_DIRECTORY, job['run_timestamp']),
+            job['log_path'],
         ])
-    print(job_table)
+    return job_table
 
 
-def show_jobs(username: Optional[str], all_jobs: bool) -> None:
-    """Show the job queue.
+def dump_job_queue(username: Optional[str], all_jobs: bool) -> str:
+    """Get the job queue in json format.
 
     Args:
         username: The username to show jobs for. Show all the users if None.
@@ -449,7 +476,23 @@ def show_jobs(username: Optional[str], all_jobs: bool) -> None:
         status_list = None
 
     jobs = _get_jobs(username, status_list=status_list)
-    _show_job_queue(jobs)
+    for job in jobs:
+        job['status'] = job['status'].value
+        job['log_path'] = os.path.join(constants.SKY_LOGS_DIRECTORY,
+                                       job.pop('run_timestamp'))
+    return json.dumps(jobs, indent=2)
+
+
+def load_job_queue(json_str: str) -> List[Dict[str, Any]]:
+    """Load the job queue from json format.
+
+    Args:
+        json_str: The json string to load.
+    """
+    jobs = json.loads(json_str)
+    for job in jobs:
+        job['status'] = JobStatus(job['status'])
+    return jobs
 
 
 def cancel_jobs(job_owner: str, jobs: Optional[List[int]]) -> None:
@@ -484,7 +527,7 @@ def stop_job(job: str):
             set_status(job['job_id'], JobStatus.CANCELLED)
 
 
-def log_dir(job_id: int) -> Optional[str]:
+def log_dir(job_id: Optional[int]) -> Optional[str]:
     """Returns the relative path to the log file for a job."""
     _CURSOR.execute(
         """\
@@ -497,22 +540,21 @@ def log_dir(job_id: int) -> Optional[str]:
     return os.path.join(constants.SKY_LOGS_DIRECTORY, run_timestamp)
 
 
-def log_dirs_with_globbing(job_id: str) -> List[str]:
+def log_dirs_with_globbing_json(job_ids: List[Optional[str]]) -> Dict[str, str]:
     """Returns the relative paths to the log files for job with globbing."""
+    query_str = ' OR '.join(['job_id GLOB (?)'] * len(job_ids))
     _CURSOR.execute(
-        """\
+        f"""\
             SELECT * FROM jobs
-            WHERE job_id GLOB (?)""", (job_id,))
+            WHERE {query_str}""", job_ids)
     rows = _CURSOR.fetchall()
-    if len(rows) == 0:
-        return None
-    log_paths = []
+    log_paths = dict()
     for row in rows:
         job_id = row[JobInfoLoc.JOB_ID.value]
         run_timestamp = row[JobInfoLoc.RUN_TIMESTAMP.value]
         log_path = os.path.join(constants.SKY_LOGS_DIRECTORY, run_timestamp)
-        log_paths.append((job_id, log_path))
-    return log_paths
+        log_paths[str(job_id)] = log_path
+    return json.dumps(log_paths)
 
 
 class JobLibCodeGen:
@@ -548,8 +590,11 @@ def update_status(cls, job_owner: str) -> str:
         return cls._build(code)
 
     @classmethod
-    def show_jobs(cls, username: Optional[str], all_jobs: bool) -> str:
-        code = [f'job_lib.show_jobs({username!r}, {all_jobs})']
+    def get_job_queue(cls, username: Optional[str], all_jobs: bool) -> str:
+        code = [
+            'job_queue = job_lib.dump_job_queue('
+            f'{username!r}, {all_jobs})', 'print(job_queue, flush=True)'
+        ]
         return cls._build(code)
 
     @classmethod
@@ -576,14 +621,13 @@ def tail_logs(cls, job_owner: str, job_id: Optional[int],
         return cls._build(code)
 
     @classmethod
-    def get_job_status(cls, job_id: Optional[int] = None) -> str:
+    def get_job_status(cls, job_ids: Optional[List[int]] = None) -> str:
         # Prints "Job <id> <status>" for UX; caller should parse the last token.
         code = [
-            f'job_id = {job_id} if {job_id} is not None '
-            'else job_lib.get_latest_job_id()',
-            'job_status = job_lib.get_status(job_id)',
-            'status_str = None if job_status is None else job_status.value',
-            'print("Job", job_id, status_str, flush=True)',
+            f'job_ids = {job_ids} if {job_ids} is not None '
+            'else [job_lib.get_latest_job_id()]',
+            'job_statuses = job_lib.get_statuses_json(job_ids)',
+            'print(job_statuses, flush=True)',
         ]
         return cls._build(code)
 
@@ -608,11 +652,11 @@ def get_log_path(cls, job_id: int) -> str:
         return cls._build(code)
 
     @classmethod
-    def get_log_path_with_globbing(cls, job_id: Optional[str]) -> str:
+    def get_log_path_with_globbing(cls, job_ids: Optional[List[str]]) -> str:
         code = [
-            f'job_id = {job_id!r} if {job_id!r} is not None '
-            'else job_lib.get_latest_job_id()',
-            'log_dirs = job_lib.log_dirs_with_globbing(str(job_id))',
+            f'job_ids = {job_ids} if {job_ids} is not None '
+            'else [job_lib.get_latest_job_id()]',
+            'log_dirs = job_lib.log_dirs_with_globbing_json(job_ids)',
             'print(log_dirs, flush=True)',
         ]
         return cls._build(code)
diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py
index 829fa63440c..f97295f81c3 100644
--- a/sky/skylet/log_lib.py
+++ b/sky/skylet/log_lib.py
@@ -16,7 +16,7 @@
 
 from sky import sky_logging
 from sky.skylet import job_lib
-from sky.skylet.utils import log_utils
+from sky.utils import log_utils
 
 SKY_REMOTE_WORKDIR = '~/sky_workdir'
 _SKY_LOG_WAITING_GAP_SECONDS = 1
diff --git a/sky/skylet/utils/__init__.py b/sky/skylet/utils/__init__.py
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/sky/spot/__init__.py b/sky/spot/__init__.py
index 29c5d314a5d..032d1bfc132 100644
--- a/sky/spot/__init__.py
+++ b/sky/spot/__init__.py
@@ -14,6 +14,8 @@
 from sky.spot.spot_utils import SpotCodeGen
 from sky.spot.spot_utils import dump_job_table_cache
 from sky.spot.spot_utils import load_job_table_cache
+from sky.spot.spot_utils import format_job_table
+from sky.spot.spot_utils import load_spot_job_queue
 
 pathlib.Path(SPOT_TASK_YAML_PREFIX).expanduser().parent.mkdir(parents=True,
                                                               exist_ok=True)
@@ -32,4 +34,6 @@
     'SpotCodeGen',
     'dump_job_table_cache',
     'load_job_table_cache',
+    'format_job_table',
+    'load_spot_job_queue',
 ]
diff --git a/sky/spot/controller.py b/sky/spot/controller.py
index 640df8b6c07..852d6fa51f2 100644
--- a/sky/spot/controller.py
+++ b/sky/spot/controller.py
@@ -2,6 +2,7 @@
 import argparse
 import pathlib
 import time
+import traceback
 from typing import Optional
 
 import colorama
@@ -133,6 +134,7 @@ def start(self):
                 self._job_id,
                 failure_type=spot_state.SpotStatus.FAILED_NO_RESOURCE)
         except (Exception, SystemExit) as e:  # pylint: disable=broad-except
+            logger.error(traceback.format_exc())
             logger.error(f'Unexpected error occurred: {type(e).__name__}: {e}')
         finally:
             self._strategy_executor.terminate_cluster()
diff --git a/sky/spot/recovery_strategy.py b/sky/spot/recovery_strategy.py
index ef7636cfc4b..0949980f486 100644
--- a/sky/spot/recovery_strategy.py
+++ b/sky/spot/recovery_strategy.py
@@ -130,7 +130,12 @@ def terminate_cluster(self):
         handle = global_user_state.get_handle_from_cluster_name(
             self.cluster_name)
         if handle is not None:
-            self.backend.teardown(handle, terminate=True)
+            try:
+                self.backend.teardown(handle, terminate=True)
+            except RuntimeError:
+                logger.error(
+                    f'Failed to terminate the spot cluster {self.cluster_name}.'
+                )
 
 
 class FailoverStrategyExecutor(StrategyExecutor, name='FAILOVER', default=True):
diff --git a/sky/spot/spot_utils.py b/sky/spot/spot_utils.py
index f90214268f3..8218f2f56cb 100644
--- a/sky/spot/spot_utils.py
+++ b/sky/spot/spot_utils.py
@@ -5,7 +5,7 @@
 import pathlib
 import shlex
 import time
-from typing import List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 import colorama
 import filelock
@@ -16,7 +16,7 @@
 from sky import sky_logging
 from sky.backends import backend_utils
 from sky.skylet import job_lib
-from sky.skylet.utils import log_utils
+from sky.utils import log_utils
 from sky.spot import constants
 from sky.spot import spot_state
 from sky.utils import subprocess_utils
@@ -53,8 +53,12 @@ def get_job_status(backend: 'backends.CloudVmRayBackend',
     status = None
     try:
         logger.info('=== Checking the job status... ===')
-        status = backend.get_job_status(handle, stream_logs=False)
-        logger.info(f'Job status: {status}')
+        statuses = backend.get_job_status(handle, stream_logs=False)
+        status = list(statuses.values())[0]
+        if status is None:
+            logger.info('No job found.')
+        else:
+            logger.info(f'Job status: {status}')
     except exceptions.CommandError:
         logger.info('Failed to connect to the cluster.')
     logger.info('=' * 34)
@@ -121,7 +125,11 @@ def cancel_jobs_by_id(job_ids: Optional[List[int]]) -> str:
                 cluster_name)
             if handle is not None:
                 backend = backend_utils.get_backend_from_handle(handle)
-                backend.teardown(handle, terminate=True)
+                try:
+                    backend.teardown(handle, terminate=True)
+                except RuntimeError:
+                    logger.error('Failed to tear down the spot cluster '
+                                 f'{cluster_name!r}.')
 
             # The controller process for this job is not running: it must
             # have exited abnormally, and we should set the job status to
@@ -216,9 +224,9 @@ def stream_logs_by_id(job_id: int) -> str:
             # SUCCEEDED or FAILED), we can safely break the loop. We use the
             # status in job queue to show the information, as the spot_state is
             # not updated yet.
-            job_status = backend.get_job_status(handle,
-                                                job_id=None,
-                                                stream_logs=False)
+            job_statuses = backend.get_job_status(handle, stream_logs=False)
+            job_status = list(job_statuses.values())[0]
+            assert job_status is not None, 'No job found.'
             logger.info(f'Logs finished for job {job_id} '
                         f'(status: {job_status.value}).')
             break
@@ -251,10 +259,32 @@ def stream_logs_by_name(job_name: str) -> str:
     return ''
 
 
-def show_jobs(show_all: bool) -> str:
-    """Show all spot jobs."""
+def dump_spot_job_queue() -> str:
     jobs = spot_state.get_spot_jobs()
 
+    for job in jobs:
+        end_at = job['end_at']
+        if end_at is None:
+            end_at = time.time()
+        job_duration = end_at - job['last_recovered_at'] + job['job_duration']
+        if job['status'] == spot_state.SpotStatus.RECOVERING:
+            # When job is recovering, the duration is exact job['job_duration']
+            job_duration = job['job_duration']
+        job['job_duration'] = job_duration
+        job['status'] = job['status'].value
+    return json.dumps(jobs, indent=2)
+
+
+def load_spot_job_queue(json_str: str) -> List[Dict[str, Any]]:
+    """Load job queue from json string."""
+    jobs = json.loads(json_str)
+    for job in jobs:
+        job['status'] = spot_state.SpotStatus(job['status'])
+    return jobs
+
+
+def format_job_table(jobs: Dict[str, Any], show_all: bool) -> str:
+    """Show all spot jobs."""
     columns = [
         'ID', 'NAME', 'RESOURCES', 'SUBMITTED', 'TOT. DURATION', 'JOB DURATION',
         '#RECOVERIES', 'STATUS'
@@ -262,6 +292,7 @@ def show_jobs(show_all: bool) -> str:
     if show_all:
         columns += ['STARTED', 'CLUSTER', 'REGION']
     job_table = log_utils.create_table(columns)
+
     status_counts = collections.defaultdict(int)
     for job in jobs:
         job_duration = log_utils.readable_time_duration(
@@ -332,10 +363,10 @@ class SpotCodeGen:
     ]
 
     @classmethod
-    def show_jobs(cls, show_all: bool) -> str:
+    def get_job_table(cls) -> str:
         code = [
-            f'job_table = spot_utils.show_jobs({show_all})',
-            'print(job_table)',
+            'job_table = spot_utils.dump_spot_job_queue()',
+            'print(job_table, flush=True)',
         ]
         return cls._build(code)
 
diff --git a/sky/utils/cli_utils/status_utils.py b/sky/utils/cli_utils/status_utils.py
index 924b5e2756b..e0ca1019f19 100644
--- a/sky/utils/cli_utils/status_utils.py
+++ b/sky/utils/cli_utils/status_utils.py
@@ -1,6 +1,5 @@
-"""Utilities for `sky status`."""
-from typing import Callable, List
-
+"""Utilities for sky status."""
+from typing import Any, Callable, Dict, List
 import click
 import colorama
 
@@ -8,7 +7,7 @@
 from sky.backends import backend_utils
 from sky.utils import common_utils
 from sky.utils.cli_utils import cli_utils
-from sky.skylet.utils import log_utils
+from sky.utils import log_utils
 
 _COMMAND_TRUNC_LENGTH = 25
 
@@ -33,10 +32,9 @@ def calc(self, record):
         return val
 
 
-def show_status_table(show_all: bool = False, refresh: bool = False):
+def show_status_table(cluster_records: List[Dict[str, Any]], show_all: bool):
     """Compute cluster table values and display."""
     # TODO(zhwu): Update the information for auto-stop clusters.
-    cluster_records = backend_utils.get_clusters(show_all, refresh)
 
     status_columns = [
         StatusColumn('NAME', _get_name),
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index 9b0ccb9639e..c64948f50d1 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -114,7 +114,7 @@ def __init__(
         Example Usage:
             runner = SSHCommandRunner(ip, ssh_user, ssh_private_key)
             runner.run('ls -l', mode=SshMode.NON_INTERACTIVE)
-            runner.rsync_up(source, target)
+            runner.rsync(source, target, up=True)
 
         Args:
             ip: The IP address of the remote machine.
@@ -261,11 +261,12 @@ def run(
                                     executable=executable,
                                     **kwargs)
 
-    def rsync_up(
+    def rsync(
         self,
         source: str,
         target: str,
         *,
+        up: bool,
         # Advanced options.
         log_path: str = os.devnull,
         stream_logs: bool = True,
@@ -291,16 +292,28 @@ def rsync_up(
         ssh_options = ' '.join(
             ssh_options_list(self.ssh_private_key, self.ssh_control_name))
         rsync_command.append(f'-e "ssh {ssh_options}"')
-        rsync_command.extend([
-            source,
-            f'{self.ssh_user}@{self.ip}:{target}',
-        ])
+        if up:
+            rsync_command.extend([
+                source,
+                f'{self.ssh_user}@{self.ip}:{target}',
+            ])
+        else:
+            rsync_command.extend([
+                f'{self.ssh_user}@{self.ip}:{source}',
+                target,
+            ])
         command = ' '.join(rsync_command)
 
-        returncode = log_lib.run_with_log(command,
-                                          log_path=log_path,
-                                          stream_logs=stream_logs,
-                                          shell=True)
+        returncode, _, stderr = log_lib.run_with_log(command,
+                                                     log_path=log_path,
+                                                     stream_logs=stream_logs,
+                                                     shell=True,
+                                                     require_outputs=True)
+
+        direction = 'up' if up else 'down'
         subprocess_utils.handle_returncode(
-            returncode, command, f'Failed to rsync up {source} -> {target}, '
-            f'see {log_path} for details.')
+            returncode,
+            command,
+            f'Failed to rsync {direction}: {source} -> {target}',
+            stderr=stderr,
+            stream_logs=stream_logs)
diff --git a/sky/skylet/utils/db_utils.py b/sky/utils/db_utils.py
similarity index 100%
rename from sky/skylet/utils/db_utils.py
rename to sky/utils/db_utils.py
diff --git a/sky/skylet/utils/log_utils.py b/sky/utils/log_utils.py
similarity index 100%
rename from sky/skylet/utils/log_utils.py
rename to sky/utils/log_utils.py
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 3b648cc3313..01b2a984019 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -131,7 +131,7 @@ def test_minimal():
             f'sky logs {name} 1 --status',  # Ensure the job succeeded.
             f'sky launch -y -c {name} examples/minimal.yaml',
             f'sky logs {name} 2 --status',
-            f'sky logs {name} --status | grep "Job 2 SUCCEEDED"',  # Equivalent.
+            f'sky logs {name} --status | grep "Job 2: SUCCEEDED"',  # Equivalent.
         ],
         f'sky down -y {name}',
     )
@@ -204,6 +204,28 @@ def test_file_mounts():
     run_one_test(test)
 
 
+# ---------- CLI logs ----------
+def test_logs():
+    name = _get_cluster_name()
+    timestamp = time.time()
+    test = Test(
+        'cli_logs',
+        [
+            f'sky launch -y -c {name} --num-nodes 2 "echo {timestamp} 1"',
+            f'sky exec {name} "echo {timestamp} 2"',
+            f'sky exec {name} "echo {timestamp} 3"',
+            f'sky exec {name} "echo {timestamp} 4"',
+            f'sky logs {name} 2 --status',
+            f'sky logs {name} 3 4 --sync-down',
+            f'sky logs {name} * --sync-down',
+            f'sky logs {name} 1 | grep "{timestamp} 1"',
+            f'sky logs {name} | grep "{timestamp} 4"',
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
 # ---------- Job Queue. ----------
 def test_job_queue():
     name = _get_cluster_name()
@@ -508,8 +530,8 @@ def test_spot():
             f'sky spot status | grep {name}-2 | head -n1 | grep STARTING',
             f'sky spot cancel -y -n {name}-1',
             'sleep 200',
-            f'sky spot status | grep {name}-1 | head -n1 | grep CANCELLED',
-            f'sky spot status | grep {name}-2 | head -n1 | grep "RUNNING\|SUCCEEDED"',
+            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name}-1 | head -n1 | grep CANCELLED',
+            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name}-2 | head -n1 | grep "RUNNING\|SUCCEEDED"',
         ],
         f'sky spot cancel -y -n {name}-1; sky spot cancel -y -n {name}-2',
     )
@@ -554,7 +576,7 @@ def test_spot_recovery():
             f'--query Reservations[].Instances[].InstanceId '
             '--output text)',
             'sleep 50',
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RECOVERING\|STARTING"',
+            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RECOVERING"',
             'sleep 200',
             f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RUNNING"',
         ],
@@ -656,6 +678,15 @@ def test_azure_start_stop_two_nodes():
     run_one_test(test)
 
 
+# ------- Testing the core API --------
+# Most of the core APIs have been tested in the CLI tests.
+# These tests are for testing the return value of the APIs not fully used in CLI.
+def test_core_api():
+    name = _get_cluster_name()
+    sky.launch
+    # TODO(zhwu): Add a test for core api.
+
+
 # ---------- Testing Storage ----------
 class TestStorageWithCredentials:
     """Storage tests which require credentials and network connection"""
diff --git a/tests/test_spot.py b/tests/test_spot.py
index 382842bc960..b9ac03b9ad9 100644
--- a/tests/test_spot.py
+++ b/tests/test_spot.py
@@ -10,7 +10,7 @@
 from sky import cli
 from sky import global_user_state
 from sky import spot
-from sky.skylet.utils import db_utils
+from sky.utils import db_utils
 
 
 def test_spot_nonexist_strategy():
@@ -153,5 +153,5 @@ def test_autostop_spot_controller(self, _mock_cluster_state):
     def test_cancel_on_spot_controller(self, _mock_cluster_state):
         cli_runner = cli_testing.CliRunner()
         result = cli_runner.invoke(cli.cancel, ['sky-spot-controller', '-a'])
-        assert isinstance(result.exception, ValueError)
-        assert 'Cancelling jobs is not allowed' in str(result.exception)
+        assert result.exit_code == click.UsageError.exit_code
+        assert 'Cancelling jobs is not allowed' in str(result.output)


Code review comments:
diff_hunk: @@ -252,6 +247,29 @@ def _parse_env_var(env_var: str) -> Tuple[str, str]:
         type=str,
         help=('The region to use. If specified, overrides the '
               '"resources.region" config. Passing "none" resets the config.')),
+    click.option(
+        '--gpus',
+        required=False,
+        type=str,
+        help=
+        ('Type and number of GPUs to use. Example values: '
+         '"V100:8", "V100" (short for a count of 1), or "V100:0.5" '
+         '(fractional counts are supported by the scheduling framework). '
+         'If a new cluster is being launched by this command, this is the '
+         'resources to provision. If an existing cluster is being reused, this'
+         ' is seen as the task demand, which must fit the cluster\'s total '
+         'resources and is used for scheduling the task. '
+         'Overrides the "accelerators" '
+         'config in the YAML if both are supplied. '
+         'Passing "none" resets the config.')),
+    click.option(
+        '--instance-type',
+        required=False,
+        type=str,
+        help=('The instance type to use. If specified, overrides the '
+              '"resources.instance_type" config. Passing "none" resets the '
+              'config.'),
+    ),
     click.option(
path: sky/cli.py
text: @user1:
Good to have this here, but `_TASK_OPTIONS` is also applied to `benchmark_launch`, which does not take an `instance_type`. Can we selectively remove it from benchmark_launch or figure out some other workaround?

@author:
Ahh, good point! I renamed the `_GPUS_OPTION` to `_EXTRA_RESOURCES_OPTIONS` and moved the `instance_type` there.

diff_hunk: @@ -1167,61 +1170,52 @@ def _show_job_queue_on_cluster(cluster: str, handle: Optional[Any],
     help=('If specified, do not show logs but exit with a status code for the '
           'job\'s status: 0 for succeeded, or 1 for all other statuses.'))
 @click.argument('cluster', required=True, type=str)
-@click.argument('job_id', required=False, type=str)
+@click.argument('job_ids', type=str, nargs=-1)
 # TODO(zhwu): support logs by job name
 @usage_lib.entrypoint
-def logs(cluster: str, job_id: Optional[str], sync_down: bool, status: bool):  # pylint: disable=redefined-outer-name
+def logs(cluster: str, job_ids: Tuple[str], sync_down: bool, status: bool):  # pylint: disable=redefined-outer-name
     """Tail the log of a job.
 
     If JOB_ID is not provided, tails the logs of the last job on the cluster.
     """
-    cluster_name = cluster
-    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
-        cluster_name)
-    if handle is None:
-        if onprem_utils.check_if_local_cloud(cluster):
-            raise click.BadParameter(
-                _UNINITIALIZED_CLUSTER_MESSAGE.format(cluster=cluster_name))
-        raise click.BadParameter(f'Cluster \'{cluster_name}\' not found'
-                                 ' (see `sky status`).')
-    backend = backend_utils.get_backend_from_handle(handle)
-    if isinstance(backend, backends.LocalDockerBackend):
-        raise click.UsageError('Logs are not available with '
-                               'LocalDockerBackend.')
-    if cluster_status != global_user_state.ClusterStatus.UP:
-        click.secho(
-            f'Cluster {cluster_name} is not up '
-            f'(status: {cluster_status.value}).',
-            fg='yellow')
-        return
-
     if sync_down and status:
         raise click.UsageError(
-            'Both --sync_down and --status are specified '
+            'Both sync_down and status are specified '
             '(ambiguous). To fix: specify at most one of them.')
 
+    if len(job_ids) > 1 and not sync_down:
+        raise click.UsageError(
+            f'Cannot stream logs of multiple jobs {job_ids}. '
+            'Set --sync_down to download them.')
path: sky/cli.py
text: @user1:
nit: `--sync-down`

@author:
Fixed. Thanks!

diff_hunk: @@ -0,0 +1,590 @@
+"""SDK functions for cluster/job management."""
+import colorama
+import getpass
+from typing import Any, Dict, List, Optional, Tuple
+
+from sky import dag
+from sky import task
+from sky import backends
+from sky import constants
+from sky import data
+from sky import exceptions
+from sky import global_user_state
+from sky import sky_logging
+from sky import spot
+from sky.backends import backend_utils
+from sky.backends import onprem_utils
+from sky.skylet import job_lib
+from sky.utils import ux_utils
+from sky.utils import subprocess_utils
+
+logger = sky_logging.init_logger(__name__)
+
+# ======================
+# = Cluster Management =
+# ======================
+
+
+# pylint: disable=redefined-builtin
+def status(all: bool, refresh: bool) -> List[Dict[str, Any]]:
+    """Get the cluster status in dict.
+
+    Please refer to the sky.cli.status for the document.
+
+    Returns:
+        List[dict]:
+        [
+            {
+                'name': (str) cluster name,
+                'launched_at': (int) timestamp of launched,
+                'last_use': (int) timestamp of last use,
+                'status': (sky.ClusterStatus) cluster status,
+                'autostop': (int) idle time before autostop,
+                'metadata': (dict) metadata of the cluster,
+            }
+        ]
+    """
+    cluster_records = backend_utils.get_clusters(all, refresh)
+    return cluster_records
+
+
+def _start(cluster_name: str,
+           idle_minutes_to_autostop: Optional[int] = None,
+           retry_until_up: bool = False) -> backends.Backend.ResourceHandle:
+
+    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
+        cluster_name)
+    if cluster_status == global_user_state.ClusterStatus.UP:
+        logger.info(f'Cluster {cluster_name!r} is already up.')
+        return
+    assert cluster_status in (
+        global_user_state.ClusterStatus.INIT,
+        global_user_state.ClusterStatus.STOPPED), cluster_status
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    if not isinstance(backend, backends.CloudVmRayBackend):
+        raise exceptions.NotSupportedError(
+            f'Starting cluster {cluster_name!r} with backend {backend.NAME} '
+            'is not supported.')
+    with dag.Dag():
+        dummy_task = task.Task().set_resources(handle.launched_resources)
+        dummy_task.num_nodes = handle.launched_nodes
+    handle = backend.provision(dummy_task,
+                               to_provision=handle.launched_resources,
+                               dryrun=False,
+                               stream_logs=True,
+                               cluster_name=cluster_name,
+                               retry_until_up=retry_until_up)
+    if idle_minutes_to_autostop is not None:
+        backend.set_autostop(handle, idle_minutes_to_autostop)
+    return handle
+
+
+def start(cluster_name: str,
+          idle_minutes_to_autostop: Optional[int] = None,
+          retry_until_up: bool = False):
+    """Start the cluster.
+
+    Please refer to the sky.cli.start for the document.
+
+    Raises:
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+    """
+    _start(cluster_name, idle_minutes_to_autostop, retry_until_up)
+
+
+def stop(cluster_name: str, purge: bool = False):
+    """Stop the cluster
+
+    Please refer to the sky.cli.stop for the document.
+
+    Raises:
+        ValueError: cluster does not exist.
+        RuntimeError: Fail to stop the cluster.
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+    """
+    if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+        raise exceptions.NotSupportedError(
+            f'Stopping sky reserved cluster {cluster_name!r} '
+            f'is not supported.')
+    handle = global_user_state.get_handle_from_cluster_name(cluster_name)
+    if handle is None:
+        raise ValueError(f'Cluster {cluster_name!r} does not exist.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    if (isinstance(backend, backends.CloudVmRayBackend) and
+            handle.launched_resources.use_spot):
+        # Disable spot instances to be stopped.
+        # TODO(suquark): enable GCP+spot to be stopped in the future.
+        raise exceptions.NotSupportedError(
+            f'{colorama.Fore.YELLOW}Stopping cluster '
+            f'{cluster_name!r}... skipped.{colorama.Style.RESET_ALL}\n'
+            '  Stopping spot instances is not supported as the attached '
+            'disks will be lost.\n'
+            '  To terminate the cluster instead, run: '
+            f'{colorama.Style.BRIGHT}sky down {cluster_name}')
+    backend.teardown(handle, terminate=False, purge=purge)
+
+
+def down(cluster_name: str, purge: bool = False):
+    """Down the cluster
+
+    Please refer to the sky.cli.down for the document.
+
+    Raises:
+        ValueError: cluster does not exist.
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+    """
+    if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+        raise exceptions.NotSupportedError(
+            f'Tearing down sky reserved cluster {cluster_name!r} '
+            f'is not supported.')
+    handle = global_user_state.get_handle_from_cluster_name(cluster_name)
+    if handle is None:
+        raise ValueError(f'Cluster {cluster_name!r} does not exist.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    backend.teardown(handle, terminate=True, purge=purge)
+
+
+def autostop(cluster_name: str, idle_minutes_to_autostop: int):
+    """Set the autostop time of the cluster.
+
+    Please refer to the sky.cli.autostop for the document.
+
+    Args:
+        cluster_name (str): name of the cluster.
+        autostop_time (int): autostop time in minutes.
+    Raises:
+        ValueError: cluster does not exist.
+        sky.exceptions.NotSupportedError: the cluster is not supported.
+        sky.exceptions.ClusterNotUpError: the cluster is not UP.
+    """
+    verb = 'Scheduling' if idle_minutes_to_autostop >= 0 else 'Cancelling'
+    operation = f'{verb} auto-stop on'
+    if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+        raise exceptions.NotSupportedError(
+            f'{operation} sky reserved cluster {cluster_name!r} '
+            f'is not supported.')
+    (cluster_status,
+     handle) = backend_utils.refresh_cluster_status_handle(cluster_name)
+    if handle is None:
+        raise ValueError(f'Cluster {cluster_name!r} does not exist.')
+
+    backend = backend_utils.get_backend_from_handle(handle)
+    if not isinstance(backend, backends.CloudVmRayBackend):
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'{colorama.Fore.YELLOW}{operation} cluster '
+                f'{cluster_name!r}... skipped{colorama.Style.RESET_ALL}'
+                '\n  Auto-stopping is only supported by backend: '
+                f'{backends.CloudVmRayBackend.NAME}')
+    if cluster_status != global_user_state.ClusterStatus.UP:
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.ClusterNotUpError(
+                f'{colorama.Fore.YELLOW}{operation} cluster '
+                f'{cluster_name!r} (status: {cluster_status.value})... skipped'
+                f'{colorama.Style.RESET_ALL}'
+                '\n  Auto-stop can only be set/unset for '
+                f'{global_user_state.ClusterStatus.UP.value} clusters.')
+    backend.set_autostop(handle, idle_minutes_to_autostop)
+
+
+# ==================
+# = Job Management =
+# ==================
+
+
+def _check_cluster_available(cluster_name: str,
+                             operation: str) -> backends.Backend.ResourceHandle:
+    """Check if the cluster is available."""
+    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
+        cluster_name)
+    backend = backend_utils.get_backend_from_handle(handle)
+    if isinstance(backend, backends.LocalDockerBackend):
+        # LocalDockerBackend does not support job queues
+        raise exceptions.NotSupportedError(
+            f'Cluster {cluster_name} with LocalDockerBackend does '
+            f'not support {operation}.')
+    if cluster_status != global_user_state.ClusterStatus.UP:
+        if onprem_utils.check_if_local_cloud(cluster_name):
+            raise exceptions.ClusterNotUpError(
+                constants.UNINITIALIZED_ONPREM_CLUSTER_MESSAGE.format(
+                    cluster_name))
+        raise exceptions.ClusterNotUpError(
+            f'{colorama.Fore.YELLOW}Cluster {cluster_name!r} is not up '
+            f'(status: {cluster_status.value}); skipped.'
+            f'{colorama.Style.RESET_ALL}')
+
+    if handle.head_ip is None:
+        raise exceptions.ClusterNotUpError(
+            f'Cluster {cluster_name!r} has been stopped or not properly set up.'
+            ' Please re-launch it with `sky start`.')
+    return handle
+
+
+def queue(cluster_name: str,
+          skip_finished: bool = False,
+          all_users: bool = False):
path: sky/core.py
text: @user1:
Add return type `-> List[dict]`?

@author:
Good catch! Fixed. Thanks!

diff_hunk: @@ -252,6 +247,29 @@ def _parse_env_var(env_var: str) -> Tuple[str, str]:
         type=str,
         help=('The region to use. If specified, overrides the '
               '"resources.region" config. Passing "none" resets the config.')),
+    click.option(
path: sky/cli.py
text: @user1:
Isn't this already included in `_GPUS_OPTION`?

@author:
Ahh, good catch! That was a merge error. Removed it. Thanks!

diff_hunk: @@ -1167,61 +1170,52 @@ def _show_job_queue_on_cluster(cluster: str, handle: Optional[Any],
     help=('If specified, do not show logs but exit with a status code for the '
           'job\'s status: 0 for succeeded, or 1 for all other statuses.'))
 @click.argument('cluster', required=True, type=str)
-@click.argument('job_id', required=False, type=str)
+@click.argument('job_ids', type=str, nargs=-1)
 # TODO(zhwu): support logs by job name
 @usage_lib.entrypoint
-def logs(cluster: str, job_id: Optional[str], sync_down: bool, status: bool):  # pylint: disable=redefined-outer-name
+def logs(cluster: str, job_ids: Tuple[str], sync_down: bool, status: bool):  # pylint: disable=redefined-outer-name
     """Tail the log of a job.
 
     If JOB_ID is not provided, tails the logs of the last job on the cluster.
     """
-    cluster_name = cluster
-    cluster_status, handle = backend_utils.refresh_cluster_status_handle(
-        cluster_name)
-    if handle is None:
-        if onprem_utils.check_if_local_cloud(cluster):
-            raise click.BadParameter(
-                _UNINITIALIZED_CLUSTER_MESSAGE.format(cluster=cluster_name))
-        raise click.BadParameter(f'Cluster \'{cluster_name}\' not found'
-                                 ' (see `sky status`).')
-    backend = backend_utils.get_backend_from_handle(handle)
-    if isinstance(backend, backends.LocalDockerBackend):
-        raise click.UsageError('Logs are not available with '
-                               'LocalDockerBackend.')
-    if cluster_status != global_user_state.ClusterStatus.UP:
-        click.secho(
-            f'Cluster {cluster_name} is not up '
-            f'(status: {cluster_status.value}).',
-            fg='yellow')
-        return
-
     if sync_down and status:
         raise click.UsageError(
-            'Both --sync_down and --status are specified '
+            'Both sync_down and status are specified '
path: sky/cli.py
text: @user1:
It is a little confusing what sync_down and status are. Either change to `Both sync_down and status flags are specified` or `Both --sync_down and --status are specified `

@author:
Oops, great catch! Fixed with `--`. Thanks!

diff_hunk: @@ -827,7 +827,7 @@ def ssh_credential_from_yaml(cluster_yaml: str) -> Tuple[str, str, str]:
     return ssh_user, ssh_private_key, ssh_control_name
 
 
-def parallel_data_transfer_to_nodes(
+def parallel_data_transfer_with_nodes(
path: sky/backends/backend_utils.py
text: @user1:
hmm isn't `to_nodes` more appropriate than `with_nodes`?

@author:
oops, that was a remnant. Reverted. Thanks!

diff_hunk: @@ -94,8 +94,8 @@ def teardown_ephemeral_storage(self, task: 'task_lib.Task') -> None:
     def teardown(self,
                  handle: ResourceHandle,
                  terminate: bool,
-                 purge: bool = False) -> bool:
-        return self._teardown(handle, terminate, purge)
+                 purge: bool = False):
path: sky/backends/backend.py
text: @user1:
Can we have this return `-> None` to be consistent with other methods in this class?

@author:
Good catch! Fixed. Thanks!