Repository: skypilot-org/skypilot
Pull request: [Core] Allow nonexistent cloud in candidate resources and speed up optimization
<!-- Describe the changes in this PR -->

Closes #3565

## Allow nonexistent cloud

### With at least one cloud enabled
Without kubernetes and IBM enabled:
```yaml
resources:
  accelerators: A100:8
  ordered:
    - cloud: kubernetes
    - cloud: aws
      use_spot: true
    - cloud: ibm
```
Master:
<details>

```
Task from YAML spec: task.yaml
I 05-20 07:20:28 optimizer.py:983] Using user-specified accelerators list (will be tried in the listed order): Kubernetes({'A100': 8}), AWS([Spot], {'A100': 8}), IBM({'A100': 8})
INFO:googleapiclient.discovery_cache:file_cache is only supported with oauth2client<4.0.0
sky.exceptions.ResourcesUnavailableError: Task requires Kubernetes which is not enabled: Task(run=<empty>)
  resources: Kubernetes({'A100': 8}).
To enable access, run sky check , or change the cloud requirement
```

</details>

This PR:
<details>

```
Task from YAML spec: task.yaml
W 05-20 07:41:11 optimizer.py:1189] Task requires IBM, Kubernetes which are not enabled. To enable access, change the task cloud requirement or run: sky check IBM Kubernetes
I 05-20 07:41:12 optimizer.py:696] == Optimizer ==
I 05-20 07:41:12 optimizer.py:707] Target: minimizing cost
I 05-20 07:41:12 optimizer.py:719] Estimated cost: $5.9 / hour
I 05-20 07:41:12 optimizer.py:719] 
I 05-20 07:41:12 optimizer.py:844] Considered resources (1 node):
I 05-20 07:41:12 optimizer.py:914] -------------------------------------------------------------------------------------------------
I 05-20 07:41:12 optimizer.py:914]  CLOUD   INSTANCE             vCPUs   Mem(GB)   ACCELERATORS   REGION/ZONE   COST ($)   CHOSEN   
I 05-20 07:41:12 optimizer.py:914] -------------------------------------------------------------------------------------------------
I 05-20 07:41:12 optimizer.py:914]  AWS     p4d.24xlarge[Spot]   96      1152      A100:8         us-east-2b    5.89          ✔     
I 05-20 07:41:12 optimizer.py:914] -------------------------------------------------------------------------------------------------
I 05-20 07:41:12 optimizer.py:914] 
Launching a new cluster 'sky-1276-azureuser'. Proceed? [Y/n]:
```

</details>

For a dag:
```yaml
resources:
  accelerators: A100:8
  ordered:
    - cloud: kubernetes
    - cloud: aws
      use_spot: true
    - cloud: ibm


run: hi

---
resources:
  accelerators: h100:8
  ordered:
    - cloud: kubernetes
    - cloud: aws
      use_spot: true
    - cloud: gcp


run: hi
```

<details>

```
Task from YAML spec: task.yaml
Managed job 'sky-1be1-azureuser' will be launched on (estimated):
W 05-20 07:47:36 optimizer.py:1189] Task 'sky-1be1-azureuser-0' requires IBM, Kubernetes which are not enabled. To enable access, change the task cloud requirement or run: sky check IBM Kubernetes
W 05-20 07:47:36 optimizer.py:1189] Task 'sky-1be1-azureuser-1' requires Kubernetes which is not enabled. To enable access, change the task cloud requirement or run: sky check Kubernetes
I 05-20 07:47:36 optimizer.py:978] 1st task is using user-specified accelerators list (will be tried in the listed order): Kubernetes({'A100': 8}), AWS([Spot], {'A100': 8}), IBM({'A100': 8})
I 05-20 07:47:36 optimizer.py:978] 2nd task is using user-specified accelerators list (will be tried in the listed order): Kubernetes({'H100': 8}), AWS([Spot], {'H100': 8}), GCP({'H100': 8})
I 05-20 07:47:38 optimizer.py:696] == Optimizer ==
I 05-20 07:47:38 optimizer.py:707] Target: minimizing cost
I 05-20 07:47:38 optimizer.py:722] Estimated total runtime: 2.0 hours
I 05-20 07:47:38 optimizer.py:722] Estimated total cost: $37.3
I 05-20 07:47:38 optimizer.py:722] 
I 05-20 07:47:38 optimizer.py:820] Best plan: 
I 05-20 07:47:38 optimizer.py:825] -------------------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:825]  TASK                   #NODES   CLOUD   INSTANCE             vCPUs   Mem(GB)   ACCELERATORS   REGION/ZONE   
I 05-20 07:47:38 optimizer.py:825] -------------------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:825]  sky-1be1-azureuser-0   1        AWS     p4d.24xlarge[Spot]   96      1152      A100:8         us-east-2b    
I 05-20 07:47:38 optimizer.py:825]  sky-1be1-azureuser-1   1        AWS     p5.48xlarge[Spot]    192     2048      H100:8         us-east-2b    
I 05-20 07:47:38 optimizer.py:825] -------------------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:825] 
I 05-20 07:47:38 optimizer.py:844] Considered resources for task 'sky-1be1-azureuser-0' (1 node):
I 05-20 07:47:38 optimizer.py:914] -------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:914]  CLOUD   INSTANCE             vCPUs   Mem(GB)   ACCELERATORS   REGION/ZONE   COST ($)   CHOSEN   
I 05-20 07:47:38 optimizer.py:914] -------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:914]  AWS     p4d.24xlarge[Spot]   96      1152      A100:8         us-east-2b    5.89          ✔     
I 05-20 07:47:38 optimizer.py:914] -------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:914] 
I 05-20 07:47:38 optimizer.py:844] Considered resources for task 'sky-1be1-azureuser-1' (1 node):
I 05-20 07:47:38 optimizer.py:914] --------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:914]  CLOUD   INSTANCE            vCPUs   Mem(GB)   ACCELERATORS   REGION/ZONE     COST ($)   CHOSEN   
I 05-20 07:47:38 optimizer.py:914] --------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:914]  GCP     a3-highgpu-8g       208     1872      H100:8         us-central1-a   87.83               
I 05-20 07:47:38 optimizer.py:914]  AWS     p5.48xlarge[Spot]   192     2048      H100:8         us-east-2b      31.44         ✔     
I 05-20 07:47:38 optimizer.py:914] --------------------------------------------------------------------------------------------------
I 05-20 07:47:38 optimizer.py:914] 
Launching a managed job 'sky-1be1-azureuser'. Proceed? [Y/n]: 
```

</details>

### With no cloud enabled

```yaml
resources:
  accelerators: A100:8
  ordered:
    - cloud: kubernetes
    - cloud: IBM
```

```
Task from YAML spec: task.yaml
sky.exceptions.ResourcesUnavailableError: Task requires IBM, Kubernetes which are not enabled. To enable access, change the task cloud requirement or run: sky check IBM Kubernetes
```


For a dag:
```yaml
resources:
  accelerators: A100:8
  ordered:
    - cloud: kubernetes
    - cloud: ibm


run: hi

---
resources:
  accelerators: h100:8
  ordered:
    - cloud: kubernetes
    - cloud: aws
      use_spot: true
    - cloud: gcp


run: hi
```

```console
$ sky jobs launch task.yaml
Task from YAML spec: task.yaml
Managed job 'sky-7f4b-azureuser' will be launched on (estimated):
sky.exceptions.ResourcesUnavailableError: Task 'sky-7f4b-azureuser-0' requires IBM, Kubernetes which are not enabled. To enable access, change the task cloud requirement or run: sky check IBM Kubernetes
```

## Speed up optimization
```yaml
resources:
  accelerators: A100:8
  ordered:
    - cloud: kubernetes
    - cloud: aws
      use_spot: true
```

`time sky launch task.yaml`
master: 12.4s
this PR: 1.99s


<!-- 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):

- [ ] Code formatting: `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
  - [x] scripts above with `ordered` and `any_of`
- [ ] 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: `conda deactivate; bash -i tests/backward_compatibility_tests.sh`


Problem statement:
Certain inputs to `sky launch` would trigger `sky check`, making it slow
<!-- Describe the bug report / feature request here -->

```
resources:
  accelerators: A100:8
  ordered:
    - cloud: kubernetes
    - cloud: aws
      use_spot: true
```
My `sky check` has AWS, GCP, ... enabled, no k8s.

On  latest main branch: 19e8ed16a, it took 16s to error out:
```
time sky launch t-multi.yaml                                                                                    
Task from YAML spec: t-multi.yaml
I 05-19 18:22:18 optimizer.py:983] Using user-specified accelerators list (will be tried in the listed order): Kubernetes({'A100': 8}), AWS([Spot], {'A100': 8})
INFO:googleapiclient.discovery_cache:file_cache is only supported with oauth2client<4.0.0
sky.exceptions.ResourcesUnavailableError: Task requires Kubernetes which is not enabled: Task(run=<empty>)
  resources: Kubernetes({'A100': 8}).
To enable access, run sky check , or change the cloud requirement
sky launch t-multi.yaml  7.12s user 2.06s system 57% cpu 16.084 total
```

py-spy:
```
sudo py-spy dump --pid  82883
Process 82883: /Users/zongheng/anaconda/envs/py39/bin/python /Users/zongheng/anaconda/envs/py39/bin/sky launch t-multi.yaml
Python v3.9.18 (/Users/zongheng/anaconda/envs/py39/bin/python3.9)

Thread 0x203261600 (idle): "MainThread"
    select (selectors.py:416)
    _communicate (subprocess.py:1995)
    communicate (subprocess.py:1134)
    run (subprocess.py:507)
    is_api_disabled (sky/clouds/gcp.py:107)
    check_credentials (sky/clouds/gcp.py:702)
    check_one_cloud (sky/check.py:34)
    check (sky/check.py:94)
    _fill_in_launchable_resources (sky/optimizer.py:1178)
    _optimize_dag (sky/optimizer.py:1000)
    optimize (sky/optimizer.py:127)
```

This appears to be caused by not having k8s enabled but it's requested in YAML. Is there a good optimization for such cases?
- At the minimum, `sky check` should not check GCP, which is not mentioned in the YAML

<!-- If relevant, fill in versioning info to help us troubleshoot -->
_Version & Commit info:_
* `sky -v`: PLEASE_FILL_IN
* `sky -c`: PLEASE_FILL_IN

Hints from issue discussion:

Commit message: revert is_same_cloud

Patch to review:
diff --git a/sky/check.py b/sky/check.py
index f4ecd5a8b18..e8a61317d63 100644
--- a/sky/check.py
+++ b/sky/check.py
@@ -18,7 +18,7 @@
 def check(
     quiet: bool = False,
     verbose: bool = False,
-    clouds: Optional[Tuple[str]] = None,
+    clouds: Optional[Iterable[str]] = None,
 ) -> None:
     echo = (lambda *_args, **_kwargs: None) if quiet else click.echo
     echo('Checking credentials to enable clouds for SkyPilot.')
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index 08045e28ab9..71397bdda56 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -247,7 +247,7 @@ def get_egress_cost(self, num_gigabytes: float):
         """
         raise NotImplementedError
 
-    def is_same_cloud(self, other: 'Cloud'):
+    def is_same_cloud(self, other: 'Cloud') -> bool:
         raise NotImplementedError
 
     def make_deploy_resources_variables(
diff --git a/sky/optimizer.py b/sky/optimizer.py
index 1cb9bc0890c..295f3428c56 100644
--- a/sky/optimizer.py
+++ b/sky/optimizer.py
@@ -4,7 +4,7 @@
 import enum
 import json
 import typing
-from typing import Any, Dict, Iterable, List, Optional, Tuple
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
 
 import colorama
 import numpy as np
@@ -120,6 +120,9 @@ def optimize(dag: 'dag_lib.Dag',
                 for a task.
             exceptions.NoCloudAccessError: if no public clouds are enabled.
         """
+        for task in dag.tasks:
+            _check_specified_clouds(task)
+
         # 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)
@@ -269,7 +272,6 @@ def _estimate_nodes_cost_or_time(
                     _fill_in_launchable_resources(
                         task=node,
                         blocked_resources=blocked_resources,
-                        try_fix_with_sky_check=True,
                         quiet=quiet))
                 node_to_candidate_map[node] = cloud_candidates
             else:
@@ -992,18 +994,18 @@ def ordinal_number(n):
         for task_id in range(len(dag.tasks)):
             task = dag.tasks[task_id]
             if isinstance(task.resources, list):
+                # For ordered resources, we try the resources in the order
+                # specified by the user.
                 local_task = local_dag.tasks[task_id]
                 for resources in task.resources:
                     # Check if there exists launchable resources
                     local_task.set_resources(resources)
-                    launchable_resources_map, _ , _ = \
+                    launchable_resources_map, _, _ = (
                         _fill_in_launchable_resources(
-                            task = local_task,
-                            blocked_resources = blocked_resources,
-                            try_fix_with_sky_check = True,
-                            quiet = False
-                    )
-                    if len(launchable_resources_map[resources]) != 0:
+                            task=local_task,
+                            blocked_resources=blocked_resources,
+                            quiet=False))
+                    if launchable_resources_map.get(resources, {}):
                         break
 
         local_graph = local_dag.get_graph()
@@ -1043,24 +1045,25 @@ def ordinal_number(n):
         if has_resources_ordered:
             best_plan = {}
             # We have to manually set the best_resources for the tasks in the
-            # original dag, to pass the optimization results
-            # to the caller, as we deep copied the dag
-            # when the dag has nodes with ordered resources.
+            # original dag, to pass the optimization results to the caller, as
+            # we deep copied the dag when the dag has nodes with ordered
+            # resources.
             for task, resources in local_best_plan.items():
                 task_idx = local_dag.tasks.index(task)
                 dag.tasks[task_idx].best_resources = resources
                 best_plan[dag.tasks[task_idx]] = resources
+
+            topo_order = list(nx.topological_sort(graph))
+            # Get the cost of each specified resources for display purpose.
+            node_to_cost_map, _ = (Optimizer._estimate_nodes_cost_or_time(
+                topo_order=topo_order,
+                minimize_cost=minimize_cost,
+                blocked_resources=blocked_resources,
+                quiet=True))
         else:
             best_plan = local_best_plan
-
-        topo_order = list(nx.topological_sort(graph)) if has_resources_ordered \
-            else local_topo_order
-        node_to_cost_map, _ = (Optimizer._estimate_nodes_cost_or_time(
-            topo_order=topo_order,
-            minimize_cost=minimize_cost,
-            blocked_resources=blocked_resources,
-            quiet=True)) if has_resources_ordered else (
-                local_node_to_cost_map, local_node_to_candidate_map)
+            topo_order = local_topo_order
+            node_to_cost_map = local_node_to_cost_map
 
         if not quiet:
             Optimizer.print_optimized_plan(graph, topo_order, best_plan,
@@ -1146,10 +1149,49 @@ def _filter_out_blocked_launchable_resources(
     return available_resources
 
 
+def _check_specified_clouds(task: task_lib.Task):
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+
+    # Recheck the enabled clouds if the task's requested resources are on a
+    # cloud that is not enabled in the cached enabled_clouds.
+    all_clouds_specified: Set[str] = set()
+    clouds_need_recheck: Set[str] = set()
+    for resources in task.resources:
+        cloud_str = str(resources.cloud)
+        if (resources.cloud is not None and
+                not clouds.cloud_in_iterable(resources.cloud, enabled_clouds)):
+            # Explicitly check again to update the enabled cloud list.
+            clouds_need_recheck.add(cloud_str)
+        all_clouds_specified.add(cloud_str)
+
+    # Explicitly check again to update the enabled cloud list.
+    sky_check.check(quiet=True, clouds=list(clouds_need_recheck))
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+    rechecked_but_disabled_clouds = (clouds_need_recheck -
+                                     {str(c) for c in enabled_clouds})
+    if rechecked_but_disabled_clouds:
+        is_or_are = 'is' if len(rechecked_but_disabled_clouds) == 1 else 'are'
+        task_name = f' {task.name!r}' if task.name is not None else ''
+        msg = (
+            f'Task{task_name} requires '
+            f'{", ".join(rechecked_but_disabled_clouds)} which {is_or_are} not '
+            'enabled. To enable access, change the task cloud requirement or '
+            f'run: {colorama.Style.BRIGHT}sky check '
+            f'{" ".join(rechecked_but_disabled_clouds)}'
+            f'{colorama.Style.RESET_ALL}')
+        if all_clouds_specified == rechecked_but_disabled_clouds:
+            # If all resources are specified with a disabled cloud, we should
+            # raise an error as no resource can satisfy the requirement.
+            # Otherwise, we should just skip the resource.
+            raise exceptions.ResourcesUnavailableError(msg)
+        logger.warning(f'{colorama.Fore.YELLOW}{msg}{colorama.Style.RESET_ALL}')
+
+
 def _fill_in_launchable_resources(
     task: task_lib.Task,
     blocked_resources: Optional[Iterable[resources_lib.Resources]],
-    try_fix_with_sky_check: bool = True,
     quiet: bool = False
 ) -> Tuple[Dict[resources_lib.Resources, List[resources_lib.Resources]],
            _PerCloudCandidates, List[str]]:
@@ -1161,10 +1203,15 @@ def _fill_in_launchable_resources(
           Resources,
         Dict mapping Cloud to a list of feasible Resources (for printing),
         Sorted list of fuzzy candidates (alternative GPU names).
+    Raises:
+      ResourcesUnavailableError: if all resources required by the task are on
+        a cloud that is not enabled.
     """
     enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
         raise_if_no_cloud_access=True)
-    launchable = collections.defaultdict(list)
+
+    launchable: Dict[resources_lib.Resources, List[resources_lib.Resources]] = (
+        collections.defaultdict(list))
     all_fuzzy_candidates = set()
     cloud_candidates: _PerCloudCandidates = collections.defaultdict(
         List[resources_lib.Resources])
@@ -1173,58 +1220,49 @@ def _fill_in_launchable_resources(
     for resources in task.resources:
         if (resources.cloud is not None and
                 not clouds.cloud_in_iterable(resources.cloud, enabled_clouds)):
-            if try_fix_with_sky_check:
-                # Explicitly check again to update the enabled cloud list.
-                sky_check.check(quiet=True)
-                return _fill_in_launchable_resources(task, blocked_resources,
-                                                     False)
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.ResourcesUnavailableError(
-                    f'Task requires {resources.cloud} which is '
-                    f'not enabled: {task}.\nTo enable access, run '
-                    f'{colorama.Style.BRIGHT}'
-                    f'sky check {colorama.Style.RESET_ALL}, or change the '
-                    'cloud requirement')
-        else:
-            clouds_list = ([resources.cloud]
-                           if resources.cloud is not None else enabled_clouds)
-            for cloud in clouds_list:
-                (feasible_resources, fuzzy_candidate_list) = (
-                    cloud.get_feasible_launchable_resources(
-                        resources, num_nodes=task.num_nodes))
-                if len(feasible_resources) > 0:
-                    # Assume feasible_resources is sorted by prices.
-                    cheapest = feasible_resources[0]
-                    # Generate region/zone-specified resources.
-                    launchable[resources].extend(
-                        _make_launchables_for_valid_region_zones(cheapest))
-                    cloud_candidates[cloud] = feasible_resources
-                else:
-                    all_fuzzy_candidates.update(fuzzy_candidate_list)
-            if len(launchable[resources]) == 0:
-                clouds_str = str(clouds_list) if len(clouds_list) > 1 else str(
-                    clouds_list[0])
-                num_node_str = ''
-                if task.num_nodes > 1:
-                    num_node_str = f'{task.num_nodes}x '
-                if not quiet:
-                    logger.info(
-                        f'No resource satisfying {num_node_str}'
-                        f'{resources.repr_with_region_zone} on {clouds_str}.')
-                if len(all_fuzzy_candidates) > 0:
-                    logger.info('Did you mean: '
-                                f'{colorama.Fore.CYAN}'
-                                f'{sorted(all_fuzzy_candidates)}'
-                                f'{colorama.Style.RESET_ALL}')
-                else:
-                    if resources.cpus is not None:
-                        logger.info('Try specifying a different CPU count, '
-                                    'or add "+" to the end of the CPU count '
-                                    'to allow for larger instances.')
-                    if resources.memory is not None:
-                        logger.info('Try specifying a different memory size, '
-                                    'or add "+" to the end of the memory size '
-                                    'to allow for larger instances.')
+            # Skip the resources that are on a cloud that is not enabled. The
+            # hint has been printed in _check_specified_clouds.
+            launchable[resources] = []
+            continue
+        clouds_list = ([resources.cloud]
+                       if resources.cloud is not None else enabled_clouds)
+        for cloud in clouds_list:
+            (feasible_resources,
+             fuzzy_candidate_list) = (cloud.get_feasible_launchable_resources(
+                 resources, num_nodes=task.num_nodes))
+            if len(feasible_resources) > 0:
+                # Assume feasible_resources is sorted by prices.
+                cheapest = feasible_resources[0]
+                # Generate region/zone-specified resources.
+                launchable[resources].extend(
+                    _make_launchables_for_valid_region_zones(cheapest))
+                cloud_candidates[cloud] = feasible_resources
+            else:
+                all_fuzzy_candidates.update(fuzzy_candidate_list)
+        if len(launchable[resources]) == 0:
+            clouds_str = str(clouds_list) if len(clouds_list) > 1 else str(
+                clouds_list[0])
+            num_node_str = ''
+            if task.num_nodes > 1:
+                num_node_str = f'{task.num_nodes}x '
+            if not quiet:
+                logger.info(
+                    f'No resource satisfying {num_node_str}'
+                    f'{resources.repr_with_region_zone} on {clouds_str}.')
+            if len(all_fuzzy_candidates) > 0:
+                logger.info('Did you mean: '
+                            f'{colorama.Fore.CYAN}'
+                            f'{sorted(all_fuzzy_candidates)}'
+                            f'{colorama.Style.RESET_ALL}')
+            else:
+                if resources.cpus is not None:
+                    logger.info('Try specifying a different CPU count, '
+                                'or add "+" to the end of the CPU count '
+                                'to allow for larger instances.')
+                if resources.memory is not None:
+                    logger.info('Try specifying a different memory size, '
+                                'or add "+" to the end of the memory size '
+                                'to allow for larger instances.')
 
         launchable[resources] = _filter_out_blocked_launchable_resources(
             launchable[resources], blocked_resources)


Code review comments:
diff_hunk: @@ -992,18 +994,18 @@ def ordinal_number(n):
         for task_id in range(len(dag.tasks)):
             task = dag.tasks[task_id]
             if isinstance(task.resources, list):
+                # For ordered resources, we try the resources in the order
+                # specified by the user.
                 local_task = local_dag.tasks[task_id]
                 for resources in task.resources:
                     # Check if there exists launchable resources
                     local_task.set_resources(resources)
-                    launchable_resources_map, _ , _ = \
+                    launchable_resources_map, _, _ = (
                         _fill_in_launchable_resources(
-                            task = local_task,
-                            blocked_resources = blocked_resources,
-                            try_fix_with_sky_check = True,
-                            quiet = False
-                    )
-                    if len(launchable_resources_map[resources]) != 0:
+                            task=local_task,
+                            blocked_resources=blocked_resources,
+                            quiet=False))
+                    if launchable_resources_map.get(resources, {}):
path: sky/optimizer.py
text: @user1:
```suggestion
                    if launchable_resources_map.get(resources, []):
```
to align with the return type?

@author:
Oops, good catch! Fixed. Thanks!

diff_hunk: @@ -1146,10 +1149,49 @@ def _filter_out_blocked_launchable_resources(
     return available_resources
 
 
+def _check_specified_clouds(task: task_lib.Task):
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+
+    # Recheck the enabled clouds if the task's requested resources are on a
+    # cloud that is not enabled in the cached enabled_clouds.
+    all_clouds_specified: Set[str] = set()
+    clouds_need_recheck: Set[str] = set()
+    for resources in task.resources:
+        cloud_str = str(resources.cloud)
+        if (resources.cloud is not None and
+                not clouds.cloud_in_iterable(resources.cloud, enabled_clouds)):
+            # Explicitly check again to update the enabled cloud list.
+            clouds_need_recheck.add(cloud_str)
+        all_clouds_specified.add(cloud_str)
+
+    # Explicitly check again to update the enabled cloud list.
+    sky_check.check(quiet=True, clouds=list(clouds_need_recheck))
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+    rechecked_but_disabled_clouds = (clouds_need_recheck -
+                                     {str(c) for c in enabled_clouds})
+    if rechecked_but_disabled_clouds:
+        is_or_are = 'is' if len(rechecked_but_disabled_clouds) == 1 else 'are'
+        task_name = f' {task.name!r}' if task.name is not None else ''
+        msg = (
+            f'Task{task_name} requires '
+            f'{", ".join(rechecked_but_disabled_clouds)} which {is_or_are} not '
+            'enabled. To enable access, change the task cloud requirement or '
+            f'run: {colorama.Style.BRIGHT}sky check '
+            f'{" ".join(rechecked_but_disabled_clouds)}'
+            f'{colorama.Style.RESET_ALL}')
+        if all_clouds_specified == rechecked_but_disabled_clouds:
+            # If all resources are specified with a disabled cloud, we should
+            # raise an error as no resource can satisfy the requirement.
+            # Otherwise, we should just skip the resource.
+            raise exceptions.ResourcesUnavailableError(msg)
+        logger.warning(f'{colorama.Fore.YELLOW}{msg}{colorama.Style.RESET_ALL}')
path: sky/optimizer.py
text: @user1:
nit: how about put this in the `if rechecked_but_disabled_clouds:` branch

@author:
It is within the branch : )

diff_hunk: @@ -1173,58 +1220,49 @@ def _fill_in_launchable_resources(
     for resources in task.resources:
         if (resources.cloud is not None and
                 not clouds.cloud_in_iterable(resources.cloud, enabled_clouds)):
-            if try_fix_with_sky_check:
-                # Explicitly check again to update the enabled cloud list.
-                sky_check.check(quiet=True)
-                return _fill_in_launchable_resources(task, blocked_resources,
-                                                     False)
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.ResourcesUnavailableError(
-                    f'Task requires {resources.cloud} which is '
-                    f'not enabled: {task}.\nTo enable access, run '
-                    f'{colorama.Style.BRIGHT}'
-                    f'sky check {colorama.Style.RESET_ALL}, or change the '
-                    'cloud requirement')
-        else:
-            clouds_list = ([resources.cloud]
-                           if resources.cloud is not None else enabled_clouds)
-            for cloud in clouds_list:
-                (feasible_resources, fuzzy_candidate_list) = (
-                    cloud.get_feasible_launchable_resources(
-                        resources, num_nodes=task.num_nodes))
-                if len(feasible_resources) > 0:
-                    # Assume feasible_resources is sorted by prices.
-                    cheapest = feasible_resources[0]
-                    # Generate region/zone-specified resources.
-                    launchable[resources].extend(
-                        _make_launchables_for_valid_region_zones(cheapest))
-                    cloud_candidates[cloud] = feasible_resources
-                else:
-                    all_fuzzy_candidates.update(fuzzy_candidate_list)
-            if len(launchable[resources]) == 0:
-                clouds_str = str(clouds_list) if len(clouds_list) > 1 else str(
-                    clouds_list[0])
-                num_node_str = ''
-                if task.num_nodes > 1:
-                    num_node_str = f'{task.num_nodes}x '
-                if not quiet:
-                    logger.info(
-                        f'No resource satisfying {num_node_str}'
-                        f'{resources.repr_with_region_zone} on {clouds_str}.')
-                if len(all_fuzzy_candidates) > 0:
-                    logger.info('Did you mean: '
-                                f'{colorama.Fore.CYAN}'
-                                f'{sorted(all_fuzzy_candidates)}'
-                                f'{colorama.Style.RESET_ALL}')
-                else:
-                    if resources.cpus is not None:
-                        logger.info('Try specifying a different CPU count, '
-                                    'or add "+" to the end of the CPU count '
-                                    'to allow for larger instances.')
-                    if resources.memory is not None:
-                        logger.info('Try specifying a different memory size, '
-                                    'or add "+" to the end of the memory size '
-                                    'to allow for larger instances.')
+            # Skip the resources that are on a cloud that is not enabled. The
+            # hint has been printed in _check_specified_clouds.
+            launchable[resources] = []
+            continue
+        clouds_list = ([resources.cloud]
+                       if resources.cloud is not None else enabled_clouds)
+        for cloud in clouds_list:
+            (feasible_resources,
+             fuzzy_candidate_list) = (cloud.get_feasible_launchable_resources(
+                 resources, num_nodes=task.num_nodes))
path: sky/optimizer.py
text: @user1:
```suggestion
             fuzzy_candidate_list) = cloud.get_feasible_launchable_resources(
                 resources, num_nodes=task.num_nodes)
```

diff_hunk: @@ -1173,58 +1220,49 @@ def _fill_in_launchable_resources(
     for resources in task.resources:
         if (resources.cloud is not None and
                 not clouds.cloud_in_iterable(resources.cloud, enabled_clouds)):
-            if try_fix_with_sky_check:
-                # Explicitly check again to update the enabled cloud list.
-                sky_check.check(quiet=True)
-                return _fill_in_launchable_resources(task, blocked_resources,
-                                                     False)
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.ResourcesUnavailableError(
-                    f'Task requires {resources.cloud} which is '
-                    f'not enabled: {task}.\nTo enable access, run '
-                    f'{colorama.Style.BRIGHT}'
-                    f'sky check {colorama.Style.RESET_ALL}, or change the '
-                    'cloud requirement')
-        else:
-            clouds_list = ([resources.cloud]
-                           if resources.cloud is not None else enabled_clouds)
-            for cloud in clouds_list:
-                (feasible_resources, fuzzy_candidate_list) = (
-                    cloud.get_feasible_launchable_resources(
-                        resources, num_nodes=task.num_nodes))
-                if len(feasible_resources) > 0:
-                    # Assume feasible_resources is sorted by prices.
-                    cheapest = feasible_resources[0]
-                    # Generate region/zone-specified resources.
-                    launchable[resources].extend(
-                        _make_launchables_for_valid_region_zones(cheapest))
-                    cloud_candidates[cloud] = feasible_resources
-                else:
-                    all_fuzzy_candidates.update(fuzzy_candidate_list)
-            if len(launchable[resources]) == 0:
-                clouds_str = str(clouds_list) if len(clouds_list) > 1 else str(
-                    clouds_list[0])
-                num_node_str = ''
-                if task.num_nodes > 1:
-                    num_node_str = f'{task.num_nodes}x '
-                if not quiet:
-                    logger.info(
-                        f'No resource satisfying {num_node_str}'
-                        f'{resources.repr_with_region_zone} on {clouds_str}.')
-                if len(all_fuzzy_candidates) > 0:
-                    logger.info('Did you mean: '
-                                f'{colorama.Fore.CYAN}'
-                                f'{sorted(all_fuzzy_candidates)}'
-                                f'{colorama.Style.RESET_ALL}')
-                else:
-                    if resources.cpus is not None:
-                        logger.info('Try specifying a different CPU count, '
-                                    'or add "+" to the end of the CPU count '
-                                    'to allow for larger instances.')
-                    if resources.memory is not None:
-                        logger.info('Try specifying a different memory size, '
-                                    'or add "+" to the end of the memory size '
-                                    'to allow for larger instances.')
+            # Skip the resources that are on a cloud that is not enabled. The
+            # hint has been printed in _check_specified_clouds.
+            launchable[resources] = []
+            continue
+        clouds_list = ([resources.cloud]
+                       if resources.cloud is not None else enabled_clouds)
+        for cloud in clouds_list:
+            (feasible_resources,
+             fuzzy_candidate_list) = (cloud.get_feasible_launchable_resources(
+                 resources, num_nodes=task.num_nodes))
+            if len(feasible_resources) > 0:
+                # Assume feasible_resources is sorted by prices.
+                cheapest = feasible_resources[0]
+                # Generate region/zone-specified resources.
+                launchable[resources].extend(
+                    _make_launchables_for_valid_region_zones(cheapest))
+                cloud_candidates[cloud] = feasible_resources
+            else:
+                all_fuzzy_candidates.update(fuzzy_candidate_list)
+        if len(launchable[resources]) == 0:
+            clouds_str = str(clouds_list) if len(clouds_list) > 1 else str(
+                clouds_list[0])
+            num_node_str = ''
+            if task.num_nodes > 1:
+                num_node_str = f'{task.num_nodes}x '
+            if not quiet:
+                logger.info(
+                    f'No resource satisfying {num_node_str}'
+                    f'{resources.repr_with_region_zone} on {clouds_str}.')
+            if len(all_fuzzy_candidates) > 0:
+                logger.info('Did you mean: '
+                            f'{colorama.Fore.CYAN}'
+                            f'{sorted(all_fuzzy_candidates)}'
+                            f'{colorama.Style.RESET_ALL}')
path: sky/optimizer.py
text: @user1:
this should be inside the `if not quiet` branch as well?

diff_hunk: @@ -120,6 +120,9 @@ def optimize(dag: 'dag_lib.Dag',
                 for a task.
             exceptions.NoCloudAccessError: if no public clouds are enabled.
         """
+        for task in dag.tasks:
+            _check_specified_clouds(task)
path: sky/optimizer.py
text: @user1:
It seems like there is a `sky check` in the `_check_specified_clouds` function, so it will be called the number of tasks times in one dag pipeline. Should we only do one sky check for a dag?

@author:
This is now fixed. Thanks for catching this!

diff_hunk: @@ -1043,24 +1045,25 @@ def ordinal_number(n):
         if has_resources_ordered:
             best_plan = {}
             # We have to manually set the best_resources for the tasks in the
-            # original dag, to pass the optimization results
-            # to the caller, as we deep copied the dag
-            # when the dag has nodes with ordered resources.
+            # original dag, to pass the optimization results to the caller, as
+            # we deep copied the dag when the dag has nodes with ordered
+            # resources.
             for task, resources in local_best_plan.items():
                 task_idx = local_dag.tasks.index(task)
                 dag.tasks[task_idx].best_resources = resources
                 best_plan[dag.tasks[task_idx]] = resources
+
+            topo_order = list(nx.topological_sort(graph))
+            # Get the cost of each specified resources for display purpose.
+            node_to_cost_map, _ = (Optimizer._estimate_nodes_cost_or_time(
+                topo_order=topo_order,
+                minimize_cost=minimize_cost,
+                blocked_resources=blocked_resources,
+                quiet=True))
path: sky/optimizer.py
text: @user1:
```suggestion
            node_to_cost_map, _ = Optimizer._estimate_nodes_cost_or_time(
                topo_order=topo_order,
                minimize_cost=minimize_cost,
                blocked_resources=blocked_resources,
                quiet=True)
```
nit

@author:
Fixed. Thanks!

diff_hunk: @@ -1146,10 +1149,49 @@ def _filter_out_blocked_launchable_resources(
     return available_resources
 
 
+def _check_specified_clouds(task: task_lib.Task):
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+
+    # Recheck the enabled clouds if the task's requested resources are on a
+    # cloud that is not enabled in the cached enabled_clouds.
+    all_clouds_specified: Set[str] = set()
+    clouds_need_recheck: Set[str] = set()
+    for resources in task.resources:
+        cloud_str = str(resources.cloud)
+        if (resources.cloud is not None and
+                not clouds.cloud_in_iterable(resources.cloud, enabled_clouds)):
+            # Explicitly check again to update the enabled cloud list.
+            clouds_need_recheck.add(cloud_str)
+        all_clouds_specified.add(cloud_str)
+
+    # Explicitly check again to update the enabled cloud list.
+    sky_check.check(quiet=True, clouds=list(clouds_need_recheck))
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+    rechecked_but_disabled_clouds = (clouds_need_recheck -
+                                     {str(c) for c in enabled_clouds})
+    if rechecked_but_disabled_clouds:
+        is_or_are = 'is' if len(rechecked_but_disabled_clouds) == 1 else 'are'
+        task_name = f' {task.name!r}' if task.name is not None else ''
+        msg = (
+            f'Task{task_name} requires '
+            f'{", ".join(rechecked_but_disabled_clouds)} which {is_or_are} not '
+            'enabled. To enable access, change the task cloud requirement or '
+            f'run: {colorama.Style.BRIGHT}sky check '
+            f'{" ".join(rechecked_but_disabled_clouds)}'
+            f'{colorama.Style.RESET_ALL}')
+        if all_clouds_specified == rechecked_but_disabled_clouds:
+            # If all resources are specified with a disabled cloud, we should
+            # raise an error as no resource can satisfy the requirement.
+            # Otherwise, we should just skip the resource.
+            raise exceptions.ResourcesUnavailableError(msg)
path: sky/optimizer.py
text: @user1:
disable traceback?

@author:
Good catch! Fixed. Thanks!

diff_hunk: @@ -1146,10 +1149,49 @@ def _filter_out_blocked_launchable_resources(
     return available_resources
 
 
+def _check_specified_clouds(task: task_lib.Task):
path: sky/optimizer.py
text: @user1:
```suggestion
def _check_specified_clouds(task: task_lib.Task) -> None:
```

diff_hunk: @@ -1146,10 +1149,49 @@ def _filter_out_blocked_launchable_resources(
     return available_resources
 
 
+def _check_specified_clouds(task: task_lib.Task):
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+
+    # Recheck the enabled clouds if the task's requested resources are on a
+    # cloud that is not enabled in the cached enabled_clouds.
+    all_clouds_specified: Set[str] = set()
+    clouds_need_recheck: Set[str] = set()
+    for resources in task.resources:
+        cloud_str = str(resources.cloud)
+        if (resources.cloud is not None and
+                not clouds.cloud_in_iterable(resources.cloud, enabled_clouds)):
+            # Explicitly check again to update the enabled cloud list.
+            clouds_need_recheck.add(cloud_str)
+        all_clouds_specified.add(cloud_str)
+
+    # Explicitly check again to update the enabled cloud list.
+    sky_check.check(quiet=True, clouds=list(clouds_need_recheck))
+    enabled_clouds = sky_check.get_cached_enabled_clouds_or_refresh(
+        raise_if_no_cloud_access=True)
+    rechecked_but_disabled_clouds = (clouds_need_recheck -
+                                     {str(c) for c in enabled_clouds})
+    if rechecked_but_disabled_clouds:
+        is_or_are = 'is' if len(rechecked_but_disabled_clouds) == 1 else 'are'
+        task_name = f' {task.name!r}' if task.name is not None else ''
+        msg = (
+            f'Task{task_name} requires '
path: sky/optimizer.py
text: @user1:
```suggestion
            f'Task{task_name} requires one of '
```

@author:
It is not actually `one of` but requested all of them : )