Repository: skypilot-org/skypilot

Issue:
[core] Failover does not respect the blocked list
Great thanks to @ewzeng for finding this bug! After the fine-grained optimizer PR #930, the blocked list in the backend is no longer respected, as it will be cleared after every retry (more details https://github.com/skypilot-org/skypilot/pull/1557#discussion_r1063198597). We should instead have the blocking all in the resources blocking set.

Discussion:

Patch:
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 548a290f..86022616 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -2403,10 +2403,14 @@ def check_cluster_name_is_valid(cluster_name: str,
 def check_cluster_name_not_reserved(
         cluster_name: Optional[str],
         operation_str: Optional[str] = None) -> None:
-    """Errors out if cluster name is reserved by sky.
+    """Errors out if the cluster is a reserved cluster (spot controller).
 
-    If the cluster name is reserved, return the error message. Otherwise,
-    return None.
+    Raises:
+      sky.exceptions.NotSupportedError: if the cluster name is reserved, raise
+        with an error message explaining 'operation_str' is not allowed.
+
+    Returns:
+      None, if the cluster name is not reserved.
     """
     if cluster_name in SKY_RESERVED_CLUSTER_NAMES:
         msg = (f'Cluster {cluster_name!r} is reserved for the '
@@ -2414,7 +2418,7 @@ def check_cluster_name_not_reserved(
         if operation_str is not None:
             msg += f' {operation_str} is not allowed.'
         with ux_utils.print_exception_no_traceback():
-            raise ValueError(msg)
+            raise exceptions.NotSupportedError(msg)
 
 
 # Handle ctrl-c
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index e593e3df..4bb3deda 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -1,5 +1,6 @@
 """Backend: runs on cloud virtual machines, managed by Ray."""
 import ast
+import copy
 import enum
 import getpass
 import inspect
@@ -574,9 +575,7 @@ class RetryingVmProvisioner(object):
     def __init__(self, log_dir: str, dag: 'dag.Dag',
                  optimize_target: OptimizeTarget,
                  local_wheel_path: pathlib.Path, wheel_hash: str):
-        self._blocked_regions = set()
-        self._blocked_zones = set()
-        self._blocked_launchable_resources = set()
+        self._blocked_resources = set()
 
         self.log_dir = os.path.expanduser(log_dir)
         self._dag = dag
@@ -586,24 +585,11 @@ class RetryingVmProvisioner(object):
 
         colorama.init()
 
-    def _in_blocklist(self, cloud, region, zones):
-        if region.name in self._blocked_regions:
-            return True
-        # We do not keep track of zones in Azure and Local,
-        # as both clouds do not have zones.
-        if isinstance(cloud, (clouds.Azure, clouds.Local)):
-            return False
-        assert zones, (cloud, region, zones)
-        for zone in zones:
-            if zone.name not in self._blocked_zones:
-                return False
-        return True
-
-    def _clear_blocklist(self):
-        self._blocked_regions.clear()
-        self._blocked_zones.clear()
+    def _update_blocklist_on_gcp_error(
+            self, launchable_resources: 'resources_lib.Resources', region,
+            zones, stdout, stderr):
 
-    def _update_blocklist_on_gcp_error(self, region, zones, stdout, stderr):
+        del region  # unused
         style = colorama.Style
         assert len(zones) == 1, zones
         zone = zones[0]
@@ -638,17 +624,13 @@ class RetryingVmProvisioner(object):
                         # This skip is only correct if we implement "first
                         # retry the region/zone of an existing cluster with the
                         # same name" correctly.
-                        for r in clouds.GCP.regions_with_offering(
-                                instance_type=None,
-                                accelerators=None,
-                                use_spot=False,
-                                region=None,
-                                zone=None):
-                            self._blocked_regions.add(r.name)
+                        self._blocked_resources.add(
+                            launchable_resources.copy(region=None, zone=None))
                     else:
                         # Per region.  Ex: Quota 'CPUS' exceeded.  Limit: 24.0
                         # in region us-west1.
-                        self._blocked_regions.add(region.name)
+                        self._blocked_resources.add(
+                            launchable_resources.copy(zone=None))
                 elif code in [
                         'ZONE_RESOURCE_POOL_EXHAUSTED',
                         'ZONE_RESOURCE_POOL_EXHAUSTED_WITH_DETAILS',
@@ -658,15 +640,18 @@ class RetryingVmProvisioner(object):
                     # However, UNSUPPORTED_OPERATION is observed empirically
                     # when VM is preempted during creation.  This seems to be
                     # not documented by GCP.
-                    self._blocked_zones.add(zone.name)
+                    self._blocked_resources.add(
+                        launchable_resources.copy(zone=zone.name))
                 elif code in ['RESOURCE_NOT_READY']:
                     # This code is returned when the VM is still STOPPING.
-                    self._blocked_zones.add(zone.name)
+                    self._blocked_resources.add(
+                        launchable_resources.copy(zone=zone.name))
                 elif code == 8:
                     # Error code 8 means TPU resources is out of
                     # capacity. Example:
                     # {'code': 8, 'message': 'There is no more capacity in the zone "europe-west4-a"; you can try in another zone where Cloud TPU Nodes are offered (see https://cloud.google.com/tpu/docs/regions) [EID: 0x1bc8f9d790be9142]'} # pylint: disable=line-too-long
-                    self._blocked_zones.add(zone.name)
+                    self._blocked_resources.add(
+                        launchable_resources.copy(zone=zone.name))
                 else:
                     assert False, error
         elif len(httperror_str) >= 1:
@@ -674,14 +659,15 @@ class RetryingVmProvisioner(object):
             if ('Requested disk size cannot be smaller than the image size'
                     in httperror_str[0]):
                 logger.info('Skipping all regions due to disk size issue.')
-                for r, _ in clouds.GCP.region_zones_provision_loop():
-                    self._blocked_regions.add(r.name)
+                self._blocked_resources.add(
+                    launchable_resources.copy(region=None, zone=None))
             else:
                 # Parse HttpError for unauthorized regions. Example:
                 # googleapiclient.errors.HttpError: <HttpError 403 when requesting ... returned "Location us-east1-d is not found or access is unauthorized.". # pylint: disable=line-too-long
                 # Details: "Location us-east1-d is not found or access is
                 # unauthorized.">
-                self._blocked_zones.add(zone.name)
+                self._blocked_resources.add(
+                    launchable_resources.copy(zone=zone.name))
         else:
             # No such structured error response found.
             assert not exception_str, stderr
@@ -690,7 +676,8 @@ class RetryingVmProvisioner(object):
                 # 'projects/<id>/zones/zone/acceleratorTypes/nvidia-tesla-v100'
                 # was not found.
                 logger.warning(f'Got \'resource not found\' in {zone.name}.')
-                self._blocked_zones.add(zone.name)
+                self._blocked_resources.add(
+                    launchable_resources.copy(zone=zone.name))
             else:
                 logger.info('====== stdout ======')
                 for s in stdout.split('\n'):
@@ -703,7 +690,9 @@ class RetryingVmProvisioner(object):
                     raise RuntimeError('Errors occurred during provision; '
                                        'check logs above.')
 
-    def _update_blocklist_on_aws_error(self, region, zones, stdout, stderr):
+    def _update_blocklist_on_aws_error(
+            self, launchable_resources: 'resources_lib.Resources', region,
+            zones, stdout, stderr):
         style = colorama.Style
         stdout_splits = stdout.split('\n')
         stderr_splits = stderr.split('\n')
@@ -748,9 +737,13 @@ class RetryingVmProvisioner(object):
             logger.warning(f'Got error(s) in {zones_str}:')
         messages = '\n\t'.join(errors)
         logger.warning(f'{style.DIM}\t{messages}{style.RESET_ALL}')
-        self._blocked_regions.add(region.name)
+        for zone in zones:
+            self._blocked_resources.add(
+                launchable_resources.copy(zone=zone.name))
 
-    def _update_blocklist_on_azure_error(self, region, zones, stdout, stderr):
+    def _update_blocklist_on_azure_error(
+            self, launchable_resources: 'resources_lib.Resources', region,
+            zones, stdout, stderr):
         del zones  # Unused.
         # The underlying ray autoscaler will try all zones of a region at once.
         style = colorama.Style
@@ -777,12 +770,14 @@ class RetryingVmProvisioner(object):
         messages = '\n\t'.join(errors)
         logger.warning(f'{style.DIM}\t{messages}{style.RESET_ALL}')
         if any('(ReadOnlyDisabledSubscription)' in s for s in errors):
-            for r in sky.Azure.regions():
-                self._blocked_regions.add(r.name)
+            self._blocked_resources.add(
+                resources_lib.Resources(cloud=clouds.Azure()))
         else:
-            self._blocked_regions.add(region.name)
+            self._blocked_resources.add(launchable_resources.copy(zone=None))
 
-    def _update_blocklist_on_local_error(self, region, zones, stdout, stderr):
+    def _update_blocklist_on_local_error(
+            self, launchable_resources: 'resources_lib.Resources', region,
+            zones, stdout, stderr):
         del zones  # Unused.
         style = colorama.Style
         stdout_splits = stdout.split('\n')
@@ -806,10 +801,12 @@ class RetryingVmProvisioner(object):
         logger.warning('Got error(s) on local cluster:')
         messages = '\n\t'.join(errors)
         logger.warning(f'{style.DIM}\t{messages}{style.RESET_ALL}')
-        self._blocked_regions.add(region.name)
+        self._blocked_resources.add(
+            launchable_resources.copy(region=region.name, zone=None))
 
-    def _update_blocklist_on_error(self, cloud, region, zones, stdout,
-                                   stderr) -> bool:
+    def _update_blocklist_on_error(
+            self, launchable_resources: 'resources_lib.Resources', region,
+            zones, stdout, stderr) -> bool:
         """Handles cloud-specific errors and updates the block list.
 
         This parses textual stdout/stderr because we don't directly use the
@@ -821,11 +818,15 @@ class RetryingVmProvisioner(object):
             launched (e.g., due to VPC errors we have never sent the provision
             request), False otherwise.
         """
+        assert launchable_resources.region == region.name, (
+            launchable_resources, region)
         if stdout is None:
             # Gang scheduling failure (head node is definitely up, but some
-            # workers' provisioning failed).  Simply block the region.
+            # workers' provisioning failed).  Simply block the zones.
             assert stderr is None, stderr
-            self._blocked_regions.add(region.name)
+            for zone in zones:
+                self._blocked_resources.add(
+                    launchable_resources.copy(zone=zone.name))
             return False  # definitely_no_nodes_launched
 
         # TODO(zongheng): refactor into Cloud interface?
@@ -835,13 +836,14 @@ class RetryingVmProvisioner(object):
             clouds.GCP: self._update_blocklist_on_gcp_error,
             clouds.Local: self._update_blocklist_on_local_error,
         }
+        cloud = launchable_resources.cloud
         cloud_type = type(cloud)
         if cloud_type not in handlers:
             raise NotImplementedError(
                 'Cloud {cloud} unknown, or has not added '
                 'support for parsing and handling provision failures.')
         handler = handlers[cloud_type]
-        handler(region, zones, stdout, stderr)
+        handler(launchable_resources, region, zones, stdout, stderr)
 
         stdout_splits = stdout.split('\n')
         stderr_splits = stderr.split('\n')
@@ -1084,12 +1086,29 @@ class RetryingVmProvisioner(object):
             global_user_state.ClusterStatus.UP
         ]
 
-        self._clear_blocklist()
         for region, zones in self._yield_region_zones(to_provision,
                                                       cluster_name,
                                                       cluster_exists):
-            if self._in_blocklist(to_provision.cloud, region, zones):
+            # Filter out zones that are blocked, if any.
+            # This optimize the provision loop by skipping zones that are
+            # indicated to be unavailable from previous provision attempts.
+            # It can happen for the provisioning on GCP, as the
+            # yield_region_zones will return zones from a region one by one,
+            # but the optimizer that does the filtering will not be involved
+            # until the next region.
+            remaining_unblocked_zones = copy.deepcopy(zones)
+            for zone in zones:
+                for blocked_resources in self._blocked_resources:
+                    if to_provision.copy(region=region.name,
+                                         zone=zone.name).should_be_blocked_by(
+                                             blocked_resources):
+                        remaining_unblocked_zones.remove(zone)
+                        break
+            if zones and not remaining_unblocked_zones:
+                # Skip the region if all zones are blocked.
                 continue
+            zones = remaining_unblocked_zones
+
             if not zones:
                 # For Azure, zones is always an empty list.
                 zone_str = 'all zones'
@@ -1199,14 +1218,14 @@ class RetryingVmProvisioner(object):
             if status == self.GangSchedulingStatus.HEAD_FAILED:
                 # ray up failed for the head node.
                 definitely_no_nodes_launched = self._update_blocklist_on_error(
-                    to_provision.cloud, region, zones, stdout, stderr)
+                    to_provision, region, zones, stdout, stderr)
             else:
                 # gang scheduling failed.
                 assert status == self.GangSchedulingStatus.GANG_FAILED, status
                 # The stdout/stderr of ray up is not useful here, since
                 # head node is successfully provisioned.
                 definitely_no_nodes_launched = self._update_blocklist_on_error(
-                    to_provision.cloud,
+                    to_provision,
                     region,
                     # Ignored and block region:
                     zones=None,
@@ -1650,7 +1669,7 @@ class RetryingVmProvisioner(object):
             if not cluster_exists:
                 # Add failed resources to the blocklist, only when it
                 # is in fallback mode.
-                self._blocked_launchable_resources.add(to_provision)
+                self._blocked_resources.add(to_provision)
             else:
                 logger.info('Retrying provisioning with requested resources '
                             f'{task.num_nodes}x {task.resources}')
@@ -1666,10 +1685,9 @@ class RetryingVmProvisioner(object):
             # (otherwise will skip re-optimizing this task).
             # TODO: set all remaining tasks' best_resources to None.
             task.best_resources = None
-            self._dag = sky.optimize(
-                self._dag,
-                minimize=self._optimize_target,
-                blocked_launchable_resources=self._blocked_launchable_resources)
+            self._dag = sky.optimize(self._dag,
+                                     minimize=self._optimize_target,
+                                     blocked_resources=self._blocked_resources)
             to_provision = task.best_resources
             assert task in self._dag.tasks, 'Internal logic error.'
             assert to_provision is not None, task
@@ -2342,9 +2360,13 @@ class CloudVmRayBackend(backends.Backend):
                             f'{colorama.Style.RESET_ALL}')
                     return err_msg
 
-                subprocess_utils.handle_returncode(returncode=returncode,
-                                                   command=setup_cmd,
-                                                   error_msg=error_message)
+                try:
+                    subprocess_utils.handle_returncode(returncode=returncode,
+                                                       command=setup_cmd,
+                                                       error_msg=error_message)
+                except exceptions.CommandError as e:
+                    with ux_utils.print_exception_no_traceback():
+                        raise exceptions.ClusterSetUpError(str(e)) from e
 
             num_nodes = len(ip_list)
             plural = 's' if num_nodes > 1 else ''
diff --git a/sky/cli.py b/sky/cli.py
index e4c9eff8..10339830 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -1180,6 +1180,15 @@ def launch(
     if backend_name is None:
         backend_name = backends.CloudVmRayBackend.NAME
 
+    # A basic check. Programmatic calls will have a proper (but less
+    # informative) error from optimizer.
+    if (cloud is not None and cloud.lower() == 'azure' and
+            use_spot is not None and use_spot):
+        raise click.UsageError(
+            'SkyPilot currently has not implemented '
+            'support for spot instances on Azure. Please file '
+            'an issue if you need this feature.')
+
     task = _make_task_from_entrypoint_with_overrides(
         entrypoint=entrypoint,
         name=name,
@@ -1613,22 +1622,48 @@ def logs(
 def cancel(cluster: str, all: bool, jobs: List[int], yes: bool):  # pylint: disable=redefined-builtin
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Cancel job(s)."""
+    bold = colorama.Style.BRIGHT
+    reset = colorama.Style.RESET_ALL
+    if not jobs and not all:
+        # Friendly message for usage like 'sky cancel 1' / 'sky cancel myclus'.
+        message = textwrap.dedent(f"""\
+          Use:
+            {bold}sky cancel <cluster_name> <job IDs>{reset}   -- cancel one or more jobs on a cluster
+            {bold}sky cancel <cluster_name> -a / --all{reset}  -- cancel all jobs on a cluster
+
+          Job IDs can be looked up by {bold}sky queue{reset}.""")
+        raise click.UsageError(message)
+
     if not yes:
         job_ids = ' '.join(map(str, jobs))
         plural = 's' if len(job_ids) > 1 else ''
-        job_identity_str = f'job{plural} with ID{plural} {job_ids}'
+        job_identity_str = f'job{plural} {job_ids}'
         if all:
-            job_identity_str = 'all managed spot jobs'
-        job_identity_str += f' on {cluster}'
+            job_identity_str = 'all jobs'
+        job_identity_str += f' on cluster {cluster!r}'
         click.confirm(f'Cancelling {job_identity_str}. Proceed?',
                       default=True,
                       abort=True,
                       show_default=True)
+
     try:
         core.cancel(cluster, all, jobs)
+    except exceptions.NotSupportedError:
+        # Friendly message for usage like 'sky cancel <spot controller> -a/<job
+        # id>'.
+        if all:
+            arg_str = '--all'
+        else:
+            arg_str = ' '.join(map(str, jobs))
+        error_str = ('Cancelling the spot controller\'s jobs is not allowed.'
+                     f'\nTo cancel spot jobs, use: sky spot cancel <spot '
+                     f'job IDs> [--all]'
+                     f'\nDo you mean: {bold}sky spot cancel {arg_str}{reset}')
+        click.echo(error_str)
+        sys.exit(1)
     except ValueError as e:
         raise click.UsageError(str(e))
-    except (exceptions.NotSupportedError, exceptions.ClusterNotUpError) as e:
+    except exceptions.ClusterNotUpError as e:
         click.echo(str(e))
         sys.exit(1)
 
@@ -2288,11 +2323,11 @@ def _down_or_stop_clusters(
                     core.down(name, purge=purge)
                 else:
                     core.stop(name, purge=purge)
-            except RuntimeError:
+            except RuntimeError as e:
                 message = (
                     f'{colorama.Fore.RED}{operation} cluster {name}...failed. '
-                    'Please check the logs and try again.'
-                    f'{colorama.Style.RESET_ALL}')
+                    f'{colorama.Style.RESET_ALL}'
+                    f'\n\tReason: {common_utils.format_exception(e)}.')
             except (exceptions.NotSupportedError,
                     exceptions.ClusterOwnerIdentityMismatchError) as e:
                 message = str(e)
@@ -2995,26 +3030,48 @@ def spot_queue(all: bool, refresh: bool, skip_finished: bool):
       watch -n60 sky spot queue
     """
     click.secho('Fetching managed spot job statuses...', fg='yellow')
+    no_jobs_found_str = '  No jobs found.'
     try:
         job_table = core.spot_queue(refresh=refresh,
                                     skip_finished=skip_finished)
     except exceptions.ClusterNotUpError:
+        # TODO(mehul): handle skip_finished for the cached case. E.g., change
+        # {load,dump}_job_table_cache() to use structured data, and let
+        # format_job_table() to take skip_finished.
         cache = spot_lib.load_job_table_cache()
         if cache is not None:
             readable_time = log_utils.readable_time_duration(cache[0])
+            if not cache[1].strip():
+                table_str = no_jobs_found_str
+                cached_hint = ''
+            else:
+                table_str = cache[1]
+                # Show a helpful hint at the end, if cached table is non-empty.
+                cached_hint = (
+                    '\n\nNOTE: cached job table is being shown '
+                    f'[last updated: {readable_time}].\nUse '
+                    f'{colorama.Style.BRIGHT}--refresh / -r'
+                    f'{colorama.Style.RESET_ALL} to restart the spot '
+                    'controller and see the latest job table.')
             table_message = (
                 f'\n{colorama.Fore.YELLOW}Cached job status table '
                 f'[last updated: {readable_time}]:{colorama.Style.RESET_ALL}\n'
-                f'{cache[1]}\n')
+                f'{table_str}{cached_hint}')
         else:
             table_message = 'No cached job status table found.'
         click.echo(table_message)
         return
     job_table = spot_lib.format_job_table(job_table, all)
 
-    # Dump cache
     spot_lib.dump_job_table_cache(job_table)
-    click.echo(f'Managed spot jobs:\n{job_table}')
+
+    if not skip_finished:
+        in_progress_only_hint = ''
+    else:
+        in_progress_only_hint = ' (showing in-progress jobs only)'
+    if not job_table:
+        job_table = no_jobs_found_str
+    click.echo(f'Managed spot jobs{in_progress_only_hint}:\n{job_table}')
 
 
 _add_command_alias_to_group(spot, spot_queue, 'status', hidden=True)
diff --git a/sky/core.py b/sky/core.py
index 89faa5ea..c31cf69a 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -454,12 +454,10 @@ def cancel(cluster_name: str,
     Please refer to the sky.cli.cancel for the document.
 
     Raises:
-        ValueError: arguments are invalid or the cluster is not supported or the
-          cluster does not exist.
-        ValueError: if the cluster does not exist.
+        ValueError: if arguments are invalid, or the cluster does not exist.
         sky.exceptions.ClusterNotUpError: if the cluster is not UP.
-        sky.exceptions.NotSupportedError: if the cluster is not based on
-          CloudVmRayBackend.
+        sky.exceptions.NotSupportedError: if the specified cluster is a
+          reserved cluster that does not support this operation.
         sky.exceptions.ClusterOwnerIdentityMismatchError: if the current user is
           not the same as the user who created the cluster.
         sky.exceptions.CloudUserIdentityError: if we fail to get the current
diff --git a/sky/exceptions.py b/sky/exceptions.py
index 308beb2e..5967a880 100644
--- a/sky/exceptions.py
+++ b/sky/exceptions.py
@@ -43,6 +43,11 @@ class ClusterNotUpError(Exception):
     pass
 
 
+class ClusterSetUpError(Exception):
+    """Raised when a cluster has setup error."""
+    pass
+
+
 class NotSupportedError(Exception):
     """Raised when a feature is not supported."""
     pass
diff --git a/sky/execution.py b/sky/execution.py
index 77554786..8164f107 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -119,6 +119,9 @@ def _execute(
     detach_run: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
     no_setup: bool = False,
+    # Internal only:
+    # pylint: disable=invalid-name
+    _is_launched_by_spot_controller: bool = False,
 ) -> None:
     """Execute a entrypoint.
 
@@ -203,18 +206,31 @@ def _execute(
                 f'Backend {backend.NAME} does not support autostop, please try '
                 f'{backends.CloudVmRayBackend.NAME}')
 
-    if not cluster_exists and 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 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)
 
@@ -299,6 +315,9 @@ def launch(
     detach_setup: bool = False,
     detach_run: bool = False,
     no_setup: bool = False,
+    # Internal only:
+    # pylint: disable=invalid-name
+    _is_launched_by_spot_controller: bool = False,
 ) -> None:
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Launch a task.
@@ -373,6 +392,7 @@ def launch(
         detach_run=detach_run,
         idle_minutes_to_autostop=idle_minutes_to_autostop,
         no_setup=no_setup,
+        _is_launched_by_spot_controller=_is_launched_by_spot_controller,
     )
 
 
@@ -428,6 +448,8 @@ def exec(  # pylint: disable=redefined-builtin
     Raises:
         ValueError: if the specified cluster does not exist or is not in UP
             status.
+        sky.exceptions.NotSupportedError: if the specified cluster is a
+            reserved cluster that does not support this operation.
     """
     entrypoint = task
     if isinstance(entrypoint, sky.Dag):
diff --git a/sky/optimizer.py b/sky/optimizer.py
index 7c606846..3a9189fa 100644
--- a/sky/optimizer.py
+++ b/sky/optimizer.py
@@ -89,11 +89,11 @@ class Optimizer:
         return egress_time
 
     @staticmethod
-    def optimize(dag: 'dag_lib.Dag',
-                 minimize=OptimizeTarget.COST,
-                 blocked_launchable_resources: Optional[List[
-                     resources_lib.Resources]] = None,
-                 quiet: bool = False):
+    def optimize(
+            dag: 'dag_lib.Dag',
+            minimize=OptimizeTarget.COST,
+            blocked_resources: Optional[List[resources_lib.Resources]] = None,
+            quiet: bool = False):
         # This function is effectful: mutates every node in 'dag' by setting
         # node.best_resources if it is None.
         Optimizer._add_dummy_source_sink_nodes(dag)
@@ -101,7 +101,7 @@ class Optimizer:
             unused_best_plan = Optimizer._optimize_objective(
                 dag,
                 minimize_cost=minimize == OptimizeTarget.COST,
-                blocked_launchable_resources=blocked_launchable_resources,
+                blocked_resources=blocked_resources,
                 quiet=quiet)
         finally:
             # Make sure to remove the dummy source/sink nodes, even if the
@@ -202,8 +202,7 @@ class Optimizer:
     def _estimate_nodes_cost_or_time(
         topo_order: List[Task],
         minimize_cost: bool = True,
-        blocked_launchable_resources: Optional[List[
-            resources_lib.Resources]] = None,
+        blocked_resources: Optional[List[resources_lib.Resources]] = None,
     ) -> Tuple[_TaskToCostMap, _TaskToPerCloudCandidates]:
         """Estimates the cost/time of each task-resource mapping in the DAG.
 
@@ -236,7 +235,7 @@ class Optimizer:
                 launchable_resources, cloud_candidates = \
                     _fill_in_launchable_resources(
                         node,
-                        blocked_launchable_resources
+                        blocked_resources
                     )
                 node_to_candidate_map[node] = cloud_candidates
             else:
@@ -765,8 +764,7 @@ class Optimizer:
     def _optimize_objective(
         dag: 'dag_lib.Dag',
         minimize_cost: bool = True,
-        blocked_launchable_resources: Optional[List[
-            resources_lib.Resources]] = None,
+        blocked_resources: Optional[List[resources_lib.Resources]] = None,
         quiet: bool = False,
     ) -> Dict[Task, resources_lib.Resources]:
         """Finds the optimal task-resource mapping for the entire DAG.
@@ -785,7 +783,7 @@ class Optimizer:
             Optimizer._estimate_nodes_cost_or_time(
                 topo_order,
                 minimize_cost,
-                blocked_launchable_resources)
+                blocked_resources)
 
         if dag.is_chain():
             best_plan, best_total_objective = Optimizer._optimize_by_dp(
@@ -874,12 +872,12 @@ def _make_launchables_for_valid_region_zones(
 
 def _filter_out_blocked_launchable_resources(
         launchable_resources: List[resources_lib.Resources],
-        blocked_launchable_resources: List[resources_lib.Resources]):
+        blocked_resources: List[resources_lib.Resources]):
     """Whether the resources are blocked."""
     available_resources = []
     for resources in launchable_resources:
-        for blocked_resources in blocked_launchable_resources:
-            if resources.should_be_blocked_by(blocked_resources):
+        for blocked in blocked_resources:
+            if resources.should_be_blocked_by(blocked):
                 break
         else:  # non-blocked launchable resources. (no break)
             available_resources.append(resources)
@@ -888,7 +886,7 @@ def _filter_out_blocked_launchable_resources(
 
 def _fill_in_launchable_resources(
     task: Task,
-    blocked_launchable_resources: Optional[List[resources_lib.Resources]],
+    blocked_resources: Optional[List[resources_lib.Resources]],
     try_fix_with_sky_check: bool = True,
 ) -> Tuple[Dict[resources_lib.Resources, List[resources_lib.Resources]],
            _PerCloudCandidates]:
@@ -896,16 +894,16 @@ def _fill_in_launchable_resources(
     enabled_clouds = global_user_state.get_enabled_clouds()
     launchable = collections.defaultdict(list)
     cloud_candidates = collections.defaultdict(resources_lib.Resources)
-    if blocked_launchable_resources is None:
-        blocked_launchable_resources = []
+    if blocked_resources is None:
+        blocked_resources = []
     for resources in task.get_resources():
         if resources.cloud is not None and not _cloud_in_list(
                 resources.cloud, enabled_clouds):
             if try_fix_with_sky_check:
                 # Explicitly check again to update the enabled cloud list.
                 check.check(quiet=True)
-                return _fill_in_launchable_resources(
-                    task, blocked_launchable_resources, False)
+                return _fill_in_launchable_resources(task, blocked_resources,
+                                                     False)
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.ResourcesUnavailableError(
                     f'Task {task} requires {resources.cloud} which is not '
@@ -930,8 +928,8 @@ def _fill_in_launchable_resources(
             launchable[resources] = _make_launchables_for_valid_region_zones(
                 resources)
         else:
-            clouds_list = [resources.cloud
-                          ] if resources.cloud is not None else enabled_clouds
+            clouds_list = ([resources.cloud]
+                           if resources.cloud is not None else enabled_clouds)
             # Hack: When >=2 cloud candidates, always remove local cloud from
             # possible candidates. This is so the optimizer will consider
             # public clouds, except local. Local will be included as part of
@@ -943,8 +941,8 @@ def _fill_in_launchable_resources(
                 ]
             all_fuzzy_candidates = set()
             for cloud in clouds_list:
-                (feasible_resources, fuzzy_candidate_list
-                ) = cloud.get_feasible_launchable_resources(resources)
+                (feasible_resources, fuzzy_candidate_list) = (
+                    cloud.get_feasible_launchable_resources(resources))
                 if len(feasible_resources) > 0:
                     # Assume feasible_resources is sorted by prices.
                     cheapest = feasible_resources[0]
@@ -955,7 +953,7 @@ def _fill_in_launchable_resources(
                 else:
                     all_fuzzy_candidates.update(fuzzy_candidate_list)
             if len(launchable[resources]) == 0:
-                logger.info(f'No resource satisfying {resources.accelerators} '
+                logger.info(f'No resource satisfying {resources} '
                             f'on {clouds_list}.')
                 if len(all_fuzzy_candidates) > 0:
                     logger.info('Did you mean: '
@@ -964,6 +962,6 @@ def _fill_in_launchable_resources(
                                 f'{colorama.Style.RESET_ALL}')
 
         launchable[resources] = _filter_out_blocked_launchable_resources(
-            launchable[resources], blocked_launchable_resources)
+            launchable[resources], blocked_resources)
 
     return launchable, cloud_candidates
diff --git a/sky/resources.py b/sky/resources.py
index e21173b9..ff4859e3 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -83,6 +83,8 @@ class Resources:
         else:
             self._disk_size = _DEFAULT_DISK_SIZE_GB
 
+        # self._image_id is a dict of {region: image_id}.
+        # The key is None if the same image_id applies for all regions.
         self._image_id = image_id
         if isinstance(image_id, str):
             self._image_id = {self._region: image_id}
@@ -775,19 +777,19 @@ class Resources:
         if version is None:
             version = -1
         if version < 0:
-            cloud = state.pop('cloud')
+            cloud = state.pop('cloud', None)
             state['_cloud'] = cloud
 
-            instance_type = state.pop('instance_type')
+            instance_type = state.pop('instance_type', None)
             state['_instance_type'] = instance_type
 
-            use_spot = state.pop('use_spot')
+            use_spot = state.pop('use_spot', False)
             state['_use_spot'] = use_spot
 
-            accelerator_args = state.pop('accelerator_args')
+            accelerator_args = state.pop('accelerator_args', None)
             state['_accelerator_args'] = accelerator_args
 
-            disk_size = state.pop('disk_size')
+            disk_size = state.pop('disk_size', _DEFAULT_DISK_SIZE_GB)
             state['_disk_size'] = disk_size
 
         if version < 2:
@@ -803,7 +805,7 @@ class Resources:
             self._zone = None
 
         if version < 6:
-            accelerators = state.pop('_accelerators')
+            accelerators = state.pop('_accelerators', None)
             if accelerators is not None:
                 accelerators = {
                     accelerator_registry.canonicalize_accelerator_name(acc):
@@ -811,4 +813,8 @@ class Resources:
                 }
             state['_accelerators'] = accelerators
 
+        image_id = state.get('_image_id', None)
+        if isinstance(image_id, str):
+            state['_image_id'] = {state.get('_region', None): image_id}
+
         self.__dict__.update(state)
diff --git a/sky/spot/controller.py b/sky/spot/controller.py
index 969069f4..1b766f3f 100644
--- a/sky/spot/controller.py
+++ b/sky/spot/controller.py
@@ -176,6 +176,11 @@ class SpotController:
             spot_state.set_failed(
                 self._job_id,
                 failure_type=spot_state.SpotStatus.FAILED_NO_RESOURCE)
+        except exceptions.ClusterSetUpError as e:
+            logger.error(f'{common_utils.class_fullname(e.__class__)}: '
+                         f'{colorama.Fore.RED}{e}{colorama.Style.RESET_ALL}')
+            spot_state.set_failed(self._job_id,
+                                  failure_type=spot_state.SpotStatus.FAILED)
         except (Exception, SystemExit) as e:  # pylint: disable=broad-except
             logger.error(traceback.format_exc())
             logger.error('Unexpected error occurred: '
diff --git a/sky/spot/recovery_strategy.py b/sky/spot/recovery_strategy.py
index 12035d35..f20ad9ec 100644
--- a/sky/spot/recovery_strategy.py
+++ b/sky/spot/recovery_strategy.py
@@ -154,11 +154,17 @@ class StrategyExecutor:
                 usage_lib.messages.usage.set_internal()
                 sky.launch(self.dag,
                            cluster_name=self.cluster_name,
-                           detach_run=True)
+                           detach_run=True,
+                           _is_launched_by_spot_controller=True)
                 logger.info('Spot cluster launched.')
             except exceptions.InvalidClusterNameError as e:
                 # The cluster name is too long.
                 raise exceptions.ResourcesUnavailableError(str(e)) from e
+            except exceptions.ClusterSetUpError as e:
+                # When setup fails, do not retry, as it will likely fail again.
+                # The exception will be handled by the caller and set the job
+                # state to FAILED.
+                raise
             except Exception as e:  # pylint: disable=broad-except
                 # If the launch fails, it will be recovered by the following
                 # code.
diff --git a/sky/task.py b/sky/task.py
index 2466bd46..613f23ed 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -381,6 +381,10 @@ class Task:
     def need_spot_recovery(self) -> bool:
         return any(r.spot_recovery is not None for r in self.resources)
 
+    @property
+    def use_spot(self) -> bool:
+        return any(r.use_spot for r in self.resources)
+
     @num_nodes.setter
     def num_nodes(self, num_nodes: Optional[int]) -> None:
         if num_nodes is None:
diff --git a/sky/templates/azure-ray.yml.j2 b/sky/templates/azure-ray.yml.j2
index 3d30ad72..dced18df 100644
--- a/sky/templates/azure-ray.yml.j2
+++ b/sky/templates/azure-ray.yml.j2
@@ -28,6 +28,8 @@ available_node_types:
     ray.head.default:
       resources: {}
       node_config:
+        tags:
+          skypilot-user: {{ user }}
         azure_arm_parameters:
             vmSize: {{instance_type}}
             # List images https://docs.microsoft.com/en-us/azure/virtual-machines/linux/cli-ps-findimage
@@ -50,6 +52,8 @@ available_node_types:
       max_workers: {{num_nodes - 1}}
       resources: {}
       node_config:
+        tags:
+          skypilot-user: {{ user }}
         azure_arm_parameters:
             vmSize: {{instance_type}}
             # List images https://docs.microsoft.com/en-us/azure/virtual-machines/linux/cli-ps-findimage
diff --git a/sky/templates/gcp-ray.yml.j2 b/sky/templates/gcp-ray.yml.j2
index 255230ee..1231c445 100644
--- a/sky/templates/gcp-ray.yml.j2
+++ b/sky/templates/gcp-ray.yml.j2
@@ -29,6 +29,8 @@ available_node_types:
   ray_head_default:
     resources: {}
     node_config:
+      labels:
+        skypilot-user: {{ user }}
 {%- if tpu_vm %}
       acceleratorType: {{tpu_type}}
       runtimeVersion: {{runtime_version}}
@@ -67,6 +69,8 @@ available_node_types:
     max_workers: {{num_nodes - 1}}
     resources: {}
     node_config:
+      labels:
+        skypilot-user: {{ user }}
   {%- if tpu_vm %}
       acceleratorType: {{tpu_type}}
       runtimeVersion: {{runtime_version}}


Test patch:
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index c3a68153..b932125e 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -277,11 +277,11 @@ def test_image_no_conda():
         'image_no_conda',
         [
             # Use image id dict.
-            f'sky launch -y -c {name} examples/tests/test_yamls/no_conda_ami.yaml',
+            f'sky launch -y -c {name} tests/test_yamls/no_conda_ami.yaml',
             f'sky logs {name} 1 --status',
             f'sky stop {name} -y',
             f'sky start {name} -y',
-            f'sky exec {name} examples/tests/test_yamls/no_conda_ami.yaml',
+            f'sky exec {name} tests/test_yamls/no_conda_ami.yaml',
             f'sky logs {name} 2 --status',
         ],
         f'sky down -y {name}',
@@ -875,6 +875,24 @@ def test_spot():
     run_one_test(test)
 
 
+def test_spot_failed_setup():
+    """Test managed spot job with failed setup."""
+    name = _get_cluster_name()
+    test = Test(
+        'spot-failed-setup',
+        [
+            f'sky spot launch -n {name} -y -d tests/test_yamls/failed_setup.yaml',
+            'sleep 200',
+            # Make sure the job failed quickly.
+            f's=$(sky spot queue); echo "$s"; echo; echo; echo "$s" | grep {name} | head -n1 | grep "FAILED"',
+        ],
+        f'sky spot cancel -y -n {name}',
+        # Increase timeout since sky spot queue -r can be blocked by other spot tests.
+        timeout=20 * 60,
+    )
+    run_one_test(test)
+
+
 # ---------- Testing managed spot ----------
 def test_spot_gcp():
     """Test managed spot on GCP."""
diff --git a/tests/test_spot.py b/tests/test_spot.py
index d24bc9b4..4e5854fc 100644
--- a/tests/test_spot.py
+++ b/tests/test_spot.py
@@ -168,5 +168,6 @@ class TestReservedClustersOperations:
         cli_runner = cli_testing.CliRunner()
         result = cli_runner.invoke(cli.cancel,
                                    [spot.SPOT_CONTROLLER_NAME, '-a'])
-        assert result.exit_code == click.UsageError.exit_code
-        assert 'Cancelling jobs is not allowed' in str(result.output)
+        assert result.exit_code == 1
+        assert 'Cancelling the spot controller\'s jobs is not allowed.' in str(
+            result.output)
diff --git a/tests/test_yamls/failed_setup.yaml b/tests/test_yamls/failed_setup.yaml
new file mode 100644
index 00000000..f6131e7c
--- /dev/null
+++ b/tests/test_yamls/failed_setup.yaml
@@ -0,0 +1,6 @@
+setup: |
+  echo failed setup
+  exit 1
+
+run: |
+  echo run