Repository: skypilot-org/skypilot
Pull request: [Core] Launching cluster autodowned or manually terminated
<!-- Describe the changes in this PR -->

Fixes #2129 

We need to refactor the code in `execution` and the `_check_existing_cluster`, as the optimization and the provision are actually quite coupled, as raised by @WoosukKwon before. This PR is a fix that will unblock the issue #2129, but not the coupled issue. In the long term, we need to rearchitect the code here.

<!-- 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)
  - [x] Reproducible script in #2129
- [ ] 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:
[Python API] NoneType issue when re-launching a cluster terminated manually on the console
An issue from our user, where our python API is not reliable for the cluster manually terminated or preempted on the cloud.

To reproduce the issue:
1. `sky launch -c min --cloud gcp -i 120`
2. terminate the cluster on the console manually
3. `python -c 'import sky; sky.launch(sky.Task(), cluster_name="min")'`

The following error will occur:
```
3721, in _check_existing_cluster
    cloud = to_provision.cloud
AttributeError: 'NoneType' object has no attribute 'cloud'
```

Hints from issue discussion:

Commit message: remove uneccessary output

Patch to review:
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 582b4cad6ce..f0030f8bead 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -3668,7 +3668,8 @@ def run_on_head(
 
     @timeline.event
     def _check_existing_cluster(
-            self, task: task_lib.Task, to_provision: resources_lib.Resources,
+            self, task: task_lib.Task,
+            to_provision: Optional[resources_lib.Resources],
             cluster_name: str) -> RetryingVmProvisioner.ToProvisionConfig:
         """Checks if the cluster exists and returns the provision config.
 
@@ -3678,6 +3679,8 @@ def _check_existing_cluster(
             exceptions.InvalidClusterNameError: If the cluster name is invalid.
             # TODO(zhwu): complete the list of exceptions.
         """
+        previous_handle = global_user_state.get_handle_from_cluster_name(
+            cluster_name)
         prev_cluster_status, handle = (
             backend_utils.refresh_cluster_status_handle(
                 cluster_name, acquire_per_cluster_status_lock=False))
@@ -3701,6 +3704,20 @@ def _check_existing_cluster(
                       if resources.cloud is not None else clouds.Cloud)
         task_cloud.check_cluster_name_is_valid(cluster_name)
 
+        if to_provision is None:
+            logger.info(
+                f'The cluster {cluster_name!r} was autodowned or manually '
+                'terminated on the cloud console. Using the original resources '
+                'to provision a new cluster.')
+            # The cluster is recently terminated either by autostop or manually
+            # terminated on the cloud. We should use the original resources to
+            # provision the cluster.
+            assert isinstance(previous_handle,
+                              CloudVmRayResourceHandle), (previous_handle,
+                                                          cluster_name)
+            to_provision = previous_handle.launched_resources
+            self.check_resources_fit_cluster(previous_handle, task)
+
         cloud = to_provision.cloud
         if isinstance(cloud, clouds.Local):
             # The field ssh_user is specified in the cluster config file.
diff --git a/sky/skylet/providers/oci/query_helper.py b/sky/skylet/providers/oci/query_helper.py
index 28aadf90341..29601192e5d 100644
--- a/sky/skylet/providers/oci/query_helper.py
+++ b/sky/skylet/providers/oci/query_helper.py
@@ -57,7 +57,7 @@ def query_instances_by_tags(cls, tag_filters, region):
 
     @classmethod
     def terminate_instances_by_tags(cls, tag_filters, region) -> int:
-        logger.info(f"Terminate instance by tags: {tag_filters}")
+        logger.debug(f"Terminate instance by tags: {tag_filters}")
         insts = cls.query_instances_by_tags(tag_filters, region)
         fail_count = 0
         for inst in insts:
@@ -73,7 +73,7 @@ def terminate_instances_by_tags(cls, tag_filters, region) -> int:
                 traceback.print_exc()
 
         if fail_count == 0:
-            logger.info(f"Instance teardown result: OK")
+            logger.debug(f"Instance teardown result: OK")
         else:
             logger.warn(f"Instance teardown result: {fail_count} failed!")
 


Code review comments:
diff_hunk: @@ -3701,6 +3704,20 @@ def _check_existing_cluster(
                       if resources.cloud is not None else clouds.Cloud)
         task_cloud.check_cluster_name_is_valid(cluster_name)
 
+        if to_provision is None:
+            logger.info(
+                f'The cluster {cluster_name!r} was autodowned or manually '
+                'terminated on the cloud console. Using the original resources '
+                'to provision a new cluster.')
+            # The cluster is recently terminated either by autostop or manually
+            # terminated on the cloud. We should use the original resources to
+            # provision the cluster.
+            assert isinstance(previous_handle,
+                              CloudVmRayResourceHandle), (previous_handle,
+                                                          cluster_name)
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
Style nit: The code looks a bit ugly tbh.
```suggestion
            is_cloud_vm_handle = isinstance(previous_handle,
                                                                      CloudVmRayResourceHandle)
            assert is_cloud_vm_handle, (previous_handle, cluster_name)
```

@author:
Unfortunately, mypy is not clear enough to have this two-line assertion. I changed the output to be a string, so as to make it look a bit better. PTAL. : )

diff_hunk: @@ -3701,6 +3704,20 @@ def _check_existing_cluster(
                       if resources.cloud is not None else clouds.Cloud)
         task_cloud.check_cluster_name_is_valid(cluster_name)
 
+        if to_provision is None:
+            logger.info(
+                f'The cluster {cluster_name!r} was autodowned or manually '
+                'terminated on the cloud console. Using the original resources '
+                'to provision a new cluster.')
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
I think the meaning of "original" here is a bit unclear. Can we change this like following?
```suggestion
                'terminated on the cloud console. Using the same resources as the previously terminated one'
                'to provision a new cluster.')
```

@author:
Good point! Fixed. Thanks!

diff_hunk: @@ -3678,6 +3679,8 @@ def _check_existing_cluster(
             exceptions.InvalidClusterNameError: If the cluster name is invalid.
             # TODO(zhwu): complete the list of exceptions.
         """
+        previous_handle = global_user_state.get_handle_from_cluster_name(
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
Maybe `prev_handle` to follow the naming convention? or `handle_before_refresh` if you want to distinguish it from `handle` below? 

@author:
Good point! I renmaed it to `handle_before_refresh`. Thanks!