Repository: skypilot-org/skypilot
Pull request: [Provisioner] Introducing `best` disk tier
<!-- Describe the changes in this PR -->

This PR introduces `best` disk tier. It will automatically select the best disk tier enabled, and compatible with clouds that do not support disk tier (e.g. Lambda); in that case, it just ignores this argument and launches normally.

This is helpful when you want to use the best disk tier and don't want to disable clouds without high disk tier enabled.

Probably fixed the problem mentioned in #2251?

<!-- 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)
     - `sky launch --cloud [aws/gcp/azure/lambda] --disk-tier best` and manually check disk type used
     - `pytest tests/test_jobs.py`
     - `pytest tests/test_optimizer_dryruns.py` except #3020 
- [x] All smoke tests: `pytest tests/test_smoke.py` 
- [x] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_[aws/gcp/azure]_disk_tier` 
- [ ] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh`


Problem statement:
[Optimizer] Treat `disk_tier` as a soft-constraint?
[For discussion] If my job has a `disk_tier` specified, then certain clouds get excluded from the candidate list (and failover). E.g.,

```
# YAML
resources:
  accelerators: A100:1
  disk_size: 1000
  disk_tier: high

# sky launch task.yaml
# Lambda gpu_1x_a100_sxm4 is not considered in failover
I 07-16 15:44:52 optimizer.py:732] Considered resources (1 node):
I 07-16 15:44:52 optimizer.py:781] --------------------------------------------------------------------------------------------
I 07-16 15:44:52 optimizer.py:781]  CLOUD   INSTANCE        vCPUs   Mem(GB)   ACCELERATORS   REGION/ZONE   COST ($)   CHOSEN   
I 07-16 15:44:52 optimizer.py:781] --------------------------------------------------------------------------------------------
I 07-16 15:44:52 optimizer.py:781]  GCP     a2-highgpu-1g   12      85        A100:1         us-central1   3.67          ✔     
I 07-16 15:44:52 optimizer.py:781] --------------------------------------------------------------------------------------------
```

This is not desirable since disk_tier is only a soft-constraint on my job - even if I don't get adequate disk performance my job will still run.  In my specific case, I would have liked my job to fail-over to lambda if GCP does not have enough capacity even though lambda has slower disks.

Hints from issue discussion:

Commit message: add test_jobs tests

Patch to review:
diff --git a/docs/source/reference/yaml-spec.rst b/docs/source/reference/yaml-spec.rst
index c8dc4b1ed4f..76b6b6b923b 100644
--- a/docs/source/reference/yaml-spec.rst
+++ b/docs/source/reference/yaml-spec.rst
@@ -74,7 +74,9 @@ Available fields:
       disk_size: 256
 
       # Disk tier to use for OS (optional).
-      # Could be one of 'low', 'medium', or 'high' (default: 'medium'). Rough performance estimate:
+      # Could be one of 'low', 'medium', 'high' or 'best' (default: 'medium').
+      # if 'best' is specified, use the best disk tier enabled.
+      # Rough performance estimate:
       #   low: 500 IOPS; read 20MB/s; write 40 MB/s
       #   medium: 3000 IOPS; read 220 MB/s; write 200 MB/s
       #   high: 6000 IOPS; 340 MB/s; write 250 MB/s
diff --git a/llm/falcon/falcon.yaml b/llm/falcon/falcon.yaml
index a0bd5f97f8f..256d936d61b 100644
--- a/llm/falcon/falcon.yaml
+++ b/llm/falcon/falcon.yaml
@@ -1,7 +1,7 @@
 resources:
   accelerators: A100-80GB:1
   disk_size: 1000
-  disk_tier: high
+  disk_tier: best
 
 workdir: .
 
diff --git a/llm/llama-2/chatbot-hf.yaml b/llm/llama-2/chatbot-hf.yaml
index 874e4453b32..14f640e654f 100644
--- a/llm/llama-2/chatbot-hf.yaml
+++ b/llm/llama-2/chatbot-hf.yaml
@@ -1,7 +1,7 @@
 resources:
   accelerators: A100:1
   disk_size: 1024
-  disk_tier: high
+  disk_tier: best
   memory: 32+
 
 envs:
diff --git a/llm/llama-2/chatbot-meta.yaml b/llm/llama-2/chatbot-meta.yaml
index 03418b222e5..a0481fe760f 100644
--- a/llm/llama-2/chatbot-meta.yaml
+++ b/llm/llama-2/chatbot-meta.yaml
@@ -2,7 +2,7 @@ resources:
   memory: 32+
   accelerators: A100:1
   disk_size: 1024
-  disk_tier: high
+  disk_tier: best
 
 envs:
   MODEL_SIZE: 7
diff --git a/llm/vicuna-llama-2/serve.yaml b/llm/vicuna-llama-2/serve.yaml
index 634c3258b9a..0a98dab5d26 100644
--- a/llm/vicuna-llama-2/serve.yaml
+++ b/llm/vicuna-llama-2/serve.yaml
@@ -4,7 +4,7 @@ envs:
 resources:
   accelerators: A100:1
   disk_size: 1024
-  disk_tier: high
+  disk_tier: best
   memory: 32+
 
 file_mounts:
diff --git a/llm/vicuna/serve.yaml b/llm/vicuna/serve.yaml
index 6e252421cec..cbfdfcd4566 100644
--- a/llm/vicuna/serve.yaml
+++ b/llm/vicuna/serve.yaml
@@ -1,7 +1,7 @@
 resources:
   accelerators: A100:1
   disk_size: 1024
-  disk_tier: high
+  disk_tier: best
 
 setup: |
   conda activate chatbot
diff --git a/llm/vicuna/train.yaml b/llm/vicuna/train.yaml
index 9a9f5c8898e..4afd93f1478 100644
--- a/llm/vicuna/train.yaml
+++ b/llm/vicuna/train.yaml
@@ -2,7 +2,7 @@ resources:
   accelerators: A100-80GB:8
   disk_size: 1000
   use_spot: true
-  disk_tier: high
+  disk_tier: best
 
 num_nodes: 1
 
diff --git a/sky/cli.py b/sky/cli.py
index 362fa0edcf4..00a6d0827e6 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -69,6 +69,7 @@
 from sky.utils import env_options
 from sky.utils import kubernetes_utils
 from sky.utils import log_utils
+from sky.utils import resources_utils
 from sky.utils import rich_utils
 from sky.utils import schemas
 from sky.utils import subprocess_utils
@@ -83,6 +84,7 @@
 
 _CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
 
+_IGNORE_OPTION_WARNING = 'WARNING: {OPT}={VAL} is ignored.'
 _CLUSTER_FLAG_HELP = """\
 A cluster name. If provided, either reuse an existing cluster with that name or
 provision a new cluster with that name. Otherwise provision a new cluster with
@@ -232,11 +234,11 @@ def _interactive_node_cli_command(cli_func):
                              help=('OS disk size in GBs.'))
     disk_tier = click.option('--disk-tier',
                              default=None,
-                             type=click.Choice(['low', 'medium', 'high'],
-                                               case_sensitive=False),
+                             type=click.Choice(
+                                 resources_utils.DiskTier.supported_tiers(),
+                                 case_sensitive=False),
                              required=False,
-                             help=('OS disk tier. Could be one of "low", '
-                                   '"medium", "high". Default: medium'))
+                             help=resources_utils.DiskTier.cli_help_message())
     ports = click.option(
         '--ports',
         required=False,
@@ -314,6 +316,9 @@ def _interactive_node_cli_command(cli_func):
         instance_type_option,
         cpus,
         memory,
+        disk_size,
+        disk_tier,
+        ports,
         *([gpus] if cli_func.__name__ == 'gpunode' else []),
         *([tpus] if cli_func.__name__ == 'tpunode' else []),
         spot_option,
@@ -322,9 +327,6 @@ def _interactive_node_cli_command(cli_func):
         # Attach options
         screen_option,
         tmux_option,
-        disk_size,
-        disk_tier,
-        ports,
     ]
     decorator = functools.reduce(lambda res, f: f(res),
                                  reversed(click_decorators), cli_func)
@@ -398,6 +400,34 @@ def _merge_env_vars(env_dict: Optional[Dict[str, str]],
         help=('Number of nodes to execute the task on. '
               'Overrides the "num_nodes" config in the YAML if both are '
               'supplied.')),
+    click.option(
+        '--cpus',
+        default=None,
+        type=str,
+        required=False,
+        help=('Number of vCPUs each instance must have (e.g., '
+              '``--cpus=4`` (exactly 4) or ``--cpus=4+`` (at least 4)). '
+              'This is used to automatically select the instance type.')),
+    click.option(
+        '--memory',
+        default=None,
+        type=str,
+        required=False,
+        help=(
+            'Amount of memory each instance must have in GB (e.g., '
+            '``--memory=16`` (exactly 16GB), ``--memory=16+`` (at least 16GB))'
+        )),
+    click.option('--disk-size',
+                 default=None,
+                 type=int,
+                 required=False,
+                 help=('OS disk size in GBs.')),
+    click.option('--disk-tier',
+                 default=None,
+                 type=click.Choice(resources_utils.DiskTier.supported_tiers(),
+                                   case_sensitive=False),
+                 required=False,
+                 help=resources_utils.DiskTier.cli_help_message()),
     click.option(
         '--use-spot/--no-use-spot',
         required=False,
@@ -1258,33 +1288,6 @@ def cli():
               default=False,
               help='If used, runs locally inside a docker container.')
 @_add_click_options(_TASK_OPTIONS + _EXTRA_RESOURCES_OPTIONS)
-@click.option('--cpus',
-              default=None,
-              type=str,
-              required=False,
-              help=('Number of vCPUs each instance must have (e.g., '
-                    '``--cpus=4`` (exactly 4) or ``--cpus=4+`` (at least 4)). '
-                    'This is used to automatically select the instance type.'))
-@click.option(
-    '--memory',
-    default=None,
-    type=str,
-    required=False,
-    help=('Amount of memory each instance must have in GB (e.g., '
-          '``--memory=16`` (exactly 16GB), ``--memory=16+`` (at least 16GB))'))
-@click.option('--disk-size',
-              default=None,
-              type=int,
-              required=False,
-              help=('OS disk size in GBs.'))
-@click.option(
-    '--disk-tier',
-    default=None,
-    type=click.Choice(['low', 'medium', 'high'], case_sensitive=False),
-    required=False,
-    help=(
-        'OS disk tier. Could be one of "low", "medium", "high". Default: medium'
-    ))
 @click.option(
     '--idle-minutes-to-autostop',
     '-i',
@@ -1316,19 +1319,20 @@ def cli():
     default=False,
     is_flag=True,
     required=False,
-    # Disabling quote check here, as there seems to be a bug in pylint,
-    # which incorrectly recognizes the help string as a docstring.
-    # pylint: disable=bad-docstring-quotes
     help=('Whether to retry provisioning infinitely until the cluster is up, '
           'if we fail to launch the cluster on any possible region/cloud due '
           'to unavailability errors.'),
 )
-@click.option('--yes',
-              '-y',
-              is_flag=True,
-              default=False,
-              required=False,
-              help='Skip confirmation prompt.')
+@click.option(
+    '--yes',
+    '-y',
+    is_flag=True,
+    default=False,
+    required=False,
+    # Disabling quote check here, as there seems to be a bug in pylint,
+    # which incorrectly recognizes the help string as a docstring.
+    # pylint: disable=bad-docstring-quotes
+    help='Skip confirmation prompt.')
 @click.option('--no-setup',
               is_flag=True,
               default=False,
@@ -1486,6 +1490,10 @@ def exec(
     image_id: Optional[str],
     env_file: Optional[Dict[str, str]],
     env: List[Tuple[str, str]],
+    cpus: Optional[str],
+    memory: Optional[str],
+    disk_size: Optional[int],
+    disk_tier: Optional[str],
 ):
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Execute a task or a command on a cluster (skip setup).
@@ -1545,6 +1553,11 @@ def exec(
       sky exec mycluster --env WANDB_API_KEY python train_gpu.py
 
     """
+    for opt, val in zip(['--cpus', '--memory', '--disk-size', '--disk-tier'],
+                        [cpus, memory, disk_size, disk_tier]):
+        if val is not None:
+            click.secho(_IGNORE_OPTION_WARNING.format(OPT=opt, VAL=val),
+                        fg='yellow')
     if ports:
         raise ValueError('`ports` is not supported by `sky exec`.')
 
@@ -3549,37 +3562,10 @@ def spot():
                 **_get_shell_complete_args(_complete_file_name))
 # TODO(zhwu): Add --dryrun option to test the launch command.
 @_add_click_options(_TASK_OPTIONS + _EXTRA_RESOURCES_OPTIONS)
-@click.option('--cpus',
-              default=None,
-              type=str,
-              required=False,
-              help=('Number of vCPUs each instance must have (e.g., '
-                    '``--cpus=4`` (exactly 4) or ``--cpus=4+`` (at least 4)). '
-                    'This is used to automatically select the instance type.'))
-@click.option(
-    '--memory',
-    default=None,
-    type=str,
-    required=False,
-    help=('Amount of memory each instance must have in GB (e.g., '
-          '``--memory=16`` (exactly 16GB), ``--memory=16+`` (at least 16GB))'))
 @click.option('--spot-recovery',
               default=None,
               type=str,
               help='Spot recovery strategy to use for the managed spot task.')
-@click.option('--disk-size',
-              default=None,
-              type=int,
-              required=False,
-              help=('OS disk size in GBs.'))
-@click.option(
-    '--disk-tier',
-    default=None,
-    type=click.Choice(['low', 'medium', 'high'], case_sensitive=False),
-    required=False,
-    help=(
-        'OS disk tier. Could be one of "low", "medium", "high". Default: medium'
-    ))
 @click.option(
     '--detach-run',
     '-d',
@@ -4046,19 +4032,6 @@ def bench():
     help=('Ports to open on the cluster. '
           'If specified, overrides the "ports" config in the YAML. '),
 )
-@click.option('--disk-size',
-              default=None,
-              type=int,
-              required=False,
-              help=('OS disk size in GBs.'))
-@click.option(
-    '--disk-tier',
-    default=None,
-    type=click.Choice(['low', 'medium', 'high'], case_sensitive=False),
-    required=False,
-    help=(
-        'OS disk tier. Could be one of "low", "medium", "high". Default: medium'
-    ))
 @click.option(
     '--idle-minutes-to-autostop',
     '-i',
@@ -4093,6 +4066,8 @@ def benchmark_launch(
     image_id: Optional[str],
     env_file: Optional[Dict[str, str]],
     env: List[Tuple[str, str]],
+    cpus: Optional[str],
+    memory: Optional[str],
     disk_size: Optional[int],
     disk_tier: Optional[str],
     ports: Tuple[str],
@@ -4106,6 +4081,10 @@ def benchmark_launch(
     Alternatively, specify the benchmarking resources in your YAML (see doc),
     which allows benchmarking on many more resource fields.
     """
+    for opt, val in zip(['--cpus', '--memory'], [cpus, memory]):
+        if val is not None:
+            click.secho(_IGNORE_OPTION_WARNING.format(OPT=opt, VAL=val),
+                        fg='yellow')
     env = _merge_env_vars(env_file, env)
     record = benchmark_state.get_benchmark_from_name(benchmark)
     if record is not None:
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index fbe3947a9d6..830efd5adb0 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -17,6 +17,7 @@
 from sky.adaptors import aws
 from sky.clouds import service_catalog
 from sky.utils import common_utils
+from sky.utils import resources_utils
 from sky.utils import rich_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -113,6 +114,8 @@ class AWS(clouds.Cloud):
         'https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-quickstart.html'  # pylint: disable=line-too-long
     )
 
+    _SUPPORTED_DISK_TIERS = set(resources_utils.DiskTier)
+
     @classmethod
     def _cloud_unsupported_features(
             cls) -> Dict[clouds.CloudImplementationFeatures, str]:
@@ -322,7 +325,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         return service_catalog.get_default_instance_type(cpus=cpus,
                                                          memory=memory,
                                                          disk_tier=disk_tier,
@@ -719,27 +723,24 @@ def accelerator_in_region_or_zone(self,
             accelerator, acc_count, region, zone, 'aws')
 
     @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
-        del instance_type, disk_tier  # unused
-
-    @classmethod
-    def _get_disk_type(cls, disk_tier: str) -> str:
-        return 'standard' if disk_tier == 'low' else 'gp3'
+    def _get_disk_type(cls, disk_tier: resources_utils.DiskTier) -> str:
+        return 'standard' if disk_tier == resources_utils.DiskTier.LOW else 'gp3'
 
     @classmethod
-    def _get_disk_specs(cls, disk_tier: Optional[str]) -> Dict[str, Any]:
-        tier = disk_tier or cls._DEFAULT_DISK_TIER
+    def _get_disk_specs(
+            cls,
+            disk_tier: Optional[resources_utils.DiskTier]) -> Dict[str, Any]:
+        tier = cls.translate_disk_tier(disk_tier)
         tier2iops = {
-            'high': 7000,
-            'medium': 3500,
-            'low': 0,  # only gp3 is required to set iops
+            resources_utils.DiskTier.HIGH: 7000,
+            resources_utils.DiskTier.MEDIUM: 3500,
+            resources_utils.DiskTier.LOW: 0,  # only gp3 is required to set iops
         }
         return {
             'disk_tier': cls._get_disk_type(tier),
             'disk_iops': tier2iops[tier],
             'disk_throughput': tier2iops[tier] // 16,
-            'custom_disk_perf': tier != 'low',
+            'custom_disk_perf': tier != resources_utils.DiskTier.LOW,
         }
 
     @classmethod
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index 41f6862d9df..4e6fa6f4281 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -19,6 +19,7 @@
 from sky.clouds import service_catalog
 from sky.skylet import log_lib
 from sky.utils import common_utils
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -57,6 +58,7 @@ class Azure(clouds.Cloud):
     # names, so the limit is 64 - 4 - 7 - 10 = 43.
     # Reference: https://azure.github.io/PSRule.Rules.Azure/en/rules/Azure.ResourceGroup.Name/ # pylint: disable=line-too-long
     _MAX_CLUSTER_NAME_LEN_LIMIT = 42
+    _BEST_DISK_TIER = resources_utils.DiskTier.MEDIUM
 
     _INDENT_PREFIX = ' ' * 4
 
@@ -127,7 +129,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         return service_catalog.get_default_instance_type(cpus=cpus,
                                                          memory=memory,
                                                          disk_tier=disk_tier,
@@ -286,8 +289,8 @@ def _get_feasible_launchable_resources(
     ) -> Tuple[List['resources.Resources'], List[str]]:
 
         def failover_disk_tier(
-                instance_type: str,
-                disk_tier: Optional[str]) -> Tuple[bool, Optional[str]]:
+            instance_type: str, disk_tier: Optional[resources_utils.DiskTier]
+        ) -> Tuple[bool, Optional[resources_utils.DiskTier]]:
             """Figure out the actual disk tier to be used.
 
             Check the disk_tier specified by the user with the instance type to
@@ -302,21 +305,27 @@ def failover_disk_tier(
                 specified configuration is not a valid combination, and should
                 not be used for launching a VM.
             """
-            if disk_tier is not None:
+            if (disk_tier is not None and
+                    disk_tier != resources_utils.DiskTier.BEST):
                 ok, _ = Azure.check_disk_tier(instance_type, disk_tier)
                 return (True, disk_tier) if ok else (False, None)
-            disk_tier = clouds.Cloud._DEFAULT_DISK_TIER
-            all_tiers = {'high', 'medium', 'low'}
-            while not Azure.check_disk_tier(instance_type, disk_tier)[0]:
-                all_tiers.remove(disk_tier)
-                if not all_tiers:
-                    # No available disk_tier found the specified instance_type
-                    return (False, None)
-                disk_tier = list(all_tiers)[0]
-            if disk_tier != clouds.Cloud._DEFAULT_DISK_TIER:
-                return True, disk_tier
+            # All tiers in reversed order, i.e. from best to worst.
+            # We will failover along this order.
+            all_tiers = list(reversed(resources_utils.DiskTier))
+            if disk_tier is None:
+                # Start failover from the default disk tier.
+                start_index = all_tiers.index(clouds.Cloud._DEFAULT_DISK_TIER)
             else:
-                return True, None
+                # Start failover from the best disk tier.
+                start_index = all_tiers.index(
+                    Azure.translate_disk_tier(disk_tier))
+            while start_index < len(all_tiers):
+                disk_tier = all_tiers[start_index]
+                ok, _ = Azure.check_disk_tier(instance_type, disk_tier)
+                if ok:
+                    return (True, disk_tier)
+                start_index += 1
+            return False, None
 
         if resources.instance_type is not None:
             assert resources.is_launchable(), resources
@@ -527,11 +536,12 @@ def _is_s_series(cls, instance_type: Optional[str]) -> bool:
         return 's' in x.group(5)
 
     @classmethod
-    def check_disk_tier(cls, instance_type: Optional[str],
-                        disk_tier: Optional[str]) -> Tuple[bool, str]:
-        if disk_tier is None:
+    def check_disk_tier(
+            cls, instance_type: Optional[str],
+            disk_tier: Optional[resources_utils.DiskTier]) -> Tuple[bool, str]:
+        if disk_tier is None or disk_tier == resources_utils.DiskTier.BEST:
             return True, ''
-        if disk_tier == 'high':
+        if disk_tier == resources_utils.DiskTier.HIGH:
             return False, ('Azure disk_tier=high is not supported now. '
                            'Please use disk_tier={low, medium} instead.')
         # Only S-series supported premium ssd
@@ -546,22 +556,23 @@ def check_disk_tier(cls, instance_type: Optional[str],
         return True, ''
 
     @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
+    def check_disk_tier_enabled(cls, instance_type: Optional[str],
+                                disk_tier: resources_utils.DiskTier) -> None:
         ok, msg = cls.check_disk_tier(instance_type, disk_tier)
         if not ok:
             with ux_utils.print_exception_no_traceback():
-                raise ValueError(msg)
+                raise exceptions.NotSupportedError(msg)
 
     @classmethod
-    def _get_disk_type(cls, disk_tier: Optional[str]) -> str:
-        tier = disk_tier or cls._DEFAULT_DISK_TIER
+    def _get_disk_type(cls,
+                       disk_tier: Optional[resources_utils.DiskTier]) -> str:
+        tier = cls.translate_disk_tier(disk_tier)
         # TODO(tian): Maybe use PremiumV2_LRS/UltraSSD_LRS? Notice these two
         # cannot be used as OS disks so we might need data disk support
         tier2name = {
-            'high': 'Disabled',
-            'medium': 'Premium_LRS',
-            'low': 'Standard_LRS',
+            resources_utils.DiskTier.HIGH: 'Disabled',
+            resources_utils.DiskTier.MEDIUM: 'Premium_LRS',
+            resources_utils.DiskTier.LOW: 'Standard_LRS',
         }
         return tier2name[tier]
 
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index d9aeca0cca2..83838837a96 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -10,6 +10,7 @@
 from sky.clouds import service_catalog
 from sky.skylet import constants
 from sky.utils import log_utils
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -58,7 +59,9 @@ class Cloud:
     """A cloud provider."""
 
     _REPR = '<Cloud>'
-    _DEFAULT_DISK_TIER = 'medium'
+    _DEFAULT_DISK_TIER = resources_utils.DiskTier.MEDIUM
+    _BEST_DISK_TIER = resources_utils.DiskTier.HIGH
+    _SUPPORTED_DISK_TIERS = {resources_utils.DiskTier.BEST}
 
     @classmethod
     def _cloud_unsupported_features(
@@ -240,7 +243,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         """Returns the default instance type with the given #vCPUs, memory and
         disk tier.
 
@@ -252,10 +256,10 @@ def get_default_instance_type(
         memory.  If 'memory=4+', this method returns the default instance
         type with 4GB or more memory.
 
-        If disk_rier='medium', this method returns the default instance type
-        that support medium disk tier.
+        If disk_tier=DiskTier.MEDIUM, this method returns the default instance
+        type that support medium disk tier.
 
-        When cpus is None, memory is None or disk tier is None, this method will
+        When cpus is None, memory is None or disk_tier is None, this method will
         never return None. This method may return None if the cloud's default
         instance family does not have a VM with the given number of vCPUs
         (e.g., when cpus='7') or does not have a VM with the give disk tier
@@ -499,14 +503,28 @@ def check_cluster_name_is_valid(cls, cluster_name: str) -> None:
                     f'{valid_regex}')
 
     @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
+    def check_disk_tier_enabled(cls, instance_type: Optional[str],
+                                disk_tier: resources_utils.DiskTier) -> None:
         """Errors out if the disk tier is not supported by the cloud provider.
 
         Raises:
             exceptions.NotSupportedError: If the disk tier is not supported.
         """
-        raise NotImplementedError
+        del instance_type  # unused
+        if disk_tier not in cls._SUPPORTED_DISK_TIERS:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    f'{disk_tier} is not supported by {cls._REPR}.')
+
+    @classmethod
+    def translate_disk_tier(
+        cls, disk_tier: Optional[resources_utils.DiskTier]
+    ) -> resources_utils.DiskTier:
+        if disk_tier is None:
+            return cls._DEFAULT_DISK_TIER
+        if disk_tier == resources_utils.DiskTier.BEST:
+            return cls._BEST_DISK_TIER
+        return disk_tier
 
     @classmethod
     def _check_instance_type_accelerators_combination(
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index 948dd716297..b171c6e7431 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -21,6 +21,7 @@
 from sky.clouds import service_catalog
 from sky.skylet import log_lib
 from sky.utils import common_utils
+from sky.utils import resources_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
 
@@ -225,6 +226,8 @@ class GCP(clouds.Cloud):
         'https://skypilot.readthedocs.io/en/latest/getting-started/installation.html#google-cloud-platform-gcp'  # pylint: disable=line-too-long
     )
 
+    _SUPPORTED_DISK_TIERS = set(resources_utils.DiskTier)
+
     def __init__(self):
         super().__init__()
 
@@ -424,7 +427,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         return service_catalog.get_default_instance_type(cpus=cpus,
                                                          memory=memory,
                                                          disk_tier=disk_tier,
@@ -978,17 +982,13 @@ def _check_instance_type_accelerators_combination(
             'gcp')
 
     @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
-        del instance_type, disk_tier  # unused
-
-    @classmethod
-    def _get_disk_type(cls, disk_tier: Optional[str]) -> str:
-        tier = disk_tier or cls._DEFAULT_DISK_TIER
+    def _get_disk_type(cls,
+                       disk_tier: Optional[resources_utils.DiskTier]) -> str:
+        tier = cls.translate_disk_tier(disk_tier)
         tier2name = {
-            'high': 'pd-ssd',
-            'medium': 'pd-balanced',
-            'low': 'pd-standard',
+            resources_utils.DiskTier.HIGH: 'pd-ssd',
+            resources_utils.DiskTier.MEDIUM: 'pd-balanced',
+            resources_utils.DiskTier.LOW: 'pd-standard',
         }
         return tier2name[tier]
 
diff --git a/sky/clouds/ibm.py b/sky/clouds/ibm.py
index 8b3376784e7..3fe784d48fc 100644
--- a/sky/clouds/ibm.py
+++ b/sky/clouds/ibm.py
@@ -12,6 +12,7 @@
 from sky.adaptors import ibm
 from sky.adaptors.ibm import CREDENTIAL_FILE
 from sky.clouds import service_catalog
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -251,7 +252,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         return service_catalog.get_default_instance_type(cpus=cpus,
                                                          memory=memory,
                                                          disk_tier=disk_tier,
@@ -421,11 +423,6 @@ def check_credentials(cls) -> Tuple[bool, Optional[str]]:
         except Exception as e:
             return (False, f'{str(e)}' + help_str)
 
-    @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
-        del instance_type, disk_tier  # unused
-
     def get_credential_file_mounts(self) -> Dict[str, str]:
         """Returns a {remote:local} credential path mapping
          written to the cluster's file_mounts segment
diff --git a/sky/clouds/kubernetes.py b/sky/clouds/kubernetes.py
index d91d2a5ecba..f80ba95bbae 100644
--- a/sky/clouds/kubernetes.py
+++ b/sky/clouds/kubernetes.py
@@ -12,6 +12,7 @@
 from sky.clouds import service_catalog
 from sky.utils import common_utils
 from sky.utils import kubernetes_utils
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -127,10 +128,11 @@ def get_port(cls, svc_name) -> int:
         return kubernetes_utils.get_port(svc_name, ns)
 
     @classmethod
-    def get_default_instance_type(cls,
-                                  cpus: Optional[str] = None,
-                                  memory: Optional[str] = None,
-                                  disk_tier: Optional[str] = None) -> str:
+    def get_default_instance_type(
+            cls,
+            cpus: Optional[str] = None,
+            memory: Optional[str] = None,
+            disk_tier: Optional[resources_utils.DiskTier] = None) -> str:
         # TODO(romilb): In the future, we may want to move the instance type
         #  selection + availability checking to a kubernetes_catalog module.
         del disk_tier  # Unused.
diff --git a/sky/clouds/lambda_cloud.py b/sky/clouds/lambda_cloud.py
index fbbcd8e8b5f..8c50a42b910 100644
--- a/sky/clouds/lambda_cloud.py
+++ b/sky/clouds/lambda_cloud.py
@@ -6,10 +6,10 @@
 import requests
 
 from sky import clouds
-from sky import exceptions
 from sky import status_lib
 from sky.clouds import service_catalog
 from sky.clouds.utils import lambda_utils
+from sky.utils import resources_utils
 
 if typing.TYPE_CHECKING:
     # Renaming to avoid shadowing variables.
@@ -127,7 +127,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         return service_catalog.get_default_instance_type(cpus=cpus,
                                                          memory=memory,
                                                          disk_tier=disk_tier,
@@ -274,12 +275,6 @@ def accelerator_in_region_or_zone(self,
     def regions(cls) -> List['clouds.Region']:
         return service_catalog.regions(clouds='lambda')
 
-    @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
-        raise exceptions.NotSupportedError(
-            'Lambda does not support disk tiers.')
-
     @classmethod
     def query_status(cls, name: str, tag_filters: Dict[str, str],
                      region: Optional[str], zone: Optional[str],
diff --git a/sky/clouds/local.py b/sky/clouds/local.py
index 3773583702b..fa2ae983d2d 100644
--- a/sky/clouds/local.py
+++ b/sky/clouds/local.py
@@ -3,7 +3,7 @@
 from typing import Dict, Iterator, List, Optional, Tuple
 
 from sky import clouds
-from sky import exceptions
+from sky.utils import resources_utils
 
 if typing.TYPE_CHECKING:
     # Renaming to avoid shadowing variables.
@@ -38,6 +38,8 @@ class Local(clouds.Cloud):
             ('Docker image is not supported in Local. '
              'You can try running docker command inside the '
              '`run` section in task.yaml.'),
+        clouds.CloudImplementationFeatures.CUSTOM_DISK_TIER:
+            ('Custom disk tier is not supported for Local.'),
         clouds.CloudImplementationFeatures.OPEN_PORTS:
             ('Opening ports is not supported for Local.'),
     }
@@ -107,10 +109,11 @@ def is_same_cloud(self, other: clouds.Cloud) -> bool:
         return isinstance(other, Local)
 
     @classmethod
-    def get_default_instance_type(cls,
-                                  cpus: Optional[str] = None,
-                                  memory: Optional[str] = None,
-                                  disk_tier: Optional[str] = None) -> str:
+    def get_default_instance_type(
+            cls,
+            cpus: Optional[str] = None,
+            memory: Optional[str] = None,
+            disk_tier: Optional[resources_utils.DiskTier] = None) -> str:
         # There is only "1" instance type for local cloud: on-prem
         del cpus, memory, disk_tier  # Unused.
         return Local._DEFAULT_INSTANCE_TYPE
@@ -196,9 +199,3 @@ def validate_region_zone(self, region: Optional[str], zone: Optional[str]):
             raise ValueError(f'Region {region!r} does not match the Local'
                              ' cloud region {Local.LOCAL_REGION.name!r}.')
         return region, zone
-
-    @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
-        raise exceptions.NotSupportedError(
-            'Local cloud does not support disk tiers.')
diff --git a/sky/clouds/oci.py b/sky/clouds/oci.py
index ed1b129d86e..a390c378878 100644
--- a/sky/clouds/oci.py
+++ b/sky/clouds/oci.py
@@ -18,6 +18,7 @@
 from sky.clouds import service_catalog
 from sky.clouds.utils import oci_utils
 from sky.utils import common_utils
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -41,6 +42,8 @@ class OCI(clouds.Cloud):
 
     _INDENT_PREFIX = '    '
 
+    _SUPPORTED_DISK_TIERS = set(resources_utils.DiskTier)
+
     @classmethod
     def _cloud_unsupported_features(
             cls) -> Dict[clouds.CloudImplementationFeatures, str]:
@@ -160,7 +163,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         return service_catalog.get_default_instance_type(cpus=cpus,
                                                          memory=memory,
                                                          disk_tier=disk_tier,
@@ -516,14 +520,13 @@ def _get_default_image(self, region_name: str, instance_type: str) -> str:
             'ERR: No image found in catalog for region '
             f'{region_name}. Try update your default image_id settings.')
 
-    @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
-        # All the disk_tier are supported for any instance_type
-        del instance_type, disk_tier  # unused
-
-    def get_vpu_from_disktier(self, cpus: Optional[float],
-                              disk_tier: Optional[str]) -> int:
+    def get_vpu_from_disktier(
+            self, cpus: Optional[float],
+            disk_tier: Optional[resources_utils.DiskTier]) -> int:
+        # Only normalize the disk_tier if it is not None, since OCI have
+        # different default disk tier according to #vCPU.
+        if disk_tier is not None:
+            disk_tier = OCI.translate_disk_tier(disk_tier)
         vpu = oci_utils.oci_config.BOOT_VOLUME_VPU[disk_tier]
         if cpus is None:
             return vpu
diff --git a/sky/clouds/scp.py b/sky/clouds/scp.py
index be792bfe0c1..d3e80674a8c 100644
--- a/sky/clouds/scp.py
+++ b/sky/clouds/scp.py
@@ -14,6 +14,7 @@
 from sky import status_lib
 from sky.clouds import service_catalog
 from sky.clouds.utils import scp_utils
+from sky.utils import resources_utils
 
 if typing.TYPE_CHECKING:
     # Renaming to avoid shadowing variables.
@@ -144,7 +145,8 @@ def get_default_instance_type(
             cls,
             cpus: Optional[str] = None,
             memory: Optional[str] = None,
-            disk_tier: Optional[str] = None) -> Optional[str]:
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
         return service_catalog.get_default_instance_type(cpus=cpus,
                                                          memory=memory,
                                                          disk_tier=disk_tier,
diff --git a/sky/clouds/service_catalog/__init__.py b/sky/clouds/service_catalog/__init__.py
index 03d62144103..7d1a074eda6 100644
--- a/sky/clouds/service_catalog/__init__.py
+++ b/sky/clouds/service_catalog/__init__.py
@@ -8,6 +8,7 @@
 from sky.clouds.service_catalog.constants import CATALOG_SCHEMA_VERSION
 from sky.clouds.service_catalog.constants import HOSTED_CATALOG_DIR_URL
 from sky.clouds.service_catalog.constants import LOCAL_CATALOG_DIR
+from sky.utils import resources_utils
 
 if typing.TYPE_CHECKING:
     from sky.clouds import cloud
@@ -176,7 +177,8 @@ def get_vcpus_mem_from_instance_type(
 
 def get_default_instance_type(cpus: Optional[str] = None,
                               memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None,
+                              disk_tier: Optional[
+                                  resources_utils.DiskTier] = None,
                               clouds: CloudFilter = None) -> Optional[str]:
     """Returns the cloud's default instance type for given #vCPUs and memory.
 
diff --git a/sky/clouds/service_catalog/aws_catalog.py b/sky/clouds/service_catalog/aws_catalog.py
index 89334d7cb21..6332a4ee810 100644
--- a/sky/clouds/service_catalog/aws_catalog.py
+++ b/sky/clouds/service_catalog/aws_catalog.py
@@ -20,6 +20,7 @@
 from sky.clouds.service_catalog import config
 from sky.clouds.service_catalog.data_fetchers import fetch_aws
 from sky.utils import common_utils
+from sky.utils import resources_utils
 
 if typing.TYPE_CHECKING:
     from sky.clouds import cloud
@@ -219,9 +220,10 @@ def get_vcpus_mem_from_instance_type(
                                                         instance_type)
 
 
-def get_default_instance_type(cpus: Optional[str] = None,
-                              memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None) -> Optional[str]:
+def get_default_instance_type(
+        cpus: Optional[str] = None,
+        memory: Optional[str] = None,
+        disk_tier: Optional[resources_utils.DiskTier] = None) -> Optional[str]:
     del disk_tier  # unused
     if cpus is None and memory is None:
         cpus = f'{_DEFAULT_NUM_VCPUS}+'
diff --git a/sky/clouds/service_catalog/azure_catalog.py b/sky/clouds/service_catalog/azure_catalog.py
index d7188cb0be5..5422eac8f5d 100644
--- a/sky/clouds/service_catalog/azure_catalog.py
+++ b/sky/clouds/service_catalog/azure_catalog.py
@@ -9,6 +9,7 @@
 from sky import clouds as cloud_lib
 from sky.clouds import Azure
 from sky.clouds.service_catalog import common
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 # The frequency of pulling the latest catalog from the cloud provider.
@@ -104,9 +105,10 @@ def _get_instance_family(instance_type: str) -> str:
     return instance_family
 
 
-def get_default_instance_type(cpus: Optional[str] = None,
-                              memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None) -> Optional[str]:
+def get_default_instance_type(
+        cpus: Optional[str] = None,
+        memory: Optional[str] = None,
+        disk_tier: Optional[resources_utils.DiskTier] = None) -> Optional[str]:
     if cpus is None and memory is None:
         cpus = f'{_DEFAULT_NUM_VCPUS}+'
     if memory is None:
diff --git a/sky/clouds/service_catalog/gcp_catalog.py b/sky/clouds/service_catalog/gcp_catalog.py
index 1e0828f6f5f..6985923bb4b 100644
--- a/sky/clouds/service_catalog/gcp_catalog.py
+++ b/sky/clouds/service_catalog/gcp_catalog.py
@@ -12,6 +12,7 @@
 from sky import exceptions
 from sky import sky_logging
 from sky.clouds.service_catalog import common
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -233,9 +234,10 @@ def get_vcpus_mem_from_instance_type(
     return common.get_vcpus_mem_from_instance_type_impl(_df, instance_type)
 
 
-def get_default_instance_type(cpus: Optional[str] = None,
-                              memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None) -> Optional[str]:
+def get_default_instance_type(
+        cpus: Optional[str] = None,
+        memory: Optional[str] = None,
+        disk_tier: Optional[resources_utils.DiskTier] = None) -> Optional[str]:
     del disk_tier  # unused
     if cpus is None and memory is None:
         cpus = f'{_DEFAULT_NUM_VCPUS}+'
diff --git a/sky/clouds/service_catalog/ibm_catalog.py b/sky/clouds/service_catalog/ibm_catalog.py
index b1bd84023e4..e0daad070b7 100644
--- a/sky/clouds/service_catalog/ibm_catalog.py
+++ b/sky/clouds/service_catalog/ibm_catalog.py
@@ -10,6 +10,7 @@
 from sky.adaptors import ibm
 from sky.clouds import cloud
 from sky.clouds.service_catalog import common
+from sky.utils import resources_utils
 
 logger = sky_logging.init_logger(__name__)
 
@@ -97,9 +98,10 @@ def list_accelerators(
                                          case_sensitive)
 
 
-def get_default_instance_type(cpus: Optional[str] = None,
-                              memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None) -> Optional[str]:
+def get_default_instance_type(
+        cpus: Optional[str] = None,
+        memory: Optional[str] = None,
+        disk_tier: Optional[resources_utils.DiskTier] = None) -> Optional[str]:
     del disk_tier  # unused
     if cpus is None and memory is None:
         cpus = f'{_DEFAULT_NUM_VCPUS}+'
diff --git a/sky/clouds/service_catalog/lambda_catalog.py b/sky/clouds/service_catalog/lambda_catalog.py
index ea9fd23ea91..9c362610ffd 100644
--- a/sky/clouds/service_catalog/lambda_catalog.py
+++ b/sky/clouds/service_catalog/lambda_catalog.py
@@ -7,6 +7,7 @@
 from typing import Dict, List, Optional, Tuple
 
 from sky.clouds.service_catalog import common
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -61,9 +62,10 @@ def get_vcpus_mem_from_instance_type(
     return common.get_vcpus_mem_from_instance_type_impl(_df, instance_type)
 
 
-def get_default_instance_type(cpus: Optional[str] = None,
-                              memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None) -> Optional[str]:
+def get_default_instance_type(
+        cpus: Optional[str] = None,
+        memory: Optional[str] = None,
+        disk_tier: Optional[resources_utils.DiskTier] = None) -> Optional[str]:
     del disk_tier  # unused
     if cpus is None and memory is None:
         cpus = f'{_DEFAULT_NUM_VCPUS}+'
diff --git a/sky/clouds/service_catalog/oci_catalog.py b/sky/clouds/service_catalog/oci_catalog.py
index 07ba864564c..29c6d9211a8 100644
--- a/sky/clouds/service_catalog/oci_catalog.py
+++ b/sky/clouds/service_catalog/oci_catalog.py
@@ -17,6 +17,7 @@
 from sky.adaptors import oci as oci_adaptor
 from sky.clouds.service_catalog import common
 from sky.clouds.utils import oci_utils
+from sky.utils import resources_utils
 
 if typing.TYPE_CHECKING:
     import pandas as pd
@@ -105,9 +106,10 @@ def get_hourly_cost(instance_type: str,
                                        region, zone)
 
 
-def get_default_instance_type(cpus: Optional[str] = None,
-                              memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None) -> Optional[str]:
+def get_default_instance_type(
+        cpus: Optional[str] = None,
+        memory: Optional[str] = None,
+        disk_tier: Optional[resources_utils.DiskTier] = None) -> Optional[str]:
     del disk_tier  # unused
     if cpus is None:
         cpus = f'{oci_utils.oci_config.DEFAULT_NUM_VCPUS}+'
diff --git a/sky/clouds/service_catalog/scp_catalog.py b/sky/clouds/service_catalog/scp_catalog.py
index c71ec177f58..244adfd542b 100644
--- a/sky/clouds/service_catalog/scp_catalog.py
+++ b/sky/clouds/service_catalog/scp_catalog.py
@@ -8,6 +8,7 @@
 from typing import Dict, List, Optional, Tuple
 
 from sky.clouds.service_catalog import common
+from sky.utils import resources_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
@@ -61,9 +62,10 @@ def get_vcpus_mem_from_instance_type(
     return common.get_vcpus_mem_from_instance_type_impl(_df, instance_type)
 
 
-def get_default_instance_type(cpus: Optional[str] = None,
-                              memory: Optional[str] = None,
-                              disk_tier: Optional[str] = None) -> Optional[str]:
+def get_default_instance_type(
+        cpus: Optional[str] = None,
+        memory: Optional[str] = None,
+        disk_tier: Optional[resources_utils.DiskTier] = None) -> Optional[str]:
     del disk_tier  # unused
     if cpus is None and memory is None:
         cpus = str(_DEFAULT_NUM_VCPUS)
diff --git a/sky/clouds/utils/oci_utils.py b/sky/clouds/utils/oci_utils.py
index 34696f77110..3d11bab24da 100644
--- a/sky/clouds/utils/oci_utils.py
+++ b/sky/clouds/utils/oci_utils.py
@@ -7,6 +7,7 @@
 import os
 
 from sky import skypilot_config
+from sky.utils import resources_utils
 
 logger = logging.getLogger(__name__)
 
@@ -69,9 +70,9 @@ class OCIConfig:
     # disk_tier to OCI VPU mapping
     BOOT_VOLUME_VPU = {
         None: DISK_TIER_MEDIUM,  # Default to medium
-        'low': DISK_TIER_LOW,
-        'medium': DISK_TIER_MEDIUM,
-        'high': DISK_TIER_HIGH,
+        resources_utils.DiskTier.LOW: DISK_TIER_LOW,
+        resources_utils.DiskTier.MEDIUM: DISK_TIER_MEDIUM,
+        resources_utils.DiskTier.HIGH: DISK_TIER_HIGH,
     }
 
     @classmethod
diff --git a/sky/resources.py b/sky/resources.py
index 3153ed24e9c..f757cde4b8a 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -3,7 +3,6 @@
 from typing import Dict, List, Optional, Set, Tuple, Union
 
 import colorama
-from typing_extensions import Literal
 
 from sky import clouds
 from sky import global_user_state
@@ -58,7 +57,7 @@ def __init__(
         zone: Optional[str] = None,
         image_id: Union[Dict[str, str], str, None] = None,
         disk_size: Optional[int] = None,
-        disk_tier: Optional[Literal['high', 'medium', 'low']] = None,
+        disk_tier: Optional[Union[str, resources_utils.DiskTier]] = None,
         ports: Optional[Union[int, str, List[str], Tuple[str]]] = None,
         # Internal use only.
         _docker_login_config: Optional[docker_utils.DockerLoginConfig] = None,
@@ -168,6 +167,8 @@ def __init__(
                 }
         self._is_image_managed = _is_image_managed
 
+        if isinstance(disk_tier, str):
+            disk_tier = resources_utils.DiskTier(str(disk_tier).lower())
         self._disk_tier = disk_tier
 
         if ports is not None:
@@ -258,7 +259,7 @@ def __repr__(self) -> str:
 
         disk_tier = ''
         if self.disk_tier is not None:
-            disk_tier = f', disk_tier={self.disk_tier}'
+            disk_tier = f', disk_tier={self.disk_tier.value}'
 
         disk_size = ''
         if self.disk_size != _DEFAULT_DISK_SIZE_GB:
@@ -375,7 +376,7 @@ def image_id(self) -> Optional[Dict[str, str]]:
         return self._image_id
 
     @property
-    def disk_tier(self) -> str:
+    def disk_tier(self) -> resources_utils.DiskTier:
         return self._disk_tier
 
     @property
@@ -788,13 +789,6 @@ def _try_validate_image_id(self) -> None:
     def _try_validate_disk_tier(self) -> None:
         if self.disk_tier is None:
             return
-        if self.disk_tier not in ['high', 'medium', 'low']:
-            with ux_utils.print_exception_no_traceback():
-                raise ValueError(
-                    f'Invalid disk_tier {self.disk_tier}. '
-                    'Please use one of "high", "medium", or "low".')
-        if self.instance_type is None:
-            return
         if self.cloud is not None:
             self.cloud.check_disk_tier_enabled(self.instance_type,
                                                self.disk_tier)
@@ -936,14 +930,6 @@ def less_demanding_than(
         if self.use_spot_specified and self.use_spot != other.use_spot:
             return False
 
-        if self.disk_tier is not None:
-            if other.disk_tier is None:
-                return False
-            if self.disk_tier != other.disk_tier:
-                types = ['low', 'medium', 'high']
-                return types.index(self.disk_tier) < types.index(
-                    other.disk_tier)
-
         if check_ports:
             if self.ports is not None:
                 if other.ports is None:
@@ -953,6 +939,18 @@ def less_demanding_than(
                 if not self_ports <= other_ports:
                     return False
 
+        if self.disk_tier is not None:
+            if other.disk_tier is None:
+                return False
+            if other.disk_tier == resources_utils.DiskTier.BEST:
+                return True
+            # Here, BEST tier means the best we can get; for a launched
+            # cluster, the best (and only) tier we can get is the launched
+            # cluster's tier.
+            if self.disk_tier == resources_utils.DiskTier.BEST:
+                return True
+            return self.disk_tier <= other.disk_tier
+
         # self <= other
         return True
 
@@ -1039,7 +1037,8 @@ def get_required_cloud_features(
         features = set()
         if self.use_spot:
             features.add(clouds.CloudImplementationFeatures.SPOT_INSTANCE)
-        if self.disk_tier is not None:
+        if (self.disk_tier is not None and
+                self.disk_tier != resources_utils.DiskTier.BEST):
             features.add(clouds.CloudImplementationFeatures.CUSTOM_DISK_TIER)
         if self.extract_docker_image() is not None:
             features.add(clouds.CloudImplementationFeatures.DOCKER_IMAGE)
diff --git a/sky/utils/resources_utils.py b/sky/utils/resources_utils.py
index d97518f6320..ec199ee8800 100644
--- a/sky/utils/resources_utils.py
+++ b/sky/utils/resources_utils.py
@@ -1,4 +1,5 @@
 """Utility functions for resources."""
+import enum
 import itertools
 from typing import List, Optional, Set
 
@@ -10,6 +11,29 @@
                   'Please use a port number between 1 and 65535.')
 
 
+class DiskTier(enum.Enum):
+    """All disk tiers supported by SkyPilot."""
+    LOW = 'low'
+    MEDIUM = 'medium'
+    HIGH = 'high'
+    BEST = 'best'
+
+    @classmethod
+    def supported_tiers(cls) -> List[str]:
+        return [tier.value for tier in cls]
+
+    @classmethod
+    def cli_help_message(cls) -> str:
+        return (
+            f'OS disk tier. Could be one of {", ".join(cls.supported_tiers())}'
+            f'. if {cls.BEST.value} is specified, use the best possible disk '
+            f'tier. Default: {cls.MEDIUM.value}')
+
+    def __le__(self, other: 'DiskTier') -> bool:
+        types = list(DiskTier)
+        return types.index(self) <= types.index(other)
+
+
 def check_port_str(port: str) -> None:
     if not port.isdigit():
         with ux_utils.print_exception_no_traceback():
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index a1be762506d..a854fa40a0d 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -5,6 +5,7 @@
 from sky import exceptions
 from sky import global_user_state
 from sky.utils import db_utils
+from sky.utils import resources_utils
 
 
 class TestExecutionOnExistingClusters:
@@ -22,10 +23,12 @@ def _mock_db_conn(self, monkeypatch, tmp_path):
     def _mock_cluster_state(self, _mock_db_conn, enable_all_clouds):
         """Add clusters to the global state.
 
-        This fixture adss three clusters to the global state:
+        This fixture adds five clusters to the global state:
         - test-cluster1: AWS, 2x p4d.24xlarge (8x A100)
         - test-cluster2: GCP, 1x n1-highmem-64, 4x V100
         - test-cluster3: Azure, 1x Standard_D4s_v3 (CPU only)
+        - test-disk-tier1: AWS, 1x m6i.2xlarge, with best disk tier
+        - test-disk-tier2: GCP, 1x n2-standard-8, with medium disk tier
         """
         assert 'state.db' not in global_user_state._DB.db_path
 
@@ -74,6 +77,38 @@ def _mock_cluster_state(self, _mock_db_conn, enable_all_clouds):
             handle,
             requested_resources={handle.launched_resources},
             ready=False)
+        handle = backends.CloudVmRayResourceHandle(
+            cluster_name='test-disk-tier1',
+            cluster_name_on_cloud='test-disk-tier1',
+            cluster_yaml='/tmp/disk-tier1.yaml',
+            launched_nodes=1,
+            launched_resources=sky.Resources(
+                sky.AWS(),
+                instance_type='m6i.2xlarge',
+                region='us-east-1',
+                zone='us-east-1a',
+                disk_tier=resources_utils.DiskTier.BEST))
+        global_user_state.add_or_update_cluster(
+            'test-disk-tier1',
+            handle,
+            requested_resources={handle.launched_resources},
+            ready=True)
+        handle = backends.CloudVmRayResourceHandle(
+            cluster_name='test-disk-tier2',
+            cluster_name_on_cloud='test-disk-tier2',
+            cluster_yaml='/tmp/disk-tier2.yaml',
+            launched_nodes=1,
+            launched_resources=sky.Resources(
+                sky.GCP(),
+                instance_type='n2-standard-8',
+                region='us-west1',
+                zone='us-west1-a',
+                disk_tier=resources_utils.DiskTier.MEDIUM))
+        global_user_state.add_or_update_cluster(
+            'test-disk-tier2',
+            handle,
+            requested_resources={handle.launched_resources},
+            ready=True)
 
     def test_launch_exec(self, _mock_cluster_state, monkeypatch):
         """Test launch and exec on existing clusters.
@@ -109,6 +144,24 @@ def test_launch_exec(self, _mock_cluster_state, monkeypatch):
         task = sky.Task(run='echo hi')
         sky.exec(task, cluster_name='test-cluster3', dryrun=True)
 
+        task = sky.Task(run='echo hi')
+        task.set_resources(
+            sky.Resources(disk_tier=resources_utils.DiskTier.BEST))
+        sky.launch(task, cluster_name='test-disk-tier1', dryrun=True)
+        sky.exec(task, cluster_name='test-disk-tier1', dryrun=True)
+        sky.launch(task, cluster_name='test-disk-tier2', dryrun=True)
+        sky.exec(task, cluster_name='test-disk-tier2', dryrun=True)
+        task.set_resources(
+            sky.Resources(disk_tier=resources_utils.DiskTier.LOW))
+        sky.launch(task, cluster_name='test-disk-tier1', dryrun=True)
+        sky.exec(task, cluster_name='test-disk-tier1', dryrun=True)
+        sky.launch(task, cluster_name='test-disk-tier2', dryrun=True)
+        sky.exec(task, cluster_name='test-disk-tier2', dryrun=True)
+        task.set_resources(
+            sky.Resources(disk_tier=resources_utils.DiskTier.HIGH))
+        sky.launch(task, cluster_name='test-disk-tier1', dryrun=True)
+        sky.exec(task, cluster_name='test-disk-tier1', dryrun=True)
+
     def _run_launch_exec_with_error(self, task, cluster_name):
         with pytest.raises(exceptions.ResourcesMismatchError) as e:
             sky.launch(task, cluster_name=cluster_name, dryrun=True)
@@ -137,3 +190,8 @@ def test_launch_exec_mismatch(self, _mock_cluster_state, monkeypatch):
 
         task.set_resources(sky.Resources(sky.GCP()))
         self._run_launch_exec_with_error(task, 'test-cluster1')
+
+        # Disk tier mismatch
+        task.set_resources(
+            sky.Resources(disk_tier=resources_utils.DiskTier.HIGH))
+        self._run_launch_exec_with_error(task, 'test-disk-tier2')
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 8f8bd78487e..ee2f41e59d3 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -51,6 +51,7 @@
 from sky.data.data_utils import Rclone
 from sky.skylet import events
 from sky.utils import common_utils
+from sky.utils import resources_utils
 from sky.utils import subprocess_utils
 
 # To avoid the second smoke test reusing the cluster launched in the first
@@ -2520,23 +2521,23 @@ def _get_aws_query_command(region, instance_id, field, expected):
                 f'--filters Name=attachment.instance-id,Values={instance_id} '
                 f'--query Volumes[*].{field} | grep {expected} ; ')
 
-    for disk_tier in ['low', 'medium', 'high']:
+    for disk_tier in list(resources_utils.DiskTier):
         specs = AWS._get_disk_specs(disk_tier)
-        name = _get_cluster_name() + '-' + disk_tier
+        name = _get_cluster_name() + '-' + disk_tier.value
         name_on_cloud = common_utils.make_cluster_name_on_cloud(
             name, sky.AWS.max_cluster_name_length())
         region = 'us-east-2'
         test = Test(
-            'aws-disk-tier',
+            'aws-disk-tier-' + disk_tier.value,
             [
                 f'sky launch -y -c {name} --cloud aws --region {region} '
-                f'--disk-tier {disk_tier} echo "hello sky"',
+                f'--disk-tier {disk_tier.value} echo "hello sky"',
                 f'id=`aws ec2 describe-instances --region {region} --filters '
                 f'Name=tag:ray-cluster-name,Values={name_on_cloud} --query '
                 f'Reservations[].Instances[].InstanceId --output text`; ' +
                 _get_aws_query_command(region, '$id', 'VolumeType',
                                        specs['disk_tier']) +
-                ('' if disk_tier == 'low' else
+                ('' if disk_tier == resources_utils.DiskTier.LOW else
                  (_get_aws_query_command(region, '$id', 'Iops',
                                          specs['disk_iops']) +
                   _get_aws_query_command(region, '$id', 'Throughput',
@@ -2550,17 +2551,17 @@ def _get_aws_query_command(region, instance_id, field, expected):
 
 @pytest.mark.gcp
 def test_gcp_disk_tier():
-    for disk_tier in ['low', 'medium', 'high']:
+    for disk_tier in list(resources_utils.DiskTier):
         type = GCP._get_disk_type(disk_tier)
-        name = _get_cluster_name() + '-' + disk_tier
+        name = _get_cluster_name() + '-' + disk_tier.value
         name_on_cloud = common_utils.make_cluster_name_on_cloud(
             name, sky.GCP.max_cluster_name_length())
         region = 'us-west2'
         test = Test(
-            'gcp-disk-tier',
+            'gcp-disk-tier-' + disk_tier.value,
             [
                 f'sky launch -y -c {name} --cloud gcp --region {region} '
-                f'--disk-tier {disk_tier} echo "hello sky"',
+                f'--disk-tier {disk_tier.value} echo "hello sky"',
                 f'name=`gcloud compute instances list --filter='
                 f'"labels.ray-cluster-name:{name_on_cloud}" '
                 '--format="value(name)"`; '
@@ -2575,17 +2576,20 @@ def test_gcp_disk_tier():
 
 @pytest.mark.azure
 def test_azure_disk_tier():
-    for disk_tier in ['low', 'medium']:
+    for disk_tier in list(resources_utils.DiskTier):
+        if disk_tier == resources_utils.DiskTier.HIGH:
+            # Azure does not support high disk tier.
+            continue
         type = Azure._get_disk_type(disk_tier)
-        name = _get_cluster_name() + '-' + disk_tier
+        name = _get_cluster_name() + '-' + disk_tier.value
         name_on_cloud = common_utils.make_cluster_name_on_cloud(
             name, sky.Azure.max_cluster_name_length())
         region = 'westus2'
         test = Test(
-            'azure-disk-tier',
+            'azure-disk-tier-' + disk_tier.value,
             [
                 f'sky launch -y -c {name} --cloud azure --region {region} '
-                f'--disk-tier {disk_tier} echo "hello sky"',
+                f'--disk-tier {disk_tier.value} echo "hello sky"',
                 f'az resource list --tag ray-cluster-name={name_on_cloud} --query '
                 f'"[?type==\'Microsoft.Compute/disks\'].sku.name" '
                 f'--output tsv | grep {type}'
@@ -2596,6 +2600,28 @@ def test_azure_disk_tier():
         run_one_test(test)
 
 
+@pytest.mark.azure
+def test_azure_best_tier_failover():
+    type = Azure._get_disk_type(resources_utils.DiskTier.LOW)
+    name = _get_cluster_name()
+    name_on_cloud = common_utils.make_cluster_name_on_cloud(
+        name, sky.Azure.max_cluster_name_length())
+    region = 'westus2'
+    test = Test(
+        'azure-best-tier-failover',
+        [
+            f'sky launch -y -c {name} --cloud azure --region {region} '
+            f'--disk-tier best --instance-type Standard_D8_v5 echo "hello sky"',
+            f'az resource list --tag ray-cluster-name={name_on_cloud} --query '
+            f'"[?type==\'Microsoft.Compute/disks\'].sku.name" '
+            f'--output tsv | grep {type}',
+        ],
+        f'sky down -y {name}',
+        timeout=20 * 60,  # 20 mins  (it takes around ~12 mins)
+    )
+    run_one_test(test)
+
+
 # ------ Testing Zero Quota Failover ------
 @pytest.mark.aws
 def test_aws_zero_quota_failover():


Code review comments:
diff_hunk: @@ -953,6 +939,18 @@ def less_demanding_than(
                 if not self_ports <= other_ports:
                     return False
 
+        if self.disk_tier is not None:
path: sky/resources.py
text: @user1:
Is there a reason, `disk_tier` check needs to go after `port` check? It would be nice to have a comment for future reference.

@author:
Originally it was for convenient (we can `return True` if it is the last field to check). Now I changed the order back 🫡

diff_hunk: @@ -168,6 +167,8 @@ def __init__(
                 }
         self._is_image_managed = _is_image_managed
 
+        if isinstance(disk_tier, str):
+            disk_tier = resources_utils.DiskTier(str(disk_tier).lower())
path: sky/resources.py
text: @user1:
Should we update `__setstate__` for backward compatibility if a previous cluster was stored in the local db with a `str` disk tier?

diff_hunk: @@ -499,14 +503,28 @@ def check_cluster_name_is_valid(cls, cluster_name: str) -> None:
                     f'{valid_regex}')
 
     @classmethod
-    def check_disk_tier_enabled(cls, instance_type: str,
-                                disk_tier: str) -> None:
+    def check_disk_tier_enabled(cls, instance_type: Optional[str],
+                                disk_tier: resources_utils.DiskTier) -> None:
         """Errors out if the disk tier is not supported by the cloud provider.
 
         Raises:
             exceptions.NotSupportedError: If the disk tier is not supported.
         """
-        raise NotImplementedError
+        del instance_type  # unused
+        if disk_tier not in cls._SUPPORTED_DISK_TIERS:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    f'{disk_tier} is not supported by {cls._REPR}.')
+
+    @classmethod
+    def translate_disk_tier(
path: sky/clouds/cloud.py
text: @user1:
Should this function be private?

diff_hunk: @@ -302,21 +305,27 @@ def failover_disk_tier(
                 specified configuration is not a valid combination, and should
                 not be used for launching a VM.
             """
-            if disk_tier is not None:
+            if (disk_tier is not None and
+                    disk_tier != resources_utils.DiskTier.BEST):
                 ok, _ = Azure.check_disk_tier(instance_type, disk_tier)
                 return (True, disk_tier) if ok else (False, None)
-            disk_tier = clouds.Cloud._DEFAULT_DISK_TIER
-            all_tiers = {'high', 'medium', 'low'}
-            while not Azure.check_disk_tier(instance_type, disk_tier)[0]:
-                all_tiers.remove(disk_tier)
-                if not all_tiers:
-                    # No available disk_tier found the specified instance_type
-                    return (False, None)
-                disk_tier = list(all_tiers)[0]
-            if disk_tier != clouds.Cloud._DEFAULT_DISK_TIER:
-                return True, disk_tier
+            # All tiers in reversed order, i.e. from best to worst.
+            # We will failover along this order.
+            all_tiers = list(reversed(resources_utils.DiskTier))
+            if disk_tier is None:
+                # Start failover from the default disk tier.
+                start_index = all_tiers.index(clouds.Cloud._DEFAULT_DISK_TIER)
             else:
-                return True, None
+                # Start failover from the best disk tier.
+                start_index = all_tiers.index(
+                    Azure.translate_disk_tier(disk_tier))
path: sky/clouds/azure.py
text: @user1:
If we start with best tier, would the following always be `True`? How does the failover work then, if it does not fill in the actual disk_tier needed? It seems the later logic will not do failover for `BEST` disk_tier.

@author:
To discuss: I'm thinking maybe we could disable the auto-failover for all non-best tiers in Azure? The current behavior is basically the same in Azure...

diff_hunk: @@ -302,21 +305,27 @@ def failover_disk_tier(
                 specified configuration is not a valid combination, and should
                 not be used for launching a VM.
             """
-            if disk_tier is not None:
+            if (disk_tier is not None and
+                    disk_tier != resources_utils.DiskTier.BEST):
                 ok, _ = Azure.check_disk_tier(instance_type, disk_tier)
                 return (True, disk_tier) if ok else (False, None)
-            disk_tier = clouds.Cloud._DEFAULT_DISK_TIER
-            all_tiers = {'high', 'medium', 'low'}
-            while not Azure.check_disk_tier(instance_type, disk_tier)[0]:
-                all_tiers.remove(disk_tier)
-                if not all_tiers:
-                    # No available disk_tier found the specified instance_type
-                    return (False, None)
-                disk_tier = list(all_tiers)[0]
-            if disk_tier != clouds.Cloud._DEFAULT_DISK_TIER:
-                return True, disk_tier
+            # All tiers in reversed order, i.e. from best to worst.
+            # We will failover along this order.
+            all_tiers = list(reversed(resources_utils.DiskTier))
+            if disk_tier is None:
+                # Start failover from the default disk tier.
+                start_index = all_tiers.index(clouds.Cloud._DEFAULT_DISK_TIER)
path: sky/clouds/azure.py
text: @user1:
Why do we make this change? Does this mean we will never try the disk_tier that is worse than default disk tier?