Repository: skypilot-org/skypilot
Pull request #1293 (2022-10-23 04:22:28)

[UX/Programmatic API] Support task in launch/exec/spot_launch
This PR is a proposal to support `sky.Task` in launch/exec/spot_launch directly in addition to our previous `sky.Dag` API. That is because none of our users is using the dag abstraction for now and we do not support dag with more than 1 task in our programmatic API.

Tested:
- [x] `sky launch -c min echo hi`
- [x] `python ./examples/resnet_app.py`
- [x] `tests/run_smoke_tests.sh`


Patch:
diff --git a/docs/source/examples/spot-jobs.rst b/docs/source/examples/spot-jobs.rst
index 04c0b270a..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
diff --git a/docs/source/reference/cli.rst b/docs/source/reference/cli.rst
index 5cd95c848..ce1d2a7d7 100644
--- a/docs/source/reference/cli.rst
+++ b/docs/source/reference/cli.rst
@@ -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/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/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 a6f9527a4..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 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)
+    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()})
+
+    # 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:
-            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']),
@@ -1077,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,
@@ -1103,7 +1109,7 @@ def launch(
             raise ValueError(f'{backend_name} backend is not supported.')
 
     _launch_with_confirm(
-        dag,
+        task,
         backend,
         cluster,
         dryrun=dryrun,
@@ -1219,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,
@@ -1236,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()
@@ -1902,10 +1908,10 @@ def _hint_for_down_spot_controller(controller_name: str):
            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 status`) will be lost.')
+           'jobs (output of `sky spot queue`) will be lost.')
     if cluster_status == global_user_state.ClusterStatus.UP:
         try:
-            spot_jobs = core.spot_status(refresh=False)
+            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
@@ -2691,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,
@@ -2713,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,
@@ -2736,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
@@ -2757,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:
@@ -2787,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 ecdd94d8a..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
@@ -87,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)
 
@@ -553,10 +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:
         [
@@ -579,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:
diff --git a/sky/execution.py b/sky/execution.py
index 244b8e3e7..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,
@@ -291,11 +299,12 @@ def launch(
     (when specified, it is synced to remote cluster).  The task undergoes job
     queue scheduling on the cluster.
 
-    Currently, the first argument must be a single-node DAG (i.e., a single
-    task). Support for pipelines/general DAGs are in experimental branches.
+    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
@@ -325,21 +334,21 @@ def launch(
 
     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,
@@ -356,7 +365,7 @@ 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,
@@ -388,7 +397,7 @@ def exec(  # pylint: disable=redefined-builtin
       Use ``ssh my_cluster`` instead.
 
     Args:
-        dag: sky.DAG containing the task to execute.
+        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
@@ -406,6 +415,12 @@ def exec(  # pylint: disable=redefined-builtin
         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')
 
@@ -417,7 +432,7 @@ 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,
@@ -433,7 +448,7 @@ 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,
@@ -445,7 +460,7 @@ def spot_launch(
     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.
 
@@ -453,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]
@@ -499,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/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:


Test patch:
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 76162a7cd..51a0861c8 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -659,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',
     )
@@ -682,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}',
     )
@@ -701,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 $('
@@ -710,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}',
@@ -736,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}',
@@ -754,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}',
     )
@@ -789,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 f80f47ea8..30f0e0e92 100644
--- a/tests/test_spot.py
+++ b/tests/test_spot.py
@@ -109,7 +109,7 @@ class TestReservedClustersOperations:
             'sky.backends.backend_utils.refresh_cluster_status_handle',
             mock_cluster_refresh_up)
 
-        monkeypatch.setattr('sky.core.spot_status', lambda refresh: [])
+        monkeypatch.setattr('sky.core.spot_queue', lambda refresh: [])
 
         cli_runner = cli_testing.CliRunner()
         result = cli_runner.invoke(cli.down, [spot.SPOT_CONTROLLER_NAME],