Repository: skypilot-org/skypilot
Pull request: [Core] Support launching a new cluster from an existing cluster's disk
<!-- Describe the changes in this PR -->

Fixes #2050. This PR only supports AWS and GCP as an experimental feature and unblocks the users.

~~It is blocked by #2087~~

`sky launch --clone-disk-from existing-cluster -c new-cluster --cloud aws --region us-west-2 task.yaml`

A problem with the image copying is that the copying can take significant of time.

TODOs:
- [x] Support GCP
- [x] Clean up the image after the target cluster being tear down.
- [x] docs for additional permissions required

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
  - AWS (with both admin and minimal permission with the new permissions):
  - [x] `sky launch -c min --cloud aws --cpus 2 --region us-east-1`; `ssh min`; `echo 'hi' > user_file.txt`
  - [x] `sky launch --clone min -c new-min --cloud aws --region us-west-2`
  - [x] `sky launch --clone min -c new-min2`
  - [x] `sky down new-min new-min2`
  - GCP:
  - [x] `sky launch -c min --cloud gcp --cpus 2 --region us-central1`; `ssh min`; `echo 'hi' > user_file.txt`
  - [x] `sky launch --clone min -c new-min --cloud gcp --region us-east1 --disk-size 50`
  - [x] `sky launch --clone min -c new-min --cloud gcp --region us-east1`
  - [x] `sky launch --clone min -c new-min2`
  - [x] `sky down new-min new-min2`
- [ ] All smoke tests: `pytest tests/test_smoke.py` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
- [ ] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh`


Problem statement:
Support migration of a stopped cluster to another region
When a cluster is STOPPED and the resource of that cluster is not available in that region/zone, the user will not be able to restart the cluster and it would be hard to migrate to another region manually as the cluster may be a dev machine.

To mitigate the problem, we should provide a way to migrate the stopped cluster across regions or even clouds.

The following is a script that can be used to migrate a cluster from one region to another on AWS:

```
# Variables need to be changed:
REGION='us-east-1'
TARGET_REGION='us-west-2'
CLUSTER_NAME='test-image-transfer'
AMI_NAME='image-transfer-ami0'

echo "Finding instance ID for cluster $CLUSTER_NAME in region $REGION..."
INSTANCE_ID=`aws ec2 describe-instances --region $REGION --filters Name=tag:ray-cluster-name,Values=$CLUSTER_NAME --query Reservations[].Instances[].InstanceId --output text`

echo "Creating AMI from instance $INSTANCE_ID in region $REGION..."
IMAGE_ID=`aws ec2 create-image --instance-id $INSTANCE_ID --name $AMI_NAME --region $REGION --output text`

echo Waiting for AMI to be available... This may take a few minutes.
aws ec2 wait image-available --image-ids $IMAGE_ID --region $REGION

echo "Copying AMI $IMAGE_ID from region $REGION to region $TARGET_REGION..."
NEW_IMAGE_ID=`aws ec2 copy-image --name $AMI_NAME --source-image-id $IMAGE_ID --source-region $REGION --region $TARGET_REGION --output text`

echo Waiting for AMI to be available... This may take a few minutes.
aws ec2 wait image-available --image-ids $NEW_IMAGE_ID --region $TARGET_REGION

echo "New AMI ID in $TARGET_REGION: $NEW_IMAGE_ID"

echo "You can now terminate the old cluster and launch a new cluster with the new AMI:"
echo "  sky down $CLUSTER_NAME -y"
echo "  sky launch -c $CLUSTER_NAME --cloud aws --region $TARGET_REGION -y"

```

Hints from issue discussion:

Commit message: fix

Patch to review:
diff --git a/docs/source/cloud-setup/cloud-permissions/aws.rst b/docs/source/cloud-setup/cloud-permissions/aws.rst
index 16f39ec3807..c30fa462cb5 100644
--- a/docs/source/cloud-setup/cloud-permissions/aws.rst
+++ b/docs/source/cloud-setup/cloud-permissions/aws.rst
@@ -103,8 +103,22 @@ AWS accounts can be attached with a policy that limits the permissions of the ac
         ]
     }
 
-4. Click **Next: Tags** and follow the instructions to finish creating the policy. You can give the policy a descriptive name, such as ``minimal-skypilot-policy``.
-5. Go back to the previous window and click on the refresh button, and you can now search for the policy you just created.
+4. **Optional**: To allow ``--clone-disk-from`` to work, you need to add the following permissions to the policy above as well.
+
+.. code-block:: json
+
+           {
+                "Effect": "Allow",
+                "Action": [
+                    "ec2:CreateImage",
+                    "ec2:CopyImage",
+                    "ec2:DeregisterImage"
+                ],
+                "Resource": "*"
+            }
+
+5. Click **Next: Tags** and follow the instructions to finish creating the policy. You can give the policy a descriptive name, such as ``minimal-skypilot-policy``.
+6. Go back to the previous window and click on the refresh button, and you can now search for the policy you just created.
 
 .. image:: ../../images/screenshots/aws/aws-add-policy.png
     :width: 80%
@@ -118,7 +132,7 @@ AWS accounts can be attached with a policy that limits the permissions of the ac
     :align: center
     :alt: AWS Add S3 Policy
 
-7. Click on **Next** and follow the instructions to create the user.
+8. Click on **Next** and follow the instructions to create the user.
 
 With the steps above you are almost ready to have the users in your organization to use SkyPilot with the minimal permissions.
 
diff --git a/docs/source/cloud-setup/cloud-permissions/gcp.rst b/docs/source/cloud-setup/cloud-permissions/gcp.rst
index 8d07e7e8c7b..f661c382152 100644
--- a/docs/source/cloud-setup/cloud-permissions/gcp.rst
+++ b/docs/source/cloud-setup/cloud-permissions/gcp.rst
@@ -112,9 +112,18 @@ User
     tpu.nodes.update
     tpu.operations.get
 
-5. Click **Create** to create the role.
-6. Go back to the "IAM" tab and click on **GRANT ACCESS**.
-7. Fill in the email address of the user in the “Add principals” section, and select ``minimal-skypilot-role`` in the “Assign roles” section. Click **Save**.
+6. **Optional**: To enable ``--clone-disk-from``, you need to have the following permissions for the role as well:
+
+.. code-block:: text
+
+    compute.disks.useReadOnly
+    compute.images.create
+    compute.images.get
+    compute.images.delete
+
+7. Click **Create** to create the role.
+8. Go back to the "IAM" tab and click on **GRANT ACCESS**.
+9. Fill in the email address of the user in the “Add principals” section, and select ``minimal-skypilot-role`` in the “Assign roles” section. Click **Save**.
 
 
 .. image:: ../../images/screenshots/gcp/create-iam.png
@@ -122,7 +131,7 @@ User
     :align: center
     :alt: GCP Grant Access
 
-8. The user should receive an invitation to the project and should be able to setup SkyPilot by following the instructions in :ref:`Installation <installation-gcp>`.
+10. The user should receive an invitation to the project and should be able to setup SkyPilot by following the instructions in :ref:`Installation <installation-gcp>`.
 
 .. note::
 
diff --git a/docs/source/reference/api.rst b/docs/source/reference/api.rst
index de8e9aea4c5..c1a4990bf40 100644
--- a/docs/source/reference/api.rst
+++ b/docs/source/reference/api.rst
@@ -58,7 +58,15 @@ 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
+  :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__
+  .. automethod:: __init__
+
+
+Resources
+-----------------
+
+.. autoclass:: sky.Resources
+  
+  .. automethod:: __init__
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index a5422663aa9..e7ed62152fa 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -1662,6 +1662,13 @@ def check_owner_identity(cluster_name: str) -> None:
                 f'is {current_user_identity!r}.')
 
 
+def tag_filter_for_cluster(cluster_name: str) -> Dict[str, str]:
+    """Returns a tag filter for the cluster."""
+    return {
+        'ray-cluster-name': cluster_name,
+    }
+
+
 def _query_cluster_status_via_cloud_api(
     handle: 'cloud_vm_ray_backend.CloudVmRayResourceHandle'
 ) -> List[status_lib.ClusterStatus]:
@@ -1680,7 +1687,7 @@ def _query_cluster_status_via_cloud_api(
 
     # Query the cloud provider.
     node_statuses = handle.launched_resources.cloud.query_status(
-        cluster_name, {'ray-cluster-name': cluster_name}, region, zone,
+        cluster_name, tag_filter_for_cluster(cluster_name), region, zone,
         **kwargs)
     # GCP does not clean up preempted TPU VMs. We remove it ourselves.
     # TODO(wei-lin): handle multi-node cases.
@@ -1704,6 +1711,89 @@ def _query_cluster_status_via_cloud_api(
     return node_statuses
 
 
+def check_can_clone_disk_and_override_task(
+    cluster_name: str, target_cluster_name: Optional[str], task: 'task_lib.Task'
+) -> Tuple['task_lib.Task', 'cloud_vm_ray_backend.CloudVmRayResourceHandle']:
+    """Check if the task is compatible to clone disk from the source cluster.
+
+    Args:
+        cluster_name: The name of the cluster to clone disk from.
+        target_cluster_name: The name of the target cluster.
+        task: The task to check.
+
+    Returns:
+        The task to use and the resource handle of the source cluster.
+
+    Raises:
+        exceptions.NotSupportedError: If the source cluster is not valid or the
+            task is not compatible to clone disk from the source cluster.
+    """
+    source_cluster_status, handle = refresh_cluster_status_handle(cluster_name)
+    if source_cluster_status is None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(
+                f'Cannot find cluster {cluster_name!r} to clone disk from.')
+
+    if not isinstance(handle, backends.CloudVmRayResourceHandle):
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from a non-cloud cluster {cluster_name!r}.')
+
+    if source_cluster_status != status_lib.ClusterStatus.STOPPED:
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from cluster {source_cluster_status!r}. Please stop the '
+                f'cluster first: sky stop {cluster_name}.')
+
+    if target_cluster_name is not None:
+        target_cluster_status, _ = refresh_cluster_status_handle(
+            target_cluster_name)
+        if target_cluster_status is not None:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    f'The target cluster {target_cluster_name!r} already exists. Cloning '
+                    'disk is only supported when creating a new cluster.')
+
+    assert len(task.resources) == 1, task.resources
+    task_resources = list(task.resources)[0]
+    if handle.launched_resources.disk_size > task_resources.disk_size:
+        # The target cluster's disk should be at least as large as the source.
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'The target cluster {target_cluster_name!r} should have a disk size '
+                f'of at least {handle.launched_resources.disk_size} GB to clone the '
+                f'disk from {cluster_name!r}.')
+    override_param = {}
+    original_cloud = handle.launched_resources.cloud
+    assert original_cloud is not None, handle.launched_resources
+    if task_resources.cloud is None:
+        override_param['cloud'] = original_cloud
+    else:
+        if not original_cloud.is_same_cloud(task_resources.cloud):
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(
+                    f'Cannot clone disk across cloud from {original_cloud} to '
+                    f'{task_resources.cloud}.')
+    original_cloud.check_features_are_supported(
+        {clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER})
+
+    if task_resources.region is None:
+        override_param['region'] = handle.launched_resources.region
+
+    if override_param:
+        logger.info(
+            f'No cloud/region specified for the task. Using the same region '
+            f'as source cluster {cluster_name!r}: '
+            f'{handle.launched_resources.cloud}'
+            f'({handle.launched_resources.region}).')
+        task_resources = task_resources.copy(**override_param)
+        task.set_resources({task_resources})
+        # Reset the best_resources to triger re-optimization
+        # later, so that the new task_resources will be used.
+        task.best_resources = None
+    return task, handle
+
+
 def _update_cluster_status_no_lock(
         cluster_name: str) -> Optional[Dict[str, Any]]:
     record = global_user_state.get_cluster_from_name(cluster_name)
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index b92c05511a4..e0af7873b57 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -3547,6 +3547,23 @@ def post_teardown_cleanup(self,
                             cluster_name=handle.cluster_name,
                             stdout=tpu_stdout,
                             stderr=tpu_stderr))
+        if handle.launched_resources.is_image_managed and terminate:
+            # Delete the image when terminating the cluster and the image is
+            # created by SkyPilot (--clone-disk-from)
+            logger.debug(f'Deleting image {handle.launched_resources.image_id}')
+            cluster_resources = handle.launched_resources
+            cluster_cloud = cluster_resources.cloud
+            image_dict = cluster_resources.image_id
+            assert cluster_cloud is not None, cluster_resources
+            assert image_dict is not None and len(image_dict) == 1
+            image_id = list(image_dict.values())[0]
+            try:
+                cluster_cloud.delete_image(image_id,
+                                           handle.launched_resources.region)
+            except exceptions.CommandError as e:
+                logger.debug(
+                    f'Failed to delete image {image_id}: '
+                    f'{common_utils.format_exception(e, use_bracket=True)}')
 
         # The cluster file must exist because the cluster_yaml will only
         # be removed after the cluster entry in the database is removed.
diff --git a/sky/cli.py b/sky/cli.py
index c69e2c9fbe5..f2a9ad65f67 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -723,12 +723,21 @@ def _launch_with_confirm(
     retry_until_up: bool = False,
     no_setup: bool = False,
     node_type: Optional[str] = None,
+    clone_disk_from: Optional[str] = None,
 ):
     """Launch a cluster with a Task."""
-    with sky.Dag() as dag:
-        dag.add(task)
     if cluster is None:
         cluster = backend_utils.generate_cluster_name()
+
+    clone_source_str = ''
+    if clone_disk_from is not None:
+        clone_source_str = f' from the disk of {clone_disk_from!r}'
+        task, _ = backend_utils.check_can_clone_disk_and_override_task(
+            clone_disk_from, cluster, task)
+
+    with sky.Dag() as dag:
+        dag.add(task)
+
     maybe_status, _ = backend_utils.refresh_cluster_status_handle(cluster)
     if maybe_status is None:
         # Show the optimize log before the prompt if the cluster does not exist.
@@ -754,7 +763,9 @@ def _launch_with_confirm(
             if onprem_utils.check_if_local_cloud(cluster):
                 prompt = f'Initializing local cluster{cluster_str}. Proceed?'
             else:
-                prompt = f'Launching a new cluster{cluster_str}. Proceed?'
+                prompt = (
+                    f'Launching a new cluster{cluster_str}{clone_source_str}. '
+                    'Proceed?')
         elif maybe_status == status_lib.ClusterStatus.STOPPED:
             prompt = f'Restarting the stopped cluster {cluster!r}. Proceed?'
         if prompt is not None:
@@ -790,6 +801,7 @@ def _launch_with_confirm(
             down=down,
             retry_until_up=retry_until_up,
             no_setup=no_setup,
+            clone_disk_from=clone_disk_from,
         )
 
 
@@ -1257,6 +1269,15 @@ def cli():
               default=False,
               required=False,
               help='Skip setup phase when (re-)launching cluster.')
+@click.option(
+    '--clone-disk-from',
+    '--clone',
+    default=None,
+    type=str,
+    **_get_shell_complete_args(_complete_cluster_name),
+    help=('[Experimental] Clone disk from an existing cluster to launch '
+          'a new one. This is useful when the new cluster needs to have '
+          'the same data on the boot disk as an existing cluster.'))
 @usage_lib.entrypoint
 def launch(
     entrypoint: List[str],
@@ -1285,6 +1306,7 @@ def launch(
     retry_until_up: bool,
     yes: bool,
     no_setup: bool,
+    clone_disk_from: Optional[str],
 ):
     # 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).
@@ -1354,7 +1376,8 @@ def launch(
                          idle_minutes_to_autostop=idle_minutes_to_autostop,
                          down=down,
                          retry_until_up=retry_until_up,
-                         no_setup=no_setup)
+                         no_setup=no_setup,
+                         clone_disk_from=clone_disk_from)
 
 
 @cli.command(cls=_DocumentedCodeCommand)
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index c32cdfbbca6..d98fb20cc4a 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -3,7 +3,6 @@
 import functools
 import json
 import os
-import random
 import re
 import subprocess
 import time
@@ -16,8 +15,9 @@
 from sky import status_lib
 from sky.adaptors import aws
 from sky.clouds import service_catalog
-from sky.skylet import log_lib
 from sky.utils import common_utils
+from sky.utils import log_utils
+from sky.utils import subprocess_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -142,7 +142,7 @@ def regions_with_offering(cls, instance_type: str,
             for r in regions:
                 assert r.zones is not None, r
                 r.set_zones([z for z in r.zones if z.name == zone])
-            regions = [r for r in regions if r.zones]
+        regions = [r for r in regions if r.zones]
         return regions
 
     @classmethod
@@ -695,6 +695,26 @@ def _get_disk_specs(cls, disk_tier: Optional[str]) -> Dict[str, Any]:
             'custom_disk_perf': tier != 'low',
         }
 
+    @classmethod
+    def _query_instance_property_and_retry(
+        cls,
+        tag_filters: Dict[str, str],
+        region: str,
+        query: str,
+    ) -> Tuple[int, str, str]:
+        filter_str = ' '.join(f'Name=tag:{key},Values={value}'
+                              for key, value in tag_filters.items())
+        query_cmd = (f'aws ec2 describe-instances --filters {filter_str} '
+                     f'--region {region} --query {query} --output json')
+        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+            query_cmd,
+            retry_returncode=[255],
+            retry_stderrs=[
+                'Unable to locate credentials. You can configure credentials by '
+                'running "aws configure"'
+            ])
+        return returncode, stdout, stderr
+
     @classmethod
     def query_status(cls, name: str, tag_filters: Dict[str, str],
                      region: Optional[str], zone: Optional[str],
@@ -711,33 +731,9 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
             'terminated': None,
         }
 
-        fitler_str = ' '.join(f'Name=tag:{key},Values={value}'
-                              for key, value in tag_filters.items())
-
-        retry_cnt = 0
-        while retry_cnt < 3:
-            query_cmd = ('aws ec2 describe-instances --filters '
-                         f'{fitler_str} '
-                         f'--region {region} '
-                         '--query "Reservations[].Instances[].State.Name" '
-                         '--output json')
-
-            returncode, stdout, stderr = log_lib.run_with_log(
-                query_cmd, '/dev/null', require_outputs=True, shell=True)
-            logger.debug(f'{query_cmd} returned {returncode}.\n'
-                         '**** STDOUT ****\n'
-                         f'{stdout}\n'
-                         '**** STDERR ****\n'
-                         f'{stderr}')
-
-            if (returncode != 0 and
-                    'Unable to locate credentials. You can configure credentials by '
-                    'running "aws configure"' in stdout + stderr):
-                retry_cnt += 1
-                time.sleep(random.uniform(0, 1) * 2)
-                continue
-            else:
-                break
+        assert region is not None, (tag_filters, region)
+        returncode, stdout, stderr = cls._query_instance_property_and_retry(
+            tag_filters, region, query='Reservations[].Instances[].State.Name')
 
         if returncode != 0:
             with ux_utils.print_exception_no_traceback():
@@ -753,3 +749,134 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
             if node_status is not None:
                 statuses.append(node_status)
         return statuses
+
+    @classmethod
+    def create_image_from_cluster(cls, cluster_name: str,
+                                  tag_filters: Dict[str,
+                                                    str], region: Optional[str],
+                                  zone: Optional[str]) -> str:
+        del zone  # unused
+        assert region is not None, (tag_filters, region)
+        image_name = f'skypilot-{cluster_name}-{int(time.time())}'
+        returncode, stdout, stderr = cls._query_instance_property_and_retry(
+            tag_filters, region, query='Reservations[].Instances[].InstanceId')
+
+        subprocess_utils.handle_returncode(
+            returncode,
+            '',
+            error_msg='Failed to find the source cluster on AWS.',
+            stderr=stderr,
+            stream_logs=False)
+
+        instance_ids = json.loads(stdout.strip())
+        if len(instance_ids) != 1:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    f'More than one instance found: '
+                    f'{stdout + stderr}')
+
+        instance_id = instance_ids[0]
+        create_image_cmd = (
+            f'aws ec2 create-image --region {region} --instance-id {instance_id} '
+            f'--name {image_name} --output text')
+        returncode, image_id, stderr = subprocess_utils.run_and_retry(
+            create_image_cmd,
+            retry_returncode=[255],
+        )
+        image_id = image_id.strip()
+        subprocess_utils.handle_returncode(
+            returncode,
+            create_image_cmd,
+            error_msg=
+            f'Failed to create image from the source instance {instance_id}.',
+            stderr=stderr,
+            stream_logs=True)
+
+        log_utils.force_update_rich_status(
+            f'Waiting for the source image {cluster_name!r} from {region} to be available on AWS.'
+        )
+        # Wait for the image to be available
+        wait_image_cmd = (
+            f'aws ec2 wait image-available --region {region} --image-ids {image_id}'
+        )
+        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+            wait_image_cmd,
+            retry_returncode=[255],
+        )
+        subprocess_utils.handle_returncode(
+            returncode,
+            wait_image_cmd,
+            error_msg=
+            f'The source image {image_id!r} creation fails to complete.',
+            stderr=stderr,
+            stream_logs=True)
+        sky_logging.print(
+            f'The source image {image_id!r} is created successfully.')
+        return image_id
+
+    @classmethod
+    def maybe_move_image(cls, image_id: str, source_region: str,
+                         target_region: str, source_zone: Optional[str],
+                         target_zone: Optional[str]) -> str:
+        del source_zone, target_zone  # unused
+        if source_region == target_region:
+            return image_id
+        image_name = f'skypilot-cloned-from-{source_region}-{image_id}'
+        copy_image_cmd = (f'aws ec2 copy-image --name {image_name} '
+                          f'--source-image-id {image_id} '
+                          f'--source-region {source_region} '
+                          f'--region {target_region} --output text')
+        returncode, target_image_id, stderr = subprocess_utils.run_and_retry(
+            copy_image_cmd,
+            retry_returncode=[255],
+        )
+        target_image_id = target_image_id.strip()
+        subprocess_utils.handle_returncode(
+            returncode,
+            copy_image_cmd,
+            error_msg=
+            f'Failed to copy image {image_id!r} from {source_region} to {target_region}.',
+            stderr=stderr,
+            stream_logs=True)
+
+        log_utils.force_update_rich_status(
+            f'Waiting for the target image {target_image_id!r} on {target_region} to be '
+            'available on AWS.')
+        wait_image_cmd = (
+            f'aws ec2 wait image-available --region {target_region} '
+            f'--image-ids {target_image_id}')
+        subprocess_utils.run_and_retry(
+            wait_image_cmd,
+            max_retry=5,
+            retry_returncode=[255],
+        )
+        subprocess_utils.handle_returncode(
+            returncode,
+            wait_image_cmd,
+            error_msg=
+            f'The target image {target_image_id!r} creation fails to complete.',
+            stderr=stderr,
+            stream_logs=True)
+
+        sky_logging.print(
+            f'The target image {target_image_id!r} is created successfully.')
+
+        log_utils.force_update_rich_status('Deleting the source image.')
+        cls.delete_image(image_id, source_region)
+        return target_image_id
+
+    @classmethod
+    def delete_image(cls, image_id: str, region: Optional[str]) -> None:
+        assert region is not None, (image_id, region)
+        delete_image_cmd = (f'aws ec2 deregister-image --region {region} '
+                            f'--image-id {image_id}')
+        returncode, _, stderr = subprocess_utils.run_and_retry(
+            delete_image_cmd,
+            retry_returncode=[255],
+        )
+        subprocess_utils.handle_returncode(
+            returncode,
+            delete_image_cmd,
+            error_msg=f'Failed to delete image {image_id!r} on {region}.',
+            stderr=stderr,
+            stream_logs=True)
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index b9db7ff7a99..9324b7e297d 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -59,7 +59,9 @@ class Azure(clouds.Cloud):
     @classmethod
     def _cloud_unsupported_features(
             cls) -> Dict[clouds.CloudImplementationFeatures, str]:
-        return dict()
+        return {
+            clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER: f'Migrating disk is not supported in {cls._REPR}.',
+        }
 
     @classmethod
     def _max_cluster_name_length(cls) -> int:
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index 4b2369378cb..cfd8d3d4dd9 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -27,6 +27,7 @@ class CloudImplementationFeatures(enum.Enum):
     STOP = 'stop'
     AUTOSTOP = 'autostop'
     MULTI_NODE = 'multi-node'
+    CLONE_DISK_FROM_CLUSTER = 'clone_disk_from_cluster'
 
 
 class Region(collections.namedtuple('Region', ['name'])):
@@ -485,5 +486,45 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
         """
         raise NotImplementedError
 
+    # === Image related methods ===
+    # These three tmethods are used to create, move and delete images. They
+    # are currently only used in `sky launch --clone-disk-from` to clone a
+    # cluster's disk to launch a new cluster.
+    # It is not required to implement these methods for clouds that do not
+    # support `--clone-disk-from`. If not implemented,
+    # CloudImplementationFeatures.CLONE_DISK_FROM should be added to the
+    # cloud._cloud_unsupported_features().
+
+    @classmethod
+    def create_image_from_cluster(cls, cluster_name: str,
+                                  tag_filters: Dict[str,
+                                                    str], region: Optional[str],
+                                  zone: Optional[str]) -> str:
+        """Creates an image from the cluster.
+
+        Returns: the image ID.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def maybe_move_image(cls, image_name: str, source_region: str,
+                         target_region: str, source_zone: Optional[str],
+                         target_zone: Optional[str]) -> str:
+        """Move an image if required.
+
+        If the image cannot be accessed in the target region, move the image
+        from the source region to the target region.
+
+        Returns: the image ID in the target region.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def delete_image(cls, image_id: str, region: Optional[str]) -> None:
+        """Deletes the image with image_id in the region."""
+        raise NotImplementedError
+
+    # === End of image related methods ===
+
     def __repr__(self):
         return self._REPR
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index 16c3027e12e..cdf06523387 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -15,6 +15,7 @@
 from sky.clouds import service_catalog
 from sky.skylet import log_lib
 from sky.utils import common_utils
+from sky.utils import subprocess_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -108,7 +109,7 @@ class GCP(clouds.Cloud):
     @classmethod
     def _cloud_unsupported_features(
             cls) -> Dict[clouds.CloudImplementationFeatures, str]:
-        return dict()
+        return {}
 
     @classmethod
     def _max_cluster_name_length(cls) -> Optional[int]:
@@ -737,6 +738,10 @@ def _get_disk_type(cls, disk_tier: Optional[str]) -> str:
         }
         return tier2name[tier]
 
+    @classmethod
+    def _label_filter_str(cls, tag_filters: Dict[str, str]) -> str:
+        return ' '.join(f'labels.{k}={v}' for k, v in tag_filters.items())
+
     @classmethod
     def query_status(cls, name: str, tag_filters: Dict[str, str],
                      region: Optional[str], zone: Optional[str],
@@ -747,8 +752,7 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
         from sky.utils import tpu_utils  # pylint: disable=import-outside-toplevel
         use_tpu_vm = kwargs.pop('use_tpu_vm', False)
 
-        label_filter_str = ' '.join(
-            f'labels.{k}={v}' for k, v in tag_filters.items())
+        label_filter_str = cls._label_filter_str(tag_filters)
         if use_tpu_vm:
             # TPU VM's state definition is different from compute VM
             # https://cloud.google.com/tpu/docs/reference/rest/v2alpha1/projects.locations.nodes#State # pylint: disable=line-too-long
@@ -813,3 +817,91 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
             status_list.append(status)
 
         return status_list
+
+    @classmethod
+    def create_image_from_cluster(cls, cluster_name: str,
+                                  tag_filters: Dict[str,
+                                                    str], region: Optional[str],
+                                  zone: Optional[str]) -> str:
+        del region  # unused
+        assert zone is not None
+        label_filter_str = cls._label_filter_str(tag_filters)
+        instance_name_cmd = ('gcloud compute instances list '
+                             f'--filter="({label_filter_str})" '
+                             '--format="json(name)"')
+        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+            instance_name_cmd,
+            retry_returncode=[255],
+        )
+        subprocess_utils.handle_returncode(
+            returncode,
+            instance_name_cmd,
+            error_msg=f'Failed to get instance name for {cluster_name!r}',
+            stderr=stderr,
+            stream_logs=True)
+        instance_names = json.loads(stdout)
+        if len(instance_names) != 1:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    'Only support creating image from single '
+                    f' instance, but got multiple instances: {instance_names}')
+        instance_name = instance_names[0]['name']
+
+        image_name = f'skypilot-{cluster_name}-{int(time.time())}'
+        create_image_cmd = (f'gcloud compute images create {image_name} '
+                            f'--source-disk  {instance_name} '
+                            f'--source-disk-zone {zone}')
+        logger.debug(create_image_cmd)
+        subprocess_utils.run_and_retry(
+            create_image_cmd,
+            retry_returncode=[255],
+        )
+        subprocess_utils.handle_returncode(
+            returncode,
+            create_image_cmd,
+            error_msg=f'Failed to create image for {cluster_name!r}',
+            stderr=stderr,
+            stream_logs=True)
+
+        image_uri_cmd = (f'gcloud compute images describe {image_name} '
+                         '--format="get(selfLink)"')
+        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+            image_uri_cmd,
+            retry_returncode=[255],
+        )
+
+        subprocess_utils.handle_returncode(
+            returncode,
+            image_uri_cmd,
+            error_msg=f'Failed to get image uri for {cluster_name!r}',
+            stderr=stderr,
+            stream_logs=True)
+
+        image_uri = stdout.strip()
+        image_id = image_uri.partition('projects/')[2]
+        image_id = 'projects/' + image_id
+        return image_id
+
+    @classmethod
+    def maybe_move_image(cls, image_id: str, source_region: str,
+                         target_region: str, source_zone: Optional[str],
+                         target_zone: Optional[str]) -> str:
+        del source_region, target_region, source_zone, target_zone  # Unused.
+        # GCP images are global, so no need to move.
+        return image_id
+
+    @classmethod
+    def delete_image(cls, image_id: str, region: Optional[str]) -> None:
+        del region  # Unused.
+        image_name = image_id.rpartition('/')[2]
+        delete_image_cmd = f'gcloud compute images delete {image_name} --quiet'
+        returncode, _, stderr = subprocess_utils.run_and_retry(
+            delete_image_cmd,
+            retry_returncode=[255],
+        )
+        subprocess_utils.handle_returncode(
+            returncode,
+            delete_image_cmd,
+            error_msg=f'Failed to delete image {image_name!r}',
+            stderr=stderr,
+            stream_logs=True)
diff --git a/sky/clouds/ibm.py b/sky/clouds/ibm.py
index a3662d26638..90b709c0589 100644
--- a/sky/clouds/ibm.py
+++ b/sky/clouds/ibm.py
@@ -35,7 +35,10 @@ class IBM(clouds.Cloud):
     @classmethod
     def _cloud_unsupported_features(
             cls) -> Dict[clouds.CloudImplementationFeatures, str]:
-        return dict()
+        return {
+            clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER:
+                (f'Migrating disk is not supported in {cls._REPR}.'),
+        }
 
     @classmethod
     def _max_cluster_name_length(cls) -> Optional[int]:
diff --git a/sky/clouds/lambda_cloud.py b/sky/clouds/lambda_cloud.py
index a82c972dc97..2bf6e484d06 100644
--- a/sky/clouds/lambda_cloud.py
+++ b/sky/clouds/lambda_cloud.py
@@ -35,6 +35,7 @@ class Lambda(clouds.Cloud):
     _CLOUD_UNSUPPORTED_FEATURES = {
         clouds.CloudImplementationFeatures.STOP: 'Lambda cloud does not support stopping VMs.',
         clouds.CloudImplementationFeatures.AUTOSTOP: 'Lambda cloud does not support stopping VMs.',
+        clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER: f'Migrating disk is not supported in {_REPR}.',
     }
 
     @classmethod
diff --git a/sky/clouds/local.py b/sky/clouds/local.py
index 898cec9b8c1..f7881933b75 100644
--- a/sky/clouds/local.py
+++ b/sky/clouds/local.py
@@ -31,7 +31,9 @@ class Local(clouds.Cloud):
         clouds.CloudImplementationFeatures.STOP:
             ('Local cloud does not support stopping instances.'),
         clouds.CloudImplementationFeatures.AUTOSTOP:
-            ('Local cloud does not support stopping instances.')
+            ('Local cloud does not support stopping instances.'),
+        clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER:
+            ('Migrating disk is not supported for Local.'),
     }
 
     @classmethod
diff --git a/sky/clouds/oci.py b/sky/clouds/oci.py
index 546fe2446e5..2fe28d21825 100644
--- a/sky/clouds/oci.py
+++ b/sky/clouds/oci.py
@@ -45,7 +45,10 @@ class OCI(clouds.Cloud):
     @classmethod
     def _cloud_unsupported_features(
             cls) -> Dict[clouds.CloudImplementationFeatures, str]:
-        return dict()
+        return {
+            clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER:
+                (f'Migrating disk is not supported in {cls._REPR}.'),
+        }
 
     @classmethod
     def _max_cluster_name_length(cls) -> Optional[int]:
diff --git a/sky/clouds/scp.py b/sky/clouds/scp.py
index feb8b0f75d3..263b392a454 100644
--- a/sky/clouds/scp.py
+++ b/sky/clouds/scp.py
@@ -42,7 +42,9 @@ class SCP(clouds.Cloud):
     # MULTI_NODE: Multi-node is not supported by the implementation yet.
     _MULTI_NODE = 'Multi-node is not supported by the SCP Cloud yet.'
     _CLOUD_UNSUPPORTED_FEATURES = {
-        clouds.CloudImplementationFeatures.MULTI_NODE: _MULTI_NODE
+        clouds.CloudImplementationFeatures.MULTI_NODE: _MULTI_NODE,
+        clouds.CloudImplementationFeatures.CLONE_DISK_FROM_CLUSTER:
+            (f'Migrating disk is not supported in {_REPR}.'),
     }
 
     _INDENT_PREFIX = '    '
diff --git a/sky/execution.py b/sky/execution.py
index b12e97effb3..3d5cbb64deb 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -40,6 +40,7 @@
 from sky.skylet import constants
 from sky.utils import common_utils
 from sky.utils import dag_utils
+from sky.utils import log_utils
 from sky.utils import env_options, timeline
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -96,6 +97,7 @@ def _convert_to_dag(entrypoint: Any) -> 'sky.Dag':
 class Stage(enum.Enum):
     """Stages for a run of a sky.Task."""
     # TODO: rename actual methods to be consistent.
+    CLONE_DISK = enum.auto()
     OPTIMIZE = enum.auto()
     PROVISION = enum.auto()
     SYNC_WORKDIR = enum.auto()
@@ -106,6 +108,55 @@ class Stage(enum.Enum):
     DOWN = enum.auto()
 
 
+def _maybe_clone_disk_from_cluster(clone_disk_from: Optional[str],
+                                   cluster_name: Optional[str],
+                                   task: 'sky.Task') -> 'sky.Task':
+    if clone_disk_from is None:
+        return task
+    task, handle = backend_utils.check_can_clone_disk_and_override_task(
+        clone_disk_from, cluster_name, task)
+    original_cloud = handle.launched_resources.cloud
+    assert original_cloud is not None, handle.launched_resources
+    task_resources = list(task.resources)[0]
+
+    with log_utils.safe_rich_status('Creating image from source cluster '
+                                    f'{clone_disk_from!r}'):
+        image_id = original_cloud.create_image_from_cluster(
+            clone_disk_from,
+            backend_utils.tag_filter_for_cluster(clone_disk_from),
+            region=handle.launched_resources.region,
+            zone=handle.launched_resources.zone,
+        )
+        log_utils.force_update_rich_status(
+            f'Migrating image {image_id} to target region '
+            f'{task_resources.region}...')
+        source_region = handle.launched_resources.region
+        target_region = task_resources.region
+        assert source_region is not None, handle.launched_resources
+        assert target_region is not None, task_resources
+
+        image_id = original_cloud.maybe_move_image(
+            image_id,
+            source_region=source_region,
+            target_region=target_region,
+            source_zone=handle.launched_resources.zone,
+            target_zone=task_resources.zone,
+        )
+    logger.info(
+        f'{colorama.Fore.GREEN}'
+        f'Successfully created image {image_id!r} for {clone_disk_from!r} '
+        f'on {original_cloud}.{colorama.Style.RESET_ALL}\n'
+        'Overriding task\'s image_id.')
+    task_resources = task_resources.copy(image_id=image_id,
+                                         _is_image_managed=True)
+    task.set_resources(task_resources)
+    # Set the best_resources to None to trigger a re-optimization, so that
+    # the new task_resources is used.
+    task.best_resources = None
+    logger.debug(f'Overridden task resources: {task.resources}')
+    return task
+
+
 def _execute(
     entrypoint: Union['sky.Task', 'sky.Dag'],
     dryrun: bool = False,
@@ -121,6 +172,7 @@ def _execute(
     detach_run: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
     no_setup: bool = False,
+    clone_disk_from: Optional[str] = None,
     # Internal only:
     # pylint: disable=invalid-name
     _is_launched_by_spot_controller: bool = False,
@@ -228,31 +280,34 @@ def _execute(
                 f'Backend {backend.NAME} does not support autostop, please try '
                 f'{backends.CloudVmRayBackend.NAME}')
 
-    if not cluster_exists:
-        if (Stage.PROVISION in stages and task.use_spot and
-                not _is_launched_by_spot_controller):
-            yellow = colorama.Fore.YELLOW
-            bold = colorama.Style.BRIGHT
-            reset = colorama.Style.RESET_ALL
-            logger.info(
-                f'{yellow}Launching an unmanaged spot task, which does not '
-                f'automatically recover from preemptions.{reset}\n{yellow}To '
-                'get automatic recovery, use managed spot instead: '
-                f'{reset}{bold}sky spot launch{reset} {yellow}or{reset} '
-                f'{bold}sky.spot_launch(){reset}.')
-
-        if Stage.OPTIMIZE in stages:
-            if task.best_resources is None:
-                # TODO: fix this for the situation where number of requested
-                # accelerators is not an integer.
-                if isinstance(backend, backends.CloudVmRayBackend):
-                    # TODO: adding this check because docker backend on a
-                    # no-credential machine should not enter optimize(), which
-                    # would directly error out ('No cloud is enabled...').  Fix
-                    # by moving `sky check` checks out of optimize()?
-                    dag = sky.optimize(dag, minimize=optimize_target)
-                    task = dag.tasks[0]  # Keep: dag may have been deep-copied.
-                    assert task.best_resources is not None, task
+    if (Stage.PROVISION in stages and task.use_spot and
+            not _is_launched_by_spot_controller and not cluster_exists):
+        yellow = colorama.Fore.YELLOW
+        bold = colorama.Style.BRIGHT
+        reset = colorama.Style.RESET_ALL
+        logger.info(
+            f'{yellow}Launching an unmanaged spot task, which does not '
+            f'automatically recover from preemptions.{reset}\n{yellow}To '
+            'get automatic recovery, use managed spot instead: '
+            f'{reset}{bold}sky spot launch{reset} {yellow}or{reset} '
+            f'{bold}sky.spot_launch(){reset}.')
+
+    if Stage.CLONE_DISK in stages:
+        task = _maybe_clone_disk_from_cluster(clone_disk_from, cluster_name,
+                                              task)
+
+    if Stage.OPTIMIZE in stages and not cluster_exists:
+        if task.best_resources is None:
+            # TODO: fix this for the situation where number of requested
+            # accelerators is not an integer.
+            if isinstance(backend, backends.CloudVmRayBackend):
+                # TODO: adding this check because docker backend on a
+                # no-credential machine should not enter optimize(), which
+                # would directly error out ('No cloud is enabled...').  Fix
+                # by moving `sky check` checks out of optimize()?
+                dag = sky.optimize(dag, minimize=optimize_target)
+                task = dag.tasks[0]  # Keep: dag may have been deep-copied.
+                assert task.best_resources is not None, task
 
     backend.register_info(dag=dag,
                           optimize_target=optimize_target,
@@ -263,7 +318,6 @@ def _execute(
         task.sync_storage_mounts()
 
     try:
-
         if Stage.PROVISION in stages:
             if handle is None:
                 handle = backend.provision(task,
@@ -341,6 +395,7 @@ def launch(
     detach_setup: bool = False,
     detach_run: bool = False,
     no_setup: bool = False,
+    clone_disk_from: Optional[str] = None,
     # Internal only:
     # pylint: disable=invalid-name
     _is_launched_by_spot_controller: bool = False,
@@ -390,6 +445,9 @@ def launch(
         detach_run: If True, as soon as a job is submitted, return from this
             function and do not stream execution logs.
         no_setup: if True, do not re-run setup commands.
+        clone_disk_from: [Experimental] if set, clone the disk from the
+            specified cluster. This is useful to migrate the cluster to a
+            different availability zone or region.
 
     Example:
         .. code-block:: python
@@ -424,6 +482,7 @@ def launch(
     entrypoint = task
     backend_utils.check_cluster_name_not_reserved(cluster_name,
                                                   operation_str='sky.launch')
+
     _execute(
         entrypoint=entrypoint,
         dryrun=dryrun,
@@ -438,6 +497,7 @@ def launch(
         detach_run=detach_run,
         idle_minutes_to_autostop=idle_minutes_to_autostop,
         no_setup=no_setup,
+        clone_disk_from=clone_disk_from,
         _is_launched_by_spot_controller=_is_launched_by_spot_controller,
     )
 
diff --git a/sky/resources.py b/sky/resources.py
index 53d76462888..78875b5f2c9 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -21,33 +21,19 @@
 
 
 class Resources:
-    """A cloud resource bundle.
+    """Resources: compute requirements of Tasks.
 
-    Used
-      * for representing resource requests for tasks/apps
-      * as a "filter" to get concrete launchable instances
-      * for calculating billing
-      * for provisioning on a cloud
+    Used:
 
-    Examples:
+    * for representing resource requests for tasks/apps
+    * as a "filter" to get concrete launchable instances
+    * for calculating billing
+    * for provisioning on a cloud
 
-        # Fully specified cloud and instance type (is_launchable() is True).
-        sky.Resources(clouds.AWS(), 'p3.2xlarge')
-        sky.Resources(clouds.GCP(), 'n1-standard-16')
-        sky.Resources(clouds.GCP(), 'n1-standard-8', 'V100')
-
-        # Specifying required resources; the system decides the cloud/instance
-        # type. The below are equivalent:
-        sky.Resources(accelerators='V100')
-        sky.Resources(accelerators='V100:1')
-        sky.Resources(accelerators={'V100': 1})
-
-        # TODO:
-        sky.Resources(requests={'mem': '16g', 'cpu': 8})
     """
     # If any fields changed, increment the version. For backward compatibility,
     # modify the __setstate__ method to handle the old version.
-    _VERSION = 9
+    _VERSION = 10
 
     def __init__(
         self,
@@ -59,12 +45,78 @@ def __init__(
         accelerator_args: Optional[Dict[str, str]] = None,
         use_spot: Optional[bool] = None,
         spot_recovery: Optional[str] = None,
-        disk_size: Optional[int] = None,
         region: Optional[str] = None,
         zone: Optional[str] = None,
         image_id: Union[Dict[str, str], str, None] = None,
+        disk_size: Optional[int] = None,
         disk_tier: Optional[Literal['high', 'medium', 'low']] = None,
+        # Internal use only.
+        _is_image_managed: Optional[bool] = None,
     ):
+        """Initialize a Resources object.
+
+        All fields are optional.  ``Resources.is_launchable`` decides whether
+        the Resources is fully specified to launch an instance.
+
+
+        Examples:
+          .. code-block:: python
+
+            # Fully specified cloud and instance type (is_launchable() is True).
+            sky.Resources(clouds.AWS(), 'p3.2xlarge')
+            sky.Resources(clouds.GCP(), 'n1-standard-16')
+            sky.Resources(clouds.GCP(), 'n1-standard-8', 'V100')
+
+            # Specifying required resources; the system decides the
+            cloud/instance
+            # type. The below are equivalent:
+            sky.Resources(accelerators='V100')
+            sky.Resources(accelerators='V100:1')
+            sky.Resources(accelerators={'V100': 1})
+            sky.Resources(cpus='2+', memory='16+', accelerators='V100')
+
+        Args:
+          cloud: the cloud to use.
+          instance_type: the instance type to use.
+          cpus: the number of CPUs required for the task.
+            If a str, must be a string of the form ``'2'`` or ``'2+'``, where
+            the ``+`` indicates that the task requires at least 2 CPUs.
+          memory: the amount of memory in GiB required. If a
+            str, must be a string of the form ``'16'`` or ``'16+'``, where
+            the ``+`` indicates that the task requires at least 16 GB of memory.
+          accelerators: the accelerators required. If a str, must be
+            a string of the form ``'V100'`` or ``'V100:2'``, where the ``:2``
+            indicates that the task requires 2 V100 GPUs. If a dict, must be a
+            dict of the form ``{'V100': 2}`` or ``{'tpu-v2-8': 1}``.
+          accelerator_args: accelerator-specific arguments. For example,
+            ``{'tpu_vm': True, 'runtime_version': 'tpu-vm-base'}`` for TPUs.
+          use_spot: whether to use spot instances. If None, defaults to
+            False.
+          spot_recovery: the spot recovery strategy to use for the managed
+            spot to recover the cluster from preemption. Refer to
+            `recovery_strategy module <https://github.com/skypilot-org/skypilot/blob/master/sky/spot/recovery_strategy.py>`__ # pylint: disable=line-too-long
+            for more details.
+          region: the region to use.
+          zone: the zone to use.
+          image_id: the image ID to use. If a str, must be a string
+            of the image id from the cloud, such as AWS:
+            ``'ami-1234567890abcdef0'``, GCP:
+            ``'projects/my-project-id/global/images/my-image-name'``;
+            Or, a image tag provided by SkyPilot, such as AWS:
+            ``'skypilot:gpu-ubuntu-2004'``. If a dict, must be a dict mapping
+            from region to image ID, such as:
+
+            .. code-block:: python
+
+              {
+                'us-west1': 'ami-1234567890abcdef0',
+                'us-east1': 'ami-1234567890abcdef0'
+              }
+
+          disk_size: the size of the OS disk in GiB.
+          disk_tier: the disk performance tier to use. If None, defaults to
+            ``'medium'``.
+        """
         self._version = self._VERSION
         self._cloud = cloud
         self._region: Optional[str] = None
@@ -100,6 +152,7 @@ def __init__(
                 self._image_id = {
                     k.strip(): v.strip() for k, v in image_id.items()
                 }
+        self._is_image_managed = _is_image_managed
 
         self._disk_tier = disk_tier
 
@@ -285,6 +338,10 @@ def image_id(self) -> Optional[Dict[str, str]]:
     def disk_tier(self) -> str:
         return self._disk_tier
 
+    @property
+    def is_image_managed(self) -> Optional[str]:
+        return self._is_image_managed
+
     def _set_cpus(
         self,
         cpus: Union[None, int, float, str],
@@ -867,6 +924,8 @@ def copy(self, **override) -> 'Resources':
             zone=override.pop('zone', self.zone),
             image_id=override.pop('image_id', self.image_id),
             disk_tier=override.pop('disk_tier', self.disk_tier),
+            _is_image_managed=override.pop('_is_image_managed',
+                                           self._is_image_managed),
         )
         assert len(override) == 0
         return resources
@@ -919,6 +978,9 @@ def from_yaml_config(cls, config: Optional[Dict[str, str]]) -> 'Resources':
             resources_fields['image_id'] = config.pop('image_id')
         if config.get('disk_tier') is not None:
             resources_fields['disk_tier'] = config.pop('disk_tier')
+        if config.get('_is_image_managed') is not None:
+            resources_fields['_is_image_managed'] = config.pop(
+                '_is_image_managed')
 
         assert not config, f'Invalid resource args: {config.keys()}'
         return Resources(**resources_fields)
@@ -946,6 +1008,8 @@ def add_if_not_none(key, value):
         add_if_not_none('zone', self.zone)
         add_if_not_none('image_id', self.image_id)
         add_if_not_none('disk_tier', self.disk_tier)
+        if self._is_image_managed is not None:
+            config['_is_image_managed'] = self._is_image_managed
         return config
 
     def __setstate__(self, state):
@@ -1008,4 +1072,7 @@ def __setstate__(self, state):
         if version < 9:
             self._disk_tier = None
 
+        if version < 10:
+            self._is_image_managed = None
+
         self.__dict__.update(state)
diff --git a/sky/utils/subprocess_utils.py b/sky/utils/subprocess_utils.py
index 8e39254b73a..e2c2684d976 100644
--- a/sky/utils/subprocess_utils.py
+++ b/sky/utils/subprocess_utils.py
@@ -1,13 +1,16 @@
 """Utility functions for subprocesses."""
 from multiprocessing import pool
 import psutil
+import random
 import subprocess
-from typing import Any, Callable, List, Optional, Union
+import time
+from typing import Any, Callable, List, Optional, Tuple, Union
 
 import colorama
 
 from sky import exceptions
 from sky import sky_logging
+from sky.skylet import log_lib
 from sky.utils import timeline
 from sky.utils import ux_utils
 
@@ -117,3 +120,49 @@ def kill_children_processes(first_pid_to_kill: Optional[int] = None,
         except psutil.NoSuchProcess:
             # The child process may have already been terminated.
             pass
+
+
+def run_and_retry(
+        cmd: str,
+        max_retry: int = 3,
+        retry_returncode: Optional[List[int]] = None,
+        retry_stderrs: Optional[List[str]] = None) -> Tuple[int, str, str]:
+    """Run a command and retry if it fails due to the specified reasons.
+
+    Args:
+        cmd: The command to run.
+        max_retry: The maximum number of retries.
+        retry_returncode: The returncodes that should be retried.
+        retry_stderr: The cmd needs to be retried if the stderr contains any of
+            the strings in this list.
+
+    Returns:
+        The returncode, stdout, and stderr of the command.
+    """
+    retry_cnt = 0
+    while True:
+        returncode, stdout, stderr = log_lib.run_with_log(cmd,
+                                                          '/dev/null',
+                                                          require_outputs=True,
+                                                          shell=True)
+        if retry_cnt < max_retry:
+            if (retry_returncode is not None and
+                    returncode in retry_returncode):
+                retry_cnt += 1
+                time.sleep(random.uniform(0, 1) * 2)
+                continue
+
+            if retry_stderrs is None:
+                break
+
+            need_retry = False
+            for retry_err in retry_stderrs:
+                if retry_err in stderr:
+                    retry_cnt += 1
+                    time.sleep(random.uniform(0, 1) * 2)
+                    need_retry = True
+                    break
+            if need_retry:
+                continue
+        break
+    return returncode, stdout, stderr


Code review comments:
diff_hunk: @@ -1704,6 +1711,89 @@ def _query_cluster_status_via_cloud_api(
     return node_statuses
 
 
+def check_can_clone_disk_and_override_task(
+    cluster_name: str, target_cluster_name: Optional[str], task: 'task_lib.Task'
+) -> Tuple['task_lib.Task', 'cloud_vm_ray_backend.CloudVmRayResourceHandle']:
+    """Check if the task is compatible to clone disk from the source cluster.
+
+    Args:
+        cluster_name: The name of the cluster to clone disk from.
+        target_cluster_name: The name of the target cluster.
+        task: The task to check.
+
+    Returns:
+        The task to use and the resource handle of the source cluster.
+
+    Raises:
+        exceptions.NotSupportedError: If the source cluster is not valid or the
path: sky/backends/backend_utils.py
text: @user1:
L1734 raises ValueError?

@author:
Added. Thanks!

diff_hunk: @@ -1704,6 +1711,89 @@ def _query_cluster_status_via_cloud_api(
     return node_statuses
 
 
+def check_can_clone_disk_and_override_task(
+    cluster_name: str, target_cluster_name: Optional[str], task: 'task_lib.Task'
+) -> Tuple['task_lib.Task', 'cloud_vm_ray_backend.CloudVmRayResourceHandle']:
+    """Check if the task is compatible to clone disk from the source cluster.
+
+    Args:
+        cluster_name: The name of the cluster to clone disk from.
+        target_cluster_name: The name of the target cluster.
+        task: The task to check.
+
+    Returns:
+        The task to use and the resource handle of the source cluster.
+
+    Raises:
+        exceptions.NotSupportedError: If the source cluster is not valid or the
+            task is not compatible to clone disk from the source cluster.
+    """
+    source_cluster_status, handle = refresh_cluster_status_handle(cluster_name)
+    if source_cluster_status is None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(
+                f'Cannot find cluster {cluster_name!r} to clone disk from.')
+
+    if not isinstance(handle, backends.CloudVmRayResourceHandle):
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from a non-cloud cluster {cluster_name!r}.')
+
+    if source_cluster_status != status_lib.ClusterStatus.STOPPED:
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from cluster {source_cluster_status!r}. Please stop the '
+                f'cluster first: sky stop {cluster_name}.')
path: sky/backends/backend_utils.py
text: @user1:
```suggestion
                f'cluster first: sky stop {cluster_name}')
```

@author:
Done. Thanks!

diff_hunk: @@ -1704,6 +1711,89 @@ def _query_cluster_status_via_cloud_api(
     return node_statuses
 
 
+def check_can_clone_disk_and_override_task(
+    cluster_name: str, target_cluster_name: Optional[str], task: 'task_lib.Task'
+) -> Tuple['task_lib.Task', 'cloud_vm_ray_backend.CloudVmRayResourceHandle']:
+    """Check if the task is compatible to clone disk from the source cluster.
+
+    Args:
+        cluster_name: The name of the cluster to clone disk from.
+        target_cluster_name: The name of the target cluster.
+        task: The task to check.
+
+    Returns:
+        The task to use and the resource handle of the source cluster.
+
+    Raises:
+        exceptions.NotSupportedError: If the source cluster is not valid or the
+            task is not compatible to clone disk from the source cluster.
+    """
+    source_cluster_status, handle = refresh_cluster_status_handle(cluster_name)
+    if source_cluster_status is None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(
+                f'Cannot find cluster {cluster_name!r} to clone disk from.')
+
+    if not isinstance(handle, backends.CloudVmRayResourceHandle):
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from a non-cloud cluster {cluster_name!r}.')
+
+    if source_cluster_status != status_lib.ClusterStatus.STOPPED:
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from cluster {source_cluster_status!r}. Please stop the '
+                f'cluster first: sky stop {cluster_name}.')
+
+    if target_cluster_name is not None:
+        target_cluster_status, _ = refresh_cluster_status_handle(
+            target_cluster_name)
+        if target_cluster_status is not None:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    f'The target cluster {target_cluster_name!r} already exists. Cloning '
+                    'disk is only supported when creating a new cluster.')
path: sky/backends/backend_utils.py
text: @user1:
```suggestion
                    'disk is only supported when creating a new cluster. To fix: specify a new target cluster name.')
```

@author:
Done. Thanks!

diff_hunk: @@ -3547,6 +3547,23 @@ def post_teardown_cleanup(self,
                             cluster_name=handle.cluster_name,
                             stdout=tpu_stdout,
                             stderr=tpu_stderr))
+        if handle.launched_resources.is_image_managed and terminate:
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
```suggestion
        if handle.launched_resources.is_image_managed is True and terminate:
```

@author:
Done. Thanks!

diff_hunk: @@ -285,6 +338,10 @@ def image_id(self) -> Optional[Dict[str, str]]:
     def disk_tier(self) -> str:
         return self._disk_tier
 
+    @property
+    def is_image_managed(self) -> Optional[str]:
path: sky/resources.py
text: @user1:
```suggestion
    def is_image_managed(self) -> Optional[bool]:
```

@author:
Good catch! Fixed. Thanks!

diff_hunk: @@ -3547,6 +3547,23 @@ def post_teardown_cleanup(self,
                             cluster_name=handle.cluster_name,
                             stdout=tpu_stdout,
                             stderr=tpu_stderr))
+        if handle.launched_resources.is_image_managed and terminate:
+            # Delete the image when terminating the cluster and the image is
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
```suggestion
            # Delete the image when terminating a "cloned" cluster, i.e., whose image is
```

@author:
Done. Thanks!

diff_hunk: @@ -485,5 +486,45 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
         """
         raise NotImplementedError
 
+    # === Image related methods ===
+    # These three tmethods are used to create, move and delete images. They
path: sky/clouds/cloud.py
text: @user1:
tmethods -> typo

@author:
Fixed. Thanks!

diff_hunk: @@ -813,3 +817,91 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
             status_list.append(status)
 
         return status_list
+
+    @classmethod
+    def create_image_from_cluster(cls, cluster_name: str,
+                                  tag_filters: Dict[str,
+                                                    str], region: Optional[str],
+                                  zone: Optional[str]) -> str:
+        del region  # unused
+        assert zone is not None
+        label_filter_str = cls._label_filter_str(tag_filters)
+        instance_name_cmd = ('gcloud compute instances list '
+                             f'--filter="({label_filter_str})" '
+                             '--format="json(name)"')
+        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+            instance_name_cmd,
+            retry_returncode=[255],
+        )
+        subprocess_utils.handle_returncode(
+            returncode,
+            instance_name_cmd,
+            error_msg=f'Failed to get instance name for {cluster_name!r}',
+            stderr=stderr,
+            stream_logs=True)
+        instance_names = json.loads(stdout)
+        if len(instance_names) != 1:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    'Only support creating image from single '
+                    f' instance, but got multiple instances: {instance_names}')
path: sky/clouds/gcp.py
text: @user1:
```suggestion
                    f' instance, but got: {instance_names}')
```
(can be 0?)

@author:
Good point! Fixed. Thanks!

diff_hunk: @@ -695,6 +695,26 @@ def _get_disk_specs(cls, disk_tier: Optional[str]) -> Dict[str, Any]:
             'custom_disk_perf': tier != 'low',
         }
 
+    @classmethod
+    def _query_instance_property_and_retry(
path: sky/clouds/aws.py
text: @user1:
_with_retries?

@author:
Done. Thanks!

diff_hunk: @@ -3547,6 +3547,23 @@ def post_teardown_cleanup(self,
                             cluster_name=handle.cluster_name,
                             stdout=tpu_stdout,
                             stderr=tpu_stderr))
+        if handle.launched_resources.is_image_managed and terminate:
+            # Delete the image when terminating the cluster and the image is
+            # created by SkyPilot (--clone-disk-from)
+            logger.debug(f'Deleting image {handle.launched_resources.image_id}')
+            cluster_resources = handle.launched_resources
+            cluster_cloud = cluster_resources.cloud
+            image_dict = cluster_resources.image_id
+            assert cluster_cloud is not None, cluster_resources
+            assert image_dict is not None and len(image_dict) == 1
+            image_id = list(image_dict.values())[0]
+            try:
+                cluster_cloud.delete_image(image_id,
+                                           handle.launched_resources.region)
+            except exceptions.CommandError as e:
+                logger.debug(
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
warn? "Failed to delete cloned image ... Please remove it manually to avoid image leakage. Details: ..."

@author:
Good point! Changed. Thanks!

diff_hunk: @@ -117,3 +120,49 @@ def kill_children_processes(first_pid_to_kill: Optional[int] = None,
         except psutil.NoSuchProcess:
             # The child process may have already been terminated.
             pass
+
+
+def run_and_retry(
path: sky/utils/subprocess_utils.py
text: @user1:
run_with_retries?

@author:
Renamed. Thanks!

diff_hunk: @@ -1704,6 +1711,89 @@ def _query_cluster_status_via_cloud_api(
     return node_statuses
 
 
+def check_can_clone_disk_and_override_task(
+    cluster_name: str, target_cluster_name: Optional[str], task: 'task_lib.Task'
+) -> Tuple['task_lib.Task', 'cloud_vm_ray_backend.CloudVmRayResourceHandle']:
+    """Check if the task is compatible to clone disk from the source cluster.
+
+    Args:
+        cluster_name: The name of the cluster to clone disk from.
+        target_cluster_name: The name of the target cluster.
+        task: The task to check.
+
+    Returns:
+        The task to use and the resource handle of the source cluster.
+
+    Raises:
+        exceptions.NotSupportedError: If the source cluster is not valid or the
+            task is not compatible to clone disk from the source cluster.
+    """
+    source_cluster_status, handle = refresh_cluster_status_handle(cluster_name)
+    if source_cluster_status is None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(
+                f'Cannot find cluster {cluster_name!r} to clone disk from.')
+
+    if not isinstance(handle, backends.CloudVmRayResourceHandle):
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from a non-cloud cluster {cluster_name!r}.')
+
+    if source_cluster_status != status_lib.ClusterStatus.STOPPED:
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'Cannot clone disk from cluster {source_cluster_status!r}. Please stop the '
+                f'cluster first: sky stop {cluster_name}.')
+
+    if target_cluster_name is not None:
+        target_cluster_status, _ = refresh_cluster_status_handle(
+            target_cluster_name)
+        if target_cluster_status is not None:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    f'The target cluster {target_cluster_name!r} already exists. Cloning '
+                    'disk is only supported when creating a new cluster.')
+
+    assert len(task.resources) == 1, task.resources
+    task_resources = list(task.resources)[0]
+    if handle.launched_resources.disk_size > task_resources.disk_size:
+        # The target cluster's disk should be at least as large as the source.
+        with ux_utils.print_exception_no_traceback():
+            raise exceptions.NotSupportedError(
+                f'The target cluster {target_cluster_name!r} should have a disk size '
path: sky/backends/backend_utils.py
text: @user1:
target_cluster_name may print None?

@author:
Good catch! Fixed. Thanks!

diff_hunk: @@ -228,31 +280,34 @@ def _execute(
                 f'Backend {backend.NAME} does not support autostop, please try '
                 f'{backends.CloudVmRayBackend.NAME}')
 
-    if not cluster_exists:
-        if (Stage.PROVISION in stages and task.use_spot and
-                not _is_launched_by_spot_controller):
-            yellow = colorama.Fore.YELLOW
-            bold = colorama.Style.BRIGHT
-            reset = colorama.Style.RESET_ALL
-            logger.info(
-                f'{yellow}Launching an unmanaged spot task, which does not '
-                f'automatically recover from preemptions.{reset}\n{yellow}To '
-                'get automatic recovery, use managed spot instead: '
-                f'{reset}{bold}sky spot launch{reset} {yellow}or{reset} '
-                f'{bold}sky.spot_launch(){reset}.')
-
-        if Stage.OPTIMIZE in stages:
-            if task.best_resources is None:
-                # TODO: fix this for the situation where number of requested
-                # accelerators is not an integer.
-                if isinstance(backend, backends.CloudVmRayBackend):
-                    # TODO: adding this check because docker backend on a
-                    # no-credential machine should not enter optimize(), which
-                    # would directly error out ('No cloud is enabled...').  Fix
-                    # by moving `sky check` checks out of optimize()?
-                    dag = sky.optimize(dag, minimize=optimize_target)
-                    task = dag.tasks[0]  # Keep: dag may have been deep-copied.
-                    assert task.best_resources is not None, task
+    if (Stage.PROVISION in stages and task.use_spot and
+            not _is_launched_by_spot_controller and not cluster_exists):
+        yellow = colorama.Fore.YELLOW
+        bold = colorama.Style.BRIGHT
+        reset = colorama.Style.RESET_ALL
+        logger.info(
+            f'{yellow}Launching an unmanaged spot task, which does not '
+            f'automatically recover from preemptions.{reset}\n{yellow}To '
+            'get automatic recovery, use managed spot instead: '
+            f'{reset}{bold}sky spot launch{reset} {yellow}or{reset} '
+            f'{bold}sky.spot_launch(){reset}.')
+
+    if Stage.CLONE_DISK in stages:
path: sky/execution.py
text: @user1:
Seems like the 3 if clauses should be placed under an "if not cluster_exists:". Not feeling strongly, just felt that may be more readable.

@author:
Actually, we can move the CLONE_DISK stage up before the if above, so that we can still have the two original if in `if not cluster_exists`. The downside with this is that, it can be a little bit confusing when the order of the stages should be `CLONE_DISK -> OPTIMIZE -> PROVISION ...`, but now it is easier to read it as `CLONE_DISK -> PROVISION -> OPTIMIZE -> PROVISION...`.