Repository: skypilot-org/skypilot
Pull request: Implement config validation
<!-- Describe the changes in this PR -->

Closes #2474 

Implements validation for the `~/.sky/config.yaml` file. If there is an error, an exception is thrown and the user is warned.

To avoid circular imports, I had to refactor the code a bit, moving validation code into `common_utils` and moving the imports in `sky.util.schemas` to within the functions.

<!-- 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] Pass existing config tests, added new unit tests
- [ ] All smoke tests: `pytest tests/test_smoke.py` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
- [ ] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh`


Problem statement:
[config] Pre-checks for `~/.sky/config.yaml`
Currently, we don't have a way to pre-check the `~/.sky/config.yaml`, making it very easy to miss spell or setting wrong configs in the yaml. We can apply the json schema check to the `~/.sky/config.yaml` to make sure there is no unexpected settings.

A user met the issue when specifying the spot controller's resources requirement, but missing one section (`resources`):
```
spot:
  controller:
    cloud: gcp
    region: us-west1
```

Hints from issue discussion:
Can I pick this up?
#2600 ran into this.
Let me know if you're still interested in picking this up @djokester! Happy to guide you.

Commit message: Add test for invalid enum in config

Patch to review:
diff --git a/sky/authentication.py b/sky/authentication.py
index 442c1abb372..4ab88fe28a4 100644
--- a/sky/authentication.py
+++ b/sky/authentication.py
@@ -42,6 +42,7 @@
 from sky.adaptors import ibm
 from sky.clouds.utils import lambda_utils
 from sky.utils import common_utils
+from sky.utils import kubernetes_enums
 from sky.utils import kubernetes_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -351,12 +352,12 @@ def _get_unique_key_name():
 def setup_kubernetes_authentication(config: Dict[str, Any]) -> Dict[str, Any]:
     # Default ssh session is established with kubectl port-forwarding with
     # ClusterIP service.
-    nodeport_mode = kubernetes_utils.KubernetesNetworkingMode.NODEPORT
-    port_forward_mode = kubernetes_utils.KubernetesNetworkingMode.PORTFORWARD
+    nodeport_mode = kubernetes_enums.KubernetesNetworkingMode.NODEPORT
+    port_forward_mode = kubernetes_enums.KubernetesNetworkingMode.PORTFORWARD
     network_mode_str = skypilot_config.get_nested(('kubernetes', 'networking'),
                                                   port_forward_mode.value)
     try:
-        network_mode = kubernetes_utils.KubernetesNetworkingMode.from_str(
+        network_mode = kubernetes_enums.KubernetesNetworkingMode.from_str(
             network_mode_str)
     except ValueError as e:
         # Add message saying "Please check: ~/.sky/config.yaml" to the error
@@ -392,14 +393,14 @@ def setup_kubernetes_authentication(config: Dict[str, Any]) -> Dict[str, Any]:
 
     ssh_jump_name = clouds.Kubernetes.SKY_SSH_JUMP_NAME
     if network_mode == nodeport_mode:
-        service_type = kubernetes_utils.KubernetesServiceType.NODEPORT
+        service_type = kubernetes_enums.KubernetesServiceType.NODEPORT
     elif network_mode == port_forward_mode:
         kubernetes_utils.check_port_forward_mode_dependencies()
         # Using `kubectl port-forward` creates a direct tunnel to jump pod and
         # does not require opening any ports on Kubernetes nodes. As a result,
         # the service can be a simple ClusterIP service which we access with
         # `kubectl port-forward`.
-        service_type = kubernetes_utils.KubernetesServiceType.CLUSTERIP
+        service_type = kubernetes_enums.KubernetesServiceType.CLUSTERIP
     else:
         # This should never happen because we check for this in from_str above.
         raise ValueError(f'Unsupported networking mode: {network_mode_str}')
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index d1e3381a800..65876ef194b 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -1,6 +1,5 @@
 """Util constants/functions for the backends."""
 from datetime import datetime
-import difflib
 import enum
 import getpass
 import json
@@ -19,7 +18,6 @@
 import colorama
 import filelock
 import jinja2
-import jsonschema
 from packaging import version
 import requests
 from requests import adapters
@@ -53,7 +51,6 @@
 from sky.utils import timeline
 from sky.utils import tpu_utils
 from sky.utils import ux_utils
-from sky.utils import validator
 
 if typing.TYPE_CHECKING:
     from sky import resources
@@ -2758,60 +2755,6 @@ def stop_handler(signum, frame):
         raise KeyboardInterrupt(exceptions.SIGTSTP_CODE)
 
 
-def validate_schema(obj, schema, err_msg_prefix='', skip_none=True):
-    """Validates an object against a given JSON schema.
-
-    Args:
-        obj: The object to validate.
-        schema: The JSON schema against which to validate the object.
-        err_msg_prefix: The string to prepend to the error message if
-          validation fails.
-        skip_none: If True, removes fields with value None from the object
-          before validation. This is useful for objects that will never contain
-          None because yaml.safe_load() loads empty fields as None.
-
-    Raises:
-        ValueError: if the object does not match the schema.
-    """
-    if skip_none:
-        obj = {k: v for k, v in obj.items() if v is not None}
-    err_msg = None
-    try:
-        validator.SchemaValidator(schema).validate(obj)
-    except jsonschema.ValidationError as e:
-        if e.validator == 'additionalProperties':
-            if tuple(e.schema_path) == ('properties', 'envs',
-                                        'additionalProperties'):
-                # Hack. Here the error is Task.envs having some invalid keys. So
-                # we should not print "unsupported field".
-                #
-                # This will print something like:
-                # 'hello world' does not match any of the regexes: <regex>
-                err_msg = (err_msg_prefix +
-                           'The `envs` field contains invalid keys:\n' +
-                           e.message)
-            else:
-                err_msg = err_msg_prefix + 'The following fields are invalid:'
-                known_fields = set(e.schema.get('properties', {}).keys())
-                for field in e.instance:
-                    if field not in known_fields:
-                        most_similar_field = difflib.get_close_matches(
-                            field, known_fields, 1)
-                        if most_similar_field:
-                            err_msg += (f'\nInstead of {field!r}, did you mean '
-                                        f'{most_similar_field[0]!r}?')
-                        else:
-                            err_msg += f'\nFound unsupported field {field!r}.'
-        else:
-            # Example e.json_path value: '$.resources'
-            err_msg = (err_msg_prefix + e.message +
-                       f'. Check problematic field(s): {e.json_path}')
-
-    if err_msg:
-        with ux_utils.print_exception_no_traceback():
-            raise ValueError(err_msg)
-
-
 def check_public_cloud_enabled():
     """Checks if any of the public clouds is enabled.
 
diff --git a/sky/backends/onprem_utils.py b/sky/backends/onprem_utils.py
index 05f015d7372..b1da3f97f79 100644
--- a/sky/backends/onprem_utils.py
+++ b/sky/backends/onprem_utils.py
@@ -73,9 +73,9 @@ def check_and_get_local_clusters(suppress_error: bool = False) -> List[str]:
         with open(path, 'r') as f:
             yaml_config = yaml.safe_load(f)
             if not suppress_error:
-                backend_utils.validate_schema(yaml_config,
-                                              schemas.get_cluster_schema(),
-                                              'Invalid cluster YAML: ')
+                common_utils.validate_schema(yaml_config,
+                                             schemas.get_cluster_schema(),
+                                             'Invalid cluster YAML: ')
             user_config = yaml_config['auth']
             cluster_name = yaml_config['cluster']['name']
         sky_local_path = SKY_USER_LOCAL_CONFIG_PATH
diff --git a/sky/cli.py b/sky/cli.py
index ca1e8a20c3e..666090f3305 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -3501,8 +3501,8 @@ def admin_deploy(clusterspec_yaml: str):
     clusterspec_yaml = ' '.join(clusterspec_yaml)
     assert clusterspec_yaml
     is_yaml, yaml_config = _check_yaml(clusterspec_yaml)
-    backend_utils.validate_schema(yaml_config, schemas.get_cluster_schema(),
-                                  'Invalid cluster YAML: ')
+    common_utils.validate_schema(yaml_config, schemas.get_cluster_schema(),
+                                 'Invalid cluster YAML: ')
     if not is_yaml:
         raise ValueError('Must specify cluster config')
     assert yaml_config is not None, (is_yaml, yaml_config)
diff --git a/sky/data/storage.py b/sky/data/storage.py
index 9dd755a1bba..1064f06f9a2 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -20,12 +20,12 @@
 from sky.adaptors import cloudflare
 from sky.adaptors import gcp
 from sky.adaptors import ibm
-from sky.backends import backend_utils
 from sky.data import data_transfer
 from sky.data import data_utils
 from sky.data import mounting_utils
 from sky.data import storage_utils
 from sky.data.data_utils import Rclone
+from sky.utils import common_utils
 from sky.utils import rich_utils
 from sky.utils import schemas
 from sky.utils import ux_utils
@@ -865,8 +865,8 @@ def warn_for_git_dir(source: str):
 
     @classmethod
     def from_yaml_config(cls, config: Dict[str, Any]) -> 'Storage':
-        backend_utils.validate_schema(config, schemas.get_storage_schema(),
-                                      'Invalid storage YAML: ')
+        common_utils.validate_schema(config, schemas.get_storage_schema(),
+                                     'Invalid storage YAML: ')
 
         name = config.pop('name', None)
         source = config.pop('source', None)
diff --git a/sky/resources.py b/sky/resources.py
index 38c677485c9..e74f25e9de7 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -11,11 +11,11 @@
 from sky import sky_logging
 from sky import skypilot_config
 from sky import spot
-from sky.backends import backend_utils
 from sky.clouds import service_catalog
 from sky.provision import docker_utils
 from sky.skylet import constants
 from sky.utils import accelerator_registry
+from sky.utils import common_utils
 from sky.utils import log_utils
 from sky.utils import resources_utils
 from sky.utils import schemas
@@ -1109,8 +1109,8 @@ def from_yaml_config(cls, config: Optional[Dict[str, str]]) -> 'Resources':
         if config is None:
             return Resources()
 
-        backend_utils.validate_schema(config, schemas.get_resources_schema(),
-                                      'Invalid resources YAML: ')
+        common_utils.validate_schema(config, schemas.get_resources_schema(),
+                                     'Invalid resources YAML: ')
 
         resources_fields = {}
         resources_fields['cloud'] = clouds.CLOUD_REGISTRY.from_str(
diff --git a/sky/skypilot_config.py b/sky/skypilot_config.py
index 11772f04ddc..d0163f77379 100644
--- a/sky/skypilot_config.py
+++ b/sky/skypilot_config.py
@@ -51,6 +51,7 @@
 from sky import sky_logging
 from sky.clouds import cloud_registry
 from sky.utils import common_utils
+from sky.utils import schemas
 
 # The config path is discovered in this order:
 #
@@ -160,6 +161,11 @@ def _try_load_config() -> None:
             logger.debug(f'Config loaded:\n{pprint.pformat(_dict)}')
         except yaml.YAMLError as e:
             logger.error(f'Error in loading config file ({config_path}):', e)
+        if _dict is not None:
+            common_utils.validate_schema(_dict,
+                                         schemas.get_config_schema(),
+                                         'Invalid config YAML: ',
+                                         skip_none=False)
 
         for cloud in cloud_registry.CLOUD_REGISTRY:
             _syntax_check_for_ssh_proxy_command(cloud)
diff --git a/sky/task.py b/sky/task.py
index 8d8b465448a..5adb3fe7ac5 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -350,8 +350,8 @@ def from_yaml_config(
         if envs is not None and isinstance(envs, dict):
             config['envs'] = {str(k): str(v) for k, v in envs.items()}
 
-        backend_utils.validate_schema(config, schemas.get_task_schema(),
-                                      'Invalid task YAML: ')
+        common_utils.validate_schema(config, schemas.get_task_schema(),
+                                     'Invalid task YAML: ')
 
         # Fill in any Task.envs into file_mounts (src/dst paths, storage
         # name/source).
diff --git a/sky/utils/common_utils.py b/sky/utils/common_utils.py
index 3d529e4be47..142b8e84b86 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -1,5 +1,6 @@
 """Utils shared between all of sky"""
 
+import difflib
 import functools
 import getpass
 import hashlib
@@ -16,10 +17,13 @@
 import uuid
 
 import colorama
+import jsonschema
 import yaml
 
 from sky import sky_logging
 from sky.skylet import constants
+from sky.utils import ux_utils
+from sky.utils import validator
 
 _USER_HASH_FILE = os.path.expanduser('~/.sky/user_hash')
 USER_HASH_LENGTH = 8
@@ -489,3 +493,57 @@ def format_float(num: Union[float, int], precision: int = 1) -> str:
     if isinstance(num, int):
         return str(num)
     return '{:.0f}'.format(num) if num.is_integer() else f'{num:.{precision}f}'
+
+
+def validate_schema(obj, schema, err_msg_prefix='', skip_none=True):
+    """Validates an object against a given JSON schema.
+
+    Args:
+        obj: The object to validate.
+        schema: The JSON schema against which to validate the object.
+        err_msg_prefix: The string to prepend to the error message if
+          validation fails.
+        skip_none: If True, removes fields with value None from the object
+          before validation. This is useful for objects that will never contain
+          None because yaml.safe_load() loads empty fields as None.
+
+    Raises:
+        ValueError: if the object does not match the schema.
+    """
+    if skip_none:
+        obj = {k: v for k, v in obj.items() if v is not None}
+    err_msg = None
+    try:
+        validator.SchemaValidator(schema).validate(obj)
+    except jsonschema.ValidationError as e:
+        if e.validator == 'additionalProperties':
+            if tuple(e.schema_path) == ('properties', 'envs',
+                                        'additionalProperties'):
+                # Hack. Here the error is Task.envs having some invalid keys. So
+                # we should not print "unsupported field".
+                #
+                # This will print something like:
+                # 'hello world' does not match any of the regexes: <regex>
+                err_msg = (err_msg_prefix +
+                           'The `envs` field contains invalid keys:\n' +
+                           e.message)
+            else:
+                err_msg = err_msg_prefix + 'The following fields are invalid:'
+                known_fields = set(e.schema.get('properties', {}).keys())
+                for field in e.instance:
+                    if field not in known_fields:
+                        most_similar_field = difflib.get_close_matches(
+                            field, known_fields, 1)
+                        if most_similar_field:
+                            err_msg += (f'\nInstead of {field!r}, did you mean '
+                                        f'{most_similar_field[0]!r}?')
+                        else:
+                            err_msg += f'\nFound unsupported field {field!r}.'
+        else:
+            # Example e.json_path value: '$.resources'
+            err_msg = (err_msg_prefix + e.message +
+                       f'. Check problematic field(s): {e.json_path}')
+
+    if err_msg:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(err_msg)
diff --git a/sky/utils/kubernetes_enums.py b/sky/utils/kubernetes_enums.py
new file mode 100644
index 00000000000..f3be89e5db9
--- /dev/null
+++ b/sky/utils/kubernetes_enums.py
@@ -0,0 +1,29 @@
+"""Kubernetes enums for SkyPilot."""
+import enum
+
+
+class KubernetesNetworkingMode(enum.Enum):
+    """Enum for the different types of networking modes for accessing
+    jump pods.
+    """
+    NODEPORT = 'nodeport'
+    PORTFORWARD = 'portforward'
+
+    @classmethod
+    def from_str(cls, mode: str) -> 'KubernetesNetworkingMode':
+        """Returns the enum value for the given string."""
+        if mode.lower() == cls.NODEPORT.value:
+            return cls.NODEPORT
+        elif mode.lower() == cls.PORTFORWARD.value:
+            return cls.PORTFORWARD
+        else:
+            raise ValueError(f'Unsupported kubernetes networking mode: '
+                             f'{mode}. The mode must be either '
+                             f'\'{cls.PORTFORWARD.value}\' or '
+                             f'\'{cls.NODEPORT.value}\'. ')
+
+
+class KubernetesServiceType(enum.Enum):
+    """Enum for the different types of services."""
+    NODEPORT = 'NodePort'
+    CLUSTERIP = 'ClusterIP'
diff --git a/sky/utils/kubernetes_utils.py b/sky/utils/kubernetes_utils.py
index afd5b0affb2..a0fc754a421 100644
--- a/sky/utils/kubernetes_utils.py
+++ b/sky/utils/kubernetes_utils.py
@@ -1,5 +1,4 @@
 """Kubernetes utilities for SkyPilot."""
-import enum
 import math
 import os
 import re
@@ -17,6 +16,7 @@
 from sky.backends import backend_utils
 from sky.utils import common_utils
 from sky.utils import env_options
+from sky.utils import kubernetes_enums
 from sky.utils import ux_utils
 
 DEFAULT_NAMESPACE = 'default'
@@ -37,33 +37,6 @@
 logger = sky_logging.init_logger(__name__)
 
 
-class KubernetesNetworkingMode(enum.Enum):
-    """Enum for the different types of networking modes for accessing
-    jump pods.
-    """
-    NODEPORT = 'nodeport'
-    PORTFORWARD = 'portforward'
-
-    @classmethod
-    def from_str(cls, mode: str) -> 'KubernetesNetworkingMode':
-        """Returns the enum value for the given string."""
-        if mode.lower() == cls.NODEPORT.value:
-            return cls.NODEPORT
-        elif mode.lower() == cls.PORTFORWARD.value:
-            return cls.PORTFORWARD
-        else:
-            raise ValueError(f'Unsupported kubernetes networking mode: '
-                             f'{mode}. The mode must be either '
-                             f'\'{cls.PORTFORWARD.value}\' or '
-                             f'\'{cls.NODEPORT.value}\'. ')
-
-
-class KubernetesServiceType(enum.Enum):
-    """Enum for the different types of services."""
-    NODEPORT = 'NodePort'
-    CLUSTERIP = 'ClusterIP'
-
-
 class GPULabelFormatter:
     """Base class to define a GPU label formatter for a Kubernetes cluster
 
@@ -435,8 +408,9 @@ def get_port(svc_name: str, namespace: str) -> int:
     return head_service.spec.ports[0].node_port
 
 
-def get_external_ip(network_mode: Optional[KubernetesNetworkingMode]):
-    if network_mode == KubernetesNetworkingMode.PORTFORWARD:
+def get_external_ip(
+        network_mode: Optional[kubernetes_enums.KubernetesNetworkingMode]):
+    if network_mode == kubernetes_enums.KubernetesNetworkingMode.PORTFORWARD:
         return '127.0.0.1'
     # Return the IP address of the first node with an external IP
     nodes = kubernetes.core_api().list_node().items
@@ -704,10 +678,10 @@ def construct_ssh_jump_command(private_key_path: str,
     return ssh_jump_proxy_command
 
 
-def get_ssh_proxy_command(private_key_path: str, ssh_jump_name: str,
-                          network_mode: KubernetesNetworkingMode,
-                          namespace: str, port_fwd_proxy_cmd_path: str,
-                          port_fwd_proxy_cmd_template: str) -> str:
+def get_ssh_proxy_command(
+        private_key_path: str, ssh_jump_name: str,
+        network_mode: kubernetes_enums.KubernetesNetworkingMode, namespace: str,
+        port_fwd_proxy_cmd_path: str, port_fwd_proxy_cmd_template: str) -> str:
     """Generates the SSH proxy command to connect through the SSH jump pod.
 
     By default, establishing an SSH connection creates a communication
@@ -755,7 +729,7 @@ def get_ssh_proxy_command(private_key_path: str, ssh_jump_name: str,
     """
     # Fetch IP to connect to for the jump svc
     ssh_jump_ip = get_external_ip(network_mode)
-    if network_mode == KubernetesNetworkingMode.NODEPORT:
+    if network_mode == kubernetes_enums.KubernetesNetworkingMode.NODEPORT:
         ssh_jump_port = get_port(ssh_jump_name, namespace)
         ssh_jump_proxy_command = construct_ssh_jump_command(
             private_key_path, ssh_jump_ip, ssh_jump_port=ssh_jump_port)
@@ -776,7 +750,7 @@ def get_ssh_proxy_command(private_key_path: str, ssh_jump_name: str,
 
 
 def setup_ssh_jump_svc(ssh_jump_name: str, namespace: str,
-                       service_type: KubernetesServiceType):
+                       service_type: kubernetes_enums.KubernetesServiceType):
     """Sets up Kubernetes service resource to access for SSH jump pod.
 
     This method acts as a necessary complement to be run along with
@@ -815,10 +789,14 @@ def setup_ssh_jump_svc(ssh_jump_name: str, namespace: str,
                     name=ssh_jump_name, namespace=namespace)
                 kubernetes.core_api().create_namespaced_service(
                     namespace, content['service_spec'])
-                port_forward_mode = KubernetesNetworkingMode.PORTFORWARD.value
-                nodeport_mode = KubernetesNetworkingMode.NODEPORT.value
-                clusterip_svc = KubernetesServiceType.CLUSTERIP.value
-                nodeport_svc = KubernetesServiceType.NODEPORT.value
+                port_forward_mode = (
+                    kubernetes_enums.KubernetesNetworkingMode.PORTFORWARD.value)
+                nodeport_mode = (
+                    kubernetes_enums.KubernetesNetworkingMode.NODEPORT.value)
+                clusterip_svc = (
+                    kubernetes_enums.KubernetesServiceType.CLUSTERIP.value)
+                nodeport_svc = (
+                    kubernetes_enums.KubernetesServiceType.NODEPORT.value)
                 curr_network_mode = port_forward_mode \
                     if curr_svc_type == clusterip_svc else nodeport_mode
                 new_network_mode = nodeport_mode \
@@ -1001,8 +979,8 @@ def check_port_forward_mode_dependencies() -> None:
             with ux_utils.print_exception_no_traceback():
                 raise RuntimeError(
                     f'`{name}` is required to setup Kubernetes cloud with '
-                    f'`{KubernetesNetworkingMode.PORTFORWARD.value}` default '
-                    'networking mode and it is not installed. '
+                    f'`{kubernetes_enums.KubernetesNetworkingMode.PORTFORWARD.value}` '  # pylint: disable=line-too-long
+                    'default networking mode and it is not installed. '
                     'On Debian/Ubuntu, install it with:\n'
                     f'  $ sudo apt install {name}\n'
                     f'On MacOS, install it with: \n'
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index 131563189dd..449a09b2b18 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -4,11 +4,11 @@
 https://json-schema.org/
 """
 
-from sky.clouds import cloud_registry
-from sky.data import storage
-
 
 def get_resources_schema():
+    # To avoid circular imports, only import when needed.
+    # pylint: disable=import-outside-toplevel
+    from sky.clouds.service_catalog import _ALL_CLOUDS
     return {
         '$schema': 'http://json-schema.org/draft-07/schema#',
         'type': 'object',
@@ -17,8 +17,7 @@ def get_resources_schema():
         'properties': {
             'cloud': {
                 'type': 'string',
-                'case_insensitive_enum': list(
-                    cloud_registry.CLOUD_REGISTRY.keys())
+                'case_insensitive_enum': list(_ALL_CLOUDS)
             },
             'region': {
                 'type': 'string',
@@ -112,6 +111,8 @@ def get_resources_schema():
 
 
 def get_storage_schema():
+    # pylint: disable=import-outside-toplevel
+    from sky.data import storage
     return {
         '$schema': 'http://json-schema.org/draft-07/schema#',
         'type': 'object',
@@ -260,3 +261,126 @@ def get_cluster_schema():
             },
         }
     }
+
+
+def get_config_schema():
+    # pylint: disable=import-outside-toplevel
+    from sky.utils import kubernetes_enums
+    return {
+        '$schema': 'https://json-schema.org/draft/2020-12/schema',
+        'type': 'object',
+        'required': [],
+        'additionalProperties': False,
+        'properties': {
+            'spot': {
+                'type': 'object',
+                'required': [],
+                'additionalProperties': False,
+                'properties': {
+                    'controller': {
+                        'type': 'object',
+                        'required': [],
+                        'additionalProperties': False,
+                        'properties': {
+                            'resources': {
+                                k: v
+                                for k, v in get_resources_schema().items()
+                                # Validation may fail if $schema is included.
+                                if k != '$schema'
+                            },
+                        }
+                    },
+                }
+            },
+            'aws': {
+                'type': 'object',
+                'required': [],
+                'additionalProperties': False,
+                'properties': {
+                    'instance_tags': {
+                        'type': 'object',
+                        'required': [],
+                        'additionalProperties': {
+                            'type': 'string',
+                        },
+                    },
+                    'vpc_name': {
+                        'oneOf': [{
+                            'type': 'string',
+                        }, {
+                            'type': 'null',
+                        }],
+                    },
+                    'use_internal_ips': {
+                        'type': 'boolean',
+                    },
+                    'ssh_proxy_command': {
+                        'oneOf': [{
+                            'type': 'string',
+                        }, {
+                            'type': 'null',
+                        }, {
+                            'type': 'object',
+                            'required': [],
+                            'additionalProperties': {
+                                'type': 'string',
+                            }
+                        }]
+                    },
+                }
+            },
+            'gcp': {
+                'type': 'object',
+                'required': [],
+                'additionalProperties': False,
+                'properties': {
+                    'specific_reservations': {
+                        'type': 'array',
+                        'items': {
+                            'type': 'string',
+                            'minItems': 1,
+                            'maxItems': 1,
+                        }
+                    },
+                }
+            },
+            'kubernetes': {
+                'type': 'object',
+                'required': [],
+                'additionalProperties': False,
+                'properties': {
+                    'networking': {
+                        'type': 'string',
+                        'case_insensitive_enum': [
+                            type.value for type in
+                            kubernetes_enums.KubernetesNetworkingMode
+                        ]
+                    },
+                }
+            },
+            'oci': {
+                'type': 'object',
+                'required': [],
+                # Properties are either 'default' or a region name.
+                'additionalProperties': {
+                    'type': 'object',
+                    'required': [],
+                    'additionalProperties': False,
+                    'properties': {
+                        'compartment_ocid': {
+                            'type': 'string',
+                        },
+                        'image_tag_general': {
+                            'type': 'string',
+                        },
+                        'image_tag_gpu': {
+                            'type': 'string',
+                        },
+                        'vcn_subnet': {
+                            'type': 'string',
+                        },
+                    }
+                },
+            },
+        }
+    }
diff --git a/tests/test_config.py b/tests/test_config.py
index 72d16ac362a..d9f2d3274e0 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -6,12 +6,12 @@
 
 from sky import skypilot_config
 from sky.utils import common_utils
-from sky.utils import kubernetes_utils
+from sky.utils import kubernetes_enums
 
 VPC_NAME = 'vpc-12345678'
 PROXY_COMMAND = 'ssh -W %h:%p -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no'
-NODEPORT_MODE_NAME = kubernetes_utils.KubernetesNetworkingMode.NODEPORT.value
-PORT_FORWARD_MODE_NAME = kubernetes_utils.KubernetesNetworkingMode.PORTFORWARD.value
+NODEPORT_MODE_NAME = kubernetes_enums.KubernetesNetworkingMode.NODEPORT.value
+PORT_FORWARD_MODE_NAME = kubernetes_enums.KubernetesNetworkingMode.PORTFORWARD.value
 
 
 def _reload_config() -> None:
@@ -58,6 +58,59 @@ def test_empty_config(monkeypatch, tmp_path) -> None:
     _check_empty_config()
 
 
+def test_invalid_field_config(monkeypatch, tmp_path) -> None:
+    """Test that the config is not loaded if the config file contains unknown field."""
+    config_path = tmp_path / 'invalid.yaml'
+    config_path.open('w').write(
+        textwrap.dedent(f"""\
+        aws:
+            vpc_name: {VPC_NAME}
+            not_a_field: 123
+        """))
+    monkeypatch.setattr(skypilot_config, 'CONFIG_PATH',
+                        tmp_path / 'invalid.yaml')
+    with pytest.raises(ValueError) as e:
+        _reload_config()
+    assert 'Invalid config YAML' in e.value.args[0]
+
+
+def test_invalid_indent_config(monkeypatch, tmp_path) -> None:
+    """Test that the config is not loaded if the config file is incorrectly indented."""
+    config_path = tmp_path / 'invalid.yaml'
+    config_path.open('w').write(
+        textwrap.dedent(f"""\
+        spot:
+            controller:
+                resources:
+                cloud: gcp
+                instance_type: n2-standard-4
+                cpus: 4
+                disk_size: 50
+        """))
+    monkeypatch.setattr(skypilot_config, 'CONFIG_PATH',
+                        tmp_path / 'invalid.yaml')
+    with pytest.raises(ValueError) as e:
+        _reload_config()
+    assert 'Invalid config YAML' in e.value.args[0]
+
+
+def test_invalid_enum_config(monkeypatch, tmp_path) -> None:
+    """Test that the config is not loaded if the config file contains an invalid enum value."""
+    config_path = tmp_path / 'invalid.yaml'
+    config_path.open('w').write(
+        textwrap.dedent(f"""\
+        spot:
+            controller:
+                resources:
+                    cloud: notacloud
+        """))
+    monkeypatch.setattr(skypilot_config, 'CONFIG_PATH',
+                        tmp_path / 'invalid.yaml')
+    with pytest.raises(ValueError) as e:
+        _reload_config()
+    assert 'Invalid config YAML' in e.value.args[0]
+
+
 def test_config_get_set_nested(monkeypatch, tmp_path) -> None:
     """Test that set_nested(), get_nested() works."""
 


Code review comments:
diff_hunk: @@ -160,6 +161,11 @@ def _try_load_config() -> None:
             logger.debug(f'Config loaded:\n{pprint.pformat(_dict)}')
         except yaml.YAMLError as e:
             logger.error(f'Error in loading config file ({config_path}):', e)
+        if _dict is not None:
+            common_utils.validate_schema(_dict,
+                                         schemas.get_config_schema(),
+                                         'Invalid config YAML: ',
path: sky/skypilot_config.py
text: @user1:
It would be nice to show the config path in the error message:
```suggestion
                                         f'Invalid config file ({config_path}): ',
```

diff_hunk: @@ -4,11 +4,11 @@
 https://json-schema.org/
 """
 
-from sky.clouds import cloud_registry
-from sky.data import storage
-
 
 def get_resources_schema():
+    # To avoid circular imports, only import when needed.
+    # pylint: disable=import-outside-toplevel
+    from sky.clouds.service_catalog import _ALL_CLOUDS
path: sky/utils/schemas.py
text: @user1:
style nit: import modules instead of object. 

Also, seems kubernetes is not included in `_ALL_CLOUDS`. Is it possible to still make the `cloud_registry` work?

@author:
I think relying on `cloud_registry` for the list of clouds for validating `skypilot_config` will be very brittle because of the many potential circular imports. This is because for `cloud_registry` to be updated with all the clouds, we need all the cloud modules in `sky.clouds.aws/gcp/...` to be imported. However, there are many modules that require `skypilot_config` before this is done.

For example, the following import path: `clouds` -> `clouds.cloud` -> `skypilot_config` occurs before any cloud module `clouds.aws/gcp/...` is imported, and changing this order will lead to a circular import, as seen from the commment in `clouds.__init__.py`. Currently, I don't see how we can avoid this.

To include kubernetes in `_ALL_CLOUDS`, I believe we can do a slight adjustment to the logic in `clouds.service_catalog.__init__.py` instead. Please take a look!