Repository: skypilot-org/skypilot
Pull request: [fast] don't skip setup if we do provisioning
Fixes #5089.

<!-- Describe the changes in this PR -->



<!-- 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: install pre-commit (auto-check on commit) or `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [x] Relevant individual tests: test_minimal, test_stale_job, test_launch_fast, test_launch_fast_with_autostop, managed jobs tests
- [x] Backward compatibility: `/quicktest-core` (CI) or `pytest tests/smoke_tests/test_backward_compat.py` (local)
- [x] new smoke test https://buildkite.com/skypilot-1/smoke-tests/builds/611/steps?jid=01961275-22a3-4ad7-b01c-6bec3e112ce3

<!-- CI commands (/-prefixed) can only be triggered by repo members -->


Problem statement:
[Jobs] Controller does not get a newly setup cloud due to missing of re-setup of cloud dependencies
We have `fast=True` for controller, which does not run setup again on the controller when the controller is UP, causing the newly configured clouds are not enabled on the controller.

This is a big issue for controllers on Kubernetes clusters, as the controller never autostop, i.e. it will never be resetup.

Hints from issue discussion:

Commit message: fix other backend.provision call

Patch to review:
diff --git a/sky/backends/backend.py b/sky/backends/backend.py
index 210496cd54c..e291944e48c 100644
--- a/sky/backends/backend.py
+++ b/sky/backends/backend.py
@@ -1,6 +1,6 @@
 """Sky backend interface."""
 import typing
-from typing import Dict, Generic, Optional
+from typing import Dict, Generic, Optional, Tuple
 
 from sky.usage import usage_lib
 from sky.utils import cluster_utils
@@ -53,7 +53,7 @@ def provision(
         cluster_name: Optional[str] = None,
         retry_until_up: bool = False,
         skip_unnecessary_provisioning: bool = False,
-    ) -> Optional[_ResourceHandleType]:
+    ) -> Tuple[Optional[_ResourceHandleType], bool]:
         """Provisions resources for the given task.
 
         Args:
@@ -66,15 +66,17 @@ def provision(
             cluster_name: Name of the cluster to provision. If None, a name will
                 be auto-generated. If the name refers to an existing cluster,
                 the existing cluster will be reused and re-provisioned.
-            retry_until_up: If True, retry provisioning until resources are
+            retry_until`_up: If True, retry provisioning until resources are
                 successfully launched.
-            skip_if_no_cluster_updates: If True, compare the cluster config to
-                the existing cluster_name's config. Skip provisioning if no
+            skip_unnecessary_provisioning: If True, compare the cluster config
+                to the existing cluster_name's config. Skip provisioning if no
                 updates are needed for the existing cluster.
 
         Returns:
-            A ResourceHandle object for the provisioned resources, or None if
-            dryrun is True.
+            - A ResourceHandle object for the provisioned resources, or None if
+              dryrun is True.
+            - A boolean that is True if the provisioning was skipped, and False
+              if provisioning actually happened. Dryrun always gives False.
         """
         if cluster_name is None:
             cluster_name = cluster_utils.generate_cluster_name()
@@ -159,7 +161,7 @@ def _provision(
         cluster_name: str,
         retry_until_up: bool = False,
         skip_unnecessary_provisioning: bool = False,
-    ) -> Optional[_ResourceHandleType]:
+    ) -> Tuple[Optional[_ResourceHandleType], bool]:
         raise NotImplementedError
 
     def _sync_workdir(self, handle: _ResourceHandleType, workdir: Path) -> None:
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 5b26d5174f6..6532a93bcb7 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -2802,7 +2802,7 @@ def _provision(
         cluster_name: str,
         retry_until_up: bool = False,
         skip_unnecessary_provisioning: bool = False,
-    ) -> Optional[CloudVmRayResourceHandle]:
+    ) -> Tuple[Optional[CloudVmRayResourceHandle], bool]:
         """Provisions the cluster, or re-provisions an existing cluster.
 
         Use the SKYPILOT provisioner if it's supported by the cloud, otherwise
@@ -2942,7 +2942,7 @@ def _provision(
                             failover_history=e.failover_history) from None
             if dryrun:
                 record = global_user_state.get_cluster_from_name(cluster_name)
-                return record['handle'] if record is not None else None
+                return record['handle'] if record is not None else None, False
 
             if config_dict['provisioning_skipped']:
                 # Skip further provisioning.
@@ -2953,7 +2953,7 @@ def _provision(
                 record = global_user_state.get_cluster_from_name(cluster_name)
                 assert record is not None and record['handle'] is not None, (
                     cluster_name, record)
-                return record['handle']
+                return record['handle'], True
 
             if 'provision_record' in config_dict:
                 # New provisioner is used here.
@@ -2995,7 +2995,7 @@ def _provision(
                 self._update_after_cluster_provisioned(
                     handle, to_provision_config.prev_handle, task,
                     prev_cluster_status, lock_path, config_hash)
-                return handle
+                return handle, False
 
             cluster_config_file = config_dict['ray']
             handle = config_dict['handle']
@@ -3067,7 +3067,7 @@ def _get_zone(runner):
             self._update_after_cluster_provisioned(
                 handle, to_provision_config.prev_handle, task,
                 prev_cluster_status, lock_path, config_hash)
-            return handle
+            return handle, False
 
     def _open_ports(self, handle: CloudVmRayResourceHandle) -> None:
         cloud = handle.launched_resources.cloud
@@ -3408,7 +3408,7 @@ def _exec_code_on_head(
         mkdir_code = (f'{cd} && mkdir -p {remote_log_dir} && '
                       f'touch {remote_log_path}')
         encoded_script = shlex.quote(codegen)
-        create_script_code = (f'{{ echo {encoded_script} > {script_path}; }}')
+        create_script_code = f'{{ echo {encoded_script} > {script_path}; }}'
         job_submit_cmd = (
             # JOB_CMD_IDENTIFIER is used for identifying the process retrieved
             # with pid is the same driver process.
@@ -4304,7 +4304,7 @@ def post_teardown_cleanup(self,
         cluster_name_on_cloud = handle.cluster_name_on_cloud
         cloud = handle.launched_resources.cloud
 
-        if (terminate and handle.launched_resources.is_image_managed is True):
+        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}')
diff --git a/sky/backends/local_docker_backend.py b/sky/backends/local_docker_backend.py
index c10e51e7975..9fcdd65bcc5 100644
--- a/sky/backends/local_docker_backend.py
+++ b/sky/backends/local_docker_backend.py
@@ -139,7 +139,7 @@ def _provision(
         cluster_name: str,
         retry_until_up: bool = False,
         skip_unnecessary_provisioning: bool = False,
-    ) -> Optional[LocalDockerResourceHandle]:
+    ) -> Tuple[Optional[LocalDockerResourceHandle], bool]:
         """Builds docker image for the task and returns cluster name as handle.
 
         Since resource demands are ignored, There's no provisioning in local
@@ -149,7 +149,7 @@ def _provision(
         assert task.name is not None, ('Task name cannot be None - have you '
                                        'specified a task name?')
         if dryrun:
-            return None
+            return None, False
         if retry_until_up:
             logger.warning(
                 f'Retrying until up is not supported in backend: {self.NAME}. '
@@ -175,7 +175,7 @@ def _provision(
                                                 requested_resources=set(
                                                     task.resources),
                                                 ready=False)
-        return handle
+        return handle, False
 
     def _sync_workdir(self, handle: LocalDockerResourceHandle,
                       workdir: Path) -> None:
diff --git a/sky/core.py b/sky/core.py
index ce61a80bd4b..6f10eecb940 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -372,12 +372,12 @@ def _start(
     with dag_lib.Dag():
         dummy_task = task_lib.Task().set_resources(handle.launched_resources)
         dummy_task.num_nodes = handle.launched_nodes
-    handle = backend.provision(dummy_task,
-                               to_provision=handle.launched_resources,
-                               dryrun=False,
-                               stream_logs=True,
-                               cluster_name=cluster_name,
-                               retry_until_up=retry_until_up)
+    (handle, _) = backend.provision(dummy_task,
+                                    to_provision=handle.launched_resources,
+                                    dryrun=False,
+                                    stream_logs=True,
+                                    cluster_name=cluster_name,
+                                    retry_until_up=retry_until_up)
     storage_mounts = backend.get_storage_mounts_metadata(handle.cluster_name)
     # Passing all_file_mounts as None ensures the local source set in Storage
     # to not redundantly sync source to the bucket.
diff --git a/sky/execution.py b/sky/execution.py
index 967c79d09d8..68e466ce23d 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -159,9 +159,9 @@ def _execute(
       no_setup: bool; whether to skip setup commands or not when (re-)launching.
       clone_disk_from: Optional[str]; if set, clone the disk from the specified
         cluster.
-      skip_unecessary_provisioning: bool; if True, compare the calculated
+      skip_unnecessary_provisioning: bool; if True, compare the calculated
         cluster config to the current cluster's config. If they match, shortcut
-        provisioning even if we have Stage.PROVISION.
+        provisioning and setup, even if we have Stage.PROVISION and Stage.SETUP.
 
     Returns:
       job_id: Optional[int]; the job ID of the submitted job. None if the
@@ -303,12 +303,13 @@ def _execute(
         task.sync_storage_mounts()
 
     try:
+        provisioning_skipped = False
         if Stage.PROVISION in stages:
             assert handle is None or skip_unnecessary_provisioning, (
                 'Provisioning requested, but handle is already set. PROVISION '
                 'should be excluded from stages or '
                 'skip_unecessary_provisioning should be set. ')
-            handle = backend.provision(
+            (handle, provisioning_skipped) = backend.provision(
                 task,
                 task.best_resources,
                 dryrun=dryrun,
@@ -341,7 +342,11 @@ def _execute(
         if no_setup:
             logger.info('Setup commands skipped.')
         elif Stage.SETUP in stages and not dryrun:
-            backend.setup(handle, task, detach_setup=detach_setup)
+            if skip_unnecessary_provisioning and provisioning_skipped:
+                logger.debug('Unnecessary provisioning was skipped, so '
+                             'skipping setup as well.')
+            else:
+                backend.setup(handle, task, detach_setup=detach_setup)
 
         if Stage.PRE_EXEC in stages and not dryrun:
             if idle_minutes_to_autostop is not None:
@@ -523,6 +528,8 @@ def launch(
                 Stage.PROVISION,
                 Stage.SYNC_WORKDIR,
                 Stage.SYNC_FILE_MOUNTS,
+                # Setup will be skipped if provisioning was skipped.
+                Stage.SETUP,
                 Stage.PRE_EXEC,
                 Stage.EXEC,
                 Stage.DOWN,


Code review comments:
diff_hunk: @@ -66,15 +66,17 @@ def provision(
             cluster_name: Name of the cluster to provision. If None, a name will
                 be auto-generated. If the name refers to an existing cluster,
                 the existing cluster will be reused and re-provisioned.
-            retry_until_up: If True, retry provisioning until resources are
+            retry_until`_up: If True, retry provisioning until resources are
path: sky/backends/backend.py
text: @user1:
```suggestion
            retry_until_up: If True, retry provisioning until resources are
```