Repository: skypilot-org/skypilot
Pull request: Fix prompt, add fallback retry for INIT cluster
Closes #514.

Main changes:
1. Remove the prompt of creating new cluster confirmation, when the previous cluster is in INIT mode.
2. Launching cluster in INIT mode will first retry the previous region/zone and launched_resources. If that fails, sky will try to launch the cluster of the current resources requirement, as discussed in #514.

Tested:
- [x] make aws us-east-1 out-of-quota; 
  `sky launch -c fallback examples/minimal.yaml` (with 2nodes) and kill it when launching on us-east-1; 
  `sky launch -c fallback examples/minimal.yaml` (with 1node). It will correctly retry other regions with the currently requested resources.
- [x] make aws us-east-1 out-of-quota; 
  `sky launch -c fallback --gpus A100:8 examples/minimal.yaml` and kill it when launching on us-east-1; 
  `sky launch -c fallback examples/minimal.yaml`. It will correctly try the us-east-1 with A100:8 again, and then try us-east-1 with no GPU.
  ```
  NAME      LAUNCHED     RESOURCES                          COMMAND                                   STATUS  
  fallback  11 secs ago  1x AWS(p4d.24xlarge, {'A100': 8})  sky launch -c fallback --gpus=A100:8 ...  INIT
  ```
  ```
  NAME      LAUNCHED     RESOURCES           COMMAND                                         STATUS  
  fallback  18 secs ago  1x AWS(m4.2xlarge)  sky launch -c fallback ./examples/minimal.yaml  UP     
  ```
- [x] `sky launch -c fallback examples/minimal.yaml` again will not create a new instance.

Problem statement:
Fix sky check showing AWS:enabled even when `aws configure` is not run
`sky check` incorrectly showed that AWS was available to use but an extra `aws configure` was required for Sky to actually view that AWS is available and include it in optimizer/provisioner.

Hints from issue discussion:
Additionally, we should be more obvious in documentation that Sky checks credentials _automatically_
Investigation below -- I'm not sure why this bug was encountered in Kevin's session.

Around 24min into the video:

![](https://i.imgur.com/7DzLbpn.png)

- We see `AWS: enabled`, which [means](https://github.com/sky-proj/sky/blob/master/sky/check.py#L17-L25) AWS is definitely added into the global_user_state as an enabled cloud.
  - (Not too relevant) From [aws.py#check_credentials()](https://github.com/sky-proj/sky/blob/master/sky/clouds/aws.py#L243), we know this means (1) `~/.aws/credentials` exists; (2) `aws_configure list` can be called and returns the correct access/secret keys.
- Then, `sky gpunode -c mygpu --gpus K80` is run.  Which means it should hit optimizer code [here](https://github.com/sky-proj/sky/blob/master/sky/optimizer.py#L408-L412), where `enabled_clouds` should've included AWS.

Now, around 29min 37s into the video:

![](https://i.imgur.com/mWnXQNb.png)

- Typed `aws configure`
- Without rerunning `sky check`, the same `sky gpunode -c mygpu --gpus K80` now shows AWS is being considered.

Any ideas why the optimizer didn't consider AWS?  @Michaelvll 
Ahhh, finally after struggling with the codes, I got back the video and found that the problem is not related to the `aws configure` he ran. 

According to the video, he did a `sky gpunode -c mygpu --gpus K80` at around 10min 40s, which failed to launch due to the azure-cli dependency issues. That failure must leave a row in the cluster status DB for the cluster `mygpu` with the `handle.resources` Azure(Standard_NC6_Promo) and the `INIT` state. Therefore, when he ran the `sky gpunode` again at 24min, sky disabled the auto-failover across the cloud (the following line will force the retry only to be done on the same cloud), since the cluster is already in the DB with `INIT` mode.

https://github.com/sky-proj/sky/blob/f0d416d78bdfa4bb9e9579bdd7f270fd590380de/sky/backends/cloud_vm_ray_backend.py#L1041-L1047

After failing to launch the cluster on Azure (tried all the regions) for the attempt in 24min, sky deleted the cluster status from the DB by design. Therefore, when he ran `sky gpunode` again at 29min 37s again, sky enabled the auto-failover across cloud again.

The problem will only appear when the `sky launch` or `sky cpu/gpu/tpunode` fails due to the failure of `ray up`, as we do not delete the cluster status row in the DB because of the unknown status of the cluster. The failure could be rare, since our `ray up` uses a minimal yaml with no user-defined code.

One possible solution is that for the cluster in `INIT` mode, we should first try the `handle.launched_resources`. If it fails to launch, we should try `handle.original_requested_resources` or `current_requested_resources` (This should be decided).

Excellent investigation!  The root cause of these complexities is because we don't go fetch the true state of VMs, and hence our state diagram is approximate.

Two problems I think we should/can fix:

----
First, around 24 min -- my previous post's first screenshot -- **the first issue is that it is confusing to show `Launching a new cluster. Proceed? [Y/n]` when in fact, a record in the cluster table exists.**  This is the key thing that confused my initial investigation.

This prompt is shown because of this line:

https://github.com/sky-proj/sky/blob/d4b0c0b66c3229f01fb0c583cc1df97a7474cc89/sky/cli.py#L318-L322

where we have `handle is not None` and `handle.head_ip is None`. 

This can be triggered by the following: `sky cpunode`, confirm the prompt, then immediately `ctrl-c`, then `sky cpunode` again.   You can see the prompt is shown again and there's no `Considered resources -> cost` the second time because it's retrying on the `handle.launched_resources`).
```
» sky cpunode
I 03-12 14:46:50 optimizer.py:325] Optimizer - plan minimizing cost (~$0.4):
I 03-12 14:46:50 optimizer.py:339]
I 03-12 14:46:50 optimizer.py:339] TASK     BEST_RESOURCE
I 03-12 14:46:50 optimizer.py:339] cpunode  AWS(m4.2xlarge)
I 03-12 14:46:50 optimizer.py:339]
I 03-12 14:46:50 optimizer.py:292] Considered resources -> cost
I 03-12 14:46:50 optimizer.py:294] {AWS(m4.2xlarge): 0.4,
I 03-12 14:46:50 optimizer.py:294]  GCP(n1-highmem-8): 0.473212,
I 03-12 14:46:50 optimizer.py:294]  Azure(Standard_D8_v4): 0.461}
I 03-12 14:46:50 optimizer.py:294]
Launching a new cluster. Proceed? [Y/n]:
I 03-12 14:46:56 cloud_vm_ray_backend.py:1050] Creating a new cluster: "sky-cpunode-zongheng" [1x AWS(m4.2xlarge)].
I 03-12 14:46:56 cloud_vm_ray_backend.py:1050] Tip: to reuse an existing cluster, specify --cluster (-c). Run `sky status` to see existing clusters.
I 03-12 14:46:56 cloud_vm_ray_backend.py:683] To view detailed progress: tail -n100 -f /Users/zongheng/sky_logs/sky-2022-03-12-14-46-50-950627/provision.log
I 03-12 14:46:58 cloud_vm_ray_backend.py:833] Launching on AWS us-east-1 (us-east-1a,us-east-1b,us-east-1c,us-east-1d,us-east-1e,us-east-1f)

Aborted!

» sky cpunode                                                                                   1 ↵
I 03-12 14:47:02 optimizer.py:325] Optimizer - plan minimizing cost (~$0.4):
I 03-12 14:47:02 optimizer.py:339]
I 03-12 14:47:02 optimizer.py:339] TASK     BEST_RESOURCE
I 03-12 14:47:02 optimizer.py:339] cpunode  AWS(m4.2xlarge)
I 03-12 14:47:02 optimizer.py:339]
Launching a new cluster. Proceed? [Y/n]:
```

Proposed fix: at

https://github.com/sky-proj/sky/blob/d4b0c0b66c3229f01fb0c583cc1df97a7474cc89/sky/cli.py#L342-L346

show the prompt only if `handle is None and not no_confirm`.  Does this sound correct?

----
Second, regarding

> One possible solution is that for the cluster in INIT mode, we should first try the handle.launched_resources. If it fails to launch, we should try handle.original_requested_resources or current_requested_resources (This should be decided).

It seems like this statement is true: if a cluster is INIT & it fails to provision all & RuntimeError isn't raised (this is for the rare failures after ray's provision step & during its file_mounts/setup steps), then it's definitely not live (correct if wrong).  Hence, this proposed fix seems ok to add, although with the above proposed fix this may not be necessary.

If we were to add this, according to 

https://github.com/sky-proj/sky/blob/f0d416d78bdfa4bb9e9579bdd7f270fd590380de/sky/backends/cloud_vm_ray_backend.py#L1025-L1035

we should probably retry with `current_requested_resources` because it may be strictly smaller than `original_requested_resources`.  Note that interactive nodes seem to do a stricter test, requiring these two to be equivalent: 

https://github.com/sky-proj/sky/blob/d4b0c0b66c3229f01fb0c583cc1df97a7474cc89/sky/cli.py#L263
Thanks for the detailed reply! Yes, only prompting on `handle is None` makes sense to me. 

> It seems like this statement is true: if a cluster is INIT & it fails to provision all & RuntimeError isn't raised (this is for the rare failures after ray's provision step & during its file_mounts/setup steps), then it's definitely not live (correct if wrong).

For this one, if ray fails during its file_mounts/setup steps, the cluster should probably live (or partially), as the provision step is finished. If it fails during the provision step, it is still possible that the cluster is alive (or partially). However, the proposed method (trying the launched_resources first then trying the requested resources) still works. 
1. If the cluster is live or partially alive, it will try to fully launch the existing cluster (and tear it down if failed) before trying other regions/clouds.
2. If the cluster is not live, starting from the previous resources and then trying the other regions / clouds is equivalent to failover by our own order.

I will think about implementing this.


> Note that interactive nodes seem to do a stricter test, requiring these two to be equivalent

For the interactive nodes, I think it may be fine to have more strict checking since it is weird to create an interactive connection for a `sky gpunode` that requires fewer gpus to a larger cluster.

Commit message: remove assert

Patch to review:
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index d0f4c2442f3..6d163d51660 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -378,6 +378,20 @@ def build(self) -> str:
 class RetryingVmProvisioner(object):
     """A provisioner that retries different cloud/regions/zones."""
 
+    class ToProvisionConfig:
+        """Resources to be provisioned"""
+
+        def __init__(self,
+                     cluster_name: str,
+                     resources: Resources,
+                     num_nodes: int,
+                     cluster_exists: bool = False) -> None:
+            assert cluster_name is not None, 'cluster_name must be specified.'
+            self.cluster_name = cluster_name
+            self.resources = resources
+            self.num_nodes = num_nodes
+            self.cluster_exists = cluster_exists
+
     class GangSchedulingStatus(enum.Enum):
         """Enum for gang scheduling status."""
         CLUSTER_READY = 0
@@ -556,7 +570,8 @@ def _update_blocklist_on_error(self, cloud, region, zones, stdout,
                 region, zones, stdout, stderr)
         assert False, f'Unknown cloud: {cloud}.'
 
-    def _yield_region_zones(self, to_provision: Resources, cluster_name: str):
+    def _yield_region_zones(self, to_provision: Resources, cluster_name: str,
+                            cluster_exists: bool):
         cloud = to_provision.cloud
         region = None
         zones = None
@@ -581,7 +596,7 @@ def _yield_region_zones(self, to_provision: Resources, cluster_name: str):
             except FileNotFoundError:
                 # Happens if no previous cluster.yaml exists.
                 pass
-        if region is not None:
+        if region is not None and cluster_exists:
             region = clouds.Region(name=region)
             if zones is not None:
                 zones = [clouds.Zone(name=zone) for zone in zones.split(',')]
@@ -589,6 +604,8 @@ def _yield_region_zones(self, to_provision: Resources, cluster_name: str):
             # Get the *previous* cluster status.
             cluster_status = global_user_state.get_status_from_cluster_name(
                 cluster_name)
+            logger.info(f'Found previous {cluster_status.value} cluster in '
+                        f'{cloud} {region.name}. Trying to launch it again...')
             yield (region, zones)  # Ok to yield again in the next loop.
 
             # If it reaches here: the cluster status gets set to INIT, since
@@ -612,6 +629,12 @@ def _yield_region_zones(self, to_provision: Resources, cluster_name: str):
 
                 raise exceptions.ResourcesUnavailableError(message,
                                                            no_retry=True)
+            assert cluster_status == global_user_state.ClusterStatus.INIT
+            message = (
+                f'Failed to launch previous INIT cluster ({cluster_name}) '
+                f'with the original {to_provision}.')
+            logger.error(message)
+            raise exceptions.ResourcesUnavailableError()
 
         for region, zones in cloud.region_zones_provision_loop(
                 instance_type=to_provision.instance_type,
@@ -671,8 +694,13 @@ def _try_provision_tpu(self, to_provision: Resources,
             logger.error(stderr)
             raise e
 
-    def _retry_region_zones(self, to_provision: Resources, num_nodes: int,
-                            dryrun: bool, stream_logs: bool, cluster_name: str):
+    def _retry_region_zones(self,
+                            to_provision: Resources,
+                            num_nodes: int,
+                            dryrun: bool,
+                            stream_logs: bool,
+                            cluster_name: str,
+                            cluster_exists: bool = False):
         """The provision retry loop."""
         style = colorama.Style
         fore = colorama.Fore
@@ -685,7 +713,8 @@ def _retry_region_zones(self, to_provision: Resources, num_nodes: int,
 
         self._clear_blocklist()
         for region, zones in self._yield_region_zones(to_provision,
-                                                      cluster_name):
+                                                      cluster_name,
+                                                      cluster_exists):
             if self._in_blocklist(to_provision.cloud, region, zones):
                 continue
             zone_str = ','.join(
@@ -893,11 +922,18 @@ def _ensure_cluster_ray_started(self,
             log_abs_path,
             stream_logs=True)
 
-    def provision_with_retries(self, task: Task, to_provision: Resources,
-                               num_nodes: int, dryrun: bool, stream_logs: bool,
-                               cluster_name: str):
+    def provision_with_retries(
+        self,
+        task: Task,
+        to_provision_config: ToProvisionConfig,
+        dryrun: bool,
+        stream_logs: bool,
+    ):
         """Provision with retries for all launchable resources."""
-        assert cluster_name is not None, 'cluster_name must be specified.'
+        cluster_name = to_provision_config.cluster_name
+        to_provision = to_provision_config.resources
+        num_nodes = to_provision_config.num_nodes
+        cluster_exists = to_provision_config.cluster_exists
         launchable_retries_disabled = (self._dag is None or
                                        self._optimize_target is None)
 
@@ -912,7 +948,8 @@ def provision_with_retries(self, task: Task, to_provision: Resources,
                     num_nodes,
                     dryrun=dryrun,
                     stream_logs=stream_logs,
-                    cluster_name=cluster_name)
+                    cluster_name=cluster_name,
+                    cluster_exists=cluster_exists)
                 if dryrun:
                     return
                 config_dict['launched_resources'] = to_provision
@@ -928,11 +965,19 @@ def provision_with_retries(self, task: Task, to_provision: Resources,
                     raise e
                 provision_failed = True
                 logger.warning(
-                    f'\n{style.BRIGHT}Provision failed for {to_provision}. '
-                    'Trying other launchable resources (if any)...'
-                    f'{style.RESET_ALL}')
-                # Add failed resources to the blocklist.
-                self._blocked_launchable_resources.add(to_provision)
+                    f'\n{style.BRIGHT}Provision failed for {num_nodes}x '
+                    f'{to_provision}. Trying other launchable resources '
+                    f'(if any)...{style.RESET_ALL}')
+                if not cluster_exists:
+                    # Add failed resources to the blocklist, only when it
+                    # is in fallback mode.
+                    self._blocked_launchable_resources.add(to_provision)
+                else:
+                    logger.info('Fallback to current requested resources: '
+                                f'{task.num_nodes}x {task.resources}')
+                    num_nodes = task.num_nodes
+                    cluster_exists = False
+
                 # Set to None so that sky.optimize() will assign a new one
                 # (otherwise will skip re-optimizing this task).
                 # TODO: set all remaining tasks' best_resources to None.
@@ -1036,15 +1081,16 @@ def _check_task_resources_smaller_than_cluster(self, handle: ResourceHandle,
 
     def _check_existing_cluster(
             self, task: Task, to_provision: Resources,
-            cluster_name: str) -> Tuple[str, Resources, int]:
+            cluster_name: str) -> RetryingVmProvisioner.ToProvisionConfig:
         handle = global_user_state.get_handle_from_cluster_name(cluster_name)
         if handle is not None:
             # Cluster already exists.
             self._check_task_resources_smaller_than_cluster(handle, task)
             # Use the existing cluster.
             assert handle.launched_resources is not None, (cluster_name, handle)
-            return (cluster_name, handle.launched_resources,
-                    handle.launched_nodes)
+            return RetryingVmProvisioner.ToProvisionConfig(
+                cluster_name, handle.launched_resources, handle.launched_nodes,
+                True)
 
         logger.info(
             f'{colorama.Fore.CYAN}Creating a new cluster: "{cluster_name}" '
@@ -1052,7 +1098,9 @@ def _check_existing_cluster(
             'Tip: to reuse an existing cluster, '
             'specify --cluster (-c). '
             'Run `sky status` to see existing clusters.')
-        return cluster_name, to_provision, task.num_nodes
+        return RetryingVmProvisioner.ToProvisionConfig(cluster_name,
+                                                       to_provision,
+                                                       task.num_nodes)
 
     def _set_tpu_name(self, cluster_config_file: str, num_nodes: int,
                       tpu_name: str) -> None:
@@ -1091,13 +1139,14 @@ def provision(self,
                                             self._optimize_target)
 
         launched_nodes = task.num_nodes
+        to_provision_config = RetryingVmProvisioner.ToProvisionConfig(
+            cluster_name, to_provision, task.num_nodes)
         if not dryrun:  # dry run doesn't need to check existing cluster.
-            cluster_name, to_provision, launched_nodes = (
-                self._check_existing_cluster(task, to_provision, cluster_name))
+            to_provision_config = (self._check_existing_cluster(
+                task, to_provision, cluster_name))
         try:
             config_dict = provisioner.provision_with_retries(
-                task, to_provision, launched_nodes, dryrun, stream_logs,
-                cluster_name)
+                task, to_provision_config, dryrun, stream_logs)
         except exceptions.ResourcesUnavailableError as e:
             # Do not remove the stopped cluster from the global state if failed
             # to start.
diff --git a/sky/cli.py b/sky/cli.py
index e9fa9a02359..b5b24c828ef 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -339,7 +339,9 @@ def _create_and_ssh_into_node(
             task = dag.tasks[0]
             backend.register_info(dag=dag)
             to_provision = task.best_resources
-            if not no_confirm:
+            if handle is None and not no_confirm:
+                # Only show the confirmation prompt if the cluster is not
+                # in INIT mode.
                 click.confirm('Launching a new cluster. Proceed?',
                               default=True,
                               abort=True,


Code review comments:
diff_hunk: @@ -612,6 +629,12 @@ def _yield_region_zones(self, to_provision: Resources, cluster_name: str):
 
                 raise exceptions.ResourcesUnavailableError(message,
                                                            no_retry=True)
+            assert cluster_status == global_user_state.ClusterStatus.INIT
+            message = (
+                f'Failed to launch previous INIT cluster ({cluster_name}) '
+                f'with the original {to_provision}.')
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
It might be clearer to say:
```
            message = (
                f'Failed to launch cluster {cluster_name!r} (previous status: INIT) '
                f'with the original resources: {to_provision}.')
```
The reason is the user probably doesn't know what INIT means.

@author:
Done. Thanks!

diff_hunk: @@ -928,11 +965,19 @@ def provision_with_retries(self, task: Task, to_provision: Resources,
                     raise e
                 provision_failed = True
                 logger.warning(
-                    f'\n{style.BRIGHT}Provision failed for {to_provision}. '
-                    'Trying other launchable resources (if any)...'
-                    f'{style.RESET_ALL}')
-                # Add failed resources to the blocklist.
-                self._blocked_launchable_resources.add(to_provision)
+                    f'\n{style.BRIGHT}Provision failed for {num_nodes}x '
+                    f'{to_provision}. Trying other launchable resources '
+                    f'(if any)...{style.RESET_ALL}')
+                if not cluster_exists:
+                    # Add failed resources to the blocklist, only when it
+                    # is in fallback mode.
+                    self._blocked_launchable_resources.add(to_provision)
+                else:
+                    logger.info('Fallback to current requested resources: '
+                                f'{task.num_nodes}x {task.resources}')
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
Is it more clear to say
```
                    logger.info('Retrying provisioning with resources '
                                f'{task.num_nodes}x {task.resources}')
```
The user may not need to know what "fallback" means here.

@author:
Fixed. Thanks!

diff_hunk: @@ -339,7 +339,9 @@ def _create_and_ssh_into_node(
             task = dag.tasks[0]
             backend.register_info(dag=dag)
             to_provision = task.best_resources
-            if not no_confirm:
+            if handle is None and not no_confirm:
+                # Only show the confirmation prompt if the cluster is not
+                # in INIT mode.
path: sky/cli.py
text: @user1:
nit: Only show the confirmation prompt if the cluster is not in the clusters table.

@author:
Done. Thanks!

diff_hunk: @@ -581,14 +596,16 @@ def _yield_region_zones(self, to_provision: Resources, cluster_name: str):
             except FileNotFoundError:
                 # Happens if no previous cluster.yaml exists.
                 pass
-        if region is not None:
+        if region is not None and cluster_exists:
             region = clouds.Region(name=region)
             if zones is not None:
                 zones = [clouds.Zone(name=zone) for zone in zones.split(',')]
                 region.set_zones(zones)
             # Get the *previous* cluster status.
             cluster_status = global_user_state.get_status_from_cluster_name(
                 cluster_name)
+            logger.info(f'Found previous {cluster_status.value} cluster in '
+                        f'{cloud} {region.name}. Trying to launch it again...')
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
Make it more friendly for restarting a STOPPED cluster?
```
f'Cluster {cluster_name!r} (status: {cluster_status.value}) was previously launched in {cloud} ({region.name}). Relaunching in that region.'
```

@author:
Done. Thanks!