Repository: skypilot-org/skypilot
Pull request #1115 (2022-08-24 04:07:09)

[Onprem] Script for Automatic setup/launching for Onprem servers
Should be tested after #1114 and #1116 are merged. This PR releases a script that automatically launches a cloud VM (acting as an onprem server) and sets up a regular user called `test` on the cloud server. It will also run `sky admin deploy` and setup the local cluster config as needed.

## Example

`python examples/local/launch_cloud_onprem.py -- local-cluster-name hello`
`sky launch -c hello -- 'echo hi'`

michaelzhiluo: @romilbhardwaj Just tested the script. It works as of now.
michaelzhiluo: @romilbhardwaj Done, think its ready to be merge 👍 

Patch:
diff --git a/docs/source/examples/spot-jobs.rst b/docs/source/examples/spot-jobs.rst
index 17656f924..8c1c4fad8 100644
--- a/docs/source/examples/spot-jobs.rst
+++ b/docs/source/examples/spot-jobs.rst
@@ -188,8 +188,8 @@ Here are some commands for managed spot jobs. Check :code:`sky spot --help` for
 .. code-block:: console
 
     # Check the status of the spot jobs
-    $ sky spot status
-    Fetching managed spot job status...
+    $ sky spot queue
+    Fetching managed spot job statuses...
     Managed spot jobs:
     ID NAME     RESOURCES     SUBMITTED   TOT. DURATION   JOB DURATION   #RECOVERIES  STATUS
     2  roberta  1x [A100:8]   2 hrs ago   2h 47m 18s      2h 36m 18s     0            RUNNING
@@ -206,13 +206,13 @@ Spot controller (Advanced)
 -------------------------------
 
 There will be a single spot controller VM (a small on-demand CPU VM) running in the background to manage all the spot jobs.
-It will be autostopped after all spot jobs finished and no new spot job is submitted for 30 minutes. Typically **no user intervention** is needed. 
-You can find the controller with :code:`sky status -a`, and refresh the status with :code:`sky status -ar`.
+It will be autostopped after all spot jobs finished and no new spot job is submitted for 10 minutes. Typically **no user intervention** is needed. 
+You can find the controller with :code:`sky status`, and refresh the status with :code:`sky status -r`.
 
 Although, the cost of the spot controller is negligible (~$0.4/hour when running and less than $0.004/hour when stopped), 
 you can still tear it down manually with 
-:code:`sky down -p sky-spot-controller-<hash>`, where the ``<hash>`` can be found in the output of :code:`sky status -a`.
+:code:`sky down <spot-controller-name>`, where the ``<spot-controller-name>`` can be found in the output of :code:`sky status`.
 
 .. note::
-  Tearing down the spot controller when there are still spot jobs running will cause resource leakage of those spot VMs.
+  Tearing down the spot controller will lose all logs and status information for the spot jobs and can cause resource leakage when there are still in-progress spot jobs.
 
diff --git a/docs/source/index.rst b/docs/source/index.rst
index bd9ae46fc..4912cec73 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -81,9 +81,10 @@ Use the clouds **easily** and **cost effectively**, without needing cloud infra
 
 .. toctree::
    :maxdepth: 1
-   :caption: SkyPilot CLI
+   :caption: API References
 
    reference/cli
+   reference/api
 
 
 .. .. toctree::
diff --git a/docs/source/reference/api.rst b/docs/source/reference/api.rst
new file mode 100644
index 000000000..923e9560b
--- /dev/null
+++ b/docs/source/reference/api.rst
@@ -0,0 +1,71 @@
+.. _pythonapi:
+
+Python API
+=================
+
+SkyPilot offers a programmatic API in Python, which is used under the hood by the :ref:`CLI <cli>`.
+
+.. note::
+
+  The Python API contains more experimental functions/classes than the CLI. That
+  said, it has been used to develop several Python libraries by users.
+
+  For questions or request for support, please reach out to the development team.
+  Your feedback is much appreciated in evolving this API!
+
+
+Core API
+-----------
+
+sky.launch
+~~~~~~~~
+
+.. autofunction:: sky.launch
+
+sky.exec
+~~~~~~~~
+
+.. autofunction:: sky.exec
+
+sky.stop
+~~~~~~~~~
+
+.. autofunction:: sky.stop
+
+sky.start
+~~~~~~~~~~~~~
+
+.. autofunction:: sky.start
+
+sky.down
+~~~~~~~~~
+
+.. autofunction:: sky.down
+
+sky.status
+~~~~~~~~~~~~~
+
+.. autofunction:: sky.status
+
+sky.autostop
+~~~~~~~~~~~~~
+
+.. autofunction:: sky.autostop
+
+.. _sky-dag-ref:
+
+Task and DAG
+-----------------
+
+
+sky.Task
+~~~~~~~~~
+
+.. autoclass:: sky.Task
+    :members:
+
+sky.Dag
+~~~~~~~~~
+
+.. autoclass:: sky.Dag
+    :members:
diff --git a/docs/source/reference/cli.rst b/docs/source/reference/cli.rst
index 53da6ae5c..ce1d2a7d7 100644
--- a/docs/source/reference/cli.rst
+++ b/docs/source/reference/cli.rst
@@ -1,6 +1,6 @@
 .. _cli:
 
-CLI Reference
+Command Line Interface
 =============
 
 Core CLI
@@ -14,14 +14,14 @@ Core CLI
    :prog: sky exec
    :nested: full
 
-.. click:: sky.cli:start
-   :prog: sky start
-   :nested: full
-
 .. click:: sky.cli:stop
    :prog: sky stop
    :nested: full
 
+.. click:: sky.cli:start
+   :prog: sky start
+   :nested: full
+
 .. click:: sky.cli:down
    :prog: sky down
    :nested: full
@@ -86,8 +86,8 @@ Managed Spot Jobs CLI
    :prog: sky spot launch
    :nested: full
 
-.. click:: sky.cli:spot_status
-   :prog: sky spot status
+.. click:: sky.cli:spot_queue
+   :prog: sky spot queue
    :nested: full
 
 .. click:: sky.cli:spot_cancel
diff --git a/examples/huggingface_glue_imdb_grid_search_app.py b/examples/huggingface_glue_imdb_grid_search_app.py
index 22adb8d92..857f21f23 100644
--- a/examples/huggingface_glue_imdb_grid_search_app.py
+++ b/examples/huggingface_glue_imdb_grid_search_app.py
@@ -17,30 +17,29 @@ with sky.Dag() as dag:
 sky.launch(dag, cluster_name='hgs', detach_run=True)
 
 for lr in [1e-5, 2e-5, 3e-5, 4e-5]:
-    with sky.Dag() as dag:
-        # To be filled in: {lr}.
-        run_format = f"""\
-            cd transformers/examples/pytorch/text-classification
-            python3 run_glue.py
-                --learning_rate {lr}
-                --output_dir /tmp/imdb-{lr}/
-                --model_name_or_path bert-base-cased
-                --dataset_name imdb
-                --do_train
-                --max_seq_length 128
-                --per_device_train_batch_size 32
-                --max_steps 50
-                --fp16 --overwrite_output_dir 2>&1 | tee run-{lr}.log'
-            """
+    # To be filled in: {lr}.
+    run_format = f"""\
+        cd transformers/examples/pytorch/text-classification
+        python3 run_glue.py
+            --learning_rate {lr}
+            --output_dir /tmp/imdb-{lr}/
+            --model_name_or_path bert-base-cased
+            --dataset_name imdb
+            --do_train
+            --max_seq_length 128
+            --per_device_train_batch_size 32
+            --max_steps 50
+            --fp16 --overwrite_output_dir 2>&1 | tee run-{lr}.log'
+        """
 
-        per_trial_resources = sky.Resources(accelerators={'V100': 1})
+    per_trial_resources = sky.Resources(accelerators={'V100': 1})
 
-        task = sky.Task(
-            # A descriptive name.
-            f'task-{lr}',
-            # Run command for each task, with different lr.
-            run=run_format.format(lr=lr)).set_resources(per_trial_resources)
+    task = sky.Task(
+        # A descriptive name.
+        f'task-{lr}',
+        # Run command for each task, with different lr.
+        run=run_format.format(lr=lr)).set_resources(per_trial_resources)
 
     # Set 'stream_logs=False' to not mix all tasks' outputs together.
     # Each task's output is redirected to run-{lr}.log and can be tail-ed.
-    sky.exec(dag, cluster_name='hgs', stream_logs=False, detach_run=True)
+    sky.exec(task, cluster_name='hgs', stream_logs=False, detach_run=True)
diff --git a/examples/local/launch_cloud_onprem.py b/examples/local/launch_cloud_onprem.py
new file mode 100644
index 000000000..b937381a1
--- /dev/null
+++ b/examples/local/launch_cloud_onprem.py
@@ -0,0 +1,106 @@
+"""Automatically launches and sets up a Sky onprem cluster on AWS.
+
+The script allocates a cloud cluster and setups up the cloud cluster to run an
+'onprem' cluster on the cloud. It performs the following steps:
+    1) Launches a cluster on the cloud with `sky launch -c
+       on-prem-cluster-[ID]`. The Sky yaml sets up another user named `test`.
+    2) Setups Sky dependencies for the admin user via `sky admin deploy ..`.
+    3) Creates the local cluster config for the regular user, stored in
+       `~/.sky/local/...`.
+To clean up the local cluster and the launched cloud cluster, run `sky down
+[LOCAL_CLUSTER] onprem-cluster-[ID]` and remove the corresponding local
+cluster config file.
+
+Usage:
+    # Creates a Sky on-premise cluster named `my-local-cluster`
+    python examples/local/launch_cloud_onprem.py my-local-cluster
+"""
+
+import argparse
+import os
+import subprocess
+import tempfile
+import textwrap
+import uuid
+import yaml
+
+from click import testing as cli_testing
+
+from sky import cli
+from sky import global_user_state
+from sky.backends import onprem_utils
+from sky.utils import common_utils
+
+parser = argparse.ArgumentParser()
+parser.add_argument('name',
+                    type=str,
+                    default='my-local-cluster',
+                    nargs='?',
+                    help='Name of the local cluster.')
+args = parser.parse_args()
+
+# Public and private keys (autogenerated by Sky)
+public_key = '~/.ssh/sky-key.pub'
+private_key = '~/.ssh/sky-key'
+
+local_cluster_name = args.name
+
+# Sky Task YAML to setup the user 'test'
+sky_yaml_config = {
+    'resources': {
+        'cloud': 'aws'
+    },
+    'file_mounts': {
+        '/user-key': public_key
+    },
+    'setup': textwrap.dedent("""\
+              sudo adduser --disabled-password --gecos '' test
+              sudo -H su test -c 'mkdir ~/.ssh'
+              sudo -H su test -c 'chmod 700 ~/.ssh'
+              sudo -H su test -c 'touch ~/.ssh/authorized_keys'
+              sudo -H su test -c 'chmod 600 ~/.ssh/authorized_keys'
+              sudo -H su test -c 'cat /user-key >> ~/.ssh/authorized_keys'
+            """)
+}
+
+cli_runner = cli_testing.CliRunner()
+onprem_name = f'onprem-cluster-{uuid.uuid4().hex[:6]}'
+with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w') as f:
+    yaml.dump(sky_yaml_config, f)
+    file_path = f.name
+    cli_runner.invoke(cli.launch, ['-c', onprem_name, file_path])
+
+handle = global_user_state.get_handle_from_cluster_name(onprem_name)
+head_ip = handle.head_ip
+
+# Cluster config YAML to setup Sky Onprem on the admin account 'ubuntu'
+cluster_yaml_config = {
+    'cluster': {
+        'ips': [head_ip],
+        'name': local_cluster_name
+    },
+    'auth': {
+        'ssh_user': 'ubuntu',
+        'ssh_private_key': private_key
+    }
+}
+
+with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w') as f:
+    yaml.dump(cluster_yaml_config, f)
+    file_path = f.name
+    subprocess.run(f'sky admin deploy {file_path}', shell=True, check=True)
+
+# Fill out the local config file in ~/.sky/local/...
+local_config = onprem_utils.get_local_cluster_config_or_error(
+    local_cluster_name)
+local_config['auth']['ssh_user'] = 'test'
+local_config['auth']['ssh_private_key'] = private_key
+local_config_path = onprem_utils.SKY_USER_LOCAL_CONFIG_PATH.format(
+    local_cluster_name)
+common_utils.dump_yaml(os.path.expanduser(local_config_path), local_config)
+
+print((
+    f'Sky onprem cluster {local_cluster_name} is now ready for use! '
+    f'You can launch jobs with `sky launch -c {local_cluster_name} -- [CMD].` '
+    f'After you are done, shut down the cluster by running `sky down {local_cluster_name} {onprem_name}` '
+    f'and removing the local cluster config in {local_config_path}.'))
diff --git a/examples/multi_echo.py b/examples/multi_echo.py
index f27117e47..933a36108 100644
--- a/examples/multi_echo.py
+++ b/examples/multi_echo.py
@@ -27,11 +27,10 @@ def run(cluster: Optional[str] = None):
 
     # Submit multiple tasks in parallel to trigger queueing behaviors.
     def _exec(i):
-        with sky.Dag() as dag:
-            task = sky.Task(run=f'echo {i}; sleep 5')
-            resources = sky.Resources(accelerators={'K80': 0.5})
-            task.set_resources(resources)
-        sky.exec(dag, cluster_name=cluster, detach_run=True)
+        task = sky.Task(run=f'echo {i}; sleep 5')
+        resources = sky.Resources(accelerators={'K80': 0.5})
+        task.set_resources(resources)
+        sky.exec(task, cluster_name=cluster, detach_run=True)
 
     with pool.ThreadPool(8) as p:
         list(p.imap(_exec, range(32)))
diff --git a/examples/resnet_app.py b/examples/resnet_app.py
index 60cf03e73..9cd3276ea 100644
--- a/examples/resnet_app.py
+++ b/examples/resnet_app.py
@@ -2,83 +2,87 @@ import subprocess
 
 import sky
 
-with sky.Dag() as dag:
-    # The working directory contains all code and will be synced to remote.
-    workdir = '~/Downloads/tpu'
-    subprocess.run(f'cd {workdir} && git checkout 222cc86',
-                   shell=True,
-                   check=True)
+# The working directory contains all code and will be synced to remote.
+workdir = '~/Downloads/tpu'
+subprocess.run(f'cd {workdir} && git checkout 222cc86', shell=True, check=True)
 
-    # The setup command.  Will be run under the working directory.
-    setup = 'pip install --upgrade pip && \
-        conda init bash && \
-        conda activate resnet || \
-          (conda create -n resnet python=3.7 -y && \
-           conda activate resnet && \
-           conda install cudatoolkit=11.0 -y && \
-           pip install tensorflow==2.4.0 pyyaml && \
-           pip install protobuf==3.20 && \
-           cd models && pip install -e .)'
+# The setup command.  Will be run under the working directory.
+setup = """\
+    set -e
+    pip install --upgrade pip
+    conda init bash
+    conda activate resnet && exists=1 || exists=0
+    if [ $exists -eq 0 ]; then
+        conda create -n resnet python=3.7 -y
+        conda activate resnet
+        conda install cudatoolkit=11.0 -y
+        pip install tensorflow==2.4.0 pyyaml
+        pip install protobuf==3.20
+        cd models && pip install -e .
+    fi
+    """
 
-    # The command to run.  Will be run under the working directory.
-    run = 'conda activate resnet && \
-        export XLA_FLAGS=\'--xla_gpu_cuda_data_dir=/usr/local/cuda/\' && \
-        python -u models/official/resnet/resnet_main.py --use_tpu=False \
+# The command to run.  Will be run under the working directory.
+run = """\
+    conda activate resnet
+    export XLA_FLAGS=\'--xla_gpu_cuda_data_dir=/usr/local/cuda/\'
+    python -u models/official/resnet/resnet_main.py --use_tpu=False \
         --mode=train --train_batch_size=256 --train_steps=250 \
         --iterations_per_loop=125 \
         --data_dir=gs://cloud-tpu-test-datasets/fake_imagenet \
         --model_dir=resnet-model-dir \
-        --amp --xla --loss_scale=128'
+        --amp --xla --loss_scale=128
+    """
 
-    ### Optional: download data to VM's local disks. ###
-    # Format: {VM paths: local paths / cloud URLs}.
-    file_mounts = {
-        # Download from GCS before training starts.
-        # '/tmp/fake_imagenet': 'gs://cloud-tpu-test-datasets/fake_imagenet',
-    }
-    # Refer to the VM local path.
-    # run = run.replace('gs://cloud-tpu-test-datasets/fake_imagenet',
-    #                   '/tmp/fake_imagenet')
-    ### Optional end ###
+### Optional: download data to VM's local disks. ###
+# Format: {VM paths: local paths / cloud URLs}.
+file_mounts = {
+    # Download from GCS before training starts.
+    # '/tmp/fake_imagenet': 'gs://cloud-tpu-test-datasets/fake_imagenet',
+}
+# Refer to the VM local path.
+# run = run.replace('gs://cloud-tpu-test-datasets/fake_imagenet',
+#                   '/tmp/fake_imagenet')
+### Optional end ###
 
-    train = sky.Task(
-        'train',
-        workdir=workdir,
-        setup=setup,
-        run=run,
-    )
-    train.set_file_mounts(file_mounts)
-    # TODO: allow option to say (or detect) no download/egress cost.
-    train.set_inputs('gs://cloud-tpu-test-datasets/fake_imagenet',
-                     estimated_size_gigabytes=70)
-    train.set_outputs('resnet-model-dir', estimated_size_gigabytes=0.1)
-    train.set_resources({
-        ##### Fully specified
-        # sky.Resources(sky.AWS(), 'p3.2xlarge'),
-        # sky.Resources(sky.GCP(), 'n1-standard-16'),
-        # sky.Resources(
-        #     sky.GCP(),
-        #     'n1-standard-8',
-        #     # Options: 'V100', {'V100': <num>}.
-        #     'V100',
-        # ),
-        ##### Partially specified
-        # sky.Resources(accelerators='T4'),
-        # sky.Resources(accelerators={'T4': 8}, use_spot=True),
-        # sky.Resources(sky.AWS(), accelerators={'T4': 8}, use_spot=True),
-        # sky.Resources(sky.AWS(), accelerators='K80'),
-        # sky.Resources(sky.AWS(), accelerators='K80', use_spot=True),
-        # sky.Resources(accelerators='tpu-v3-8'),
-        # sky.Resources(accelerators='V100', use_spot=True),
-        # sky.Resources(accelerators={'T4': 4}),
-        sky.Resources(sky.AWS(), accelerators='V100'),
-        # sky.Resources(sky.GCP(), accelerators={'V100': 4}),
-        # sky.Resources(sky.AWS(), accelerators='V100', use_spot=True),
-        # sky.Resources(sky.AWS(), accelerators={'V100': 8}),
-    })
+task = sky.Task(
+    'train',
+    workdir=workdir,
+    setup=setup,
+    run=run,
+)
+task.set_file_mounts(file_mounts)
+# TODO: allow option to say (or detect) no download/egress cost.
+task.set_inputs('gs://cloud-tpu-test-datasets/fake_imagenet',
+                estimated_size_gigabytes=70)
+task.set_outputs('resnet-model-dir', estimated_size_gigabytes=0.1)
+task.set_resources({
+    ##### Fully specified
+    # sky.Resources(sky.AWS(), 'p3.2xlarge'),
+    # sky.Resources(sky.GCP(), 'n1-standard-16'),
+    # sky.Resources(
+    #     sky.GCP(),
+    #     'n1-standard-8',
+    #     # Options: 'V100', {'V100': <num>}.
+    #     'V100',
+    # ),
+    ##### Partially specified
+    # sky.Resources(accelerators='T4'),
+    # sky.Resources(accelerators={'T4': 8}, use_spot=True),
+    # sky.Resources(sky.AWS(), accelerators={'T4': 8}, use_spot=True),
+    # sky.Resources(sky.AWS(), accelerators='K80'),
+    # sky.Resources(sky.AWS(), accelerators='K80', use_spot=True),
+    # sky.Resources(accelerators='tpu-v3-8'),
+    # sky.Resources(accelerators='V100', use_spot=True),
+    # sky.Resources(accelerators={'T4': 4}),
+    sky.Resources(sky.AWS(), accelerators='V100'),
+    # sky.Resources(sky.GCP(), accelerators={'V100': 4}),
+    # sky.Resources(sky.AWS(), accelerators='V100', use_spot=True),
+    # sky.Resources(sky.AWS(), accelerators={'V100': 8}),
+})
 
-    # Optionally, specify a time estimator: Resources -> time in seconds.
-    # train.set_time_estimator(time_estimators.resnet50_estimate_runtime)
+# Optionally, specify a time estimator: Resources -> time in seconds.
+# task.set_time_estimator(time_estimators.resnet50_estimate_runtime)
 
-# sky.launch(dag, dryrun=True)
-sky.launch(dag)
+# sky.launch(task, dryrun=True)
+sky.launch(task)
diff --git a/examples/resnet_distributed_torch_app.py b/examples/resnet_distributed_torch_app.py
index 6fc489287..25a922284 100644
--- a/examples/resnet_distributed_torch_app.py
+++ b/examples/resnet_distributed_torch_app.py
@@ -2,52 +2,53 @@ from typing import List, Optional
 
 import sky
 
-with sky.Dag() as dag:
-    # Total Nodes, INCLUDING Head Node
-    num_nodes = 2
-
-    # The setup command.  Will be run under the working directory.
-    setup = 'echo \"alias python=python3\" >> ~/.bashrc && pip3 install --upgrade pip && \
-      [ -d pytorch-distributed-resnet ] || \
-      (git clone https://github.com/michaelzhiluo/pytorch-distributed-resnet && \
-      cd pytorch-distributed-resnet && pip3 install -r requirements.txt && \
-      mkdir -p data  && mkdir -p saved_models && cd data && \
-      wget -c --quiet https://www.cs.toronto.edu/~kriz/cifar-10-python.tar.gz && \
-      tar -xvzf cifar-10-python.tar.gz)'
-
-    # The command to run.  Will be run under the working directory.
-    def run_fn(node_rank: int, ip_list: List[str]) -> Optional[str]:
-        num_nodes = len(ip_list)
-        return f"""\
-        cd pytorch-distributed-resnet
-        python3 -m torch.distributed.launch --nproc_per_node=1 \
-        --nnodes={num_nodes} --node_rank={node_rank} --master_addr={ip_list[0]} \
-        --master_port=8008 resnet_ddp.py --num_epochs 20
-        """
-
-    train = sky.Task(
-        'train',
-        setup=setup,
-        num_nodes=num_nodes,
-        run=run_fn,
-    )
-
-    train.set_resources({
-        ##### Fully specified
-        sky.Resources(sky.AWS(), 'p3.2xlarge'),
-        # sky.Resources(sky.GCP(), 'n1-standard-16'),
-        #sky.Resources(
-        #     sky.GCP(),
-        #     'n1-standard-8',
-        # Options: 'V100', {'V100': <num>}.
-        #     'V100',
-        #),
-        ##### Partially specified
-        #sky.Resources(accelerators='V100'),
-        # sky.Resources(accelerators='tpu-v3-8'),
-        # sky.Resources(sky.AWS(), accelerators={'V100': 4}),
-        # sky.Resources(sky.AWS(), accelerators='V100'),
-    })
-
-sky.launch(dag, cluster_name='dth')
-# sky.exec(dag, cluster_name='dth')
+# Total Nodes, INCLUDING Head Node
+num_nodes = 2
+
+# The setup command.  Will be run under the working directory.
+setup = 'echo \"alias python=python3\" >> ~/.bashrc && pip3 install --upgrade pip && \
+    [ -d pytorch-distributed-resnet ] || \
+    (git clone https://github.com/michaelzhiluo/pytorch-distributed-resnet && \
+    cd pytorch-distributed-resnet && pip3 install -r requirements.txt && \
+    mkdir -p data  && mkdir -p saved_models && cd data && \
+    wget -c --quiet https://www.cs.toronto.edu/~kriz/cifar-10-python.tar.gz && \
+    tar -xvzf cifar-10-python.tar.gz)'
+
+
+# The command to run.  Will be run under the working directory.
+def run_fn(node_rank: int, ip_list: List[str]) -> Optional[str]:
+    num_nodes = len(ip_list)
+    return f"""\
+    cd pytorch-distributed-resnet
+    python3 -m torch.distributed.launch --nproc_per_node=1 \
+    --nnodes={num_nodes} --node_rank={node_rank} --master_addr={ip_list[0]} \
+    --master_port=8008 resnet_ddp.py --num_epochs 20
+    """
+
+
+train = sky.Task(
+    'train',
+    setup=setup,
+    num_nodes=num_nodes,
+    run=run_fn,
+)
+
+train.set_resources({
+    ##### Fully specified
+    sky.Resources(sky.AWS(), 'p3.2xlarge'),
+    # sky.Resources(sky.GCP(), 'n1-standard-16'),
+    #sky.Resources(
+    #     sky.GCP(),
+    #     'n1-standard-8',
+    # Options: 'V100', {'V100': <num>}.
+    #     'V100',
+    #),
+    ##### Partially specified
+    #sky.Resources(accelerators='V100'),
+    # sky.Resources(accelerators='tpu-v3-8'),
+    # sky.Resources(sky.AWS(), accelerators={'V100': 4}),
+    # sky.Resources(sky.AWS(), accelerators='V100'),
+})
+
+sky.launch(train, cluster_name='dth')
+# sky.exec(train, cluster_name='dth')
diff --git a/examples/resnet_distributed_torch_with_script.yaml b/examples/resnet_distributed_torch_with_script.yaml
index 6614cc6e1..a492e4878 100644
--- a/examples/resnet_distributed_torch_with_script.yaml
+++ b/examples/resnet_distributed_torch_with_script.yaml
@@ -7,10 +7,10 @@ resources:
 
 num_nodes: 2
 
-workdir: .
+workdir: ./examples/resnet_distributed_torch_scripts
 
 setup: |
-    bash examples/resnet_distributed_torch_scripts/setup.sh
+    bash ./setup.sh
 
 run: |
-    bash examples/resnet_distributed_torch_scripts/run.sh
+    bash ./run.sh
diff --git a/examples/resnet_inference_app.yaml b/examples/resnet_inference_app.yaml
new file mode 100644
index 000000000..7f62f7195
--- /dev/null
+++ b/examples/resnet_inference_app.yaml
@@ -0,0 +1,49 @@
+# Runs ResNet in inference mode on the fake_imagenet dataset
+#
+# Usage:
+#   sky launch -c infer resnet_inference_app.yaml
+#   sky down infer
+
+name: resnet-inference
+
+resources:
+  accelerators:
+    V100: 1
+
+file_mounts:
+  /tmp/resnet-model-dir: 
+    source: s3://mluo-resnet-model-dir
+
+
+setup: |
+  git clone https://github.com/concretevitamin/tpu || true
+  cd tpu
+  git checkout gpu_train
+
+  . $(conda info --base)/etc/profile.d/conda.sh
+  pip install --upgrade pip
+
+  conda activate resnet
+
+  if [ $? -eq 0 ]; then
+    echo "conda env exists"
+  else
+    conda create -n resnet python=3.7 -y
+    conda activate resnet
+    conda install cudatoolkit=11.0 -y
+    pip install tensorflow==2.4.0 pyyaml
+    pip install protobuf==3.20
+    cd models
+    pip install -e .
+  fi
+
+run: |
+  cd tpu
+  . $(conda info --base)/etc/profile.d/conda.sh
+  conda activate resnet
+
+  export XLA_FLAGS='--xla_gpu_cuda_data_dir=/usr/local/cuda/'
+  python -u models/official/resnet/resnet_main.py --use_tpu=False \
+  --mode=infer --data_dir=gs://cloud-tpu-test-datasets/fake_imagenet \
+  --model_dir=/tmp/resnet-model-dir --amp --xla --loss_scale=128 \
+  --infer_batch_size=8 --infer_steps=10000
diff --git a/examples/tensorboard_app.py b/examples/tensorboard_app.py
index 6b27b6df0..5f102a277 100644
--- a/examples/tensorboard_app.py
+++ b/examples/tensorboard_app.py
@@ -25,40 +25,37 @@ with sky.Dag() as dag:
 sky.launch(dag, cluster_name='tb', detach_run=True)
 
 # Run the training task.
-with sky.Dag() as dag:
-
-    # The command to run.  Will be run under the working directory.
-    run = 'conda activate resnet && mkdir -p resnet-model-dir && \
-        export XLA_FLAGS=\'--xla_gpu_cuda_data_dir=/usr/local/cuda/\' && \
-        python -u models/official/resnet/resnet_main.py --use_tpu=False \
-        --mode=train --train_batch_size=256 --train_steps=250 \
-        --iterations_per_loop=125 \
-        --data_dir=gs://cloud-tpu-test-datasets/fake_imagenet \
-        --model_dir=resnet-model-dir \
-        --amp --xla --loss_scale=128'
-
-    train = sky.Task(
-        'train',
-        workdir=workdir,
-        run=run,
-    )
-
-    train.set_resources({
-        sky.Resources(accelerators='V100'),
-    })
-sky.exec(dag, cluster_name='tb', detach_run=True)
+# The command to run.  Will be run under the working directory.
+run = 'conda activate resnet && mkdir -p resnet-model-dir && \
+    export XLA_FLAGS=\'--xla_gpu_cuda_data_dir=/usr/local/cuda/\' && \
+    python -u models/official/resnet/resnet_main.py --use_tpu=False \
+    --mode=train --train_batch_size=256 --train_steps=250 \
+    --iterations_per_loop=125 \
+    --data_dir=gs://cloud-tpu-test-datasets/fake_imagenet \
+    --model_dir=resnet-model-dir \
+    --amp --xla --loss_scale=128'
+
+train = sky.Task(
+    'train',
+    workdir=workdir,
+    run=run,
+)
+
+train.set_resources({
+    sky.Resources(accelerators='V100'),
+})
+sky.exec(train, cluster_name='tb', detach_run=True)
 
 # Run the tensorboard task.
-with sky.Dag() as dag:
-    # Use 'ssh -L 4650:localhost:4650 <cluster_name>' to forward port to local.
-    # 'ssh -L 4650:localhost:4650 tb'
-    tensorboard = sky.Task(
-        'tensorboard',
-        workdir=workdir,
-        setup=setup,
-        run='conda activate resnet && \
-            tensorboard --logdir resnet-model-dir --port 4650',
-    )
-    tensorboard.set_resources(sky.Resources())
-
-sky.exec(dag, cluster_name='tb', detach_run=True)
+# Use 'ssh -L 4650:localhost:4650 <cluster_name>' to forward port to local.
+# 'ssh -L 4650:localhost:4650 tb'
+tensorboard = sky.Task(
+    'tensorboard',
+    workdir=workdir,
+    setup=setup,
+    run='conda activate resnet && \
+        tensorboard --logdir resnet-model-dir --port 4650',
+)
+tensorboard.set_resources(sky.Resources())
+
+sky.exec(tensorboard, cluster_name='tb', detach_run=True)
diff --git a/sky/__init__.py b/sky/__init__.py
index 2f6d95e0d..db912af15 100644
--- a/sky/__init__.py
+++ b/sky/__init__.py
@@ -20,8 +20,8 @@ from sky.data import Storage, StorageMode, 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,
-                      tail_logs, download_logs, job_status, spot_status,
-                      spot_cancel, storage_ls, storage_delete)
+                      tail_logs, download_logs, job_status, spot_queue,
+                      spot_status, spot_cancel, storage_ls, storage_delete)
 
 # Aliases.
 AWS = clouds.AWS
@@ -68,7 +68,8 @@ __all__ = [
     'download_logs',
     'job_status',
     # core APIs Spot Job Management
-    'spot_status',
+    'spot_queue',
+    'spot_status',  # Deprecated (alias for spot_queue)
     'spot_cancel',
     # core APIs Storage Management
     'storage_ls',
diff --git a/sky/backends/backend.py b/sky/backends/backend.py
index 272ffb7e3..0a6d816f2 100644
--- a/sky/backends/backend.py
+++ b/sky/backends/backend.py
@@ -82,9 +82,9 @@ class Backend:
         return self._execute(handle, task, detach_run)
 
     @timeline.event
-    def post_execute(self, handle: ResourceHandle, teardown: bool) -> None:
+    def post_execute(self, handle: ResourceHandle, down: bool) -> None:
         """Post execute(): e.g., print helpful inspection messages."""
-        return self._post_execute(handle, teardown)
+        return self._post_execute(handle, down)
 
     @timeline.event
     def teardown_ephemeral_storage(self, task: 'task_lib.Task') -> None:
@@ -130,7 +130,7 @@ class Backend:
                  detach_run: bool) -> None:
         raise NotImplementedError
 
-    def _post_execute(self, handle: ResourceHandle, teardown: bool) -> None:
+    def _post_execute(self, handle: ResourceHandle, down: bool) -> None:
         raise NotImplementedError
 
     def _teardown_ephemeral_storage(self, task: 'task_lib.Task') -> None:
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 2f0739321..02df9e19e 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -101,7 +101,11 @@ _MAX_CLUSTER_NAME_LEN_FOR_GCP = 35
 # Note: This value cannot be too small, otherwise OOM issue may occur.
 DEFAULT_TASK_CPU_DEMAND = 0.5
 
-SKY_RESERVED_CLUSTER_NAMES = [spot_lib.SPOT_CONTROLLER_NAME]
+# Mapping from reserved cluster names to the corresponding group name (logging purpose).
+# NOTE: each group can only have one reserved cluster name for now.
+SKY_RESERVED_CLUSTER_NAMES = {
+    spot_lib.SPOT_CONTROLLER_NAME: 'Managed spot controller'
+}
 
 # Filelocks for the cluster status change.
 CLUSTER_STATUS_LOCK_PATH = os.path.expanduser('~/.sky/.{}.lock')
@@ -1742,17 +1746,17 @@ def get_clusters(
 ) -> List[Dict[str, Any]]:
     """Returns a list of cached cluster records.
 
-    Combs through the Sky database (in ~/.sky/state.db) to get a list of records
+    Combs through the database (in ~/.sky/state.db) to get a list of records
     corresponding to launched clusters.
 
     Args:
-        include_reserved: Whether to include sky-reserved clusters, e.g. spot
+        include_reserved: Whether to include reserved clusters, e.g. spot
             controller.
         refresh: Whether to refresh the status of the clusters. (Refreshing will
             set the status to STOPPED if the cluster cannot be pinged.)
         cloud_filter: Sets which clouds to filer through from the global user
-            state. Supports three values, 'all' for all clouds, 'public' for public
-            clouds only, and 'local' for only local clouds.
+            state. Supports three values, 'all' for all clouds, 'public' for
+            public clouds only, and 'local' for only local clouds.
 
     Returns:
         A list of cluster records.
@@ -1927,13 +1931,11 @@ def check_cluster_name_not_reserved(
     If the cluster name is reserved, return the error message. Otherwise,
     return None.
     """
-    usage = 'internal use'
-    if cluster_name == spot_lib.SPOT_CONTROLLER_NAME:
-        usage = 'spot controller'
-    msg = f'Cluster {cluster_name!r} is reserved for {usage}.'
-    if operation_str is not None:
-        msg += f' {operation_str} is not allowed.'
     if cluster_name in SKY_RESERVED_CLUSTER_NAMES:
+        msg = (f'Cluster {cluster_name!r} is reserved for '
+               f'{SKY_RESERVED_CLUSTER_NAMES[cluster_name].lower()}.')
+        if operation_str is not None:
+            msg += f' {operation_str} is not allowed.'
         with ux_utils.print_exception_no_traceback():
             raise ValueError(msg)
 
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index fc22bdf91..229a51557 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -198,7 +198,7 @@ class RayCodeGen:
             'run_bash_command_with_log = ray.remote(run_bash_command_with_log)',
         ]
         if spot_task is not None:
-            # Add the spot job to spot status table.
+            # Add the spot job to spot queue table.
             resources_str = backend_utils.get_task_resources_str(spot_task)
             self._code += [
                 'from sky.spot import spot_state',
@@ -2092,17 +2092,33 @@ class CloudVmRayBackend(backends.Backend):
                     self.tail_logs(handle, job_id)
         finally:
             name = handle.cluster_name
-            logger.info(f'{fore.CYAN}Job ID: '
-                        f'{style.BRIGHT}{job_id}{style.RESET_ALL}'
-                        '\nTo cancel the job:\t'
-                        f'{backend_utils.BOLD}sky cancel {name} {job_id}'
-                        f'{backend_utils.RESET_BOLD}'
-                        '\nTo stream the logs:\t'
-                        f'{backend_utils.BOLD}sky logs {name} {job_id}'
-                        f'{backend_utils.RESET_BOLD}'
-                        '\nTo view the job queue:\t'
-                        f'{backend_utils.BOLD}sky queue {name}'
-                        f'{backend_utils.RESET_BOLD}')
+            if name == spot_lib.SPOT_CONTROLLER_NAME:
+                logger.info(f'{fore.CYAN}Spot Job ID: '
+                            f'{style.BRIGHT}{job_id}{style.RESET_ALL}'
+                            '\nTo cancel the job:\t\t'
+                            f'{backend_utils.BOLD}sky spot cancel {job_id}'
+                            f'{backend_utils.RESET_BOLD}'
+                            '\nTo stream the logs:\t\t'
+                            f'{backend_utils.BOLD}sky spot logs {job_id}'
+                            f'{backend_utils.RESET_BOLD}'
+                            f'\nTo stream controller logs:\t'
+                            f'{backend_utils.BOLD}sky logs {name} {job_id}'
+                            f'{backend_utils.RESET_BOLD}'
+                            '\nTo view all spot jobs:\t\t'
+                            f'{backend_utils.BOLD}sky spot queue'
+                            f'{backend_utils.RESET_BOLD}')
+            else:
+                logger.info(f'{fore.CYAN}Job ID: '
+                            f'{style.BRIGHT}{job_id}{style.RESET_ALL}'
+                            '\nTo cancel the job:\t'
+                            f'{backend_utils.BOLD}sky cancel {name} {job_id}'
+                            f'{backend_utils.RESET_BOLD}'
+                            '\nTo stream the logs:\t'
+                            f'{backend_utils.BOLD}sky logs {name} {job_id}'
+                            f'{backend_utils.RESET_BOLD}'
+                            '\nTo view the job queue:\t'
+                            f'{backend_utils.BOLD}sky queue {name}'
+                            f'{backend_utils.RESET_BOLD}')
 
     def _setup_and_create_job_cmd_on_local_head(
         self,
@@ -2190,31 +2206,32 @@ class CloudVmRayBackend(backends.Backend):
             # Case: task_lib.Task(run, num_nodes=1)
             self._execute_task_one_node(handle, task, job_id, detach_run)
 
-    def _post_execute(self, handle: ResourceHandle, teardown: bool) -> None:
+    def _post_execute(self, handle: ResourceHandle, down: bool) -> None:
         colorama.init()
         fore = colorama.Fore
         style = colorama.Style
         name = handle.cluster_name
+        if name == spot_lib.SPOT_CONTROLLER_NAME or down:
+            return
         stop_str = ('\nTo stop the cluster:'
                     f'\t{backend_utils.BOLD}sky stop {name}'
                     f'{backend_utils.RESET_BOLD}')
         if isinstance(handle.launched_resources.cloud, clouds.Local):
             stop_str = ''
-        if not teardown:
-            logger.info(f'\n{fore.CYAN}Cluster name: '
-                        f'{style.BRIGHT}{name}{style.RESET_ALL}'
-                        '\nTo log into the head VM:\t'
-                        f'{backend_utils.BOLD}ssh {name}'
-                        f'{backend_utils.RESET_BOLD}'
-                        '\nTo submit a job:'
-                        f'\t\t{backend_utils.BOLD}sky exec {name} yaml_file'
-                        f'{backend_utils.RESET_BOLD}'
-                        f'{stop_str}'
-                        '\nTo teardown the cluster:'
-                        f'\t{backend_utils.BOLD}sky down {name}'
-                        f'{backend_utils.RESET_BOLD}')
-            if handle.tpu_delete_script is not None:
-                logger.info('Tip: `sky down` will delete launched TPU(s) too.')
+        logger.info(f'\n{fore.CYAN}Cluster name: '
+                    f'{style.BRIGHT}{name}{style.RESET_ALL}'
+                    '\nTo log into the head VM:\t'
+                    f'{backend_utils.BOLD}ssh {name}'
+                    f'{backend_utils.RESET_BOLD}'
+                    '\nTo submit a job:'
+                    f'\t\t{backend_utils.BOLD}sky exec {name} yaml_file'
+                    f'{backend_utils.RESET_BOLD}'
+                    f'{stop_str}'
+                    '\nTo teardown the cluster:'
+                    f'\t{backend_utils.BOLD}sky down {name}'
+                    f'{backend_utils.RESET_BOLD}')
+        if handle.tpu_delete_script is not None:
+            logger.info('Tip: `sky down` will delete launched TPU(s) too.')
 
     def _teardown_ephemeral_storage(self, task: task_lib.Task) -> None:
         storage_mounts = task.storage_mounts
diff --git a/sky/backends/local_docker_backend.py b/sky/backends/local_docker_backend.py
index cb0058283..9aaf8b114 100644
--- a/sky/backends/local_docker_backend.py
+++ b/sky/backends/local_docker_backend.py
@@ -268,7 +268,8 @@ class LocalDockerBackend(backends.Backend):
 
         self._execute_task_one_node(handle, task)
 
-    def _post_execute(self, handle: ResourceHandle, teardown: bool) -> None:
+    def _post_execute(self, handle: ResourceHandle, down: bool) -> None:
+        del down  # unused
         colorama.init()
         style = colorama.Style
         container = self.containers[handle]
diff --git a/sky/cli.py b/sky/cli.py
index cf71d82fb..025010bf6 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -27,6 +27,7 @@ NOTE: the order of command definitions in this file corresponds to how they are
 listed in "sky --help".  Take care to put logically connected commands close to
 each other.
 """
+import copy
 import datetime
 import functools
 import getpass
@@ -607,7 +608,7 @@ def _check_resources_match(backend: backends.Backend,
 
 
 def _launch_with_confirm(
-    dag: sky.Dag,
+    task: sky.Task,
     backend: backends.Backend,
     cluster: str,
     *,
@@ -621,7 +622,9 @@ def _launch_with_confirm(
     node_type: Optional[str] = None,
     is_local_cloud: Optional[bool] = False,
 ):
-    """Launch a cluster with a DAG."""
+    """Launch a cluster with a Task."""
+    with sky.Dag() as dag:
+        dag.add(task)
     if cluster is None:
         cluster = backend_utils.generate_cluster_name()
     maybe_status, _ = backend_utils.refresh_cluster_status_handle(cluster)
@@ -701,17 +704,17 @@ def _create_and_ssh_into_node(
         raise click.BadParameter(
             f'Name {cluster_name!r} taken by a local cluster and cannot '
             f'be used for a {node_type}.')
-    with sky.Dag() as dag:
-        # TODO: Add conda environment replication
-        # should be setup =
-        # 'conda env export | grep -v "^prefix: " > environment.yml'
-        # && conda env create -f environment.yml
-        task = sky.Task(
-            node_type,
-            workdir=None,
-            setup=None,
-        )
-        task.set_resources(resources)
+
+    # TODO: Add conda environment replication
+    # should be setup =
+    # 'conda env export | grep -v "^prefix: " > environment.yml'
+    # && conda env create -f environment.yml
+    task = sky.Task(
+        node_type,
+        workdir=None,
+        setup=None,
+    )
+    task.set_resources(resources)
 
     backend = backend if backend is not None else backends.CloudVmRayBackend()
     maybe_status, _ = backend_utils.refresh_cluster_status_handle(cluster_name)
@@ -726,7 +729,7 @@ def _create_and_ssh_into_node(
             f'sky {node_type}{name_arg}{colorama.Style.RESET_ALL}')
 
     _launch_with_confirm(
-        dag,
+        task,
         backend,
         cluster_name,
         dryrun=False,
@@ -812,7 +815,7 @@ def _check_yaml(entrypoint: str) -> Tuple[bool, dict]:
     return is_yaml, config
 
 
-def _make_dag_from_entrypoint_with_overrides(
+def _make_task_from_entrypoint_with_overrides(
     entrypoint: List[str],
     *,
     name: Optional[str] = None,
@@ -830,93 +833,72 @@ def _make_dag_from_entrypoint_with_overrides(
     env: List[Dict[str, str]] = None,
     # spot launch specific
     spot_recovery: Optional[str] = None,
-) -> sky.Dag:
+) -> sky.Task:
     entrypoint = ' '.join(entrypoint)
-    with sky.Dag() as dag:
-        is_yaml, yaml_config = _check_yaml(entrypoint)
-        if is_yaml:
-            # Treat entrypoint as a yaml.
-            click.secho('Task from YAML spec: ', fg='yellow', nl=False)
-            click.secho(entrypoint, bold=True)
+    is_yaml, yaml_config = _check_yaml(entrypoint)
+    if is_yaml:
+        # Treat entrypoint as a yaml.
+        click.secho('Task from YAML spec: ', fg='yellow', nl=False)
+        click.secho(entrypoint, bold=True)
+    else:
+        if not entrypoint:
+            entrypoint = None
         else:
-            if not entrypoint:
-                entrypoint = None
-            else:
-                # Treat entrypoint as a bash command.
-                click.secho('Task from command: ', fg='yellow', nl=False)
-                click.secho(entrypoint, bold=True)
+            # Treat entrypoint as a bash command.
+            click.secho('Task from command: ', fg='yellow', nl=False)
+            click.secho(entrypoint, bold=True)
+
+    if onprem_utils.check_local_cloud_args(cloud, cluster, yaml_config):
+        cloud = 'local'
+
+    if is_yaml:
+        usage_lib.messages.usage.update_user_task_yaml(entrypoint)
+        task = sky.Task.from_yaml(entrypoint)
+    else:
+        task = sky.Task(name='sky-cmd', run=entrypoint)
+        task.set_resources({sky.Resources()})
 
-        if onprem_utils.check_local_cloud_args(cloud, cluster, yaml_config):
-            cloud = 'local'
+    # Override.
+    if workdir is not None:
+        task.workdir = workdir
 
-        if is_yaml:
-            usage_lib.messages.usage.update_user_task_yaml(entrypoint)
-            task = sky.Task.from_yaml(entrypoint)
+    override_params = _parse_override_params(cloud=cloud,
+                                             region=region,
+                                             zone=zone,
+                                             gpus=gpus,
+                                             instance_type=instance_type,
+                                             use_spot=use_spot,
+                                             image_id=image_id,
+                                             disk_size=disk_size)
+    # Spot launch specific.
+    if spot_recovery is not None:
+        if spot_recovery.lower() == 'none':
+            override_params['spot_recovery'] = None
         else:
-            task = sky.Task(name='sky-cmd', run=entrypoint)
-            task.set_resources({sky.Resources()})
-
-        # Override.
-        if workdir is not None:
-            task.workdir = workdir
-
-        override_params = _parse_override_params(cloud=cloud,
-                                                 region=region,
-                                                 zone=zone,
-                                                 gpus=gpus,
-                                                 instance_type=instance_type,
-                                                 use_spot=use_spot,
-                                                 image_id=image_id,
-                                                 disk_size=disk_size)
-        # Spot launch specific.
-        if spot_recovery is not None:
-            if spot_recovery.lower() == 'none':
-                override_params['spot_recovery'] = None
-            else:
-                override_params['spot_recovery'] = spot_recovery
-
-        assert len(task.resources) == 1
-        old_resources = list(task.resources)[0]
-        new_resources = old_resources.copy(**override_params)
-
-        task.set_resources({new_resources})
-
-        if num_nodes is not None:
-            task.num_nodes = num_nodes
-        if name is not None:
-            task.name = name
-        task.set_envs(env)
-        # TODO(wei-lin): move this validation into Python API.
-        if new_resources.accelerators is not None:
-            acc, _ = list(new_resources.accelerators.items())[0]
-            if acc.startswith('tpu-') and task.num_nodes > 1:
-                raise ValueError('Multi-node TPU cluster is not supported. '
-                                 f'Got num_nodes={task.num_nodes}.')
-    return dag
-
-
-def _start_cluster(cluster_name: str,
-                   idle_minutes_to_autostop: Optional[int] = None,
-                   retry_until_up: bool = False):
-    handle = global_user_state.get_handle_from_cluster_name(cluster_name)
-    backend = backend_utils.get_backend_from_handle(handle)
-    assert isinstance(backend, backends.CloudVmRayBackend)
-    with sky.Dag():
-        dummy_task = sky.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
+            override_params['spot_recovery'] = spot_recovery
+
+    assert len(task.resources) == 1
+    old_resources = list(task.resources)[0]
+    new_resources = old_resources.copy(**override_params)
+
+    task.set_resources({new_resources})
+
+    if num_nodes is not None:
+        task.num_nodes = num_nodes
+    if name is not None:
+        task.name = name
+    task.set_envs(env)
+    # TODO(wei-lin): move this validation into Python API.
+    if new_resources.accelerators is not None:
+        acc, _ = list(new_resources.accelerators.items())[0]
+        if acc.startswith('tpu-') and task.num_nodes > 1:
+            raise ValueError('Multi-node TPU cluster is not supported. '
+                             f'Got num_nodes={task.num_nodes}.')
+    return task
 
 
 class _NaturalOrderGroup(click.Group):
-    """Lists commands in the order they are defined in this script.
+    """Lists commands in the order defined in this script.
 
     Reference: https://github.com/pallets/click/issues/513
     """
@@ -940,6 +922,30 @@ class _DocumentedCodeCommand(click.Command):
         return super().get_help(ctx)
 
 
+def _with_deprecation_warning(f, original_name, alias_name):
+
+    @functools.wraps(f)
+    def wrapper(self, *args, **kwargs):
+        click.secho(
+            f'WARNING: `{alias_name}` is deprecated and will be removed in a '
+            f'future release. Please use `{original_name}` instead.\n',
+            err=True,
+            fg='yellow')
+        return f(self, *args, **kwargs)
+
+    return wrapper
+
+
+def _add_command_alias_to_group(group, command, name, hidden):
+    """Add a alias of a command to a group."""
+    new_command = copy.deepcopy(command)
+    new_command.hidden = hidden
+    new_command.name = name
+    new_command.invoke = _with_deprecation_warning(new_command.invoke,
+                                                   command.name, name)
+    group.add_command(new_command, name=name)
+
+
 @click.group(cls=_NaturalOrderGroup, context_settings=_CONTEXT_SETTINGS)
 @click.option('--install-shell-completion',
               type=click.Choice(['bash', 'zsh', 'fish', 'auto']),
@@ -1011,9 +1017,9 @@ def cli():
     is_flag=True,
     required=False,
     help=
-    ('Tear down the cluster after all jobs finish (successfully or '
-     'abnormally). If --idle-minutes-to-autostop is also set, the cluster will '
-     'be torn down after the specified idle time. '
+    ('Autodown the cluster: tear down the cluster after all jobs finish '
+     '(successfully or abnormally). If --idle-minutes-to-autostop is also set, '
+     'the cluster will be torn down after the specified idle time. '
      'Note that if errors occur during provisioning/data syncing/setting up, '
      'the cluster will not be torn down for debugging purposes.'),
 )
@@ -1063,6 +1069,7 @@ def launch(
     yes: bool,
     no_setup: bool,
 ):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Launch a task from a YAML or a command (rerun setup if cluster exists).
 
     If ENTRYPOINT points to a valid YAML file, it is read in as the task
@@ -1076,7 +1083,7 @@ def launch(
     if backend_name is None:
         backend_name = backends.CloudVmRayBackend.NAME
 
-    dag = _make_dag_from_entrypoint_with_overrides(
+    task = _make_task_from_entrypoint_with_overrides(
         entrypoint=entrypoint,
         name=name,
         cluster=cluster,
@@ -1102,7 +1109,7 @@ def launch(
             raise ValueError(f'{backend_name} backend is not supported.')
 
     _launch_with_confirm(
-        dag,
+        task,
         backend,
         cluster,
         dryrun=dryrun,
@@ -1150,6 +1157,7 @@ def exec(
     image_id: Optional[str],
     env: List[Dict[str, str]],
 ):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Execute a task or a command on a cluster (skip setup).
 
     If ENTRYPOINT points to a valid YAML file, it is read in as the task
@@ -1217,7 +1225,7 @@ def exec(
                                  'Use `sky launch` to provision first.')
     backend = backend_utils.get_backend_from_handle(handle)
 
-    dag = _make_dag_from_entrypoint_with_overrides(
+    task = _make_task_from_entrypoint_with_overrides(
         entrypoint=entrypoint,
         name=name,
         cluster=cluster,
@@ -1234,7 +1242,7 @@ def exec(
     )
 
     click.secho(f'Executing task on cluster {cluster}...', fg='yellow')
-    sky.exec(dag, backend=backend, cluster_name=cluster, detach_run=detach_run)
+    sky.exec(task, backend=backend, cluster_name=cluster, detach_run=detach_run)
 
 
 @cli.command()
@@ -1244,14 +1252,16 @@ def exec(
               is_flag=True,
               required=False,
               help='Show all information in full.')
-@click.option('--refresh',
-              '-r',
-              default=False,
-              is_flag=True,
-              required=False,
-              help='Query cluster status from the cloud provider.')
+@click.option(
+    '--refresh',
+    '-r',
+    default=False,
+    is_flag=True,
+    required=False,
+    help='Query the latest cluster statuses from the cloud provider(s).')
 @usage_lib.entrypoint
 def status(all: bool, refresh: bool):  # pylint: disable=redefined-builtin
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Show clusters.
 
     The following fields for each cluster are recorded: cluster name, time
@@ -1279,10 +1289,35 @@ def status(all: bool, refresh: bool):  # pylint: disable=redefined-builtin
     '(down)', e.g. '1m (down)', the cluster will be autodowned, rather than
     autostopped.
     """
-    cluster_records = core.status(all=all, refresh=refresh)
+    cluster_records = core.status(refresh=refresh)
+    nonreserved_cluster_records = []
+    reserved_clusters = dict()
+    for cluster_record in cluster_records:
+        cluster_name = cluster_record['name']
+        if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
+            cluster_group_name = backend_utils.SKY_RESERVED_CLUSTER_NAMES[
+                cluster_name]
+            reserved_clusters[cluster_group_name] = cluster_record
+        else:
+            nonreserved_cluster_records.append(cluster_record)
     local_clusters = onprem_utils.check_and_get_local_clusters(
         suppress_error=True)
-    status_utils.show_status_table(cluster_records, all)
+
+    num_pending_autostop = 0
+    num_pending_autostop += status_utils.show_status_table(
+        nonreserved_cluster_records, all)
+    for cluster_group_name, cluster_record in reserved_clusters.items():
+        num_pending_autostop += status_utils.show_status_table(
+            [cluster_record], all, reserved_group_name=cluster_group_name)
+    if num_pending_autostop > 0:
+        plural = ' has'
+        if num_pending_autostop > 1:
+            plural = 's have'
+        click.echo('\n'
+                   f'{num_pending_autostop} cluster{plural} '
+                   'auto{stop,down} scheduled. Refresh statuses with: '
+                   f'{colorama.Style.BRIGHT}sky status --refresh'
+                   f'{colorama.Style.RESET_ALL}')
     status_utils.show_local_status_table(local_clusters)
 
 
@@ -1306,6 +1341,7 @@ def status(all: bool, refresh: bool):  # pylint: disable=redefined-builtin
                 **_get_shell_complete_args(_complete_cluster_name))
 @usage_lib.entrypoint
 def queue(clusters: Tuple[str], skip_finished: bool, all_users: bool):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Show the job queue for cluster(s)."""
     click.secho('Fetching and parsing job queue...', fg='yellow')
     show_local_clusters = False
@@ -1379,6 +1415,7 @@ def logs(
     status: bool,  # pylint: disable=redefined-outer-name
     follow: bool,
 ):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Tail the log of a job.
 
     If JOB_ID is not provided, the latest job on the cluster will be used.
@@ -1446,6 +1483,7 @@ def logs(
 @click.argument('jobs', required=False, type=int, nargs=-1)
 @usage_lib.entrypoint
 def cancel(cluster: str, all: bool, jobs: List[int]):  # pylint: disable=redefined-builtin
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Cancel job(s)."""
     try:
         core.cancel(cluster, all, jobs)
@@ -1475,14 +1513,15 @@ def stop(
     all: Optional[bool],  # pylint: disable=redefined-builtin
     yes: bool,
 ):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Stop cluster(s).
 
     CLUSTER is the name (or glob pattern) of the cluster to stop.  If both
     CLUSTER and ``--all`` are supplied, the latter takes precedence.
 
     Data on attached disks is not lost when a cluster is stopped.  Billing for
-    the instances will stop while the disks will still be charged.  Those disks
-    will be reattached when restarting the cluster.
+    the instances will stop, while the disks will still be charged.  Those
+    disks will be reattached when restarting the cluster.
 
     Currently, spot instance clusters cannot be stopped.
 
@@ -1554,6 +1593,7 @@ def autostop(
     down: bool,  # pylint: disable=redefined-outer-name
     yes: bool,
 ):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Schedule or cancel an autostop or autodown for cluster(s).
 
     CLUSTERS are the name (or glob pattern) of the clusters to stop.  If both
@@ -1639,9 +1679,9 @@ def autostop(
     is_flag=True,
     required=False,
     help=
-    ('Tear down the cluster after all jobs finish (successfully or '
-     'abnormally). If --idle-minutes-to-autostop is also set, the cluster will '
-     'be torn down after the specified idle time.'),
+    ('Autodown the cluster: tear down the cluster after specified minutes of '
+     'idle time after all jobs finish (successfully or abnormally). Requires '
+     ' --idle-minutes-to-autostop to be set.'),
 )
 @click.option(
     '--retry-until-up',
@@ -1660,18 +1700,19 @@ def start(
         idle_minutes_to_autostop: Optional[int],
         down: bool,  # pylint: disable=redefined-outer-name
         retry_until_up: bool):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Restart cluster(s).
 
     If a cluster is previously stopped (status is STOPPED) or failed in
-    provisioning/runtime setup (status is INIT), this command will attempt to
-    start the cluster.  (In the second case, any failed setup steps are not
-    performed and only a request to start the machines is attempted.)
+    provisioning/runtime installation (status is INIT), this command will
+    attempt to start the cluster.  In the latter case, provisioning and runtime
+    installation will be retried.
 
-    Auto-failover provisioning is not used when restarting stopped
-    clusters. They will be started on the same cloud and region that was chosen
-    before.
+    Auto-failover provisioning is not used when restarting a stopped
+    cluster. It will be started on the same cloud, region, and zone that were
+    chosen before.
 
-    If a cluster is already in the UP status, this command has no effect on it.
+    If a cluster is already in the UP status, this command has no effect.
 
     Examples:
 
@@ -1817,17 +1858,19 @@ def down(
     yes: bool,
     purge: bool,
 ):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Tear down cluster(s).
 
     CLUSTER is the name of the cluster (or glob pattern) to tear down.  If both
     CLUSTER and ``--all`` are supplied, the latter takes precedence.
 
     Tearing down a cluster will delete all associated resources (all billing
-    stops), and any data on the attached disks will be lost. For local clusters,
-    `sky down` does not terminate the local cluster, but instead removes the
-    cluster from `sky status` and terminates the calling user's running jobs.
+    stops), and any data on the attached disks will be lost.  Accelerators
+    (e.g., TPUs) that are part of the cluster will be deleted too.
 
-    Accelerators (e.g., TPUs) that are part of the cluster will be deleted too.
+    For local on-prem clusters, this command does not terminate the local
+    cluster, but instead removes the cluster from the status table and
+    terminates the calling user's running jobs.
 
     Examples:
 
@@ -1853,6 +1896,50 @@ def down(
                            purge=purge)
 
 
+def _hint_for_down_spot_controller(controller_name: str):
+    # spot_jobs will be empty when the spot cluster is not running.
+    cluster_status, _ = backend_utils.refresh_cluster_status_handle(
+        controller_name)
+    if cluster_status is None:
+        click.echo('Managed spot controller has already been torn down.')
+        return
+
+    msg = (f'{colorama.Fore.YELLOW}WARNING: Tearing down the managed '
+           f'spot controller ({cluster_status.value}). Please be '
+           f'aware of the following:{colorama.Style.RESET_ALL}'
+           '\n * All logs and status information of the spot '
+           'jobs (output of `sky spot queue`) will be lost.')
+    if cluster_status == global_user_state.ClusterStatus.UP:
+        try:
+            spot_jobs = core.spot_queue(refresh=False)
+        except exceptions.ClusterNotUpError:
+            # The spot controller cluster status changed during querying
+            # the spot jobs, use the latest cluster status, so that the
+            # message for INIT and STOPPED states will be correctly
+            # added to the message.
+            cluster_status = backend_utils.refresh_cluster_status_handle(
+                controller_name)
+            spot_jobs = []
+
+        # Find in-progress spot jobs, and hint users to cancel them.
+        non_terminal_jobs = [
+            job for job in spot_jobs if not job['status'].is_terminal()
+        ]
+        if (cluster_status == global_user_state.ClusterStatus.UP and
+                non_terminal_jobs):
+            msg += ('\n * In-progress spot jobs found, their resources '
+                    'will not be terminated and require manual cleanup:\n')
+            job_table = spot_lib.format_job_table(non_terminal_jobs,
+                                                  show_all=False)
+            # Add prefix to each line to align with the bullet point.
+            msg += '\n'.join(
+                ['   ' + line for line in job_table.split('\n') if line != ''])
+    if cluster_status == global_user_state.ClusterStatus.INIT:
+        msg += ('\n * If there are pending/in-progress spot jobs, those '
+                'resources will not be terminated and require manual cleanup.')
+    click.echo(msg)
+
+
 def _down_or_stop_clusters(
         names: Tuple[str],
         apply_to_all: Optional[bool],
@@ -1903,16 +1990,8 @@ def _down_or_stop_clusters(
                     '`sky stop/autostop`.'))
             ]
         # Make sure the reserved clusters are explicitly specified without other
-        # normal clusters and purge is True.
+        # normal clusters.
         if len(reserved_clusters) > 0:
-            if not purge:
-                msg = (f'{operation} reserved cluster(s) '
-                       f'{reserved_clusters_str} is not supported.')
-                if down:
-                    msg += (
-                        '\nPlease specify --purge (-p) to force-terminate the '
-                        'reserved cluster(s).')
-                raise click.UsageError(msg)
             if len(names) != 0:
                 names_str = ', '.join(map(repr, names))
                 raise click.UsageError(
@@ -1920,6 +1999,21 @@ def _down_or_stop_clusters(
                     f'{reserved_clusters_str} with multiple other cluster(s) '
                     f'{names_str} is not supported.\n'
                     f'Please omit the reserved cluster(s) {reserved_clusters}.')
+            if not down:
+                raise click.UsageError(
+                    f'{operation} reserved cluster(s) '
+                    f'{reserved_clusters_str} is not supported.')
+            else:
+                # TODO(zhwu): We can only have one reserved cluster (spot
+                # controller).
+                assert len(reserved_clusters) == 1, reserved_clusters
+                _hint_for_down_spot_controller(reserved_clusters[0])
+
+                click.confirm('Proceed?',
+                              default=False,
+                              abort=True,
+                              show_default=True)
+                no_confirm = True
         names += reserved_clusters
 
     if apply_to_all:
@@ -2603,7 +2697,7 @@ def spot_launch(
     else:
         backend_utils.check_cluster_name_is_valid(name)
 
-    dag = _make_dag_from_entrypoint_with_overrides(
+    task = _make_task_from_entrypoint_with_overrides(
         entrypoint,
         name=name,
         workdir=workdir,
@@ -2625,13 +2719,13 @@ def spot_launch(
         if prompt is not None:
             click.confirm(prompt, default=True, abort=True, show_default=True)
 
-    sky.spot_launch(dag,
+    sky.spot_launch(task,
                     name,
                     detach_run=detach_run,
                     retry_until_up=retry_until_up)
 
 
-@spot.command('status', cls=_DocumentedCodeCommand)
+@spot.command('queue', cls=_DocumentedCodeCommand)
 @click.option('--all',
               '-a',
               default=False,
@@ -2648,7 +2742,7 @@ def spot_launch(
 )
 @usage_lib.entrypoint
 # pylint: disable=redefined-builtin
-def spot_status(all: bool, refresh: bool):
+def spot_queue(all: bool, refresh: bool):
     """Show statuses of managed spot jobs.
 
     \b
@@ -2669,17 +2763,17 @@ def spot_status(all: bool, refresh: bool):
 
     If the job failed, either due to user code or spot unavailability, the error
     log can be found with ``sky logs sky-spot-controller-<user_hash> job_id``.
-    Please find your exact spot controller name with ``sky status -a``.
+    Please find your exact spot controller name with ``sky status``.
 
     (Tip) To fetch job statuses every 60 seconds, use ``watch``:
 
     .. code-block:: bash
 
-      watch -n60 sky spot status
+      watch -n60 sky spot queue
     """
     click.secho('Fetching managed spot job statuses...', fg='yellow')
     try:
-        job_table = core.spot_status(refresh=refresh)
+        job_table = core.spot_queue(refresh=refresh)
     except exceptions.ClusterNotUpError:
         cache = spot_lib.load_job_table_cache()
         if cache is not None:
@@ -2699,6 +2793,9 @@ def spot_status(all: bool, refresh: bool):
     click.echo(f'Managed spot jobs:\n{job_table}')
 
 
+_add_command_alias_to_group(spot, spot_queue, 'status', hidden=True)
+
+
 @spot.command('cancel', cls=_DocumentedCodeCommand)
 @click.option('--name',
               '-n',
diff --git a/sky/core.py b/sky/core.py
index 15d20fbaf..74c275a2d 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -1,6 +1,7 @@
 """SDK functions for cluster/job management."""
 import colorama
 import getpass
+import sys
 from typing import Any, Dict, List, Optional, Tuple
 
 from sky import dag
@@ -30,29 +31,36 @@ logger = sky_logging.init_logger(__name__)
 
 
 @usage_lib.entrypoint
-def status(all: bool, refresh: bool) -> List[Dict[str, Any]]:
-    """Get the cluster status in dict.
+def status(refresh: bool = False) -> List[Dict[str, Any]]:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Get all cluster statuses.
+
+    Each returned value has the following fields:
+
+    .. code-block:: python
+
+        {
+            'name': (str) cluster name,
+            'launched_at': (int) timestamp of last launch on this cluster,
+            'handle': (ResourceHandle) an internal handle to the cluster,
+            'last_use': (str) the last command/entrypoint that affected this
+              cluster,
+            'status': (sky.ClusterStatus) cluster status,
+            'autostop': (int) idle time before autostop,
+            'to_down': (bool) whether autodown is used instead of autostop,
+            'metadata': (dict) metadata of the cluster,
+        }
 
-    Please refer to the sky.cli.status for the document.
+    Args:
+        refresh: whether to query the latest cluster statuses from the cloud
+            provider(s).
 
     Returns:
-        A list of dicts, each dict contains the information of a cluster.
-
-        [
-            {
-                '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,
-                'to_down': (bool) whether autodown is used instead of
-                    autostop,
-                'metadata': (dict) metadata of the cluster,
-            }
-        ]
-
+        A list of dicts, with each dict containing the information of a
+        cluster.
     """
-    cluster_records = backend_utils.get_clusters(all, refresh)
+    cluster_records = backend_utils.get_clusters(include_reserved=True,
+                                                 refresh=refresh)
     return cluster_records
 
 
@@ -80,7 +88,7 @@ def _start(
             f'Starting cluster {cluster_name!r} with backend {backend.NAME} '
             'is not supported.')
 
-    # NOTE: if spot_status() calls _start() and hits here, that entrypoint
+    # NOTE: if spot_queue() calls _start() and hits here, that entrypoint
     # would have a cluster name (the controller) filled in.
     usage_lib.record_cluster_name_for_current_operation(cluster_name)
 
@@ -104,28 +112,70 @@ def start(
         idle_minutes_to_autostop: Optional[int] = None,
         retry_until_up: bool = False,
         down: bool = False,  # pylint: disable=redefined-outer-name
-):
-    """Start the cluster.
+) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Restart a cluster.
+
+    If a cluster is previously stopped (status is STOPPED) or failed in
+    provisioning/runtime installation (status is INIT), this function will
+    attempt to start the cluster.  In the latter case, provisioning and runtime
+    installation will be retried.
+
+    Auto-failover provisioning is not used when restarting a stopped
+    cluster. It will be started on the same cloud, region, and zone that were
+    chosen before.
 
-    Please refer to the sky.cli.start for the document.
+    If a cluster is already in the UP status, this function has no effect.
+
+    Args:
+        cluster_name: name of the cluster to start.
+        idle_minutes_to_autostop: automatically stop the cluster after this
+            many minute 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(..., detach_run=True, ...)``
+            and then ``sky.autostop(idle_minutes=<minutes>)``. If not set,
+            the cluster will not be autostopped.
+        retry_until_up: whether to retry launching the cluster until it is
+            up.
+        down: Autodown the cluster: tear down the cluster after specified
+            minutes of idle time after all jobs finish (successfully or
+            abnormally). Requires ``idle_minutes_to_autostop`` to be set.
 
     Raises:
-        ValueError: cluster does not exist.
-        sky.exceptions.NotSupportedError: the cluster is not supported.
+        ValueError: the specified cluster does not exist; or if ``down`` is set
+          to True but ``idle_minutes_to_autostop`` is None.
+        sky.exceptions.NotSupportedError: if the cluster to restart was
+          launched using a non-default backend that does not support this
+          operation.
     """
+    if down and idle_minutes_to_autostop is None:
+        raise ValueError(
+            '`idle_minutes_to_autostop` must be set if `down` is True.')
     _start(cluster_name, idle_minutes_to_autostop, retry_until_up, down)
 
 
 @usage_lib.entrypoint
-def stop(cluster_name: str, purge: bool = False):
-    """Stop the cluster.
+def stop(cluster_name: str, purge: bool = False) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Stop a cluster.
 
-    Please refer to the sky.cli.stop for the document.
+    Data on attached disks is not lost when a cluster is stopped.  Billing for
+    the instances will stop, while the disks will still be charged.  Those
+    disks will be reattached when restarting the cluster.
+
+    Currently, spot instance clusters cannot be stopped.
+
+    Args:
+        cluster_name: name of the cluster to stop.
+        purge: whether to ignore cloud provider errors (if any).
 
     Raises:
-        ValueError: cluster does not exist.
-        RuntimeError: Fail to stop the cluster.
-        sky.exceptions.NotSupportedError: the cluster is not supported.
+        ValueError: the specified cluster does not exist.
+        RuntimeError: failed to stop the cluster.
+        sky.exceptions.NotSupportedError: if the specified cluster is a spot
+          cluster, or a TPU VM Pod cluster, or the managed spot controller.
     """
     if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
         raise exceptions.NotSupportedError(
@@ -158,19 +208,27 @@ def stop(cluster_name: str, purge: bool = False):
 
 
 @usage_lib.entrypoint
-def down(cluster_name: str, purge: bool = False):
-    """Down the cluster.
+def down(cluster_name: str, purge: bool = False) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Tear down a cluster.
 
-    Please refer to the sky.cli.down for the document.
+    Tearing down a cluster will delete all associated resources (all billing
+    stops), and any data on the attached disks will be lost.  Accelerators
+    (e.g., TPUs) that are part of the cluster will be deleted too.
+
+    For local on-prem clusters, this function does not terminate the local
+    cluster, but instead removes the cluster from the status table and
+    terminates the calling user's running jobs.
+
+    Args:
+        cluster_name: name of the cluster to down.
+        purge: whether to ignore cloud provider errors (if any).
 
     Raises:
-        ValueError: cluster does not exist.
-        sky.exceptions.NotSupportedError: the cluster is not supported.
+        ValueError: the specified cluster does not exist.
+        sky.exceptions.NotSupportedError: the specified cluster is the managed
+          spot controller.
     """
-    if (cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES and not purge):
-        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.')
@@ -183,22 +241,31 @@ def down(cluster_name: str, purge: bool = False):
 @usage_lib.entrypoint
 def autostop(
         cluster_name: str,
-        idle_minutes_to_autostop: int,
+        idle_minutes: int,
         down: bool = False,  # pylint: disable=redefined-outer-name
-):
-    """Set the autostop time of the cluster.
+) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Schedule or cancel an autostop/autodown for a cluster.
 
-    Please refer to the sky.cli.autostop for the document.
+    When multiple configurations are specified for the same cluster (e.g., this
+    function is called multiple times), the last one takes precedence.
 
     Args:
-        cluster_name (str): name of the cluster.
-        autostop_time (int): autostop time in minutes.
+        cluster_name: name of the cluster.
+        idle_minutes: the number of minutes of idleness (no pending/running
+          jobs) after which the cluster will be stopped automatically. Setting
+          to a negative number means cancel any autostop/autodown setting.
+        down: if true, use autodown (tear down the cluster; non-restartable),
+          rather than autostop (restartable).
+
     Raises:
-        ValueError: cluster does not exist.
-        sky.exceptions.NotSupportedError: the cluster is not supported.
+        ValueError: the specified cluster does not exist.
+        sky.exceptions.NotSupportedError: if the specified cluster is a TPU VM
+          Pod cluster, or the managed spot controller, or was launched using a
+          non-default backend.
         sky.exceptions.ClusterNotUpError: the cluster is not UP.
     """
-    is_cancel = idle_minutes_to_autostop < 0
+    is_cancel = idle_minutes < 0
     verb = 'Cancelling' if is_cancel else 'Scheduling'
     option_str = 'down' if down else 'stop'
     if is_cancel:
@@ -236,7 +303,7 @@ def autostop(
                 f'\n  auto{option_str} can only be set/unset for '
                 f'{global_user_state.ClusterStatus.UP.value} clusters.')
     usage_lib.record_cluster_name_for_current_operation(cluster_name)
-    backend.set_autostop(handle, idle_minutes_to_autostop, down)
+    backend.set_autostop(handle, idle_minutes, down)
 
 
 # ==================
@@ -281,7 +348,8 @@ def _check_cluster_available(cluster_name: str,
 def queue(cluster_name: str,
           skip_finished: bool = False,
           all_users: bool = False) -> List[dict]:
-    """Get the job queue in List[dict].
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Get the job queue of a cluster.
 
     Please refer to the sky.cli.queue for the document.
 
@@ -330,8 +398,9 @@ def queue(cluster_name: str,
 # pylint: disable=redefined-builtin
 def cancel(cluster_name: str,
            all: bool = False,
-           job_ids: Optional[List[int]] = None):
-    """Cancel jobs.
+           job_ids: Optional[List[int]] = None) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Cancel jobs on a cluster.
 
     Please refer to the sky.cli.cancel for the document.
 
@@ -368,7 +437,10 @@ def cancel(cluster_name: str,
 
 
 @usage_lib.entrypoint
-def tail_logs(cluster_name: str, job_id: Optional[str], follow: bool = True):
+def tail_logs(cluster_name: str,
+              job_id: Optional[str],
+              follow: bool = True) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Tail the logs of a job.
 
     Please refer to the sky.cli.tail_logs for the document.
@@ -398,6 +470,7 @@ def download_logs(
         cluster_name: str,
         job_ids: Optional[List[str]],
         local_dir: str = constants.SKY_LOGS_DIRECTORY) -> Dict[str, str]:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Download the logs of jobs.
 
     Args:
@@ -426,6 +499,7 @@ def job_status(
         cluster_name: str,
         job_ids: Optional[List[str]],
         stream_logs: bool = False) -> Dict[str, Optional[job_lib.JobStatus]]:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Get the status of jobs.
 
     Args:
@@ -480,9 +554,20 @@ def _is_spot_controller_up(
 
 @usage_lib.entrypoint
 def spot_status(refresh: bool) -> List[Dict[str, Any]]:
+    """[Deprecated] (alias of spot_queue) Get statuses of managed spot jobs."""
+    print(
+        f'{colorama.Fore.YELLOW}WARNING: `spot_status()` is deprecated. '
+        f'Instead, use: spot_queue(){colorama.Style.RESET_ALL}',
+        file=sys.stderr)
+    return spot_queue(refresh=refresh)
+
+
+@usage_lib.entrypoint
+def spot_queue(refresh: bool) -> List[Dict[str, Any]]:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Get statuses of managed spot jobs.
 
-    Please refer to the sky.cli.spot_status for the document.
+    Please refer to the sky.cli.spot_queue for the documentation.
 
     Returns:
         [
@@ -505,7 +590,7 @@ def spot_status(refresh: bool) -> List[Dict[str, Any]]:
 
     stop_msg = ''
     if not refresh:
-        stop_msg = 'To view the latest job table: sky spot status --refresh'
+        stop_msg = 'To view the latest job table: sky spot queue --refresh'
     controller_status, handle = _is_spot_controller_up(stop_msg)
 
     if controller_status is None:
@@ -550,8 +635,9 @@ def spot_status(refresh: bool) -> List[Dict[str, Any]]:
 # pylint: disable=redefined-builtin
 def spot_cancel(name: Optional[str] = None,
                 job_ids: Optional[Tuple[int]] = None,
-                all: bool = False):
-    """Cancel managed spot jobs
+                all: bool = False) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Cancel managed spot jobs.
 
     Please refer to the sky.cli.spot_cancel for the document.
 
@@ -601,7 +687,9 @@ def spot_cancel(name: Optional[str] = None,
 
 
 @usage_lib.entrypoint
-def spot_tail_logs(name: Optional[str], job_id: Optional[int], follow: bool):
+def spot_tail_logs(name: Optional[str], job_id: Optional[int],
+                   follow: bool) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Tail logs of managed spot jobs.
 
     Please refer to the sky.cli.spot_logs for the document.
@@ -633,6 +721,7 @@ def spot_tail_logs(name: Optional[str], job_id: Optional[int], follow: bool):
 # ======================
 @usage_lib.entrypoint
 def storage_ls() -> List[Dict[str, Any]]:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Get the storages.
 
     Returns:
@@ -653,7 +742,8 @@ def storage_ls() -> List[Dict[str, Any]]:
 
 
 @usage_lib.entrypoint
-def storage_delete(name: str):
+def storage_delete(name: str) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Delete a storage.
 
     Raises:
diff --git a/sky/execution.py b/sky/execution.py
index 95db48a98..4a2bbdb17 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -17,7 +17,7 @@ import enum
 import getpass
 import tempfile
 import os
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Union
 
 import colorama
 
@@ -48,38 +48,46 @@ OptimizeTarget = optimizer.OptimizeTarget
 # instead of a Dag.  CLI (cli.py) is implemented by us so should not trigger
 # this.
 _ENTRYPOINT_STRING_AS_DAG_MESSAGE = """\
-Expected a sky.Dag but received a string.
+Expected a sky.Task or sky.Dag but received a string.
 
-If you meant to run a command, make it a Task's run command and wrap as a Dag:
+If you meant to run a command, make it a Task's run command:
 
-  def to_dag(command, gpu=None):
-      with sky.Dag() as dag:
-          sky.Task(run=command).set_resources(
-              sky.Resources(accelerators=gpu))
-      return dag
+    task = sky.Task(run=command)
 
 The command can then be run as:
 
-  sky.exec(to_dag(cmd), cluster_name=..., ...)
+  sky.exec(task, cluster_name=..., ...)
   # Or use {'V100': 1}, 'V100:0.5', etc.
-  sky.exec(to_dag(cmd, gpu='V100'), cluster_name=..., ...)
+  task.set_resources(sky.Resources(accelerators='V100:1'))
+  sky.exec(task, cluster_name=..., ...)
 
-  sky.launch(to_dag(cmd), ...)
+  sky.launch(task, ...)
 
-  sky.spot_launch(to_dag(cmd), ...)
+  sky.spot_launch(task, ...)
 """.strip()
 
 
-def _type_check_dag(dag):
-    """Raises TypeError if 'dag' is not a 'sky.Dag'."""
+def _convert_to_dag(entrypoint: Any) -> 'sky.Dag':
+    """Convert the entrypoint to a sky.Dag.
+
+    Raises TypeError if 'entrypoint' is not a 'sky.Task' or 'sky.Dag'.
+    """
     # Not suppressing stacktrace: when calling this via API user may want to
     # see their own program in the stacktrace. Our CLI impl would not trigger
     # these errors.
-    if isinstance(dag, str):
+    if isinstance(entrypoint, str):
         raise TypeError(_ENTRYPOINT_STRING_AS_DAG_MESSAGE)
-    elif not isinstance(dag, sky.Dag):
-        raise TypeError('Expected a sky.Dag but received argument of type: '
-                        f'{type(dag)}')
+    elif isinstance(entrypoint, sky.Dag):
+        return copy.deepcopy(entrypoint)
+    elif isinstance(entrypoint, task_lib.Task):
+        entrypoint = copy.deepcopy(entrypoint)
+        with sky.Dag() as dag:
+            dag.add(entrypoint)
+        return dag
+    else:
+        raise TypeError(
+            'Expected a sky.Task or sky.Dag but received argument of type: '
+            f'{type(entrypoint)}')
 
 
 class Stage(enum.Enum):
@@ -96,7 +104,7 @@ class Stage(enum.Enum):
 
 
 def _execute(
-    dag: sky.Dag,
+    entrypoint: Union['sky.Task', 'sky.Dag'],
     dryrun: bool = False,
     down: bool = False,
     stream_logs: bool = True,
@@ -110,13 +118,13 @@ def _execute(
     idle_minutes_to_autostop: Optional[int] = None,
     no_setup: bool = False,
 ) -> None:
-    """Runs a DAG.
+    """Execute a entrypoint.
 
-    If the DAG has not been optimized yet, this will call sky.optimize() for
-    the caller.
+    If sky.Task is given or DAG has not been optimized yet, this will call
+    sky.optimize() for the caller.
 
     Args:
-      dag: sky.Dag.
+      entrypoint: sky.Task or sky.Dag.
       dryrun: bool; if True, only print the provision info (e.g., cluster
         yaml).
       down: bool; whether to tear down the launched resources after all jobs
@@ -143,7 +151,7 @@ def _execute(
         autostop after this many minutes of idleness.
       no_setup: bool; whether to skip setup commands or not when (re-)launching.
     """
-    _type_check_dag(dag)
+    dag = _convert_to_dag(entrypoint)
     assert len(dag) == 1, f'We support 1 task for now. {dag}'
     task = dag.tasks[0]
 
@@ -272,7 +280,7 @@ def _execute(
 @timeline.event
 @usage_lib.entrypoint
 def launch(
-    dag: sky.Dag,
+    task: Union['sky.Task', 'sky.Dag'],
     cluster_name: Optional[str] = None,
     retry_until_up: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
@@ -283,11 +291,20 @@ def launch(
     optimize_target: OptimizeTarget = OptimizeTarget.COST,
     detach_run: bool = False,
     no_setup: bool = False,
-):
-    """Launch a sky.DAG (rerun setup if cluster exists).
+) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Launch a task.
+
+    The task's setup and run commands are executed under the task's workdir
+    (when specified, it is synced to remote cluster).  The task undergoes job
+    queue scheduling on the cluster.
+
+    Currently, the first argument must be a sky.Task, or (more advanced usage)
+    a sky.Dag; in the latter case, currently it must be a single task).
+    Support for pipelines/general DAGs are in experimental branches.
 
     Args:
-        dag: sky.DAG to launch.
+        task: sky.Task or sky.Dag to launch.
         cluster_name: name of the cluster to create/reuse.  If None,
             auto-generate a name.
         retry_until_up: whether to retry launching the cluster until it is
@@ -313,25 +330,25 @@ def launch(
             OptimizeTarget.TIME.
         detach_run: If True, run setup first (blocking), then detach from the
             job's execution.
-        no_setup: if true, the cluster will not re-run setup instructions
+        no_setup: if True, do not re-run setup commands.
 
-    Examples:
+    Example:
         >>> import sky
-        >>> with sky.Dag() as dag:
-        >>>     task = sky.Task(run='echo hello SkyPilot')
-        >>>     task.set_resources(
-        ...             sky.Resources(
-        ...                 cloud=sky.AWS(),
-        ...                 accelerators='V100:4'
-        ...             )
-        ...     )
-        >>> sky.launch(dag, cluster_name='my-cluster')
+        >>> task = sky.Task(run='echo hello SkyPilot')
+        >>> task.set_resources(
+        ...         sky.Resources(
+        ...             cloud=sky.AWS(),
+        ...             accelerators='V100:4'
+        ...         )
+        ... )
+        >>> sky.launch(task, cluster_name='my-cluster')
 
     """
+    entrypoint = task
     backend_utils.check_cluster_name_not_reserved(cluster_name,
                                                   operation_str='sky.launch')
     _execute(
-        dag=dag,
+        entrypoint=entrypoint,
         dryrun=dryrun,
         down=down,
         stream_logs=stream_logs,
@@ -348,15 +365,62 @@ def launch(
 
 @usage_lib.entrypoint
 def exec(  # pylint: disable=redefined-builtin
-    dag: sky.Dag,
+    task: Union['sky.Task', 'sky.Dag'],
     cluster_name: str,
     dryrun: bool = False,
     down: bool = False,
     stream_logs: bool = True,
     backend: Optional[backends.Backend] = None,
-    optimize_target: OptimizeTarget = OptimizeTarget.COST,
     detach_run: bool = False,
-):
+) -> None:
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
+    """Execute a task on an existing cluster.
+
+    This function performs two actions:
+
+    (1) workdir syncing, if the task has a workdir specified;
+    (2) executing the task's ``run`` commands.
+
+    All other steps (provisioning, setup commands, file mounts syncing) are
+    skipped.  If any of those specifications changed in the task, this function
+    will not reflect those changes.  To ensure a cluster's setup is up to date,
+    use ``sky.launch()`` instead.
+
+    Execution and scheduling behavior:
+
+    - The task will undergo job queue scheduling, respecting any specified
+      resource requirement. It can be executed on any node of the cluster with
+      enough resources.
+    - The task is run under the workdir (if specified).
+    - The task is run non-interactively (without a pseudo-terminal or
+      pty), so interactive commands such as ``htop`` do not work.
+      Use ``ssh my_cluster`` instead.
+
+    Args:
+        task: sky.Task or sky.Dag containing the task to execute.
+        cluster_name: name of an existing cluster to execute the task.
+        down: Tear down the cluster after all jobs finish (successfully or
+            abnormally). If --idle-minutes-to-autostop is also set, the
+            cluster will be torn down after the specified idle time.
+            Note that if errors occur during provisioning/data syncing/setting
+            up, the cluster will not be torn down for debugging purposes.
+        dryrun: if True, do not actually execute the task.
+        stream_logs: if True, show the logs in the terminal.
+        backend: backend to use.  If None, use the default backend
+            (CloudVMRayBackend).
+        detach_run: if True, detach from logging once the task has been
+            submitted.
+
+    Raises:
+        ValueError: if the specified cluster does not exist or is not in UP
+            status.
+    """
+    entrypoint = task
+    if isinstance(entrypoint, sky.Dag):
+        logger.warning(
+            f'{colorama.Fore.YELLOW}Passing a sky.Dag to sky.exec() is '
+            'deprecated. Pass sky.Task instead.'
+            f'{colorama.Style.RESET_ALL}')
     backend_utils.check_cluster_name_not_reserved(cluster_name,
                                                   operation_str='sky.exec')
 
@@ -368,13 +432,12 @@ def exec(  # pylint: disable=redefined-builtin
         if status != global_user_state.ClusterStatus.UP:
             raise ValueError(f'Cluster {cluster_name!r} is not up.  '
                              'Use `sky status` to check the status.')
-    _execute(dag=dag,
+    _execute(entrypoint=entrypoint,
              dryrun=dryrun,
              down=down,
              stream_logs=stream_logs,
              handle=handle,
              backend=backend,
-             optimize_target=optimize_target,
              stages=[
                  Stage.SYNC_WORKDIR,
                  Stage.EXEC,
@@ -385,18 +448,19 @@ def exec(  # pylint: disable=redefined-builtin
 
 @usage_lib.entrypoint
 def spot_launch(
-    dag: sky.Dag,
+    task: Union['sky.Task', 'sky.Dag'],
     name: Optional[str] = None,
     stream_logs: bool = True,
     detach_run: bool = False,
     retry_until_up: bool = False,
 ):
+    # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Launch a managed spot job.
 
     Please refer to the sky.cli.spot_launch for the document.
 
     Args:
-        dag: DAG to launch.
+        task: sky.Task or sky.Dag to launch as a managed spot job.
         name: Name of the spot job.
         detach_run: Whether to detach the run.
 
@@ -404,10 +468,11 @@ def spot_launch(
         ValueError: cluster does not exist.
         sky.exceptions.NotSupportedError: the feature is not supported.
     """
+    entrypoint = task
     if name is None:
         name = backend_utils.generate_cluster_name()
 
-    _type_check_dag(dag)
+    dag = _convert_to_dag(entrypoint)
     assert len(dag.tasks) == 1, ('Only one task is allowed in a spot launch.',
                                  dag)
     task = dag.tasks[0]
@@ -450,16 +515,15 @@ def spot_launch(
                 'retry_until_up': retry_until_up,
             },
             output_prefix=spot.SPOT_CONTROLLER_YAML_PREFIX)
-        with sky.Dag() as spot_dag:
-            controller_task = task_lib.Task.from_yaml(yaml_path)
-            controller_task.spot_task = task
-            assert len(controller_task.resources) == 1
+        controller_task = task_lib.Task.from_yaml(yaml_path)
+        controller_task.spot_task = task
+        assert len(controller_task.resources) == 1
         print(f'{colorama.Fore.YELLOW}'
               f'Launching managed spot job {name} from spot controller...'
               f'{colorama.Style.RESET_ALL}')
         print('Launching spot controller...')
         _execute(
-            dag=spot_dag,
+            entrypoint=controller_task,
             stream_logs=stream_logs,
             cluster_name=controller_name,
             detach_run=detach_run,
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index ddd8ca957..54991a4ca 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -14,7 +14,6 @@ import filelock
 from sky import sky_logging
 from sky.skylet import constants
 from sky.utils import common_utils
-from sky.utils import subprocess_utils
 from sky.utils import db_utils
 from sky.utils import log_utils
 
@@ -503,21 +502,22 @@ def cancel_jobs(job_owner: str, jobs: Optional[List[int]]) -> None:
     else:
         job_records = _get_jobs_by_ids(jobs)
 
-    jobs = [make_ray_job_id(job['job_id'], job_owner) for job in job_records]
     # TODO(zhwu): `job_client.stop_job` will wait for the jobs to be killed, but
     # when the memory is not enough, this will keep waiting.
     job_client = _create_ray_job_submission_client()
 
-    def stop_job(job: str):
+    # Sequentially cancel the jobs to avoid the resource number bug caused by
+    # ray cluster (tracked in #1262).
+    for job in job_records:
+        job_id = make_ray_job_id(job['job_id'], job_owner)
         try:
-            job_client.stop_job(job)
+            job_client.stop_job(job_id)
         except RuntimeError as e:
             # If the job does not exist or if the request to the
             # job server fails.
             logger.warning(str(e))
+            continue
 
-    subprocess_utils.run_in_parallel(stop_job, jobs)
-    for job in job_records:
         if job['status'] in [JobStatus.PENDING, JobStatus.RUNNING]:
             set_status(job['job_id'], JobStatus.CANCELLED)
 
diff --git a/sky/skylet/ray_patches/__init__.py b/sky/skylet/ray_patches/__init__.py
index 8ab9b73e2..86f9d5490 100644
--- a/sky/skylet/ray_patches/__init__.py
+++ b/sky/skylet/ray_patches/__init__.py
@@ -27,6 +27,8 @@ Example workflow:
 import os
 import subprocess
 
+import pkg_resources
+
 from sky.skylet import constants
 
 
@@ -34,11 +36,12 @@ def _to_absolute(pwd_file):
     return os.path.join(os.path.dirname(os.path.abspath(__file__)), pwd_file)
 
 
-def _run_patch(target_file, patch_file):
+def _run_patch(target_file,
+               patch_file,
+               version=constants.SKY_REMOTE_RAY_VERSION):
     """Applies a patch if it has not been applied already."""
     # .orig is the original file that is not patched.
-    orig_file = os.path.abspath(
-        f'{target_file}-v{constants.SKY_REMOTE_RAY_VERSION}.orig')
+    orig_file = os.path.abspath(f'{target_file}-v{version}.orig')
     script = f"""\
     which patch >/dev/null 2>&1 || sudo yum install -y patch || true
     which patch >/dev/null 2>&1 || (echo "`patch` is not found. Failed to setup ray." && exit 1)
@@ -78,3 +81,16 @@ def patch() -> None:
     from ray.autoscaler._private import resource_demand_scheduler
     _run_patch(resource_demand_scheduler.__file__,
                _to_absolute('resource_demand_scheduler.py.patch'))
+
+    # Fix the Azure get-access-token (used by ray azure node_provider) timeout issue,
+    # by increasing the timeout.
+    # Tracked in https://github.com/Azure/azure-cli/issues/20404#issuecomment-1249575110
+    # Only patch it if azure cli is installed.
+    try:
+        import azure
+        from azure.identity._credentials import azure_cli
+        version = pkg_resources.get_distribution('azure-cli').version
+        _run_patch(azure_cli.__file__, _to_absolute('azure_cli.py.patch'),
+                   version)
+    except ImportError:
+        pass
diff --git a/sky/skylet/ray_patches/azure_cli.py.patch b/sky/skylet/ray_patches/azure_cli.py.patch
new file mode 100644
index 000000000..527289d29
--- /dev/null
+++ b/sky/skylet/ray_patches/azure_cli.py.patch
@@ -0,0 +1,4 @@
+147c147
+<             kwargs["timeout"] = 10
+---
+>             kwargs["timeout"] = 30
diff --git a/sky/spot/constants.py b/sky/spot/constants.py
index 9accab396..33d069327 100644
--- a/sky/spot/constants.py
+++ b/sky/spot/constants.py
@@ -1,6 +1,6 @@
 """Constants used for Managed Spot."""
 
-SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP = 30
+SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP = 10
 
 SPOT_CONTROLLER_TEMPLATE = 'spot-controller.yaml.j2'
 SPOT_CONTROLLER_YAML_PREFIX = '~/.sky/spot_controller'
diff --git a/sky/spot/spot_utils.py b/sky/spot/spot_utils.py
index 1784d986a..5dc402e1b 100644
--- a/sky/spot/spot_utils.py
+++ b/sky/spot/spot_utils.py
@@ -259,7 +259,7 @@ def stream_logs_by_id(job_id: int, follow: bool = True) -> str:
             f'in {JOB_STATUS_CHECK_GAP_SECONDS} seconds.')
         # If the tailing fails, it is likely that the cluster fails, so we wait
         # a while to make sure the spot state is updated by the controller, and
-        # check the spot status again.
+        # check the spot queue again.
         time.sleep(JOB_STATUS_CHECK_GAP_SECONDS)
         spot_status = spot_state.get_status(job_id)
     else:
diff --git a/sky/utils/cli_utils/status_utils.py b/sky/utils/cli_utils/status_utils.py
index 6931ee875..78e34014f 100644
--- a/sky/utils/cli_utils/status_utils.py
+++ b/sky/utils/cli_utils/status_utils.py
@@ -1,9 +1,10 @@
 """Utilities for sky status."""
-from typing import Any, Callable, Dict, List
+from typing import Any, Callable, Dict, List, Optional
 import click
 import colorama
 
 from sky import backends
+from sky import spot
 from sky.backends import backend_utils
 from sky.utils import common_utils
 from sky.utils.cli_utils import cli_utils
@@ -32,8 +33,14 @@ class StatusColumn:
         return val
 
 
-def show_status_table(cluster_records: List[Dict[str, Any]], show_all: bool):
-    """Compute cluster table values and display."""
+def show_status_table(cluster_records: List[Dict[str, Any]],
+                      show_all: bool,
+                      reserved_group_name: Optional[str] = None) -> int:
+    """Compute cluster table values and display.
+
+    Returns:
+        Number of pending auto{stop,down} clusters.
+    """
     # TODO(zhwu): Update the information for autostop clusters.
 
     status_columns = [
@@ -68,14 +75,20 @@ def show_status_table(cluster_records: List[Dict[str, Any]], show_all: bool):
         pending_autostop += _is_pending_autostop(record)
 
     if cluster_records:
+        if reserved_group_name is not None:
+            autostop_minutes = spot.SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP
+            click.echo(f'\n{colorama.Fore.CYAN}{colorama.Style.BRIGHT}'
+                       f'{reserved_group_name}{colorama.Style.RESET_ALL}'
+                       f'{colorama.Style.DIM} (will be autostopped if idle for '
+                       f'{autostop_minutes}min)'
+                       f'{colorama.Style.RESET_ALL}')
+        else:
+            click.echo(f'{colorama.Fore.CYAN}{colorama.Style.BRIGHT}Clusters'
+                       f'{colorama.Style.RESET_ALL}')
         click.echo(cluster_table)
-        if pending_autostop:
-            click.echo(
-                '\n'
-                f'You have {pending_autostop} clusters with auto{{stop,down}} '
-                'scheduled. Refresh statuses with: `sky status --refresh`.')
     else:
         click.echo('No existing clusters.')
+    return pending_autostop
 
 
 def show_local_status_table(local_clusters: List[str]):


Test patch:
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 6263a7c0f..51a0861c8 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -285,11 +285,15 @@ def test_job_queue():
         'job_queue',
         [
             f'sky launch -y -c {name} examples/job_queue/cluster.yaml',
-            f'sky exec {name} -d examples/job_queue/job.yaml',
-            f'sky exec {name} -d examples/job_queue/job.yaml',
-            f'sky exec {name} -d examples/job_queue/job.yaml',
-            f'sky logs {name} 2',
-            f'sky queue {name}',
+            f'sky exec {name} -n {name}-1 -d examples/job_queue/job.yaml',
+            f'sky exec {name} -n {name}-2 -d examples/job_queue/job.yaml',
+            f'sky exec {name} -n {name}-3 -d examples/job_queue/job.yaml',
+            f'sky queue {name} | grep {name}-1 | grep RUNNING',
+            f'sky queue {name} | grep {name}-2 | grep RUNNING',
+            f'sky queue {name} | grep {name}-3 | grep PENDING',
+            f'sky cancel {name} 2',
+            'sleep 5',
+            f'sky queue {name} | grep {name}-3 | grep RUNNING',
         ],
         f'sky down -y {name}',
     )
@@ -302,12 +306,15 @@ def test_n_node_job_queue():
         'job_queue_multinode',
         [
             f'sky launch -y -c {name} examples/job_queue/cluster_multinode.yaml',
-            f'sky exec {name} -d examples/job_queue/job_multinode.yaml',
-            f'sky exec {name} -d examples/job_queue/job_multinode.yaml',
-            f'sky exec {name} -d examples/job_queue/job_multinode.yaml',
+            f'sky exec {name} -n {name}-1 -d examples/job_queue/job_multinode.yaml',
+            f'sky exec {name} -n {name}-2 -d examples/job_queue/job_multinode.yaml',
+            f'sky exec {name} -n {name}-3 -d examples/job_queue/job_multinode.yaml',
+            f'sky queue {name} | grep {name}-1 | grep RUNNING',
+            f'sky queue {name} | grep {name}-2 | grep RUNNING',
+            f'sky queue {name} | grep {name}-3 | grep PENDING',
             f'sky cancel {name} 1',
-            f'sky logs {name} 2',
-            f'sky queue {name}',
+            'sleep 5',
+            f'sky queue {name} | grep {name}-3 | grep RUNNING',
         ],
         f'sky down -y {name}',
     )
@@ -319,7 +326,7 @@ def test_large_job_queue():
     test = Test(
         'large_job_queue',
         [
-            f'sky launch -y -c {name} --cloud gcp ""',
+            f'sky launch -y -c {name} --cloud gcp',
             f'for i in `seq 1 75`; do sky exec {name} -d "echo $i; sleep 100000000"; done',
             f'sky cancel {name} 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16',
             'sleep 20',
@@ -652,12 +659,12 @@ def test_spot():
             f'sky spot launch -n {name}-1 examples/managed_spot.yaml -y -d',
             f'sky spot launch -n {name}-2 examples/managed_spot.yaml -y -d',
             'sleep 5',
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name}-1 | head -n1 | grep "STARTING\|RUNNING"',
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name}-2 | head -n1 | grep "STARTING\|RUNNING"',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name}-1 | head -n1 | grep "STARTING\|RUNNING"',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name}-2 | head -n1 | grep "STARTING\|RUNNING"',
             f'sky spot cancel -y -n {name}-1',
             'sleep 200',
-            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's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name}-1 | head -n1 | grep CANCELLED',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name}-2 | head -n1 | grep "RUNNING\|SUCCEEDED"',
         ],
         f'sky spot cancel -y -n {name}-1; sky spot cancel -y -n {name}-2',
     )
@@ -675,9 +682,9 @@ def test_gcp_spot():
             'sleep 5',
             # Captures & prints the table for easier debugging. Two echo's to
             # separate the table from the grep output.
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep STARTING',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep STARTING',
             'sleep 200',
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep RUNNING',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep RUNNING',
         ],
         f'sky spot cancel -y -n {name}',
     )
@@ -694,7 +701,7 @@ def test_spot_recovery():
         [
             f'sky spot launch --cloud aws --region {region} -n {name} "echo SKYPILOT_RUN_ID: $SKYPILOT_RUN_ID; sleep 1000"  -y -d',
             'sleep 300',
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RUNNING"',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RUNNING"',
             f'RUN_ID=$(sky spot logs -n {name} --no-follow | grep SKYPILOT_RUN_ID | cut -d: -f2); echo $RUN_ID',
             # Terminate the cluster manually.
             (f'aws ec2 terminate-instances --region {region} --instance-ids $('
@@ -703,9 +710,9 @@ 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"',
+            f's=$(sky spot queue); 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"',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RUNNING"',
             f'sky spot logs -n {name} --no-follow | grep SKYPILOT_RUN_ID | grep "$RUN_ID"',
         ],
         f'sky spot cancel -y -n {name}',
@@ -729,8 +736,8 @@ def test_spot_storage():
             'managed-spot-storage',
             [
                 f'sky spot launch -n {name} {file_path} -y',
-                'sleep 60',  # Wait the spot status to be updated
-                f'sky spot status | grep {name} | grep SUCCEEDED',
+                'sleep 60',  # Wait the spot queue to be updated
+                f'sky spot queue | grep {name} | grep SUCCEEDED',
                 f'[ $(aws s3api list-buckets --query "Buckets[?contains(Name, \'{storage_name}\')].Name" --output text | wc -l) -eq 0 ]'
             ],
             f'sky spot cancel -y -n {name}',
@@ -747,9 +754,9 @@ def test_spot_tpu():
         [
             f'sky spot launch -n {name} examples/tpu/tpuvm_mnist.yaml -y -d',
             'sleep 5',
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep STARTING',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep STARTING',
             'sleep 600',  # TPU takes a while to launch
-            f's=$(sky spot status); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RUNNING\|SUCCEEDED"',
+            f's=$(sky spot queue); printf "$s"; echo; echo; printf "$s" | grep {name} | head -n1 | grep "RUNNING\|SUCCEEDED"',
         ],
         f'sky spot cancel -y -n {name}',
     )
@@ -782,7 +789,7 @@ def test_inline_spot_env():
         [
             f'sky spot launch -n {name} -y --env TEST_ENV="hello world" -- "([[ ! -z \\"\$TEST_ENV\\" ]] && [[ ! -z \\"\$SKY_NODE_IPS\\" ]] && [[ ! -z \\"\$SKY_NODE_RANK\\" ]]) || exit 1"',
             'sleep 10',
-            f's=$(sky spot status) && printf "$s" && echo "$s"  | grep {name} | grep SUCCEEDED',
+            f's=$(sky spot queue) && printf "$s" && echo "$s"  | grep {name} | grep SUCCEEDED',
         ],
         f'sky spot cancel -y -n {name}',
     )
diff --git a/tests/test_spot.py b/tests/test_spot.py
index 549411a4d..30f0e0e92 100644
--- a/tests/test_spot.py
+++ b/tests/test_spot.py
@@ -94,14 +94,29 @@ class TestReservedClustersOperations:
                                                 ready=True)
 
     @pytest.mark.timeout(60)
-    def test_down_spot_controller(self, _mock_cluster_state):
-        cli_runner = cli_testing.CliRunner()
+    def test_down_spot_controller(self, _mock_cluster_state, monkeypatch):
 
-        result = cli_runner.invoke(cli.down, [spot.SPOT_CONTROLLER_NAME])
-        assert result.exit_code == click.UsageError.exit_code
-        assert (
-            f'Terminating reserved cluster(s) \'{spot.SPOT_CONTROLLER_NAME}\' '
-            'is not supported' in result.output)
+        def mock_cluster_refresh_up(
+            cluster_name: str,
+            *,
+            force_refresh: bool = False,
+            acquire_per_cluster_status_lock: bool = True,
+        ):
+            record = global_user_state.get_cluster_from_name(cluster_name)
+            return record['status'], record['handle']
+
+        monkeypatch.setattr(
+            'sky.backends.backend_utils.refresh_cluster_status_handle',
+            mock_cluster_refresh_up)
+
+        monkeypatch.setattr('sky.core.spot_queue', lambda refresh: [])
+
+        cli_runner = cli_testing.CliRunner()
+        result = cli_runner.invoke(cli.down, [spot.SPOT_CONTROLLER_NAME],
+                                   input='n')
+        assert 'WARNING: Tearing down the managed spot controller (UP).' in result.output
+        assert isinstance(result.exception,
+                          SystemExit), (result.exception, result.output)
 
         result = cli_runner.invoke(cli.down, ['sky-spot-con*'])
         assert not result.exception