[Core] Support launching a new cluster from an existing cluster's disk

docs/source/cloud-setup/cloud-permissions/gcp.rst

@@ -39,7 +39,7 @@ You can grant those accesses via GCP's `IAM & Admin console <https://console.clo
 Minimal Permissions
 -----------------------
 
-The :ref:`Medium Permissions <medium-permissions>` assigns admin permissions for some GCP services to the user.  If you would like to grant finer-grained and more minimal permissions to your users in your organization / project, you can create a custom role by following the steps below:
+The :ref:`Medium Permissions <gcp-medium-permissions>` assigns admin permissions for some GCP services to the user.  If you would like to grant finer-grained and more minimal permissions to your users in your organization / project, you can create a custom role by following the steps below:
 
 User
 ~~~~~~~~~~~~

sky/backends/backend_utils.py

@@ -1725,6 +1725,7 @@ def check_can_clone_disk_and_override_task(
         The task to use and the resource handle of the source cluster.
 
     Raises:
+        ValueError: If the source cluster does not exist.
         exceptions.NotSupportedError: If the source cluster is not valid or the
             task is not compatible to clone disk from the source cluster.
     """
@@ -1742,8 +1743,9 @@ def check_can_clone_disk_and_override_task(
     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}.')
+                f'Cannot clone disk from cluster {cluster_name!r} '
+                f'({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(
@@ -1752,15 +1754,19 @@ def check_can_clone_disk_and_override_task(
             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.')
+                    'disk is only supported when creating a new cluster. To fix: specify '
+                    'a new target cluster name.')
 
     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():
+            target_cluster_name_str = f' {target_cluster_name!r}'
+            if target_cluster_name is None:
+                target_cluster_name_str = ''
             raise exceptions.NotSupportedError(
-                f'The target cluster {target_cluster_name!r} should have a disk size '
+                f'The target cluster {target_cluster_name_str} 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 = {}

sky/backends/cloud_vm_ray_backend.py

@@ -3577,9 +3577,9 @@ 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)
+        if (terminate and handle.launched_resources.is_image_managed is True):
+            # Delete the image when terminating a "cloned" cluster, i.e.,
+            # whose 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
@@ -3591,8 +3591,9 @@ def post_teardown_cleanup(self,
                 cluster_cloud.delete_image(image_id,
                                            handle.launched_resources.region)
             except exceptions.CommandError as e:
-                logger.debug(
-                    f'Failed to delete image {image_id}: '
+                logger.warning(
+                    f'Failed to delete cloned image {image_id}. Please '
+                    'remove it manually to avoid image leakage. Details: '
                     f'{common_utils.format_exception(e, use_bracket=True)}')
 
         # The cluster file must exist because the cluster_yaml will only

sky/clouds/aws.py

@@ -689,7 +689,48 @@ def _get_disk_specs(cls, disk_tier: Optional[str]) -> Dict[str, Any]:
         }
 
     @classmethod
-    def _query_instance_property_and_retry(
+    def check_quota_available(cls,
+                              region: str,
+                              instance_type: str,
+                              use_spot: bool = False) -> bool:
+        """Check if AWS quota is available for `instance_type` in `region`.
+
+        AWS-specific implementation of check_quota_available. The function works by
+        matching the instance_type to the corresponding AWS quota code, and then using
+        the boto3 Python API to query the region for the specific quota code.
+
+        Returns:
+            False if the quota is found to be zero, and True otherwise.
+        Raises:
+            ImportError: if the dependencies for AWS are not able to be installed.
+            botocore.exceptions.ClientError: error in Boto3 client request.
+        """
+
+        from sky.clouds.service_catalog import aws_catalog  # pylint: disable=import-outside-toplevel,unused-import
+
+        quota_code = aws_catalog.get_quota_code(instance_type, use_spot)
+
+        if quota_code is None:
+            # Quota code not found in the catalog for the chosen instance_type, try provisioning anyway
+            return True
+
+        client = aws.client('service-quotas', region_name=region)
+        try:
+            response = client.get_service_quota(ServiceCode='ec2',
+                                                QuotaCode=quota_code)
+        except aws.botocore_exceptions().ClientError:
+            # Botocore client connection not established, try provisioning anyways
+            return True
+
+        if response['Quota']['Value'] == 0:
+            # Quota found to be zero, do not try provisioning
+            return False
+
+        # Quota found to be greater than zero, try provisioning
+        return True
+
+    @classmethod
+    def _query_instance_property_with_retries(
         cls,
         tag_filters: Dict[str, str],
         region: str,
@@ -698,8 +739,8 @@ def _query_instance_property_and_retry(
         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(
+                     f'--region {region} --query "{query}" --output json')
+        returncode, stdout, stderr = subprocess_utils.run_with_retries(
             query_cmd,
             retry_returncode=[255],
             retry_stderrs=[
@@ -725,7 +766,7 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
         }
 
         assert region is not None, (tag_filters, region)
-        returncode, stdout, stderr = cls._query_instance_property_and_retry(
+        returncode, stdout, stderr = cls._query_instance_property_with_retries(
             tag_filters, region, query='Reservations[].Instances[].State.Name')
 
         if returncode != 0:
@@ -751,7 +792,7 @@ def create_image_from_cluster(cls, cluster_name: 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(
+        returncode, stdout, stderr = cls._query_instance_property_with_retries(
             tag_filters, region, query='Reservations[].Instances[].InstanceId')
 
         subprocess_utils.handle_returncode(
@@ -765,14 +806,14 @@ def create_image_from_cluster(cls, cluster_name: str,
         if len(instance_ids) != 1:
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.NotSupportedError(
-                    f'More than one instance found: '
-                    f'{stdout + stderr}')
+                    'Only support creating image from single '
+                    f'instance, but got: {instance_ids}')
 
         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(
+        returncode, image_id, stderr = subprocess_utils.run_with_retries(
             create_image_cmd,
             retry_returncode=[255],
         )
@@ -792,7 +833,7 @@ def create_image_from_cluster(cls, cluster_name: str,
         wait_image_cmd = (
             f'aws ec2 wait image-available --region {region} --image-ids {image_id}'
         )
-        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+        returncode, stdout, stderr = subprocess_utils.run_with_retries(
             wait_image_cmd,
             retry_returncode=[255],
         )
@@ -819,7 +860,7 @@ def maybe_move_image(cls, image_id: str, source_region: str,
                           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(
+        returncode, target_image_id, stderr = subprocess_utils.run_with_retries(
             copy_image_cmd,
             retry_returncode=[255],
         )
@@ -838,7 +879,7 @@ def maybe_move_image(cls, image_id: str, source_region: str,
         wait_image_cmd = (
             f'aws ec2 wait image-available --region {target_region} '
             f'--image-ids {target_image_id}')
-        subprocess_utils.run_and_retry(
+        subprocess_utils.run_with_retries(
             wait_image_cmd,
             max_retry=5,
             retry_returncode=[255],
@@ -863,7 +904,7 @@ 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(
+        returncode, _, stderr = subprocess_utils.run_with_retries(
             delete_image_cmd,
             retry_returncode=[255],
         )

sky/clouds/cloud.py

@@ -539,7 +539,7 @@ 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
+    # These three methods 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

sky/clouds/gcp.py

@@ -838,7 +838,7 @@ def create_image_from_cluster(cls, cluster_name: str,
         instance_name_cmd = ('gcloud compute instances list '
                              f'--filter="({label_filter_str})" '
                              '--format="json(name)"')
-        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+        returncode, stdout, stderr = subprocess_utils.run_with_retries(
             instance_name_cmd,
             retry_returncode=[255],
         )
@@ -853,15 +853,15 @@ def create_image_from_cluster(cls, cluster_name: str,
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.NotSupportedError(
                     'Only support creating image from single '
-                    f' instance, but got multiple instances: {instance_names}')
+                    f'instance, but got: {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(
+        subprocess_utils.run_with_retries(
             create_image_cmd,
             retry_returncode=[255],
         )
@@ -874,7 +874,7 @@ def create_image_from_cluster(cls, cluster_name: str,
 
         image_uri_cmd = (f'gcloud compute images describe {image_name} '
                          '--format="get(selfLink)"')
-        returncode, stdout, stderr = subprocess_utils.run_and_retry(
+        returncode, stdout, stderr = subprocess_utils.run_with_retries(
             image_uri_cmd,
             retry_returncode=[255],
         )
@@ -904,7 +904,7 @@ 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(
+        returncode, _, stderr = subprocess_utils.run_with_retries(
             delete_image_cmd,
             retry_returncode=[255],
         )

sky/execution.py

@@ -280,34 +280,35 @@ def _execute(
                 f'Backend {backend.NAME} does not support autostop, please try '
                 f'{backends.CloudVmRayBackend.NAME}')
 
-    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
+    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
 
     backend.register_info(dag=dag,
                           optimize_target=optimize_target,

sky/resources.py

@@ -58,7 +58,6 @@ def __init__(
         All fields are optional.  ``Resources.is_launchable`` decides whether
         the Resources is fully specified to launch an instance.
 
-
         Examples:
           .. code-block:: python
 
@@ -339,7 +338,7 @@ def disk_tier(self) -> str:
         return self._disk_tier
 
     @property
-    def is_image_managed(self) -> Optional[str]:
+    def is_image_managed(self) -> Optional[bool]:
         return self._is_image_managed
 
     def _set_cpus(

sky/utils/subprocess_utils.py

@@ -122,7 +122,7 @@ def kill_children_processes(first_pid_to_kill: Optional[int] = None,
             pass
 
 
-def run_and_retry(
+def run_with_retries(
         cmd: str,
         max_retry: int = 3,
         retry_returncode: Optional[List[int]] = None,