Repository: skypilot-org/skypilot
Pull request #1291 (2022-10-22 19:09:47)

Fix SKY_NODE_RANK environment variable
As discussed in #1206, this PR ensures that SKY_NODE_RANK is always 0 on the head node, and that it is assigned to the worker nodes in a stable fashion when executing tasks on multi-node clusters. 

To accomplish this, in each call to `_execute_task_n_nodes`, we create a map of internal node ips to their correct rank, and pass this map into the Ray code generator, which assigns to correct rank to each node.

Tested:
- [x] Single-node cluster
- [x] Multi-node cluster using the following config
```
resources:
  cloud: aws
  disk_size: 1024
  instance_type: m6i.large

workdir: .

num_nodes: 4

run: |
  echo $SKY_NODE_RANK
```




Patch:
diff --git a/README.md b/README.md
index 3d751c547..036046384 100644
--- a/README.md
+++ b/README.md
@@ -5,26 +5,45 @@
   </picture>
 </p>
 
-[![Join Slack](https://img.shields.io/badge/SkyPilot-Join%20Slack-blue?logo=slack)](https://join.slack.com/t/skypilot-org/shared_invite/zt-1i4pa7lyc-g6Lo4_rqqCFWOSXdvwTs3Q)
-![pytest](https://github.com/skypilot-org/skypilot/actions/workflows/pytest.yml/badge.svg)
-[![Documentation Status](https://readthedocs.org/projects/skypilot/badge/?version=latest)](https://skypilot.readthedocs.io/en/latest/?badge=latest)
+<p align="center">
+  <a href="https://skypilot.readthedocs.io/en/latest/"> 
+    <img alt="Documentation" src="https://readthedocs.org/projects/skypilot/badge/?version=latest">
+  </a>
+  
+  <a href="https://github.com/skypilot-org/skypilot/releases"> 
+    <img alt="GitHub Release" src="https://img.shields.io/github/release/skypilot-org/skypilot.svg">
+  </a>
+  
+  <a href="https://join.slack.com/t/skypilot-org/shared_invite/zt-1i4pa7lyc-g6Lo4_rqqCFWOSXdvwTs3Q"> 
+    <img alt="Join Slack" src="https://img.shields.io/badge/SkyPilot-Join%20Slack-blue?logo=slack">
+  </a>
+  
+</p>
+
 
-SkyPilot is a framework for easily running machine learning workloads[^1] on any cloud. 
+<h3 align="center">
+    Run jobs on any cloud, easily and cost effectively
+</h3>
 
-Use the clouds **easily** and **cost effectively**, without needing cloud infra expertise.
+SkyPilot is a framework for easily and cost effectively running ML workloads[^1] on any cloud. 
 
-_Ease of use_
-* **Run existing projects on the cloud** with zero code changes
-* Use a **unified interface** to run on any cloud, without vendor lock-in (currently AWS, Azure, GCP)
-* **Queue jobs** on one or multiple clusters
-* **Automatic failover** to find scarce resources (GPUs) across regions and clouds
-* **Use datasets on the cloud** like you would on a local file system 
+SkyPilot abstracts away cloud infra burden:
+- Launch jobs & clusters on any cloud (AWS, Azure, GCP)
+- Find scarce resources across zones/regions/clouds
+- Queue jobs & use cloud object stores
 
-_Cost saving_
-* Run jobs on **spot instances** with **automatic recovery** from preemptions
-* Hands-free cluster management: **automatically stopping idle clusters**
-* One-click use of **TPUs**, for high-performance, cost-effective training
-* Automatically benchmark and find the cheapest hardware for your job
+SkyPilot cuts your cloud costs:
+* [Managed Spot](https://skypilot.readthedocs.io/en/latest/examples/spot-jobs.html): **3x cost savings** using spot VMs, with auto-recovery from preemptions
+* [Autostop](https://skypilot.readthedocs.io/en/latest/reference/auto-stop.html): hands-free cleanup of idle clusters 
+* [Benchmark](https://skypilot.readthedocs.io/en/latest/reference/benchmark/index.html): find best VM types for your jobs
+* Optimizer: **2x cost savings** by auto-picking best prices across zones/regions/clouds
+
+SkyPilot supports your existing GPU, TPU, and CPU workloads, with no code changes. 
+
+Install with pip (choose your clouds) or [from source](https://skypilot.readthedocs.io/en/latest/getting-started/installation.html):
+```
+pip install "skypilot[aws,gcp,azure]"
+```
 
 ## Getting Started
 You can find our documentation [here](https://skypilot.readthedocs.io/en/latest/).
@@ -32,40 +51,37 @@ You can find our documentation [here](https://skypilot.readthedocs.io/en/latest/
 - [Quickstart](https://skypilot.readthedocs.io/en/latest/getting-started/quickstart.html)
 - [CLI reference](https://skypilot.readthedocs.io/en/latest/reference/cli.html)
 
-## Example SkyPilot Task
+## SkyPilot in 1 minute
 
 A SkyPilot task specifies: resource requirements, data to be synced, setup commands, and the task commands. 
 
-Once written in this [**unified interface**](https://skypilot.readthedocs.io/en/latest/reference/yaml-spec.html) (YAML or Python API), the task can be launched on any available cloud. 
+Once written in this [**unified interface**](https://skypilot.readthedocs.io/en/latest/reference/yaml-spec.html) (YAML or Python API), the task can be launched on any available cloud.  This avoids vendor lock-in, and allows easily moving jobs to a different provider.
 
-Example:
+Paste the following into a file `my_task.yaml`:
 
 ```yaml
-# my_task.yaml
 resources:
-  # 1x NVIDIA V100 GPU
-  accelerators: V100:1
+  accelerators: V100:1  # 1x NVIDIA V100 GPU
 
-# Number of VMs to launch in the cluster
-num_nodes: 1
+num_nodes: 1  # Number of VMs to launch
 
 # Working directory (optional) containing the project codebase.
 # Its contents are synced to ~/sky_workdir/ on the cluster.
 workdir: ~/torch_examples
 
-# Commands to be run before executing the job
+# Commands to be run before executing the job.
 # Typical use: pip install -r requirements.txt, git clone, etc.
 setup: |
   pip install torch torchvision
 
-# Commands to run as a job
-# Typical use: make use of resources, such as running training.
+# Commands to run as a job.
+# Typical use: launch the main program.
 run: |
   cd mnist
   python main.py --epochs 1
 ```
 
-Prepare the workdir by cloning locally:
+Prepare the workdir by cloning:
 ```bash
 git clone https://github.com/pytorch/examples.git ~/torch_examples
 ```
@@ -74,10 +90,11 @@ Launch with `sky launch`:
 ```bash
 sky launch my_task.yaml
 ```
-SkyPilot will perform multiple actions for you:
+
+SkyPilot then performs the heavy-lifting for you, including:
 1. Find the lowest priced VM instance type across different clouds
-2. Provision the VM
-3. Copy the local contents of `workdir` to the VM
+2. Provision the VM, with auto-failover if the cloud returned capacity errors
+3. Sync the local `workdir` to the VM
 4. Run the task's `setup` commands to prepare the VM for running the task 
 5. Run the task's `run` commands
 
@@ -86,20 +103,26 @@ SkyPilot will perform multiple actions for you:
 </p>
 
 
-See [**`examples`**](./examples) for more YAMLs that run popular ML frameworks on the cloud with one command (PyTorch/Distributed PyTorch, TensorFlow/Distributed TensorFlow, HuggingFace, JAX, Flax, Docker).  
+Refer to [Quickstart](https://skypilot.readthedocs.io/en/latest/getting-started/quickstart.html) to get started with SkyPilot.
+
+## Learn more
 
-Besides YAML, SkyPilot offers a corresponding [**Python API**](https://github.com/skypilot-org/skypilot/blob/master/sky/core.py) for programmatic use.
+- [Documentation](https://skypilot.readthedocs.io/en/latest/)
+- [Example: HuggingFace](https://skypilot.readthedocs.io/en/latest/getting-started/tutorial.html) 
+- [Tutorials](https://github.com/skypilot-org/skypilot-tutorial) 
+- [YAML reference](https://skypilot.readthedocs.io/en/latest/reference/yaml-spec.html)
+- Framework examples: [PyTorch DDP](https://github.com/skypilot-org/skypilot/blob/master/examples/resnet_distributed_torch.yaml),  [Distributed](https://github.com/skypilot-org/skypilot/blob/master/examples/resnet_distributed_tf_app.py) [TensorFlow](https://github.com/skypilot-org/skypilot/blob/master/examples/resnet_app_storage.yaml), [JAX/Flax on TPU](https://github.com/skypilot-org/skypilot/blob/master/examples/tpu/tpuvm_mnist.yaml), [Stable Diffusion](https://github.com/skypilot-org/skypilot/tree/master/examples/stable_diffusion), [Detectron2](https://github.com/skypilot-org/skypilot/blob/master/examples/detectron2_docker.yaml), [programmatic grid search](https://github.com/skypilot-org/skypilot/blob/master/examples/huggingface_glue_imdb_grid_search_app.py), [Docker](https://github.com/skypilot-org/skypilot/blob/master/examples/docker/echo_app.yaml), and [many more](./examples).
 
-Refer to [Quickstart](https://skypilot.readthedocs.io/en/latest/getting-started/quickstart.html) for more on how to get started with SkyPilot.
 
+## Issues, feature requests, and questions
+We are excited to hear your feedback! 
+* For issues and feature requests, please [open a GitHub issue](https://github.com/skypilot-org/skypilot/issues/new).
+* For questions, please use [GitHub Discussions](https://github.com/skypilot-org/skypilot/discussions).
 
-## Issues, feature requests and questions
-We are excited to hear your feedback! SkyPilot has two channels for engaging with the community - [GitHub Issues](https://github.com/skypilot-org/skypilot/issues) and [GitHub Discussions](https://github.com/skypilot-org/skypilot/discussions).
-* For bug reports and issues, please [open an issue](https://github.com/skypilot-org/skypilot/issues/new).
-* For feature requests or general questions, please join us on [GitHub Discussions](https://github.com/skypilot-org/skypilot/discussions).
+For general discussions, join us on the [SkyPilot Slack](https://join.slack.com/t/skypilot-org/shared_invite/zt-1i4pa7lyc-g6Lo4_rqqCFWOSXdvwTs3Q).
 
 ## Contributing
-We welcome and value all contributions to the project! Please refer to the [contribution guide](CONTRIBUTING.md) for more on how to get involved.
+We welcome and value all contributions to the project! Please refer to [CONTRIBUTING](CONTRIBUTING.md) for how to get involved.
 
 <!-- Footnote -->
-[^1]: While SkyPilot is currently targeted at machine learning workloads, it supports and has been used for other general workloads. We're excited to hear about your use case and how we can better support your requirements - please join us in [this discussion](https://github.com/skypilot-org/skypilot/discussions/1016)!
+[^1]: While SkyPilot is currently targeted at machine learning workloads, it supports and has been used for other general batch workloads. We're excited to hear about your use case and how we can better support your requirements; please join us in [this discussion](https://github.com/skypilot-org/skypilot/discussions/1016)!
diff --git a/ROADMAP.md b/ROADMAP.md
new file mode 100644
index 000000000..bd62d08f8
--- /dev/null
+++ b/ROADMAP.md
@@ -0,0 +1,59 @@
+# SkyPilot Roadmap
+
+This doc lists general directions of interest to facilitate community contributions. 
+
+Note that
+- This list is not meant to be comprehensive (i.e., new work items of interest may pop up)
+- Even though listed under a specific version, not all items need to be completed before we ship that version (i.e., some items can go into future versions)
+
+## v0.3
+
+### Managed Spot
+- Minimize the cost of the controller
+  - Support running spot controller on an existing/local cluster
+  - Reducing the fixed cost of the controller (e.g., allow setting controller VM type)
+- Supporting a higher number of pending/concurrent jobs
+- Framework-specific guides to add checkpointing/reloading using SkyPilot Storage
+
+### Smarter Optimizer
+- Fine-grained optimizer: pick by cheapest zone order
+- Better consider data egress time/cost
+  - Consider buckets/Storage objects in file_mounts
+- Optimizing the data placement for SkyPilot Storage local uploads
+  - Use the optimizer to decide the bucket location
+
+### Programmatic API
+- Refactor/extend the current API to *make it easy to programmatically use SkyPilot*
+- Expose core classes in docs
+
+### Support more clouds
+- Refactoring of interfaces to ease adding new clouds
+- IBM Cloud
+- Explore support for low-cost clouds (e.g., lambda labs/runpod/jarvis labs)
+
+### On-prem
+- Robustify the on-prem feature
+- Design for switching between cloud and on-prem
+- Explore/design of "local mode" to run SkyPilot tasks locally
+
+### Faster launching speed
+- Consider a more minimal image
+- Azure speed investigation
+
+### k8s support
+- Ray-on-k8s backend
+  - To figure out: Launch a new k8s cluster? Launch SkyPilot Tasks to an existing k8s cluster?
+
+### Cost: Optimization, Tracking, and Reporting
+- Track and show costs related to a job/cluster
+- For managed spot jobs, track and show %savings vs. on-demand
+- Optimizer: take into account disk costs
+
+### Serverless
+- Design and prototype of a "serverless jobs" submission API and CLI
+  - Initial use case: hundreds of hyperparameter tuning trials
+
+### Backend
+- Support heterogeneous node types in a cluster (e.g., in RL, CPU actor(s) and GPU learner(s) in the same cluster)
+- Support CPUs as resource requirements
+- General robustness/UX improvements
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 8d19677b6..797ea9f94 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -52,19 +52,28 @@ pygments_style = None
 autosummary_generate = True
 napolean_use_rtype = False
 
+# -- Options for autodoc
+
+# Python methods should be presented in source code order
+autodoc_member_order = 'bysource'
+
 # -- Options for HTML output
 
 html_theme = 'sphinx_book_theme'
 html_theme_options = {
     # 'show_toc_level': 2,
     'logo_only': True,
-    "repository_url": "https://github.com/skypilot-org/skypilot",
-    "use_repository_button": True,
-    "use_issues_button": True,
-    "use_edit_page_button": True,
-    "path_to_docs": "docs/source",
+    'repository_url': 'https://github.com/skypilot-org/skypilot',
+    'use_repository_button': True,
+    'use_issues_button': True,
+    'use_edit_page_button': True,
+    'path_to_docs': 'docs/source',
 }
 
+# The name for this set of Sphinx documents.  If None, it defaults to
+# "<project> v<release> documentation".
+html_title = 'SkyPilot documentation'
+
 # -- Options for EPUB output
 epub_show_urls = 'footnote'
 
@@ -76,13 +85,13 @@ html_show_sourcelink = False
 
 # The name of an image file (relative to this directory) to place at the top
 # of the sidebar.
-html_logo = "images/skypilot-wide-light-1k.png"
+html_logo = 'images/skypilot-wide-light-1k.png'
 
 # The name of an image file (within the static path) to use as favicon of the
 # docs. This file should be a Windows icon file (.ico), 16x16 or 32x32 pixels.
-html_favicon = "_static/favicon.ico"
+html_favicon = '_static/favicon.ico'
 
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,
-# so a file named "default.css" will overwrite the builtin "default.css".
-html_static_path = ["_static"]
+# so a file named 'default.css' will overwrite the builtin 'default.css'.
+html_static_path = ['_static']
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/index.rst b/docs/source/index.rst
index bd9ae46fc..17381cb29 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -1,4 +1,4 @@
-SkyPilot Documentation
+Welcome to SkyPilot!
 =========================
 
 .. figure:: ./images/skypilot-wide-light-1k.png
@@ -17,24 +17,29 @@ SkyPilot Documentation
    <a class="github-button" href="https://github.com/skypilot-org/skypilot/fork" data-icon="octicon-repo-forked" data-size="large" aria-label="Fork skypilot-org/skypilot on GitHub">Fork</a>
    </p>
 
-SkyPilot is a framework for easily running machine learning workloads on any cloud.
+   <p style="text-align:center">
+   <strong>Run jobs on any cloud, easily and cost effectively</strong>
+   </p>
+
+SkyPilot is a framework for easily and cost effectively running ML workloads on any cloud.
+
+SkyPilot abstracts away cloud infra burden:
 
-Use the clouds **easily** and **cost effectively**, without needing cloud infra expertise.
+- Launch jobs & clusters on any cloud (AWS, Azure, GCP)
+- Find scarce resources across zones/regions/clouds
+- Queue jobs & use cloud object stores
 
-*Ease of use*
+SkyPilot cuts your cloud costs:
 
-- **Run existing projects on the cloud** with zero code changes
-- Use a **unified interface** to run on any cloud, without vendor lock-in (currently AWS, Azure, GCP)
-- **Queue jobs** on one or multiple clusters
-- **Automatic failover** to find scarce resources (GPUs) across regions and clouds
-- **Use datasets on the cloud** like you would on a local file system
+* :ref:`Managed Spot <Managed Spot Jobs>`: **3x cost savings** using spot VMs, with auto-recovery from preemptions
+* :ref:`Autostop <Auto-stopping>`: hands-free cleanup of idle clusters
+* :ref:`Benchmark <Benchmark>`: find best VM types for your jobs
+* Optimizer: **2x cost savings** by auto-picking best prices across zones/regions/clouds
 
-*Cost saving*
+SkyPilot supports your existing GPU, TPU, and CPU workloads, with no code changes.
 
-- Run jobs on **spot instances** with **automatic recovery** from preemptions
-- Hands-free cluster management: **automatically stopping idle clusters**
-- One-click use of **TPUs**, for high-performance, cost-effective training
-- Automatically benchmark and find the cheapest hardware for your job
+Documentation
+--------------------------
 
 .. toctree::
    :maxdepth: 1
@@ -81,9 +86,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..de8e9aea4
--- /dev/null
+++ b/docs/source/reference/api.rst
@@ -0,0 +1,64 @@
+.. _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
+-----------------
+
+.. autoclass:: sky.Task
+    :members:
+    :exclude-members: estimate_runtime, get_cloud_to_remote_file_mounts, get_inputs_cloud, get_local_to_remote_file_mounts, set_time_estimator, sync_storage_mounts, to_yaml_config
+
+    .. automethod:: __init__
diff --git a/docs/source/reference/cli.rst b/docs/source/reference/cli.rst
index 53da6ae5c..7b131e36b 100644
--- a/docs/source/reference/cli.rst
+++ b/docs/source/reference/cli.rst
@@ -1,7 +1,7 @@
 .. _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
@@ -34,24 +34,6 @@ Core CLI
    :prog: sky autostop
    :nested: full
 
-
-Interactive Node CLI
------------------------
-
-.. click:: sky.cli:cpunode
-   :prog: sky cpunode
-   :nested: full
-
-.. _sky-gpunode:
-.. click:: sky.cli:gpunode
-   :prog: sky gpunode
-   :nested: full
-
-.. click:: sky.cli:tpunode
-   :prog: sky tpunode
-   :nested: full
-
-
 Job Queue CLI
 --------------
 
@@ -68,17 +50,6 @@ Job Queue CLI
    :nested: full
 
 
-Storage CLI
-------------
-
-.. click:: sky.cli:storage_ls
-   :prog: sky storage ls
-   :nested: full
-
-.. click:: sky.cli:storage_delete
-   :prog: sky storage delete
-   :nested: full
-
 Managed Spot Jobs CLI
 ---------------------------
 
@@ -86,8 +57,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
@@ -98,14 +69,42 @@ Managed Spot Jobs CLI
    :prog: sky spot logs
    :nested: full
 
-Miscellaneous
--------------
+Interactive Node CLI
+-----------------------
 
-.. click:: sky.cli:check
-   :prog: sky check
+.. click:: sky.cli:cpunode
+   :prog: sky cpunode
+   :nested: full
+
+.. _sky-gpunode:
+.. click:: sky.cli:gpunode
+   :prog: sky gpunode
+   :nested: full
+
+.. click:: sky.cli:tpunode
+   :prog: sky tpunode
    :nested: full
 
 
+Storage CLI
+------------
+
+.. click:: sky.cli:storage_ls
+   :prog: sky storage ls
+   :nested: full
+
+.. click:: sky.cli:storage_delete
+   :prog: sky storage delete
+   :nested: full
+
+Utils: ``show-gpus``, ``check``
+---------------------------------------
+
+
 .. click:: sky.cli:show_gpus
    :prog: sky show-gpus
    :nested: full
+
+.. click:: sky.cli:check
+   :prog: sky check
+   :nested: full
diff --git a/docs/source/reference/local/setup.rst b/docs/source/reference/local/setup.rst
index 3a32b8519..c737143ed 100644
--- a/docs/source/reference/local/setup.rst
+++ b/docs/source/reference/local/setup.rst
@@ -14,13 +14,13 @@ For further reference, `here <https://docs.ray.io/en/latest/ray-core/configure.h
 Installing SkyPilot dependencies
 -----------------------------------
 
-SkyPilot On-prem requires :code:`python3`, :code:`ray==1.13.0`, and :code:`sky` to be setup on all local nodes and globally available to all users.
+SkyPilot On-prem requires :code:`python3`, :code:`ray==2.0.1`, and :code:`sky` to be setup on all local nodes and globally available to all users.
 
 To install Ray and SkyPilot for all users, run the following commands on all local nodes:
 
 .. code-block:: console
 
-   $ pip3 install ray[default]==1.13.0
+   $ pip3 install ray[default]==2.0.1
 
    $ # SkyPilot requires python >= 3.6.
    $ pip3 install skypilot
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/cluster-config.yaml b/examples/local/cluster-config.yaml
index 1bff98245..e30992a60 100644
--- a/examples/local/cluster-config.yaml
+++ b/examples/local/cluster-config.yaml
@@ -4,7 +4,7 @@
 # The system administrator must have `sudo` access to the local nodes.
 # Requirements:
 #   1) Python (> 3.6) on all nodes.
-#   2) Ray CLI (= 1.13.0) on all nodes.
+#   2) Ray CLI (= 2.0.1) on all nodes.
 #
 # Example usage:
 #   >> sky admin deploy cluster-config.yaml
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/examples/using_file_mounts.yaml b/examples/using_file_mounts.yaml
index 328e0b398..2cee62017 100644
--- a/examples/using_file_mounts.yaml
+++ b/examples/using_file_mounts.yaml
@@ -57,6 +57,7 @@ file_mounts:
 
   # Relative paths are under ~/ (after sync, ~/relative_dir/ exists).
   relative_dir: ~/tmp-workdir
+  ./dotslash_relative_dir: ~/tmp-workdir
 
   #############################################
   #  Sync from Google Cloud Storage (gsutil)  #
@@ -102,8 +103,9 @@ run: |
   ls -lthr ~/.ssh/
 
   ls -lthr ~/
-  ls -lthr ~/relative_dir
-  ls -lthr ~/relative_dir/
+  ls -lthr ~/sky_workdir/relative_dir
+  ls -lthr ~/sky_workdir/relative_dir/
+  ls -lthr ~/sky_workdir/dotslash_relative_dir/
 
   du -schL /train-00001-of-01024
   wc -l /train-00001-of-01024
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 fe7a613d5..7de75ffa5 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -756,12 +756,10 @@ def write_cluster_config(
     region_name = resources_vars.get('region')
 
     yaml_path = _get_yaml_path_from_cluster_name(cluster_name)
-    old_yaml_content = None
-    if os.path.exists(yaml_path) and keep_launch_fields_in_existing_config:
-        with open(yaml_path, 'r') as f:
-            old_yaml_content = f.read()
 
-    yaml_path = fill_template(
+    # Use a tmp file path to avoid incomplete YAML file being re-used in the future.
+    tmp_yaml_path = yaml_path + '.tmp'
+    tmp_yaml_path = fill_template(
         cluster_config_template,
         dict(
             resources_vars,
@@ -799,28 +797,35 @@ def write_cluster_config(
                 'ssh_private_key': (None if auth_config is None else
                                     auth_config['ssh_private_key']),
             }),
-    )
+        output_path=tmp_yaml_path)
     config_dict['cluster_name'] = cluster_name
     config_dict['ray'] = yaml_path
     if dryrun:
+        # If dryrun, return the unfinished tmp yaml path.
+        config_dict['ray'] = tmp_yaml_path
         return config_dict
-    _add_auth_to_cluster_config(cloud, yaml_path)
+    _add_auth_to_cluster_config(cloud, tmp_yaml_path)
     # Delay the optimization of the config until the authentication files is added.
     if not isinstance(cloud, clouds.Local):
         # Only optimize the file mounts for public clouds now, as local has not
         # been fully tested yet.
-        _optimize_file_mounts(yaml_path)
+        _optimize_file_mounts(tmp_yaml_path)
 
     # Restore the old yaml content for backward compatibility.
-    if old_yaml_content is not None:
+    if os.path.exists(yaml_path) and keep_launch_fields_in_existing_config:
         with open(yaml_path, 'r') as f:
+            old_yaml_content = f.read()
+        with open(tmp_yaml_path, 'r') as f:
             new_yaml_content = f.read()
         restored_yaml_content = _replace_yaml_dicts(
             new_yaml_content, old_yaml_content,
             _RAY_YAML_KEYS_TO_RESTORE_FOR_BACK_COMPATIBILITY)
-        with open(yaml_path, 'w') as f:
+        with open(tmp_yaml_path, 'w') as f:
             f.write(restored_yaml_content)
 
+    # Rename the tmp file to the final YAML path.
+    os.rename(tmp_yaml_path, yaml_path)
+
     usage_lib.messages.usage.update_ray_yaml(yaml_path)
     # For TPU nodes. TPU VMs do not need TPU_NAME.
     if (resources_vars.get('tpu_type') is not None and
@@ -1129,7 +1134,8 @@ def get_node_ips(cluster_yaml: str,
                  expected_num_nodes: int,
                  handle: Optional[backends.Backend.ResourceHandle] = None,
                  head_ip_max_attempts: int = 1,
-                 worker_ip_max_attempts: int = 1) -> List[str]:
+                 worker_ip_max_attempts: int = 1,
+                 get_internal_ips: bool = False) -> List[str]:
     """Returns the IPs of all nodes in the cluster."""
 
     # When ray up launches TPU VM Pod, Pod workers (except for the head)
@@ -1146,6 +1152,12 @@ def get_node_ips(cluster_yaml: str,
             handle.head_ip is not None):
         return [handle.head_ip]
 
+    if get_internal_ips:
+        with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
+            ray_config['provider']['use_internal_ips'] = True
+            yaml.dump(ray_config, f)
+            cluster_yaml = f.name
+
     # Check the network connection first to avoid long hanging time for
     # ray get-head-ip below, if a long-lasting network connection failure
     # happens.
@@ -1418,7 +1430,7 @@ def _ray_launch_hash(cluster_name: str, ray_config: Dict[str, Any]) -> Set[str]:
         return set(ray_launch_hashes)
     with suppress_output():
         ray_config = ray_commands._bootstrap_config(ray_config)  # pylint: disable=protected-access
-    # Adopted from https://github.com/ray-project/ray/blob/ray-1.13.0/python/ray/autoscaler/_private/node_launcher.py#L56-L64
+    # Adopted from https://github.com/ray-project/ray/blob/ray-2.0.1/python/ray/autoscaler/_private/node_launcher.py#L87-L97
     # TODO(zhwu): this logic is duplicated from the ray code above (keep in sync).
     launch_hashes = set()
     head_node_type = ray_config['head_node_type']
@@ -1746,17 +1758,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.
@@ -1991,9 +2003,11 @@ def kill_children_processes():
 # Handle ctrl-c
 def interrupt_handler(signum, frame):
     del signum, frame
-    logger.warning(f'{colorama.Style.DIM}The job will keep '
-                   f'running after Ctrl-C.{colorama.Style.RESET_ALL}')
     kill_children_processes()
+    # Avoid using logger here, as it will print the stack trace for broken
+    # pipe, when the output is piped to another program.
+    print(f'{colorama.Style.DIM}Tip: The job will keep '
+          f'running after Ctrl-C.{colorama.Style.RESET_ALL}')
     with ux_utils.print_exception_no_traceback():
         raise KeyboardInterrupt(exceptions.KEYBOARD_INTERRUPT_CODE)
 
@@ -2001,9 +2015,11 @@ def interrupt_handler(signum, frame):
 # Handle ctrl-z
 def stop_handler(signum, frame):
     del signum, frame
-    logger.warning(f'{colorama.Style.DIM}The job will keep '
-                   f'running after Ctrl-Z.{colorama.Style.RESET_ALL}')
     kill_children_processes()
+    # Avoid using logger here, as it will print the stack trace for broken
+    # pipe, when the output is piped to another program.
+    print(f'{colorama.Style.DIM}Tip: The job will keep '
+          f'running after Ctrl-Z.{colorama.Style.RESET_ALL}')
     with ux_utils.print_exception_no_traceback():
         raise KeyboardInterrupt(exceptions.SIGTSTP_CODE)
 
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index fc22bdf91..31c217194 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -154,7 +154,7 @@ class RayCodeGen:
         # Should use 'auto' or 'ray://<internal_head_ip>:10001' rather than
         # 'ray://localhost:10001', or 'ray://127.0.0.1:10001', for public cloud.
         # Otherwise, it will a bug of ray job failed to get the placement group
-        # in ray <= 2.0.0.
+        # in ray <= 2.0.1.
         # TODO(mluo): Check why 'auto' not working with on-prem cluster and
         # whether the placement group issue also occurs in on-prem cluster.
         ray_address = 'ray://localhost:10001' if is_local else 'auto'
@@ -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',
@@ -210,8 +210,14 @@ class RayCodeGen:
         self,
         num_nodes: int,
         accelerator_dict: Dict[str, int],
+        cluster_ips_sorted: Optional[List[str]] = None,
     ) -> None:
-        """Create the gang scheduling placement group for a Task."""
+        """Create the gang scheduling placement group for a Task.
+
+        cluster_ips_sorted is used to ensure that the SKY_NODE_RANK environment
+        variable is assigned in a deterministic order whenever a new task is
+        added.
+        """
         assert self._has_prologue, ('Call add_prologue() before '
                                     'add_gang_scheduling_placement_group().')
         self._has_gang_scheduling = True
@@ -269,14 +275,23 @@ class RayCodeGen:
                 @ray.remote
                 def check_ip():
                     return ray.util.get_node_ip_address()
-                ip_list = ray.get([
+                gang_scheduling_id_to_ip = ray.get([
                     check_ip.options(num_cpus={backend_utils.DEFAULT_TASK_CPU_DEMAND},
                                      placement_group=pg,
                                      placement_group_bundle_index=i).remote()
                     for i in range(pg.bundle_count)
                 ])
-                print('INFO: Reserved IPs:', ip_list)
-                ip_list_str = '\\n'.join(ip_list)
+                print('INFO: Reserved IPs:', gang_scheduling_id_to_ip)
+
+                if {cluster_ips_sorted!r} is not None:
+                    cluster_ips_map = {{ip: i for i, ip in enumerate({cluster_ips_sorted!r})}}
+                    ip_rank_list = sorted(gang_scheduling_id_to_ip, key=cluster_ips_map.get)
+                    ip_rank_map = {{ip: i for i, ip in enumerate(ip_rank_list)}}
+                    ip_list_str = '\\n'.join(ip_rank_list)
+                else:
+                    ip_rank_map = {{ip: i for i, ip in enumerate(gang_scheduling_id_to_ip)}}
+                    ip_list_str = '\\n'.join(gang_scheduling_id_to_ip)
+
                 sky_env_vars_dict['SKY_NODE_IPS'] = ip_list_str
                 """),
         ]
@@ -299,16 +314,14 @@ class RayCodeGen:
             f'run_fn = {run_fn_name}',
         ]
 
-    def add_ray_task(
-        self,
-        bash_script: str,
-        task_name: Optional[str],
-        ray_resources_dict: Optional[Dict[str, float]],
-        log_path: str,
-        env_vars: Dict[str, str] = None,
-        gang_scheduling_id: int = 0,
-        use_sudo: bool = False,
-    ) -> None:
+    def add_ray_task(self,
+                     bash_script: str,
+                     task_name: Optional[str],
+                     ray_resources_dict: Optional[Dict[str, float]],
+                     log_path: str,
+                     env_vars: Dict[str, str] = None,
+                     gang_scheduling_id: int = 0,
+                     use_sudo: bool = False) -> None:
         """Generates code for a ray remote task that runs a bash command."""
         assert self._has_gang_scheduling, (
             'Call add_gang_schedule_placement_group() before add_ray_task().')
@@ -358,12 +371,13 @@ class RayCodeGen:
             textwrap.dedent(f"""\
         script = {bash_script!r}
         if run_fn is not None:
-            script = run_fn({gang_scheduling_id}, ip_list)
+            script = run_fn({gang_scheduling_id}, gang_scheduling_id_to_ip)
 
         log_path = os.path.expanduser({log_path!r})
 
         if script is not None:
-            sky_env_vars_dict['SKY_NODE_RANK'] = {gang_scheduling_id!r}
+            ip = gang_scheduling_id_to_ip[{gang_scheduling_id!r}]
+            sky_env_vars_dict['SKY_NODE_RANK'] = ip_rank_map[ip]
             sky_env_vars_dict['SKY_JOB_ID'] = {self.job_id}
 
             futures.append(run_bash_command_with_log \\
@@ -401,6 +415,8 @@ class RayCodeGen:
                 # Need this to set the job status in ray job to be FAILED.
                 sys.exit(1)
             else:
+                sys.stdout.flush()
+                sys.stderr.flush()
                 job_lib.set_status({self.job_id!r}, job_lib.JobStatus.SUCCEEDED)
                 # This waits for all streaming logs to finish.
                 time.sleep(1)
@@ -1341,7 +1357,7 @@ class RetryingVmProvisioner(object):
         if isinstance(launched_resources.cloud, clouds.Local):
             raise RuntimeError(
                 'The command `ray status` errored out on the head node '
-                'of the local cluster. Check if ray[default]==1.13.0 '
+                'of the local cluster. Check if ray[default]==2.0.1 '
                 'is installed or running correctly.')
         backend.run_on_head(handle, 'ray stop', use_cached_head_ip=False)
 
@@ -1810,8 +1826,13 @@ class CloudVmRayBackend(backends.Backend):
                 get_zone_cmd = (
                     handle.launched_resources.cloud.get_zone_shell_cmd())
                 if get_zone_cmd is not None:
-                    returncode, stdout, _ = self.run_on_head(
+                    returncode, stdout, stderr = self.run_on_head(
                         handle, get_zone_cmd, require_outputs=True)
+                    subprocess_utils.handle_returncode(returncode,
+                                                       get_zone_cmd,
+                                                       'Failed to get zone',
+                                                       stderr=stderr,
+                                                       stream_logs=stream_logs)
                     # zone will be checked during Resources cls
                     # initialization.
                     handle.launched_resources = handle.launched_resources.copy(
@@ -2066,8 +2087,8 @@ class CloudVmRayBackend(backends.Backend):
         else:
             job_submit_cmd = (
                 f'{cd} && mkdir -p {remote_log_dir} && ray job submit '
-                f'--address=http://127.0.0.1:8265 --job-id {ray_job_id} '
-                '--no-wait -- '
+                f'--address=http://127.0.0.1:8265 --submission-id {ray_job_id} '
+                '--no-wait '
                 f'"{executable} -u {script_path} > {remote_log_path} 2>&1"')
 
         returncode, stdout, stderr = self.run_on_head(handle,
@@ -2092,17 +2113,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,
@@ -2135,8 +2172,8 @@ class CloudVmRayBackend(backends.Backend):
         switch_user_cmd = ' '.join(switch_user_cmd)
         job_submit_cmd = (
             'ray job submit '
-            f'--address=http://127.0.0.1:8265 --job-id {ray_job_id} --no-wait '
-            f'-- {switch_user_cmd}')
+            f'--address=http://127.0.0.1:8265 --submission-id {ray_job_id} '
+            f'--no-wait -- {switch_user_cmd}')
         return job_submit_cmd
 
     def _add_job(self, handle: ResourceHandle, job_name: str,
@@ -2190,31 +2227,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
@@ -2801,7 +2839,7 @@ class CloudVmRayBackend(backends.Backend):
             # alternatives (smart_open, each provider's own sdk), a
             # data-transfer container etc.
             if not os.path.isabs(dst) and not dst.startswith('~/'):
-                dst = f'~/{dst}'
+                dst = f'{SKY_REMOTE_WORKDIR}/{dst}'
             # Sync 'src' to 'wrapped_dst', a safe-to-write "wrapped" path.
             wrapped_dst = dst
             if not dst.startswith('~/') and not dst.startswith('/tmp/'):
@@ -2921,7 +2959,7 @@ class CloudVmRayBackend(backends.Backend):
 
         for dst, storage_obj in storage_mounts.items():
             if not os.path.isabs(dst) and not dst.startswith('~/'):
-                dst = f'~/{dst}'
+                dst = f'{SKY_REMOTE_WORKDIR}/{dst}'
             # Get the first store and use it to mount
             store = list(storage_obj.stores.values())[0]
             mount_cmd = store.mount_command(dst)
@@ -2996,13 +3034,24 @@ class CloudVmRayBackend(backends.Backend):
         else:
             num_actual_nodes = task.num_nodes
 
+        cluster_internal_ips = backend_utils.get_node_ips(handle.cluster_yaml,
+                                                          handle.launched_nodes,
+                                                          handle=handle,
+                                                          get_internal_ips=True)
+        # Ensure head node is the first element, then sort the remaining IPs
+        # for stableness
+        cluster_ips_sorted = [cluster_internal_ips[0]] + sorted(
+            cluster_internal_ips[1:])
+
         codegen = RayCodeGen()
         is_local = isinstance(handle.launched_resources.cloud, clouds.Local)
         codegen.add_prologue(job_id,
                              spot_task=task.spot_task,
                              is_local=is_local)
-        codegen.add_gang_scheduling_placement_group(num_actual_nodes,
-                                                    accelerator_dict)
+        codegen.add_gang_scheduling_placement_group(
+            num_actual_nodes,
+            accelerator_dict,
+            cluster_ips_sorted=cluster_ips_sorted)
 
         if callable(task.run):
             run_fn_code = textwrap.dedent(inspect.getsource(task.run))
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 365adecd0..1ccb421c0 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)
+    else:
+        task = sky.Task(name='sky-cmd', run=entrypoint)
+        task.set_resources({sky.Resources()})
 
-        if is_yaml:
-            usage_lib.messages.usage.update_user_task_yaml(entrypoint)
-            task = sky.Task.from_yaml(entrypoint)
+    # 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']),
@@ -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
@@ -1331,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
@@ -1404,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.
@@ -1471,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)
@@ -1500,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.
 
@@ -1579,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
@@ -1664,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',
@@ -1685,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:
 
@@ -1842,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:
 
@@ -1890,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
@@ -2343,10 +2361,15 @@ def check():
 def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pylint: disable=redefined-builtin
     """Show supported GPU/TPU/accelerators.
 
+    The names and counts shown can be set in the ``accelerators`` field in task
+    YAMLs, or in the ``--gpus`` flag in CLI commands. For example, if this
+    table shows 8x V100s are supported, then the string ``V100:8`` will be
+    accepted by the above.
+
     To show the detailed information of a GPU/TPU type (which clouds offer it,
     the quantity in each VM type, etc.), use ``sky show-gpus <gpu>``.
 
-    To show all GPUs, including less common ones and their detailed
+    To show all accelerators, including less common ones and their detailed
     information, use ``sky show-gpus --all``.
 
     NOTE: The price displayed for each instance type is the lowest across all
@@ -2673,13 +2696,26 @@ def spot_launch(
     retry_until_up: bool,
     yes: bool,
 ):
-    """Launch a managed spot job."""
+    """Launch a managed spot job from a YAML or a command.
+
+    If ENTRYPOINT points to a valid YAML file, it is read in as the task
+    specification. Otherwise, it is interpreted as a bash command.
+
+    Examples:
+
+    .. code-block:: bash
+
+      # You can use normal task YAMLs.
+      sky spot launch task.yaml
+
+      sky spot launch 'echo hello!'
+    """
     if name is None:
         name = backend_utils.generate_cluster_name()
     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,
@@ -2701,13 +2737,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,
@@ -2724,7 +2760,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
@@ -2745,17 +2781,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:
@@ -2775,6 +2811,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 414c3e7db..1db8ee8fa 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,27 +31,33 @@ logger = sky_logging.init_logger(__name__)
 
 
 @usage_lib.entrypoint
-def status(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(include_reserved=True,
                                                  refresh=refresh)
@@ -81,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)
 
@@ -105,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.
+
+    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.
 
-    Please refer to the sky.cli.stop for the document.
+    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(
@@ -159,14 +208,26 @@ 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.
+
+    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.
 
-    Please refer to the sky.cli.down for the document.
+    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.
     """
     handle = global_user_state.get_handle_from_cluster_name(cluster_name)
     if handle is None:
@@ -180,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:
@@ -233,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)
 
 
 # ==================
@@ -278,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.
 
@@ -327,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.
 
@@ -365,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.
@@ -395,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:
@@ -423,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:
@@ -477,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:
         [
@@ -502,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:
@@ -547,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.
 
@@ -598,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.
@@ -630,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:
@@ -650,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:
@@ -660,7 +753,6 @@ def storage_delete(name: str):
     if handle is None:
         raise ValueError(f'Storage name {name!r} not found.')
     else:
-        print(f'Deleting storage object {name!r}.')
         store_object = data.Storage(name=handle.storage_name,
                                     source=handle.source,
                                     sync_on_reconstruction=False)
diff --git a/sky/data/mounting_utils.py b/sky/data/mounting_utils.py
index ccaf7374b..338dfdc91 100644
--- a/sky/data/mounting_utils.py
+++ b/sky/data/mounting_utils.py
@@ -34,7 +34,7 @@ def get_mounting_command(
         # Check if path is already mounted
         if grep -q $MOUNT_PATH /proc/mounts ; then
             echo "Path already mounted - unmounting..."
-            fusermount -u "$MOUNT_PATH"
+            fusermount -uz "$MOUNT_PATH"
             echo "Successfully unmounted $MOUNT_PATH."
         fi
 
diff --git a/sky/data/storage.py b/sky/data/storage.py
index bffa5cb46..66c56c20f 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -6,6 +6,8 @@ import time
 from typing import Any, Dict, Optional, Tuple, Union
 import urllib.parse
 
+import colorama
+
 from sky import clouds
 from sky.adaptors import aws
 from sky.adaptors import gcp
@@ -108,6 +110,7 @@ class AbstractStore:
     _STAT_CACHE_TTL = '5s'
     _STAT_CACHE_CAPACITY = 4096
     _TYPE_CACHE_TTL = '5s'
+    _RENAME_DIR_LIMIT = 10000
 
     class StoreMetadata:
         """A pickle-able representation of Store
@@ -398,8 +401,8 @@ class Storage(object):
         self.handle = global_user_state.get_handle_from_storage_name(self.name)
         if self.handle:
             # Reconstruct the Storage object from the global_user_state
-            logger.info('Detected existing storage object, '
-                        f'loading Storage: {self.name}')
+            logger.debug('Detected existing storage object, '
+                         f'loading Storage: {self.name}')
             for s_type, s_metadata in self.handle.sky_stores.items():
                 # When initializing from global_user_state, we override the
                 # source from the YAML
@@ -834,7 +837,8 @@ class S3Store(AbstractStore):
 
     def delete(self) -> None:
         self._delete_s3_bucket(self.name)
-        logger.info(f'Deleted S3 bucket {self.name}.')
+        logger.info(f'{colorama.Fore.GREEN}Deleted S3 bucket {self.name}.'
+                    f'{colorama.Style.RESET_ALL}')
 
     def get_handle(self) -> StorageHandle:
         return aws.resource('s3').Bucket(self.name)
@@ -1002,7 +1006,7 @@ class S3Store(AbstractStore):
         remove_command = f'aws s3 rb s3://{bucket_name} --force'
         try:
             with backend_utils.safe_console_status(
-                    f'[bold cyan]Deleting [green]bucket {bucket_name}'):
+                    f'[bold cyan]Deleting S3 bucket {bucket_name}[/]'):
                 subprocess.check_output(remove_command.split(' '))
         except subprocess.CalledProcessError as e:
             logger.error(e.output)
@@ -1088,7 +1092,8 @@ class GcsStore(AbstractStore):
 
     def delete(self) -> None:
         self._delete_gcs_bucket(self.name)
-        logger.info(f'Deleted GCS bucket {self.name}.')
+        logger.info(f'{colorama.Fore.GREEN}Deleted GCS bucket {self.name}.'
+                    f'{colorama.Style.RESET_ALL}')
 
     def get_handle(self) -> StorageHandle:
         return self.client.get_bucket(self.name)
@@ -1187,9 +1192,11 @@ class GcsStore(AbstractStore):
                        '-O /tmp/gcsfuse.deb && '
                        'sudo dpkg --install /tmp/gcsfuse.deb')
         mount_cmd = ('gcsfuse -o allow_other '
+                     '--implicit-dirs '
                      f'--stat-cache-capacity {self._STAT_CACHE_CAPACITY} '
                      f'--stat-cache-ttl {self._STAT_CACHE_TTL} '
                      f'--type-cache-ttl {self._TYPE_CACHE_TTL} '
+                     f'--rename-dir-limit {self._RENAME_DIR_LIMIT} '
                      f'{self.bucket.name} {mount_path}')
         return mounting_utils.get_mounting_command(mount_path, install_cmd,
                                                    mount_cmd)
@@ -1234,7 +1241,7 @@ class GcsStore(AbstractStore):
           bucket_name: str; Name of bucket
         """
         try:
-            bucket = self.client.get_bucket(bucket_name)
+            self.client.get_bucket(bucket_name)
         except gcp.forbidden_exception() as e:
             # Try public bucket to see if bucket exists
             with ux_utils.print_exception_no_traceback():
@@ -1242,17 +1249,13 @@ class GcsStore(AbstractStore):
                     'External Bucket detected. User not allowed to delete '
                     'external bucket.') from e
 
-        num_files = subprocess.check_output(
-            f'gsutil du gs://{bucket_name} | wc -l', shell=True)
-        num_files = int(num_files)
-
         try:
             with backend_utils.safe_console_status(
-                    f'[bold cyan]Deleting [green]bucket {bucket_name}'):
-                if num_files >= _GCS_RM_MAX_OBJS:
-                    remove_obj_command = f'gsutil -m rm -a gs://{bucket_name}/*'
-                    subprocess.check_output(remove_obj_command.split(' '))
-                bucket.delete(force=True)
+                    f'[bold cyan]Deleting GCS bucket {bucket_name}[/]'):
+                remove_obj_command = ('gsutil -m rm -r'
+                                      f' gs://{bucket_name}')
+                subprocess.check_output(remove_obj_command.split(' '),
+                                        stderr=subprocess.STDOUT)
         except subprocess.CalledProcessError as e:
             logger.error(e.output)
             with ux_utils.print_exception_no_traceback():
diff --git a/sky/design_docs/onprem-design.md b/sky/design_docs/onprem-design.md
index aca15bb35..fa9401ad0 100644
--- a/sky/design_docs/onprem-design.md
+++ b/sky/design_docs/onprem-design.md
@@ -9,7 +9,7 @@
 - Does not support different types of accelerators within the same node (intranode).
 
 ## Installing Ray and SkyPilot 
-- Admin installs Ray==1.13.0 and SkyPilot globally on all machines. It is assumed that the admin regularly keeps SkyPilot updated on the cluster.
+- Admin installs Ray==2.0.1 and SkyPilot globally on all machines. It is assumed that the admin regularly keeps SkyPilot updated on the cluster.
 - Python >= 3.6 for all users.
 - When a regular user runs `sky launch`, a local version of SkyPilot will be installed on the machine for each user. The local installation of Ray is specified in `sky/templates/local-ray.yml.j2`.
 
@@ -36,7 +36,7 @@ ray.get(ray.remote(f).remote())
 ```
   
 - Therefore, SkyPilot On-prem transparently includes user-switching so that SkyPilot tasks are still run as the calling, unprivileged user. This user-switching (`sudo -H su --login [USER]` in appropriate places) works as follows:
-    - In `sky/backends/cloud_vm_ray_backend.py::_setup_and_create_job_cmd_on_local_head`, switching between users is called during Ray job submission. The command `ray job submit --address=http://127.0.0.1:8265 --job-id {ray_job_id} -- sudo -H su --login [SSH_USER] -c \"[JOB_COMMAND]\"` switches job submission execution from admin back to the original user `SSH_USER`. The `JOB_COMMAND` argument runs a bash script with the user's run commands.
+    - In `sky/backends/cloud_vm_ray_backend.py::_setup_and_create_job_cmd_on_local_head`, switching between users is called during Ray job submission. The command `ray job submit --address=http://127.0.0.1:8265 --submission-id {ray_job_id} -- sudo -H su --login [SSH_USER] -c \"[JOB_COMMAND]\"` switches job submission execution from admin back to the original user `SSH_USER`. The `JOB_COMMAND` argument runs a bash script with the user's run commands.
     - In `sky/skylet/log_lib.py::run_bash_command_with_log`, there is also another `sudo -H su` command to switch users. The function `run_bash_command_with_log` is part of the `RayCodeGen` job execution script uploaded to remote for job submission (located in `~/.sky/sky_app/sky_app_[JOB_ID].py`). This program initially runs under the calling user, but it executes the function `run_bash_command_with_log` from the context of the admin, as the function is executed within the Ray cluster as a Ray remote function (see above for why all Ray remote functions are run under admin).
 - SkyPilot ensures Ray-related environment variables (that are critical for execution) are preserved across switching users (check with `examples/env_check.yaml`).
 
diff --git a/sky/execution.py b/sky/execution.py
index 95db48a98..5cf12e6d9 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 (EXPERIMENTAL advanced
+    usage) a sky.Dag. In the latter case, currently it must contain a single
+    task; support for pipelines/general DAGs are in experimental branches.
 
     Args:
-        dag: sky.DAG to launch.
+        task: sky.Task, or sky.Dag (experimental; 1-task only) 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,23 @@ 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
-
-    Examples:
-        >>> 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')
+        no_setup: if True, do not re-run setup commands.
+
+    Example:
+        .. code-block:: python
+
+            import sky
+            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 +363,63 @@ 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 (experimental; 1-task only) 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 +431,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 +447,20 @@ 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 (experimental; 1-task only) 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/setup_files/setup.py b/sky/setup_files/setup.py
index 1d0a054af..7f3ea2f3d 100644
--- a/sky/setup_files/setup.py
+++ b/sky/setup_files/setup.py
@@ -56,10 +56,12 @@ def parse_footnote(readme: str) -> str:
 
 install_requires = [
     'wheel',
-    # NOTE: ray 1.13.0 requires click<=8.0.4,>=7.0; We disable the
+    # NOTE: ray 2.0.1 requires click<=8.0.4,>=7.0; We disable the
     # shell completion for click<8.0 for backward compatibility.
     'click<=8.0.4,>=7.0',
-    'colorama',
+    # NOTE: required by awscli. To avoid ray automatically installing
+    # the latest version.
+    'colorama<0.4.5',
     'cryptography',
     'jinja2',
     'jsonschema',
@@ -70,14 +72,14 @@ install_requires = [
     'PrettyTable',
     # Lower local ray version is not fully supported, due to the
     # autoscaler issues (also tracked in #537).
-    'ray[default]>=1.9.0,<=1.13.0',
+    'ray[default]>=1.9.0,<=2.0.1',
     'rich',
     'tabulate',
     'filelock',  # TODO(mraheja): Enforce >=3.6.0 when python version is >= 3.7
     # This is used by ray. The latest 1.44.0 will generate an error
     # `Fork support is only compatible with the epoll1 and poll
     # polling strategies`
-    'grpcio<=1.43.0',
+    'grpcio>=1.32.0,<=1.43.0',
     'packaging',
     # The latest 4.21.1 will break ray. Enforce < 4.0.0 until Ray releases the
     # fix.
@@ -98,6 +100,8 @@ extras_require = {
     ],
     # TODO(zongheng): azure-cli is huge and takes a long time to install.
     # Tracked in: https://github.com/Azure/azure-cli/issues/7387
+    # azure-cli need to be pinned to 2.31.0 due to later versions
+    # do not have azure-identity (used in node_provider) installed
     'azure': ['azure-cli==2.31.0', 'azure-core'],
     'gcp': ['google-api-python-client', 'google-cloud-storage'],
     'docker': ['docker'],
diff --git a/sky/skylet/LICENCE b/sky/skylet/LICENCE
index 0dc90ab2c..c9867ce19 100644
--- a/sky/skylet/LICENCE
+++ b/sky/skylet/LICENCE
@@ -203,16 +203,16 @@
 --------------------------------------------------------------------------------
 
 Code in providers/azure from
-https://github.com/ray-project/ray/tree/ray-1.13.0/python/ray/autoscaler/_private/_azure
-Git commit of the release 1.13.0: 4ce38d001dbbe09cd21c497fedd03d692b2be3e
+https://github.com/ray-project/ray/tree/ray-2.0.1/python/ray/autoscaler/_private/_azure
+Git commit of the release 2.0.1: 03b6bc7b5a305877501110ec04710a9c57011479
 
 Code in providers/gcp from
-https://github.com/ray-project/ray/tree/ray-1.13.0/python/ray/autoscaler/_private/gcp
-Git commit of the release 1.13.0: 4ce38d001dbbe09cd21c497fedd03d692b2be3e
+https://github.com/ray-project/ray/tree/ray-2.0.1/python/ray/autoscaler/_private/gcp
+Git commit of the release 2.0.1: 03b6bc7b5a305877501110ec04710a9c57011479
 
 Code in providers/aws from
-https://github.com/ray-project/ray/tree/ray-1.13.0/python/ray/autoscaler/_private/aws
-Git commit of the release 1.13.0: 4ce38d001dbbe09cd21c497fedd03d692b2be3e
+https://github.com/ray-project/ray/tree/ray-2.0.1/python/ray/autoscaler/_private/aws
+Git commit of the release 2.0.1: 03b6bc7b5a305877501110ec04710a9c57011479
 
 
 Copyright 2016-2022 Ray developers
diff --git a/sky/skylet/constants.py b/sky/skylet/constants.py
index c0d98ecef..e256dcf98 100644
--- a/sky/skylet/constants.py
+++ b/sky/skylet/constants.py
@@ -2,7 +2,7 @@
 
 SKY_LOGS_DIRECTORY = '~/sky_logs'
 SKY_REMOTE_WORKDIR = '~/sky_workdir'
-SKY_REMOTE_RAY_VERSION = '1.13.0'
+SKY_REMOTE_RAY_VERSION = '2.0.1'
 
 # TODO(mluo): Make explicit `sky launch -c <name> ''` optional.
 UNINITIALIZED_ONPREM_CLUSTER_MESSAGE = (
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index 54991a4ca..245146806 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -7,6 +7,7 @@ import os
 import pathlib
 import shlex
 import time
+import typing
 from typing import Any, Dict, List, Optional
 
 import filelock
@@ -17,6 +18,9 @@ from sky.utils import common_utils
 from sky.utils import db_utils
 from sky.utils import log_utils
 
+if typing.TYPE_CHECKING:
+    from ray.dashboard.modules.job import pydantic_models as ray_pydantic
+
 logger = sky_logging.init_logger(__name__)
 
 _JOB_STATUS_LOCK = '~/.sky/locks/.job_{}.lock'
@@ -331,7 +335,7 @@ def update_job_status(job_owner: str,
     we still need this to handle staleness problem, caused by instance
     restarting and other corner cases (if any).
 
-    This function should only be run on the remote instance with ray==1.13.0.
+    This function should only be run on the remote instance with ray==2.0.1.
     """
     if len(job_ids) == 0:
         return []
@@ -341,13 +345,19 @@ def update_job_status(job_owner: str,
 
     job_client = _create_ray_job_submission_client()
 
-    # In ray 1.13.0, job_client.list_jobs returns a dict of job_id to job_info,
-    # where job_info contains the job status (str).
-    ray_job_infos = job_client.list_jobs()
+    # In ray 2.0.1, job_client.list_jobs returns a list of JobDetails,
+    # which contains the job status (str) and submission_id (str).
+    job_detail_lists: List['ray_pydantic.JobDetails'] = job_client.list_jobs()
+
+    job_details = dict()
+    ray_job_ids_set = set(ray_job_ids)
+    for job_detail in job_detail_lists:
+        if job_detail.submission_id in ray_job_ids_set:
+            job_details[job_detail.submission_id] = job_detail
     job_statuses: List[JobStatus] = [None] * len(ray_job_ids)
     for i, ray_job_id in enumerate(ray_job_ids):
-        if ray_job_id in ray_job_infos:
-            ray_status = ray_job_infos[ray_job_id].status
+        if ray_job_id in job_details:
+            ray_status = job_details[ray_job_id].status
             job_statuses[i] = _RAY_TO_JOB_STATUS_MAP[ray_status]
 
     assert len(job_statuses) == len(job_ids), (job_statuses, job_ids)
diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py
index 0d2bf96b7..a8ede3f57 100644
--- a/sky/skylet/log_lib.py
+++ b/sky/skylet/log_lib.py
@@ -244,6 +244,7 @@ def make_task_bash_script(codegen: str,
     # set -a is used for exporting all variables functions to the environment
     # so that bash `user_script` can access `conda activate`. Detail: #436.
     # Reference: https://www.gnu.org/software/bash/manual/html_node/The-Set-Builtin.html # pylint: disable=line-too-long
+    # PYTHONUNBUFFERED is used to disable python output buffering.
     script = [
         textwrap.dedent(f"""\
             #!/bin/bash
@@ -251,6 +252,7 @@ def make_task_bash_script(codegen: str,
             set -a
             . $(conda info --base 2> /dev/null)/etc/profile.d/conda.sh > /dev/null 2>&1 || true
             set +a
+            export PYTHONUNBUFFERED=1
             cd {constants.SKY_REMOTE_WORKDIR}"""),
     ]
     if env_vars is not None:
@@ -415,6 +417,7 @@ def tail_logs(job_owner: str,
         time.sleep(_SKY_LOG_WAITING_GAP_SECONDS)
         status = job_lib.update_job_status(job_owner, [job_id], silent=True)[0]
 
+    start_stream_at = 'INFO: Tip: use Ctrl-C to exit log'
     if follow and status in [
             job_lib.JobStatus.RUNNING, job_lib.JobStatus.PENDING
     ]:
@@ -423,15 +426,19 @@ def tail_logs(job_owner: str,
         with open(log_path, 'r', newline='') as log_file:
             # Using `_follow` instead of `tail -f` to streaming the whole
             # log and creating a new process for tail.
-            for line in _follow_job_logs(
-                    log_file,
-                    job_id=job_id,
-                    start_streaming_at='INFO: Tip: use Ctrl-C to exit log'):
+            for line in _follow_job_logs(log_file,
+                                         job_id=job_id,
+                                         start_streaming_at=start_stream_at):
                 print(line, end='', flush=True)
     else:
         try:
+            start_stream = False
             with open(log_path, 'r') as f:
-                print(f.read())
+                for line in f.readlines():
+                    if start_stream_at in line:
+                        start_stream = True
+                    if start_stream:
+                        print(line, end='', flush=True)
         except FileNotFoundError:
             print(f'{colorama.Fore.RED}ERROR: Logs for job {job_id} (status:'
                   f' {status.value}) does not exist.{colorama.Style.RESET_ALL}')
diff --git a/sky/skylet/providers/aws/cloudwatch/cloudwatch_helper.py b/sky/skylet/providers/aws/cloudwatch/cloudwatch_helper.py
index b87805d6d..526cd19d1 100644
--- a/sky/skylet/providers/aws/cloudwatch/cloudwatch_helper.py
+++ b/sky/skylet/providers/aws/cloudwatch/cloudwatch_helper.py
@@ -1,13 +1,15 @@
-import botocore
 import copy
+import hashlib
 import json
-import os
 import logging
+import os
 import time
-import hashlib
-from typing import Any, Dict, List, Union, Tuple
+from typing import Any, Dict, List, Tuple, Union
+
+import botocore
+
 from sky.skylet.providers.aws.utils import client_cache, resource_cache
-from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME, NODE_KIND_HEAD, TAG_RAY_NODE_KIND
+from ray.autoscaler.tags import NODE_KIND_HEAD, TAG_RAY_CLUSTER_NAME, TAG_RAY_NODE_KIND
 
 logger = logging.getLogger(__name__)
 
diff --git a/sky/skylet/providers/aws/config.py b/sky/skylet/providers/aws/config.py
index abdd48756..6daf8c951 100644
--- a/sky/skylet/providers/aws/config.py
+++ b/sky/skylet/providers/aws/config.py
@@ -1,30 +1,29 @@
-from distutils.version import StrictVersion
-from functools import lru_cache
-from functools import partial
 import copy
 import itertools
 import json
+import logging
 import os
 import time
+from distutils.version import StrictVersion
+from functools import lru_cache, partial
 from typing import Any, Dict, List, Optional, Set, Tuple
-import logging
 
 import boto3
 import botocore
 
-from ray.autoscaler._private.util import check_legacy_fields
-from ray.autoscaler.tags import NODE_TYPE_LEGACY_HEAD, NODE_TYPE_LEGACY_WORKER
-from ray.autoscaler._private.providers import _PROVIDER_PRETTY_NAMES
+from sky.skylet.providers.aws.cloudwatch.cloudwatch_helper import (
+    CloudwatchHelper as cwh,
+)
 from sky.skylet.providers.aws.utils import (
     LazyDefaultDict,
     handle_boto_error,
     resource_cache,
 )
-from ray.autoscaler._private.cli_logger import cli_logger, cf
+from ray.autoscaler._private.cli_logger import cf, cli_logger
 from ray.autoscaler._private.event_system import CreateClusterEvent, global_event_system
-from sky.skylet.providers.aws.cloudwatch.cloudwatch_helper import (
-    CloudwatchHelper as cwh,
-)
+from ray.autoscaler._private.providers import _PROVIDER_PRETTY_NAMES
+from ray.autoscaler._private.util import check_legacy_fields
+from ray.autoscaler.tags import NODE_TYPE_LEGACY_HEAD, NODE_TYPE_LEGACY_WORKER
 
 logger = logging.getLogger(__name__)
 
@@ -33,20 +32,22 @@ DEFAULT_RAY_INSTANCE_PROFILE = RAY + "-v1"
 DEFAULT_RAY_IAM_ROLE = RAY + "-v1"
 SECURITY_GROUP_TEMPLATE = RAY + "-{}"
 
-DEFAULT_AMI_NAME = "AWS Deep Learning AMI (Ubuntu 18.04) V30.0"
+# V61.0 has CUDA 11.2
+DEFAULT_AMI_NAME = "AWS Deep Learning AMI (Ubuntu 18.04) V61.0"
 
-# Obtained from https://aws.amazon.com/marketplace/pp/B07Y43P7X5 on 8/4/2020.
+# Obtained from https://aws.amazon.com/marketplace/pp/B07Y43P7X5 on 6/10/2022.
+# NOTE(skypilot): these are not used; skypilot instead uses the default AMIs in aws.py.
 DEFAULT_AMI = {
-    "us-east-1": "ami-029510cec6d69f121",  # US East (N. Virginia)
-    "us-east-2": "ami-08bf49c7b3a0c761e",  # US East (Ohio)
-    "us-west-1": "ami-0cc472544ce594a19",  # US West (N. California)
-    "us-west-2": "ami-0a2363a9cff180a64",  # US West (Oregon)
-    "ca-central-1": "ami-0a871851b2ab39f01",  # Canada (Central)
-    "eu-central-1": "ami-049fb1ea198d189d7",  # EU (Frankfurt)
-    "eu-west-1": "ami-0abcbc65f89fb220e",  # EU (Ireland)
-    "eu-west-2": "ami-0755b39fd4dab7cbe",  # EU (London)
-    "eu-west-3": "ami-020485d8df1d45530",  # EU (Paris)
-    "sa-east-1": "ami-058a6883cbdb4e599",  # SA (Sao Paulo)
+    "us-east-1": "ami-0dd6adfad4ad37eec",  # US East (N. Virginia)
+    "us-east-2": "ami-0c77cd5ca05bf1281",  # US East (Ohio)
+    "us-west-1": "ami-020ab1b368a5ed1db",  # US West (N. California)
+    "us-west-2": "ami-0387d929287ab193e",  # US West (Oregon)
+    "ca-central-1": "ami-07dbafdbd38f18d98",  # Canada (Central)
+    "eu-central-1": "ami-0383bd0c1fc4c63ec",  # EU (Frankfurt)
+    "eu-west-1": "ami-0a074b0a311a837ac",  # EU (Ireland)
+    "eu-west-2": "ami-094ba2b4651f761ca",  # EU (London)
+    "eu-west-3": "ami-031da10fbf225bf5f",  # EU (Paris)
+    "sa-east-1": "ami-0be7c1f1dd96d7337",  # SA (Sao Paulo)
 }
 
 # todo: cli_logger should handle this assert properly
diff --git a/sky/skylet/providers/aws/node_provider.py b/sky/skylet/providers/aws/node_provider.py
index 96335a7be..7203ab70f 100644
--- a/sky/skylet/providers/aws/node_provider.py
+++ b/sky/skylet/providers/aws/node_provider.py
@@ -1,36 +1,39 @@
 import copy
-import threading
-from collections import defaultdict, OrderedDict
 import logging
+import threading
 import time
+from collections import defaultdict, OrderedDict
 from typing import Any, Dict, List
 
 import botocore
+from boto3.resources.base import ServiceResource
 
-from ray.autoscaler.node_provider import NodeProvider
-from ray.autoscaler.tags import (
-    TAG_RAY_CLUSTER_NAME,
-    TAG_RAY_NODE_NAME,
-    TAG_RAY_LAUNCH_CONFIG,
-    TAG_RAY_NODE_KIND,
-    TAG_RAY_USER_NODE_TYPE,
+try:
+    import ray._private.ray_constants as ray_constants
+except ImportError:
+    # SkyPilot: for local ray version lower than 2.0.1
+    import ray.ray_constants as ray_constants
+from sky.skylet.providers.aws.cloudwatch.cloudwatch_helper import (
+    CloudwatchHelper,
+    CLOUDWATCH_AGENT_INSTALLED_AMI_TAG,
+    CLOUDWATCH_AGENT_INSTALLED_TAG,
 )
-from ray.autoscaler._private.constants import BOTO_MAX_RETRIES, BOTO_CREATE_MAX_RETRIES
 from sky.skylet.providers.aws.config import bootstrap_aws
-from ray.autoscaler._private.log_timer import LogTimer
-
 from sky.skylet.providers.aws.utils import (
     boto_exception_handler,
     resource_cache,
     client_cache,
 )
 from ray.autoscaler._private.cli_logger import cli_logger, cf
-import ray.ray_constants as ray_constants
-
-from sky.skylet.providers.aws.cloudwatch.cloudwatch_helper import (
-    CloudwatchHelper,
-    CLOUDWATCH_AGENT_INSTALLED_AMI_TAG,
-    CLOUDWATCH_AGENT_INSTALLED_TAG,
+from ray.autoscaler._private.constants import BOTO_MAX_RETRIES, BOTO_CREATE_MAX_RETRIES
+from ray.autoscaler._private.log_timer import LogTimer
+from ray.autoscaler.node_provider import NodeProvider
+from ray.autoscaler.tags import (
+    TAG_RAY_CLUSTER_NAME,
+    TAG_RAY_LAUNCH_CONFIG,
+    TAG_RAY_NODE_KIND,
+    TAG_RAY_NODE_NAME,
+    TAG_RAY_USER_NODE_TYPE,
 )
 
 logger = logging.getLogger(__name__)
@@ -56,7 +59,7 @@ def from_aws_format(tags):
     return tags
 
 
-def make_ec2_client(region, max_retries, aws_credentials=None):
+def make_ec2_resource(region, max_retries, aws_credentials=None):
     """Make client, retrying requests up to `max_retries`."""
     aws_credentials = aws_credentials or {}
     return resource_cache("ec2", region, max_retries, **aws_credentials)
@@ -67,7 +70,7 @@ def list_ec2_instances(
 ) -> List[Dict[str, Any]]:
     """Get all instance-types/resources available in the user's AWS region.
     Args:
-        region (str): the region of the AWS provider. e.g., "us-west-2".
+        region: the region of the AWS provider. e.g., "us-west-2".
     Returns:
         final_instance_types: a list of instances. An example of one element in
         the list:
@@ -101,12 +104,12 @@ class AWSNodeProvider(NodeProvider):
         self.cache_stopped_nodes = provider_config.get("cache_stopped_nodes", True)
         aws_credentials = provider_config.get("aws_credentials")
 
-        self.ec2 = make_ec2_client(
+        self.ec2 = make_ec2_resource(
             region=provider_config["region"],
             max_retries=BOTO_MAX_RETRIES,
             aws_credentials=aws_credentials,
         )
-        self.ec2_fail_fast = make_ec2_client(
+        self.ec2_fail_fast = make_ec2_resource(
             region=provider_config["region"],
             max_retries=0,
             aws_credentials=aws_credentials,
@@ -494,7 +497,6 @@ class AWSNodeProvider(NodeProvider):
         # asyncrhonous or error, which would result in a use after free error.
         # If this leak becomes bad, we can garbage collect the tag cache when
         # the node cache is updated.
-        pass
 
     def _check_ami_cwa_installation(self, config):
         response = self.ec2.meta.client.describe_images(ImageIds=[config["ImageId"]])
diff --git a/sky/skylet/providers/aws/utils.py b/sky/skylet/providers/aws/utils.py
index 154931548..72dda831a 100644
--- a/sky/skylet/providers/aws/utils.py
+++ b/sky/skylet/providers/aws/utils.py
@@ -1,11 +1,13 @@
 from collections import defaultdict
 from functools import lru_cache
 
+import boto3
 from boto3.exceptions import ResourceNotExistsError
+from boto3.resources.base import ServiceResource
+from botocore.client import BaseClient
 from botocore.config import Config
-import boto3
 
-from ray.autoscaler._private.cli_logger import cli_logger, cf
+from ray.autoscaler._private.cli_logger import cf, cli_logger
 from ray.autoscaler._private.constants import BOTO_MAX_RETRIES
 
 
@@ -141,7 +143,9 @@ def boto_exception_handler(msg, *args, **kwargs):
 
 
 @lru_cache()
-def resource_cache(name, region, max_retries=BOTO_MAX_RETRIES, **kwargs):
+def resource_cache(
+    name, region, max_retries=BOTO_MAX_RETRIES, **kwargs
+) -> ServiceResource:
     cli_logger.verbose(
         "Creating AWS resource `{}` in `{}`", cf.bold(name), cf.bold(region)
     )
@@ -157,7 +161,7 @@ def resource_cache(name, region, max_retries=BOTO_MAX_RETRIES, **kwargs):
 
 
 @lru_cache()
-def client_cache(name, region, max_retries=BOTO_MAX_RETRIES, **kwargs):
+def client_cache(name, region, max_retries=BOTO_MAX_RETRIES, **kwargs) -> BaseClient:
     try:
         # try to re-use a client from the resource cache first
         return resource_cache(name, region, max_retries, **kwargs).meta.client
diff --git a/sky/skylet/providers/azure/config.py b/sky/skylet/providers/azure/config.py
index a8a9ca088..530d15e4b 100644
--- a/sky/skylet/providers/azure/config.py
+++ b/sky/skylet/providers/azure/config.py
@@ -1,7 +1,7 @@
 import json
 import logging
-from pathlib import Path
 import random
+from pathlib import Path
 from typing import Any, Callable
 
 from azure.common.credentials import get_cli_profile
diff --git a/sky/skylet/providers/azure/node_provider.py b/sky/skylet/providers/azure/node_provider.py
index 6bd7a9f23..630b59934 100644
--- a/sky/skylet/providers/azure/node_provider.py
+++ b/sky/skylet/providers/azure/node_provider.py
@@ -10,18 +10,18 @@ from azure.mgmt.network import NetworkManagementClient
 from azure.mgmt.resource import ResourceManagementClient
 from azure.mgmt.resource.resources.models import DeploymentMode
 
+from sky.skylet.providers.azure.config import (
+    bootstrap_azure,
+    get_azure_sdk_function,
+)
 from ray.autoscaler.node_provider import NodeProvider
 from ray.autoscaler.tags import (
     TAG_RAY_CLUSTER_NAME,
-    TAG_RAY_NODE_NAME,
-    TAG_RAY_NODE_KIND,
     TAG_RAY_LAUNCH_CONFIG,
+    TAG_RAY_NODE_KIND,
+    TAG_RAY_NODE_NAME,
     TAG_RAY_USER_NODE_TYPE,
 )
-from sky.skylet.providers.azure.config import (
-    bootstrap_azure,
-    get_azure_sdk_function,
-)
 
 VM_NAME_MAX_LEN = 64
 VM_NAME_UUID_LEN = 8
diff --git a/sky/skylet/providers/gcp/config.py b/sky/skylet/providers/gcp/config.py
index 539f9d677..42ab7caf5 100644
--- a/sky/skylet/providers/gcp/config.py
+++ b/sky/skylet/providers/gcp/config.py
@@ -1,19 +1,19 @@
 import copy
-from functools import partial
 import json
-import os
 import logging
+import os
 import time
+from functools import partial
 
+from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
-from cryptography.hazmat.backends import default_backend
-from googleapiclient import discovery, errors
 from google.oauth2 import service_account
 from google.oauth2.credentials import Credentials as OAuthCredentials
+from googleapiclient import discovery, errors
 
+from sky.skylet.providers.gcp.node import MAX_POLLS, POLL_INTERVAL, GCPNodeType
 from ray.autoscaler._private.util import check_legacy_fields
-from sky.skylet.providers.gcp.node import GCPNodeType, MAX_POLLS, POLL_INTERVAL
 
 logger = logging.getLogger(__name__)
 
diff --git a/sky/skylet/providers/gcp/node.py b/sky/skylet/providers/gcp/node.py
index aac8e5f7f..5dec7451c 100644
--- a/sky/skylet/providers/gcp/node.py
+++ b/sky/skylet/providers/gcp/node.py
@@ -23,16 +23,17 @@ update the ``_generate_node_name`` method and finally update the
 node provider.
 """
 
-from copy import deepcopy
-from typing import Any, Dict, List, Optional, Tuple, Union
-import logging
 import abc
-import time
+import logging
 import re
-from uuid import uuid4
+import time
+
 from collections import UserDict
+from copy import deepcopy
 from enum import Enum
 from functools import wraps
+from typing import Any, Dict, List, Optional, Tuple, Union
+from uuid import uuid4
 
 from googleapiclient.discovery import Resource
 from googleapiclient.errors import HttpError
diff --git a/sky/skylet/providers/gcp/node_provider.py b/sky/skylet/providers/gcp/node_provider.py
index de1da31f7..7257a87c8 100644
--- a/sky/skylet/providers/gcp/node_provider.py
+++ b/sky/skylet/providers/gcp/node_provider.py
@@ -1,37 +1,39 @@
-from typing import Dict
+import logging
+import time
 import copy
 from functools import wraps
 from threading import RLock
-import time
-import logging
+from typing import Dict, List, Tuple
 
 import googleapiclient
 
-from ray.autoscaler.node_provider import NodeProvider
-from ray.autoscaler.tags import (TAG_RAY_LAUNCH_CONFIG, TAG_RAY_NODE_KIND,
-                                 TAG_RAY_USER_NODE_TYPE)
-from ray.autoscaler._private.cli_logger import cli_logger, cf
-
 from sky.skylet.providers.gcp.config import (
     bootstrap_gcp,
     construct_clients_from_provider_config,
     get_node_type,
 )
 
+from ray.autoscaler.tags import (TAG_RAY_LAUNCH_CONFIG, TAG_RAY_NODE_KIND,
+                                 TAG_RAY_USER_NODE_TYPE)
+from ray.autoscaler._private.cli_logger import cf, cli_logger
+
+
 # The logic has been abstracted away here to allow for different GCP resources
 # (API endpoints), which can differ widely, making it impossible to use
 # the same logic for everything.
 from sky.skylet.providers.gcp.node import (  # noqa
-    GCPResource,
-    GCPNode,
     GCPCompute,
-    GCPTPU,
+    GCPNode,
     GCPNodeType,
+    GCPResource,
+    GCPTPU,
+    # Added by SkyPilot
     INSTANCE_NAME_MAX_LEN,
     INSTANCE_NAME_UUID_LEN,
     MAX_POLLS_STOP,
     POLL_INTERVAL,
 )
+from ray.autoscaler.node_provider import NodeProvider
 
 logger = logging.getLogger(__name__)
 
@@ -173,8 +175,14 @@ class GCPNodeProvider(NodeProvider):
             return ip
 
     @_retry
-    def create_node(self, base_config: dict, tags: dict, count: int) -> None:
+    def create_node(self, base_config: dict, tags: dict, count: int) -> Dict[str, dict]:
+        """Creates instances.
+
+        Returns dict mapping instance id to each create operation result for the created
+        instances.
+        """
         with self.lock:
+            result_dict = dict()
             labels = tags  # gcp uses "labels" instead of aws "tags"
             labels = dict(sorted(copy.deepcopy(labels).items()))
 
@@ -211,12 +219,16 @@ class GCPNodeProvider(NodeProvider):
                         "under `provider` in the cluster configuration."
                     )
                     for node_id in reuse_node_ids:
-                        resource.start_instance(node_id)
+                        result = resource.start_instance(node_id)
+                        result_dict[node_id] = {node_id: result}
                     for node_id in reuse_node_ids:
                         self.set_node_tags(node_id, tags)
                     count -= len(reuse_node_ids)
             if count:
-                resource.create_instances(base_config, labels, count)
+                results = resource.create_instances(base_config, labels, count)
+                result_dict.update({instance_id: result for result, instance_id in results})
+            return result_dict
+
 
     @_retry
     def terminate_node(self, node_id: str):
diff --git a/sky/skylet/ray_patches/__init__.py b/sky/skylet/ray_patches/__init__.py
index 8ab9b73e2..4f745e126 100644
--- a/sky/skylet/ray_patches/__init__.py
+++ b/sky/skylet/ray_patches/__init__.py
@@ -14,7 +14,7 @@ To get original versions, go to the Ray branch with version:
 
 Example workflow:
 
-  >> wget https://raw.githubusercontent.com/ray-project/ray/releases/1.13.0/python/ray/autoscaler/_private/command_runner.py
+  >> wget https://raw.githubusercontent.com/ray-project/ray/releases/2.0.1/python/ray/autoscaler/_private/command_runner.py
   >> cp command_runner.py command_runner.py.1
 
   >> # Make some edits to command_runner.py.1...
@@ -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)
@@ -60,7 +63,7 @@ def patch() -> None:
     from ray._private import log_monitor
     _run_patch(log_monitor.__file__, _to_absolute('log_monitor.py.patch'))
 
-    from ray import worker
+    from ray._private import worker
     _run_patch(worker.__file__, _to_absolute('worker.py.patch'))
 
     from ray.dashboard.modules.job import cli
@@ -78,3 +81,19 @@ def patch() -> None:
     from ray.autoscaler._private import resource_demand_scheduler
     _run_patch(resource_demand_scheduler.__file__,
                _to_absolute('resource_demand_scheduler.py.patch'))
+
+    from ray.autoscaler._private import updater
+    _run_patch(updater.__file__, _to_absolute('updater.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/autoscaler.py.patch b/sky/skylet/ray_patches/autoscaler.py.patch
index 123979934..2d466dc17 100644
--- a/sky/skylet/ray_patches/autoscaler.py.patch
+++ b/sky/skylet/ray_patches/autoscaler.py.patch
@@ -1,9 +1,9 @@
 0a1,4
-> # From https://github.com/ray-project/ray/blob/releases/1.13.0/python/ray/autoscaler/_private/autoscaler.py
+> # From https://github.com/ray-project/ray/blob/ray-2.0.1/python/ray/autoscaler/_private/autoscaler.py
 > # Sky patch changes:
 > #  - enable upscaling_speed to be 0.0
 > 
-915c919
+1022c1026
 <             if upscaling_speed:
 ---
 >             if upscaling_speed is not None: # NOTE(sky): enable 0.0
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/skylet/ray_patches/cli.py.patch b/sky/skylet/ray_patches/cli.py.patch
index ae7ee2bd2..7eef4cd19 100644
--- a/sky/skylet/ray_patches/cli.py.patch
+++ b/sky/skylet/ray_patches/cli.py.patch
@@ -1,11 +1,11 @@
 0a1,4
-> # From https://github.com/ray-project/ray/blob/ray-1.13.0/dashboard/modules/job/cli.py
+> # Adapted from https://github.com/ray-project/ray/blob/ray-2.0.1/dashboard/modules/job/cli.py
 > # Fixed the problem in ray's issue https://github.com/ray-project/ray/issues/26514
 > # Otherwise, the output redirection ">" will not work.
 > 
-4d7
+5d8
 < from subprocess import list2cmdline
-153c156
+182c185
 <         entrypoint=list2cmdline(entrypoint),
 ---
 >         entrypoint=" ".join(entrypoint),
diff --git a/sky/skylet/ray_patches/command_runner.py.patch b/sky/skylet/ray_patches/command_runner.py.patch
index 2f47022c6..899f2e02d 100644
--- a/sky/skylet/ray_patches/command_runner.py.patch
+++ b/sky/skylet/ray_patches/command_runner.py.patch
@@ -1,4 +1,4 @@
-330c330
+329c329
 <                     "ControlPersist": "10s",
 ---
 >                     "ControlPersist": "300s",
diff --git a/sky/skylet/ray_patches/job_manager.py.patch b/sky/skylet/ray_patches/job_manager.py.patch
index 30500cca6..82c54fd90 100644
--- a/sky/skylet/ray_patches/job_manager.py.patch
+++ b/sky/skylet/ray_patches/job_manager.py.patch
@@ -1,9 +1,9 @@
 0a1,4
-> # Adapted from https://github.com/ray-project/ray/blob/ray-1.13.0/dashboard/modules/job/job_manager.py
+> # Adapted from https://github.com/ray-project/ray/blob/ray-2.0.1/dashboard/modules/job/job_manager.py
 > # Fixed the problem where the _monitor_job thread is leaked, due to `await job_supervisor.ping.remote()`
 > # does not raise an exception after the job_supervisor is exited, causing the dashboard to hang.
 > 
-334c338,349
+384c388,398
 <                 await job_supervisor.ping.remote()
 ---
 >                 # Simulate the await behavior, in case some unexpected exception happens
@@ -17,4 +17,3 @@
 >                         ray.get(ready)
 >                     else:
 >                         await asyncio.sleep(0)
-> 
diff --git a/sky/skylet/ray_patches/log_monitor.py.patch b/sky/skylet/ray_patches/log_monitor.py.patch
index fa63f2e73..f06d150e0 100644
--- a/sky/skylet/ray_patches/log_monitor.py.patch
+++ b/sky/skylet/ray_patches/log_monitor.py.patch
@@ -1,9 +1,8 @@
-0a1,4
-> # Adapted from https://github.com/ray-project/ray/blob/ray-1.13.0/python/ray/_private/log_monitor.py
+0a1,3
+> # Adapted from https://github.com/ray-project/ray/blob/ray-2.0.1/python/ray/_private/log_monitor.py
 > # Fixed the problem for progress bar, as the latest version does not preserve \r for progress bar.
 > # The change is adapted from https://github.com/ray-project/ray/blob/ray-1.10.0/python/ray/_private/log_monitor.py#L299-L300
-> 
-306c310,311
+351c354,355
 <                     next_line = next_line.rstrip("\r\n")
 ---
 >                     if next_line[-1] == "\n":
diff --git a/sky/skylet/ray_patches/resource_demand_scheduler.py.patch b/sky/skylet/ray_patches/resource_demand_scheduler.py.patch
index 59e2e3642..41a1719d6 100644
--- a/sky/skylet/ray_patches/resource_demand_scheduler.py.patch
+++ b/sky/skylet/ray_patches/resource_demand_scheduler.py.patch
@@ -1,17 +1,17 @@
 0a1,5
-> # From https://github.com/ray-project/ray/blob/releases/1.13.0/python/ray/autoscaler/_private/resource_demand_scheduler.py
+> # From https://github.com/ray-project/ray/blob/ray-2.0.1/python/ray/autoscaler/_private/resource_demand_scheduler.py
 > # Sky patch changes:
 > #  - no new nodes are allowed to be launched launched when the upscaling_speed is 0
 > #  - comment out "assert not unfulfilled": this seems a buggy assert
 > 
-493c498,501
+509c514,517
 <             if upper_bound > 0:
 ---
 >             # NOTE(sky): do not autoscale when upsclaing speed is 0.
 >             if self.upscaling_speed == 0:
 >                 upper_bound = 0
 >             if upper_bound >= 0:
-630c638
+646c654
 <             assert not unfulfilled
 ---
 >             # assert not unfulfilled  # NOTE(sky): buggy assert.
diff --git a/sky/skylet/ray_patches/updater.py.patch b/sky/skylet/ray_patches/updater.py.patch
new file mode 100644
index 000000000..77cfce04f
--- /dev/null
+++ b/sky/skylet/ray_patches/updater.py.patch
@@ -0,0 +1,7 @@
+0a1,4
+> # From https://github.com/ray-project/ray/blob/releases/2.0.1/python/ray/autoscaler/_private/updater.py
+> # Sky patch changes:
+> #  - Ensure the node state is refreshed before checking the node is terminated.
+> 
+318a319
+>                     self.provider.non_terminated_nodes({})
diff --git a/sky/skylet/ray_patches/worker.py.patch b/sky/skylet/ray_patches/worker.py.patch
index 66135729d..256f09f92 100644
--- a/sky/skylet/ray_patches/worker.py.patch
+++ b/sky/skylet/ray_patches/worker.py.patch
@@ -1,9 +1,9 @@
 0a1,4
-> # Adapted from https://github.com/ray-project/ray/blob/ray-1.13.0/python/ray/worker.py
+> # Adapted from https://github.com/ray-project/ray/blob/ray-2.0.1/python/ray/worker.py
 > # Fixed the problem in ray's issue https://github.com/ray-project/ray/issues/9233
 > # Tracked in PR https://github.com/ray-project/ray/pull/21977/files.
 > 
-1323a1328,1334
+1748a1753,1759
 >     
 >     def end_for(line: str) -> str:
 >         if sys.platform == "win32":
@@ -11,7 +11,7 @@
 >         if line.endswith("\r"):
 >             return ""
 >         return "\n"
-1343a1355
+1768a1780
 >                 end=end_for(line),
-1357a1370
+1782a1795
 >                 end=end_for(line),
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/task.py b/sky/task.py
index f4207cded..416300bb2 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -70,7 +70,7 @@ def _is_valid_env_var(name: str) -> bool:
 
 
 class Task:
-    """Task: a coarse-grained stage in an application."""
+    """Task: a computation to be run on the cloud."""
 
     def __init__(
         self,
@@ -86,36 +86,57 @@ class Task:
     ):
         """Initializes a Task.
 
-        All fields are optional.  `Task.run` is the actual program: either a
+        All fields are optional.  ``Task.run`` is the actual program: either a
         shell command to run (str) or a command generator for different nodes
         (lambda; see below).
 
-        Before executing a Task, it is required to call Task.set_resources() to
-        assign resource requirements to this task.
+        Optionally, call ``Task.set_resources()`` to set the resource
+        requirements for this task.  If not set, a default CPU-only requirement
+        is assumed (the same as ``sky cpunode``).
+
+        All setters of this class, ``Task.set_*()``, return ``self``, i.e.,
+        they are fluent APIs and can be chained together.
+
+        Example:
+            .. code-block:: python
+
+                # A Task that will sync up local workdir '.', containing
+                # requirements.txt and train.py.
+                sky.Task(setup='pip install requirements.txt',
+                         run='python train.py',
+                         workdir='.')
+
+                # An empty Task for provisioning a cluster.
+                task = sky.Task(num_nodes=n).set_resources(...)
+
+                # Chaining setters.
+                sky.Task().set_resources(...).set_file_mounts(...)
 
         Args:
-          name: A string name for the Task.
-          setup: A setup command, run under 'workdir' and before actually
-            executing the run command, 'run'.
-          run: Either a shell command (str) or a command generator (callable).
-            If latter, it must take a node rank and a list of node addresses as
-            input and return a shell command (str) (valid to return None for
-            some nodes, in which case no commands are run on them).  Commands
-            will be run under 'workdir'. Note the command generator should be
-            self-contained.
+          name: A string name for the Task for display purposes.
+          setup: A setup command, which will be run before executing the run
+            commands ``run``, and executed under ``workdir``.
+          run: The actual command for the task. If not None, either a shell
+            command (str) or a command generator (callable).  If latter, it
+            must take a node rank and a list of node addresses as input and
+            return a shell command (str) (valid to return None for some nodes,
+            in which case no commands are run on them).  Run commands will be
+            run under ``workdir``. Note the command generator should be a
+            self-contained lambda.
           envs: A dictionary of environment variables to set before running the
-            setup and run command.
-          workdir: The local working directory.  This directory and its files
-            will be synced to a location on the remote VM(s), and 'setup' and
-            'run' commands will be run under that location (thus, they can rely
-            on relative paths when invoking binaries).
+            setup and run commands.
+          workdir: The local working directory.  This directory will be synced
+            to a location on the remote VM(s), and ``setup`` and ``run``
+            commands will be run under that location (thus, they can rely on
+            relative paths when invoking binaries).
           num_nodes: The number of nodes to provision for this Task.  If None,
             treated as 1 node.  If > 1, each node will execute its own
-            setup/run command; 'run' can either be a str, meaning all nodes get
-            the same command, or a lambda, as documented above.
-          docker_image: The base docker image that this Task will be built on.
-            In effect when LocalDockerBackend is used.  Defaults to
-            'gpuci/miniforge-cuda:11.4-devel-ubuntu18.04'.
+            setup/run command, where ``run`` can either be a str, meaning all
+            nodes get the same command, or a lambda, with the semantics
+            documented above.
+          docker_image: (EXPERIMENTAL: Only in effect when LocalDockerBackend
+            is used.) The base docker image that this Task will be built on.
+            Defaults to 'gpuci/miniforge-cuda:11.4-devel-ubuntu18.04'.
         """
         self.name = name
         self.run = run
@@ -207,7 +228,21 @@ class Task:
                         f'a symlink to a directory). {self.workdir} not found.')
 
     @staticmethod
-    def from_yaml(yaml_path):
+    def from_yaml(yaml_path: str) -> 'Task':
+        """Initializes a task from a task YAML.
+
+        Example:
+            .. code-block:: python
+
+                task = sky.Task.from_yaml('/path/to/task.yaml')
+
+        Args:
+          yaml_path: file path to a valid task yaml file.
+
+        Raises:
+          ValueError: if the path gets loaded into a str instead of a dict; or
+            if there are any other parsing errors.
+        """
         with open(os.path.expanduser(yaml_path), 'r') as f:
             # TODO(zongheng): use
             #  https://github.com/yaml/pyyaml/issues/165#issuecomment-430074049
@@ -295,47 +330,6 @@ class Task:
         assert not config, f'Invalid task args: {config.keys()}'
         return task
 
-    def to_yaml_config(self) -> Dict[str, Any]:
-        """Returns a yaml-style dict representation of the task."""
-        config = dict()
-
-        def add_if_not_none(key, value):
-            if value is not None:
-                config[key] = value
-
-        add_if_not_none('name', self.name)
-
-        if self.resources is not None:
-            assert len(self.resources) == 1
-            resources = list(self.resources)[0]
-            add_if_not_none('resources', resources.to_yaml_config())
-        add_if_not_none('num_nodes', self.num_nodes)
-
-        if self.inputs is not None:
-            add_if_not_none('inputs',
-                            {self.inputs: self.estimated_inputs_size_gigabytes})
-        if self.outputs is not None:
-            add_if_not_none(
-                'outputs',
-                {self.outputs: self.estimated_outputs_size_gigabytes})
-
-        add_if_not_none('setup', self.setup)
-        add_if_not_none('workdir', self.workdir)
-        add_if_not_none('run', self.run)
-        add_if_not_none('envs', self.envs)
-
-        add_if_not_none('file_mounts', dict())
-
-        if self.file_mounts is not None:
-            config['file_mounts'].update(self.file_mounts)
-
-        if self.storage_mounts is not None:
-            config['file_mounts'].update({
-                mount_path: storage.to_yaml_config()
-                for mount_path, storage in self.storage_mounts.items()
-            })
-        return config
-
     @property
     def num_nodes(self) -> int:
         return self._num_nodes
@@ -344,11 +338,24 @@ class Task:
     def envs(self) -> Dict[str, str]:
         return self._envs
 
-    def set_envs(self, envs: Union[None, Tuple[Tuple[str, str]], Dict[str,
-                                                                      str]]):
+    def set_envs(
+            self, envs: Union[None, Tuple[Tuple[str, str]],
+                              Dict[str, str]]) -> 'Task':
+        """Sets the environment variables for use inside the setup/run commands.
+
+        Args:
+          envs: (optional) either a list of ``(env_name, value)`` or a dict
+            ``{env_name: value}``.
+
+        Returns:
+          self: The current task, with envs set.
+
+        Raises:
+          ValueError: if various invalid inputs errors are detected.
+        """
         if envs is None:
             self._envs = None
-            return
+            return self
         if isinstance(envs, (list, tuple)):
             keys = set(env[0] for env in envs)
             if len(keys) != len(envs):
@@ -369,6 +376,7 @@ class Task:
                     'envs must be List[Tuple[str, str]] or Dict[str, str]: '
                     f'{envs}')
         self._envs = envs
+        return self
 
     @property
     def need_spot_recovery(self) -> bool:
@@ -384,8 +392,8 @@ class Task:
                     f'num_nodes should be a positive int. Got: {num_nodes}')
         self._num_nodes = num_nodes
 
-    # E.g., 's3://bucket', 'gs://bucket', or None.
-    def set_inputs(self, inputs, estimated_size_gigabytes):
+    def set_inputs(self, inputs, estimated_size_gigabytes) -> 'Task':
+        # E.g., 's3://bucket', 'gs://bucket', or None.
         self.inputs = inputs
         self.estimated_inputs_size_gigabytes = estimated_size_gigabytes
         return self
@@ -397,7 +405,7 @@ class Task:
         return self.estimated_inputs_size_gigabytes
 
     def get_inputs_cloud(self):
-        """Returns the cloud my inputs live in."""
+        """EXPERIMENTAL: Returns the cloud my inputs live in."""
         assert isinstance(self.inputs, str), self.inputs
         if self.inputs.startswith('s3:'):
             return clouds.AWS()
@@ -407,7 +415,7 @@ class Task:
             with ux_utils.print_exception_no_traceback():
                 raise ValueError(f'cloud path not supported: {self.inputs}')
 
-    def set_outputs(self, outputs, estimated_size_gigabytes):
+    def set_outputs(self, outputs, estimated_size_gigabytes) -> 'Task':
         self.outputs = outputs
         self.estimated_outputs_size_gigabytes = estimated_size_gigabytes
         return self
@@ -418,14 +426,22 @@ class Task:
     def get_estimated_outputs_size_gigabytes(self):
         return self.estimated_outputs_size_gigabytes
 
-    def set_resources(self, resources: Union['resources_lib.Resources',
-                                             Set['resources_lib.Resources']]):
+    def set_resources(
+        self, resources: Union['resources_lib.Resources',
+                               Set['resources_lib.Resources']]
+    ) -> 'Task':
         """Sets the required resources to execute this task.
 
+        If this function is not called for a Task, default resource
+        requirements will be used (8 vCPUs).
+
         Args:
           resources: either a sky.Resources, or a set of them.  The latter case
-            indicates the user intent "pick any one of these resources" to run
-            a task.
+            is EXPERIMENTAL and indicates asking the optimizer to "pick the
+            best of these resources" to run this task.
+
+        Returns:
+          self: The current task, with resources set.
         """
         if isinstance(resources, sky.Resources):
             resources = {resources}
@@ -435,39 +451,161 @@ class Task:
     def get_resources(self):
         return self.resources
 
-    def set_time_estimator(self, func):
-        """Sets a func mapping resources to estimated time (secs)."""
+    def set_time_estimator(self, func: Callable[['sky.Resources'],
+                                                int]) -> 'Task':
+        """Sets a func mapping resources to estimated time (secs).
+
+        This is EXPERIMENTAL.
+        """
         self.time_estimator_func = func
         return self
 
     def estimate_runtime(self, resources):
-        """Returns a func mapping resources to estimated time (secs)."""
+        """Returns a func mapping resources to estimated time (secs).
+
+        This is EXPERIMENTAL.
+        """
         if self.time_estimator_func is None:
             raise NotImplementedError(
                 'Node [{}] does not have a cost model set; '
                 'call set_time_estimator() first'.format(self))
         return self.time_estimator_func(resources)
 
+    def set_file_mounts(self, file_mounts: Optional[Dict[str, str]]) -> 'Task':
+        """Sets the file mounts for this task.
+
+        Useful for syncing datasets, dotfiles, etc.
+
+        File mounts are a dictionary: ``{remote_path: local_path/cloud URI}``.
+        Local (or cloud) files/directories will be synced to the specified
+        paths on the remote VM(s) where this Task will run.
+
+        Neither source or destimation paths can end with a slash.
+
+        Example:
+            .. code-block:: python
+
+                task.set_file_mounts({
+                    '~/.dotfile': '/local/.dotfile',
+                    # /remote/dir/ will contain the contents of /local/dir/.
+                    '/remote/dir': '/local/dir',
+                })
+
+        Args:
+          file_mounts: an optional dict of ``{remote_path: local_path/cloud
+            URI}``, where remote means the VM(s) on which this Task will
+            eventually run on, and local means the node from which the task is
+            launched.
+
+        Returns:
+          self: the current task, with file mounts set.
+
+        Raises:
+          ValueError: if input paths are invalid.
+        """
+        if file_mounts is None:
+            self.file_mounts = None
+            return self
+        for target, source in file_mounts.items():
+            if target.endswith('/') or source.endswith('/'):
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        'File mount paths cannot end with a slash '
+                        '(try "/mydir: /mydir" or "/myfile: /myfile"). '
+                        f'Found: target={target} source={source}')
+            if data_utils.is_cloud_store_url(target):
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        'File mount destination paths cannot be cloud storage')
+            if not data_utils.is_cloud_store_url(source):
+                if not os.path.exists(
+                        os.path.abspath(os.path.expanduser(source))):
+                    with ux_utils.print_exception_no_traceback():
+                        raise ValueError(
+                            f'File mount source {source!r} does not exist '
+                            'locally. To fix: check if it exists, and correct '
+                            'the path.')
+            # TODO(zhwu): /home/username/sky_workdir as the target path need
+            # to be filtered out as well.
+            if (target == constants.SKY_REMOTE_WORKDIR and
+                    self.workdir is not None):
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        f'Cannot use {constants.SKY_REMOTE_WORKDIR!r} as a '
+                        'destination path of a file mount, as it will be used '
+                        'by the workdir. If uploading a file/folder to the '
+                        'workdir is needed, please specify the full path to '
+                        'the file/folder.')
+
+        self.file_mounts = file_mounts
+        return self
+
+    def update_file_mounts(self, file_mounts: Dict[str, str]) -> 'Task':
+        """Updates the file mounts for this task.
+
+        Different from set_file_mounts(), this function updates into the
+        existing file_mounts (calls ``dict.update()``), rather than
+        overwritting it.
+
+        This should be called before provisioning in order to take effect.
+
+        Example:
+            .. code-block:: python
+
+                task.update_file_mounts({
+                    '~/.config': '~/Documents/config',
+                    '/tmp/workdir': '/local/workdir/cnn-cifar10',
+                })
+
+        Args:
+          file_mounts: a dict of ``{remote_path: local_path/cloud URI}``, where
+            remote means the VM(s) on which this Task will eventually run on,
+            and local means the node from which the task is launched.
+
+        Returns:
+          self: the current task, with file mounts updated.
+
+        Raises:
+          ValueError: if input paths are invalid.
+        """
+        if self.file_mounts is None:
+            self.file_mounts = {}
+        self.file_mounts.update(file_mounts)
+        # For validation logic:
+        return self.set_file_mounts(self.file_mounts)
+
     def set_storage_mounts(
         self,
         storage_mounts: Optional[Dict[str, storage_lib.Storage]],
-    ):
-        """Sets the storage mounts for this Task
+    ) -> 'Task':
+        """Sets the storage mounts for this task.
 
-        Advanced method for users. Storage mounts map a mount path on the Cloud
-        VM to a Storage object (see data/storage.py). The storage object can be
-        from a local folder or from an existing cloud bucket.
+        Storage mounts are a dictionary: ``{mount_path: sky.Storage object}``,
+        each of which mounts a sky.Storage object (a cloud object store bucket)
+        to a path inside the remote cluster.
+
+        A sky.Storage object can be created by uploading from a local directory
+        (setting ``source``), or backed by an existing cloud bucket (setting
+        ``name`` to the bucket name; or setting ``source`` to the bucket URI).
 
         Example:
-            task.set_storage_mounts({
-                '/tmp/imagenet/': \
-                Storage(name='imagenet', source='s3://imagenet-bucket'):
-            })
+            .. code-block:: python
+
+                task.set_storage_mounts({
+                    '/remote/imagenet/': sky.Storage(name='my-bucket',
+                                                     source='/local/imagenet'),
+                })
 
         Args:
-            storage_mounts: a dict of {mount_path: Storage}, where mount_path
-            is the path on the Cloud VM where the Storage object will be
-            mounted on
+          storage_mounts: an optional dict of ``{mount_path: sky.Storage
+            object}``, where mount_path is the path inside the remote VM(s)
+            where the Storage object will be mounted on.
+
+        Returns:
+          self: The current task, with storage mounts set.
+
+        Raises:
+          ValueError: if input paths are invalid.
         """
         if storage_mounts is None:
             self.storage_mounts = None
@@ -494,9 +632,27 @@ class Task:
         self.storage_mounts = storage_mounts
         return self
 
-    def update_storage_mounts(self, storage_mounts: Dict[str,
-                                                         storage_lib.Storage]):
-        """Updates the storage mounts for this Task"""
+    def update_storage_mounts(
+            self, storage_mounts: Dict[str, storage_lib.Storage]) -> 'Task':
+        """Updates the storage mounts for this task.
+
+        Different from set_storage_mounts(), this function updates into the
+        existing storage_mounts (calls ``dict.update()``), rather than
+        overwritting it.
+
+        This should be called before provisioning in order to take effect.
+
+        Args:
+          storage_mounts: an optional dict of ``{mount_path: sky.Storage
+            object}``, where mount_path is the path inside the remote VM(s)
+            where the Storage object will be mounted on.
+
+        Returns:
+          self: The current task, with storage mounts updated.
+
+        Raises:
+          ValueError: if input paths are invalid.
+        """
         if not storage_mounts:
             return self
         task_storage_mounts = self.storage_mounts if self.storage_mounts else {}
@@ -539,8 +695,12 @@ class Task:
         return store_type
 
     def sync_storage_mounts(self) -> None:
-        """Syncs storage mounts: sync files/dirs to cloud storage."""
+        """(INTERNAL) Eagerly syncs storage mounts to cloud storage.
 
+        After syncing up, COPY-mode storage mounts are translated into regular
+        file_mounts of the form ``{ /remote/path: {s3,gs,..}://<bucket path>
+        }``.
+        """
         for storage in self.storage_mounts.values():
             if len(storage.stores) == 0:
                 store_type = self.get_preferred_store_type()
@@ -581,95 +741,13 @@ class Task:
                         raise ValueError(f'Storage Type {store_type} '
                                          'does not exist!')
 
-    def set_file_mounts(self, file_mounts: Optional[Dict[str, str]]) -> None:
-        """Sets the file mounts for this Task.
-
-        File mounts are a dictionary of { remote_path: local_path/cloud URI }.
-        Local (or cloud) files/directories will be synced to the specified
-        paths on the remote VM(s) where this Task will run.
-
-        Used for syncing datasets, dotfiles, etc.
-
-        Paths cannot end with a slash (for clarity).
-
-        Example:
-
-            task.set_file_mounts({
-                '~/.dotfile': '/local/.dotfile',
-                # /remote/dir/ will contain the contents of /local/dir/.
-                '/remote/dir': '/local/dir',
-            })
-
-        Args:
-          file_mounts: either None or a dict of { remote_path: local_path/cloud
-            URI }, where remote is the VM on which this Task will eventually
-            run on, and local is the node from which the task is launched.
-        """
-        if file_mounts is None:
-            self.file_mounts = None
-            return self
-        for target, source in file_mounts.items():
-            if target.endswith('/') or source.endswith('/'):
-                with ux_utils.print_exception_no_traceback():
-                    raise ValueError(
-                        'File mount paths cannot end with a slash '
-                        '(try "/mydir: /mydir" or "/myfile: /myfile"). '
-                        f'Found: target={target} source={source}')
-            if data_utils.is_cloud_store_url(target):
-                with ux_utils.print_exception_no_traceback():
-                    raise ValueError(
-                        'File mount destination paths cannot be cloud storage')
-            if not data_utils.is_cloud_store_url(source):
-                if not os.path.exists(
-                        os.path.abspath(os.path.expanduser(source))):
-                    with ux_utils.print_exception_no_traceback():
-                        raise ValueError(
-                            f'File mount source {source!r} does not exist '
-                            'locally. To fix: check if it exists, and correct '
-                            'the path.')
-            # TODO(zhwu): /home/username/sky_workdir as the target path need
-            # to be filtered out as well.
-            if (target == constants.SKY_REMOTE_WORKDIR and
-                    self.workdir is not None):
-                with ux_utils.print_exception_no_traceback():
-                    raise ValueError(
-                        f'Cannot use {constants.SKY_REMOTE_WORKDIR!r} as a '
-                        'destination path of a file mount, as it will be used '
-                        'by the workdir. If uploading a file/folder to the '
-                        'workdir is needed, please specify the full path to '
-                        'the file/folder.')
-
-        self.file_mounts = file_mounts
-        return self
-
-    def update_file_mounts(self, file_mounts: Dict[str, str]):
-        """Updates the file mounts for this Task.
-
-        This should be run before provisioning.
-
-        Example:
-
-            task.update_file_mounts({
-                '~/.config': '~/Documents/config',
-                '/tmp/workdir': '/local/workdir/cnn-cifar10',
-            })
-
-        Args:
-          file_mounts: a dict of { remote_path: local_path }, where remote is
-            the VM on which this Task will eventually run on, and local is the
-            node from which the task is launched.
-        """
-        if self.file_mounts is None:
-            self.file_mounts = {}
-        self.file_mounts.update(file_mounts)
-        # For validation logic:
-        return self.set_file_mounts(self.file_mounts)
-
     def get_local_to_remote_file_mounts(self) -> Optional[Dict[str, str]]:
         """Returns file mounts of the form (dst=VM path, src=local path).
 
-        Any cloud object store URLs (gs://, s3://, etc.), either as source or
+        Any cloud object store URIs (gs://, s3://, etc.), either as source or
         destination, are not included.
+
+        INTERNAL: this method is internal-facing.
         """
         if self.file_mounts is None:
             return None
@@ -685,6 +763,8 @@ class Task:
 
         Local-to-remote file mounts are excluded (handled by
         get_local_to_remote_file_mounts()).
+
+        INTERNAL: this method is internal-facing.
         """
         if self.file_mounts is None:
             return None
@@ -695,6 +775,50 @@ class Task:
                 d[k] = v
         return d
 
+    def to_yaml_config(self) -> Dict[str, Any]:
+        """Returns a yaml-style dict representation of the task.
+
+        INTERNAL: this method is internal-facing.
+        """
+        config = dict()
+
+        def add_if_not_none(key, value):
+            if value is not None:
+                config[key] = value
+
+        add_if_not_none('name', self.name)
+
+        if self.resources is not None:
+            assert len(self.resources) == 1
+            resources = list(self.resources)[0]
+            add_if_not_none('resources', resources.to_yaml_config())
+        add_if_not_none('num_nodes', self.num_nodes)
+
+        if self.inputs is not None:
+            add_if_not_none('inputs',
+                            {self.inputs: self.estimated_inputs_size_gigabytes})
+        if self.outputs is not None:
+            add_if_not_none(
+                'outputs',
+                {self.outputs: self.estimated_outputs_size_gigabytes})
+
+        add_if_not_none('setup', self.setup)
+        add_if_not_none('workdir', self.workdir)
+        add_if_not_none('run', self.run)
+        add_if_not_none('envs', self.envs)
+
+        add_if_not_none('file_mounts', dict())
+
+        if self.file_mounts is not None:
+            config['file_mounts'].update(self.file_mounts)
+
+        if self.storage_mounts is not None:
+            config['file_mounts'].update({
+                mount_path: storage.to_yaml_config()
+                for mount_path, storage in self.storage_mounts.items()
+            })
+        return config
+
     def __rshift__(self, b):
         sky.dag.get_current_dag().add_edge(self, b)
 
diff --git a/sky/templates/aws-ray.yml.j2 b/sky/templates/aws-ray.yml.j2
index 5732d61dc..4ed178b44 100644
--- a/sky/templates/aws-ray.yml.j2
+++ b/sky/templates/aws-ray.yml.j2
@@ -128,7 +128,7 @@ head_start_ray_commands:
   # NOTE: --disable-usage-stats in `ray start` saves 10 seconds of idle wait.
   # Line "which prlimit ..": increase the limit of the number of open files for the raylet process, as the `ulimit` may not take effect at this point, because it requires
   # all the sessions to be reloaded. This is a workaround.
-  - ((ps aux | grep "-m sky.skylet.skylet" | grep -q python3) || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
+  - ((ps aux | grep -v nohup | grep -v grep | grep -q -- "python3 -m sky.skylet.skylet") || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
     ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
     which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 
diff --git a/sky/templates/azure-ray.yml.j2 b/sky/templates/azure-ray.yml.j2
index 88f442a4d..66c97cebf 100644
--- a/sky/templates/azure-ray.yml.j2
+++ b/sky/templates/azure-ray.yml.j2
@@ -124,7 +124,7 @@ setup_commands:
 head_start_ray_commands:
   # Start skylet daemon. (Should not place it in the head_setup_commands, otherwise it will run before skypilot is installed.)
   # NOTE: --disable-usage-stats in `ray start` saves 10 seconds of idle wait.
-  - ((ps aux | grep "-m sky.skylet.skylet" | grep -q python3) || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
+  - ((ps aux | grep -v nohup | grep -v grep | grep -q -- "python3 -m sky.skylet.skylet") || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
     ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
     which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
 
diff --git a/sky/templates/gcp-ray.yml.j2 b/sky/templates/gcp-ray.yml.j2
index 892b4986b..a0964d8c2 100644
--- a/sky/templates/gcp-ray.yml.j2
+++ b/sky/templates/gcp-ray.yml.j2
@@ -162,7 +162,7 @@ head_start_ray_commands:
   # NOTE: --disable-usage-stats in `ray start` saves 10 seconds of idle wait.
   # Line "which prlimit ..": increase the limit of the number of open files for the raylet process, as the `ulimit` may not take effect at this point, because it requires
   # all the sessions to be reloaded. This is a workaround.
-  - ((ps aux | grep "-m sky.skylet.skylet" | grep -q python3) || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
+  - ((ps aux | grep -v nohup | grep -v grep | grep -q -- "python3 -m sky.skylet.skylet") || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
     export SKY_NUM_GPUS=0 && which nvidia-smi > /dev/null && SKY_NUM_GPUS=$(nvidia-smi --query-gpu=index,name --format=csv,noheader | wc -l);
     ray stop; ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} --num-gpus=$SKY_NUM_GPUS || exit 1;
     which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;


Test patch:
diff --git a/tests/backward_comaptibility_tests.sh b/tests/backward_comaptibility_tests.sh
old mode 100644
new mode 100755
index fcdd4bd5d..228c4e4ce
--- a/tests/backward_comaptibility_tests.sh
+++ b/tests/backward_comaptibility_tests.sh
@@ -1,6 +1,9 @@
 #!/bin/bash
 set -ev
 
+need_launch=${1:-0}
+start_from=${2:-0}
+
 source ~/.bashrc 
 CLUSTER_NAME="test-back-compat-$USER"
 source $(conda info --base 2> /dev/null)/etc/profile.d/conda.sh
@@ -30,6 +33,7 @@ pip install -e ".[all]"
 
 
 # exec + launch
+if [ "$start_from" -le 1 ]; then
 mamba activate sky-back-compat-master
 rm -r  ~/.sky/wheels || true
 which sky
@@ -37,13 +41,19 @@ sky launch --cloud gcp -y -c ${CLUSTER_NAME} examples/minimal.yaml
 
 mamba activate sky-back-compat-current
 rm -r  ~/.sky/wheels || true
+if [ "$need_launch" -eq "1" ]; then
+  sky launch --cloud gcp -y -c ${CLUSTER_NAME}
+fi
 sky exec --cloud gcp ${CLUSTER_NAME} examples/minimal.yaml
 s=$(sky launch --cloud gcp -d -c ${CLUSTER_NAME} examples/minimal.yaml)
 echo $s
-echo $s | grep "Job ID: 3" || exit 1
+# remove color and find the job id
+echo $s | sed -r "s/\x1B\[([0-9]{1,3}(;[0-9]{1,2})?)?[mGK]//g" | grep "Job ID: 3" || exit 1
 sky queue ${CLUSTER_NAME}
+fi
 
 # sky stop + sky start + sky exec
+if [ "$start_from" -le 2 ]; then
 mamba activate sky-back-compat-master
 rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-2 examples/minimal.yaml
@@ -53,20 +63,24 @@ sky stop -y ${CLUSTER_NAME}-2
 sky start -y ${CLUSTER_NAME}-2
 s=$(sky exec --cloud gcp -d ${CLUSTER_NAME}-2 examples/minimal.yaml)
 echo $s
-echo $s | grep "Job ID: 2" || exit 1
+echo $s | sed -r "s/\x1B\[([0-9]{1,3}(;[0-9]{1,2})?)?[mGK]//g" | grep "Job ID: 2" || exit 1
+fi
 
 # `sky autostop` + `sky status -r`
+if [ "$start_from" -le 3 ]; then
 mamba activate sky-back-compat-master
 rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-3 examples/minimal.yaml
 mamba activate sky-back-compat-current
 rm -r  ~/.sky/wheels || true
 sky autostop -y -i0 ${CLUSTER_NAME}-3
-sleep 100
+sleep 120
 sky status -r | grep ${CLUSTER_NAME}-3 | grep STOPPED || exit 1
+fi
 
 
 # (1 node) sky launch + sky exec + sky queue + sky logs
+if [ "$start_from" -le 4 ]; then
 mamba activate sky-back-compat-master
 rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-4 examples/minimal.yaml
@@ -79,8 +93,10 @@ sky logs ${CLUSTER_NAME}-4 1 --status
 sky logs ${CLUSTER_NAME}-4 2 --status
 sky logs ${CLUSTER_NAME}-4 1
 sky logs ${CLUSTER_NAME}-4 2
+fi
 
 # (1 node) sky start + sky exec + sky queue + sky logs
+if [ "$start_form" -le 5 ]; then
 mamba activate sky-back-compat-master
 rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-5 examples/minimal.yaml
@@ -95,8 +111,10 @@ sky launch --cloud gcp -y -c ${CLUSTER_NAME}-5 examples/minimal.yaml
 sky queue ${CLUSTER_NAME}-5
 sky logs ${CLUSTER_NAME}-5 2 --status
 sky logs ${CLUSTER_NAME}-5 2
+fi
 
 # (2 nodes) sky launch + sky exec + sky queue + sky logs
+if [ "$start_from" -le 6 ]; then
 mamba activate sky-back-compat-master
 rm -r  ~/.sky/wheels || true
 sky launch --cloud gcp -y -c ${CLUSTER_NAME}-6 examples/multi_hostname.yaml
@@ -111,5 +129,6 @@ sky exec --cloud gcp ${CLUSTER_NAME}-6 examples/multi_hostname.yaml
 sky queue ${CLUSTER_NAME}-6
 sky logs ${CLUSTER_NAME}-6 2 --status
 sky logs ${CLUSTER_NAME}-6 2
+fi
 
 sky down ${CLUSTER_NAME}* -y
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 76162a7cd..e2af948a6 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}',
     )
@@ -955,6 +955,29 @@ class TestStorageWithCredentials:
         out = subprocess.check_output(['sky', 'storage', 'ls'])
         assert tmp_local_storage_obj.name not in out.decode('utf-8')
 
+    @pytest.mark.parametrize(
+        'store_type', [storage_lib.StoreType.S3, storage_lib.StoreType.GCS])
+    def test_bucket_bulk_deletion(self, store_type):
+        # Create a temp folder with over 256 files and folders, upload
+        # files and folders to a new bucket, then delete bucket.
+        with tempfile.TemporaryDirectory() as tmpdir:
+            subprocess.check_output(f'mkdir -p {tmpdir}/folder{{000..255}}',
+                                    shell=True)
+            subprocess.check_output(f'touch {tmpdir}/test{{000..255}}.txt',
+                                    shell=True)
+            subprocess.check_output(
+                f'touch {tmpdir}/folder{{000..255}}/test.txt', shell=True)
+
+            timestamp = str(time.time()).replace('.', '')
+            store_obj = storage_lib.Storage(name=f'sky-test-{timestamp}',
+                                            source=tmpdir)
+            store_obj.add_store(store_type)
+
+        subprocess.check_output(['sky', 'storage', 'delete', store_obj.name])
+
+        output = subprocess.check_output(['sky', 'storage', 'ls'])
+        assert store_obj.name not in output.decode('utf-8')
+
     @pytest.mark.parametrize(
         'tmp_public_storage_obj, store_type',
         [('s3://tcga-2-open', storage_lib.StoreType.S3),
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],