Repository: skypilot-org/skypilot
Pull request #1640 (2023-01-27 18:22:13)

Make GCP VPC creation more robust
<!-- Describe the changes in this PR -->

In this PR, we
1. reduce one required firewall rule (ICMP) as SkyPilot doesn't necessary need it to function
2. make the VPC creation idempotent to avoid user accidentally touch the VPC and robust against failure during VPC creation. 

<!-- 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] `sky launch` with no VPC
- [x] `sky launch` with an existing skypilot-vpc with no firewall rules
- [x] `sky launch` with an existing "default" VPC that includes two required firewall rules + some additional rules
- [x] All smoke tests: `pytest tests/test_smoke.py --gcp` 



Patch:
diff --git a/docs/source/reference/yaml-spec.rst b/docs/source/reference/yaml-spec.rst
index 586e0366c..1465deff8 100644
--- a/docs/source/reference/yaml-spec.rst
+++ b/docs/source/reference/yaml-spec.rst
@@ -47,6 +47,12 @@ Available fields:
       # Format: <name>:<count> (or simply <name>, short for a count of 1).
       accelerators: V100:4
 
+      # Number of vCPUs per node (optional).
+      #
+      # Format: <count> (exactly <count> vCPUs) or <count>+
+      # (at least <count> vCPUs).
+      cpus: 32
+
       # Instance type to use (optional). If 'accelerators' is specified,
       # the corresponding instance type is automatically inferred.
       instance_type: p3.8xlarge
diff --git a/examples/example_app.py b/examples/example_app.py
index 4c48147ea..c26c49703 100644
--- a/examples/example_app.py
+++ b/examples/example_app.py
@@ -43,7 +43,7 @@ def make_application():
             sky.Resources(sky.AWS(), 'p3.2xlarge'),  # 1 V100, EC2.
             sky.Resources(sky.AWS(), 'p3.8xlarge'),  # 4 V100s, EC2.
             # Tuples mean all resources are required.
-            sky.Resources(sky.GCP(), 'n1-standard-8', 'tpu-v3-8'),
+            sky.Resources(sky.GCP(), 'n1-standard-8', accelerators='tpu-v3-8'),
         })
 
         train_op.set_time_estimator(time_estimators.resnet50_estimate_runtime)
@@ -60,8 +60,8 @@ def make_application():
         infer_op.set_resources({
             sky.Resources(sky.AWS(), 'inf1.2xlarge'),
             sky.Resources(sky.AWS(), 'p3.2xlarge'),
-            sky.Resources(sky.GCP(), 'n1-standard-4', 'T4'),
-            sky.Resources(sky.GCP(), 'n1-standard-8', 'T4'),
+            sky.Resources(sky.GCP(), 'n1-standard-4', accelerators='T4'),
+            sky.Resources(sky.GCP(), 'n1-standard-8', accelerators='T4'),
         })
 
         infer_op.set_time_estimator(
diff --git a/pyproject.toml b/pyproject.toml
index 565a91243..53a88edda 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,7 +16,7 @@ env = [
     "SKYPILOT_DEBUG=1",
     "SKYPILOT_DISABLE_USAGE_COLLECTION=1"
 ]
-addopts = "-s -n 16 -q --tb=short --disable-warnings"
+addopts = "-s -n 16 -q --tb=short --dist loadgroup --disable-warnings"
 
 [tool.mypy]
 python_version = "3.8"
diff --git a/sky/__init__.py b/sky/__init__.py
index 21eae8508..3462fd67d 100644
--- a/sky/__init__.py
+++ b/sky/__init__.py
@@ -28,6 +28,7 @@ from sky.core import (status, start, stop, down, autostop, queue, cancel,
 AWS = clouds.AWS
 Azure = clouds.Azure
 GCP = clouds.GCP
+Lambda = clouds.Lambda
 Local = clouds.Local
 optimize = Optimizer.optimize
 
@@ -36,6 +37,7 @@ __all__ = [
     'AWS',
     'Azure',
     'GCP',
+    'Lambda',
     'Local',
     'Optimizer',
     'OptimizeTarget',
diff --git a/sky/authentication.py b/sky/authentication.py
index 981d25a5b..ae1aaa66c 100644
--- a/sky/authentication.py
+++ b/sky/authentication.py
@@ -21,6 +21,7 @@ from sky.adaptors import gcp
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
+from sky.skylet.providers.lambda_cloud import lambda_utils
 
 logger = sky_logging.init_logger(__name__)
 
@@ -310,3 +311,26 @@ def setup_azure_authentication(config: Dict[str, Any]) -> Dict[str, Any]:
     config['file_mounts'] = file_mounts
 
     return config
+
+
+def setup_lambda_authentication(config: Dict[str, Any]) -> Dict[str, Any]:
+    get_or_generate_keys()
+
+    # Ensure ssh key is registered with Lambda Cloud
+    lambda_client = lambda_utils.LambdaCloudClient()
+    if lambda_client.ssh_key_name is None:
+        public_key_path = os.path.expanduser(PUBLIC_SSH_KEY_PATH)
+        with open(public_key_path, 'r') as f:
+            public_key = f.read()
+        name = f'sky-key-{common_utils.get_user_hash()}'
+        lambda_client.set_ssh_key(name, public_key)
+
+    # Need to use ~ relative path because Ray uses the same
+    # path for finding the public key path on both local and head node.
+    config['auth']['ssh_public_key'] = PUBLIC_SSH_KEY_PATH
+
+    file_mounts = config['file_mounts']
+    file_mounts[PUBLIC_SSH_KEY_PATH] = PUBLIC_SSH_KEY_PATH
+    config['file_mounts'] = file_mounts
+
+    return config
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index b83dc5d8a..c033bb788 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -44,6 +44,7 @@ from sky import spot as spot_lib
 from sky.backends import onprem_utils
 from sky.skylet import constants
 from sky.skylet import log_lib
+from sky.skylet.providers.lambda_cloud import lambda_utils
 from sky.utils import common_utils
 from sky.utils import command_runner
 from sky.utils import env_options
@@ -89,15 +90,6 @@ WAIT_HEAD_NODE_IP_MAX_ATTEMPTS = 3
 # Refer to: https://stackoverflow.com/questions/3764291/how-can-i-see-if-theres-an-available-and-active-network-connection-in-python # pylint: disable=line-too-long
 _TEST_IP = 'https://8.8.8.8'
 
-# GCP has a 63 char limit; however, Ray autoscaler adds many
-# characters. Through testing, this is the maximum length for the Sky cluster
-# name on GCP.  Ref:
-# https://cloud.google.com/compute/docs/naming-resources#resource-name-format
-# NOTE: actually 37 is maximum for a single-node cluster which gets the suffix
-# '-head', but 35 for a multinode cluster because workers get the suffix
-# '-worker'. Here we do not distinguish these cases and take the lower limit.
-_MAX_CLUSTER_NAME_LEN_FOR_GCP = 35
-
 # Allow each CPU thread take 2 tasks.
 # Note: This value cannot be too small, otherwise OOM issue may occur.
 DEFAULT_TASK_CPU_DEMAND = 0.5
@@ -951,6 +943,8 @@ def _add_auth_to_cluster_config(cloud: clouds.Cloud, cluster_config_file: str):
         config = auth.setup_gcp_authentication(config)
     elif isinstance(cloud, clouds.Azure):
         config = auth.setup_azure_authentication(config)
+    elif isinstance(cloud, clouds.Lambda):
+        config = auth.setup_lambda_authentication(config)
     else:
         assert isinstance(cloud, clouds.Local), cloud
         # Local cluster case, authentication is already filled by the user
@@ -1407,10 +1401,11 @@ def _get_tpu_vm_pod_ips(ray_config: Dict[str, Any],
         tpuvm_json = json.loads(stdout)
         if tpuvm_json['state'] != 'READY':
             # May be a leaked preempted resource, or terminated by user in the
-            # console.
+            # console, or still in the process of being created.
             ux_utils.console_newline()
-            logger.warning(f'TPU VM {tpu_id} is not in READY state. '
-                           'Skipping...')
+            logger.debug(f'TPU VM {tpu_id} is in {tpuvm_json["state"]} '
+                         'state. Skipping IP query... '
+                         'Hint: make sure it is not leaked.')
             continue
 
         if not get_internal_ips:
@@ -1802,10 +1797,29 @@ def _query_status_azure(
     return _process_cli_query('Azure', cluster, query_cmd, '\t', status_map)
 
 
+def _query_status_lambda(
+        cluster: str,
+        ray_config: Dict[str, Any],  # pylint: disable=unused-argument
+) -> List[global_user_state.ClusterStatus]:
+    status_map = {
+        'booting': global_user_state.ClusterStatus.INIT,
+        'active': global_user_state.ClusterStatus.UP,
+        'unhealthy': global_user_state.ClusterStatus.INIT,
+        'terminated': None,
+    }
+    # TODO(ewzeng): filter by hash_filter_string to be safe
+    vms = lambda_utils.LambdaCloudClient().list_instances()
+    for node in vms:
+        if node['name'] == cluster:
+            return [status_map[node['status']]]
+    return []
+
+
 _QUERY_STATUS_FUNCS = {
     'AWS': _query_status_aws,
     'GCP': _query_status_gcp,
     'Azure': _query_status_azure,
+    'Lambda': _query_status_lambda,
 }
 
 
@@ -1950,24 +1964,34 @@ def _update_cluster_status_no_lock(
     #     abnormal.
     #
     # An abnormal cluster will transition to INIT and have any autostop setting
-    # reset.
+    # reset (unless it's autostopping/autodowning.).
     is_abnormal = ((0 < len(node_statuses) < handle.launched_nodes) or
                    any(status != global_user_state.ClusterStatus.STOPPED
                        for status in node_statuses))
     if is_abnormal:
-        # Reset the autostop to avoid false information with best effort.
-        # Side effect: if the status is refreshed during autostopping, the
-        # autostop field in the local cache will be reset, even though the
-        # cluster will still be correctly stopped.
-        try:
-            backend = backends.CloudVmRayBackend()
-            backend.set_autostop(handle, -1, stream_logs=False)
-        except (Exception, SystemExit) as e:  # pylint: disable=broad-except
-            logger.debug(f'Failed to reset autostop. Due to '
-                         f'{common_utils.format_exception(e)}')
-        global_user_state.set_cluster_autostop_value(handle.cluster_name,
-                                                     -1,
-                                                     to_down=False)
+        backend = get_backend_from_handle(handle)
+        if isinstance(backend,
+                      backends.CloudVmRayBackend) and record['autostop'] >= 0:
+            if not backend.is_definitely_autostopping(handle,
+                                                      stream_logs=False):
+                # Reset the autostopping as the cluster is abnormal, and may
+                # not correctly autostop. Resetting the autostop will let
+                # the user know that the autostop may not happen to avoid
+                # leakages from the assumption that the cluster will autostop.
+                try:
+                    backend.set_autostop(handle, -1, stream_logs=False)
+                except (Exception, SystemExit) as e:  # pylint: disable=broad-except
+                    logger.debug(f'Failed to reset autostop. Due to '
+                                 f'{common_utils.format_exception(e)}')
+                global_user_state.set_cluster_autostop_value(
+                    handle.cluster_name, -1, to_down=False)
+            else:
+                ux_utils.console_newline()
+                operation_str = 'autodowning' if record[
+                    'to_down'] else 'autostopping'
+                logger.info(
+                    f'Cluster {cluster_name!r} is {operation_str}. Setting to '
+                    'INIT status; try refresh again in a while.')
 
         # If the user starts part of a STOPPED cluster, we still need a status
         # to represent the abnormal status. For spot cluster, it can also
@@ -2390,35 +2414,6 @@ def get_task_resources_str(task: 'task_lib.Task') -> str:
     return resources_str
 
 
-def check_cluster_name_is_valid(cluster_name: str,
-                                cloud: Optional[clouds.Cloud] = None) -> None:
-    """Errors out on invalid cluster names not supported by cloud providers.
-
-    Bans (including but not limited to) names that:
-    - are digits-only
-    - contain underscore (_)
-    """
-    if cluster_name is None:
-        return
-    # GCP errors return this exact regex.  An informal description is at:
-    # https://cloud.google.com/compute/docs/naming-resources#resource-name-format
-    valid_regex = '[a-z]([-a-z0-9]{0,61}[a-z0-9])?'
-    if re.fullmatch(valid_regex, cluster_name) is None:
-        with ux_utils.print_exception_no_traceback():
-            raise exceptions.InvalidClusterNameError(
-                f'Cluster name "{cluster_name}" is invalid; '
-                f'ensure it is fully matched by regex: {valid_regex}')
-    if isinstance(cloud, clouds.GCP):
-        # GCP has too restrictive of a length limit. Don't check for other
-        # clouds.
-        if len(cluster_name) > _MAX_CLUSTER_NAME_LEN_FOR_GCP:
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.InvalidClusterNameError(
-                    f'Cluster name {cluster_name!r} has {len(cluster_name)} '
-                    f'chars; maximum length is {_MAX_CLUSTER_NAME_LEN_FOR_GCP} '
-                    'chars.')
-
-
 def check_cluster_name_not_reserved(
         cluster_name: Optional[str],
         operation_str: Optional[str] = None) -> None:
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 6ea6d4e78..787c12f3d 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -114,6 +114,7 @@ def _get_cluster_config_template(cloud):
         clouds.AWS: 'aws-ray.yml.j2',
         clouds.Azure: 'azure-ray.yml.j2',
         clouds.GCP: 'gcp-ray.yml.j2',
+        clouds.Lambda: 'lambda-ray.yml.j2',
         clouds.Local: 'local-ray.yml.j2',
     }
     return cloud_to_template[type(cloud)]
@@ -574,12 +575,14 @@ class RetryingVmProvisioner(object):
 
     def __init__(self, log_dir: str, dag: 'dag.Dag',
                  optimize_target: OptimizeTarget,
+                 requested_features: Set[clouds.CloudImplementationFeatures],
                  local_wheel_path: pathlib.Path, wheel_hash: str):
         self._blocked_resources = set()
 
         self.log_dir = os.path.expanduser(log_dir)
         self._dag = dag
         self._optimize_target = optimize_target
+        self._requested_features = requested_features
         self._local_wheel_path = local_wheel_path
         self._wheel_hash = wheel_hash
 
@@ -775,6 +778,42 @@ class RetryingVmProvisioner(object):
         else:
             self._blocked_resources.add(launchable_resources.copy(zone=None))
 
+    def _update_blocklist_on_lambda_error(
+            self, launchable_resources: 'resources_lib.Resources', region,
+            zones, stdout, stderr):
+        del zones  # Unused.
+        style = colorama.Style
+        stdout_splits = stdout.split('\n')
+        stderr_splits = stderr.split('\n')
+        errors = [
+            s.strip()
+            for s in stdout_splits + stderr_splits
+            if 'LambdaCloudError:' in s.strip()
+        ]
+        if not errors:
+            logger.info('====== stdout ======')
+            for s in stdout_splits:
+                print(s)
+            logger.info('====== stderr ======')
+            for s in stderr_splits:
+                print(s)
+            with ux_utils.print_exception_no_traceback():
+                raise RuntimeError('Errors occurred during provision; '
+                                   'check logs above.')
+
+        logger.warning(f'Got error(s) in {region.name}:')
+        messages = '\n\t'.join(errors)
+        logger.warning(f'{style.DIM}\t{messages}{style.RESET_ALL}')
+        self._blocked_resources.add(launchable_resources.copy(zone=None))
+
+        # Sometimes, LambdaCloudError will list available regions.
+        for e in errors:
+            if e.find('Regions with capacity available:') != -1:
+                for r in clouds.Lambda.regions():
+                    if e.find(r.name) == -1:
+                        self._blocked_resources.add(
+                            launchable_resources.copy(region=r.name, zone=None))
+
     def _update_blocklist_on_local_error(
             self, launchable_resources: 'resources_lib.Resources', region,
             zones, stdout, stderr):
@@ -834,6 +873,7 @@ class RetryingVmProvisioner(object):
             clouds.AWS: self._update_blocklist_on_aws_error,
             clouds.Azure: self._update_blocklist_on_azure_error,
             clouds.GCP: self._update_blocklist_on_gcp_error,
+            clouds.Lambda: self._update_blocklist_on_lambda_error,
             clouds.Local: self._update_blocklist_on_local_error,
         }
         cloud = launchable_resources.cloud
@@ -888,6 +928,9 @@ class RetryingVmProvisioner(object):
                     elif cloud.is_same_cloud(clouds.Azure()):
                         region = config['provider']['location']
                         zones = None
+                    elif cloud.is_same_cloud(clouds.Lambda()):
+                        region = config['provider']['region']
+                        zones = None
                     elif cloud.is_same_cloud(clouds.Local()):
                         local_regions = clouds.Local.regions()
                         region = local_regions[0].name
@@ -1234,12 +1277,7 @@ class RetryingVmProvisioner(object):
                 # The stdout/stderr of ray up is not useful here, since
                 # head node is successfully provisioned.
                 definitely_no_nodes_launched = self._update_blocklist_on_error(
-                    to_provision,
-                    region,
-                    # Ignored and block region:
-                    zones=None,
-                    stdout=None,
-                    stderr=None)
+                    to_provision, region, zones=zones, stdout=None, stderr=None)
                 # GANG_FAILED means head is up, workers failed.
                 assert definitely_no_nodes_launched is False, (
                     definitely_no_nodes_launched)
@@ -1630,12 +1668,15 @@ class RetryingVmProvisioner(object):
             try:
                 # Recheck cluster name as the 'except:' block below may
                 # change the cloud assignment.
-                backend_utils.check_cluster_name_is_valid(
-                    cluster_name, to_provision.cloud)
+                to_provision.cloud.check_cluster_name_is_valid(cluster_name)
                 if dryrun:
                     cloud_user = None
                 else:
                     cloud_user = to_provision.cloud.get_current_user_identity()
+                # Skip if to_provision.cloud does not support requested features
+                to_provision.cloud.check_features_are_supported(
+                    self._requested_features)
+
                 config_dict = self._retry_region_zones(
                     to_provision,
                     num_nodes,
@@ -1647,6 +1688,17 @@ class RetryingVmProvisioner(object):
                     cluster_exists=cluster_exists)
                 if dryrun:
                     return
+            except (exceptions.InvalidClusterNameError,
+                    exceptions.NotSupportedError,
+                    exceptions.CloudUserIdentityError) as e:
+                # InvalidClusterNameError: cluster name is invalid,
+                # NotSupportedError: cloud does not support requested features,
+                # CloudUserIdentityError: cloud user identity is invalid.
+                # The exceptions above should be applicable to the whole
+                # cloud, so we do add the cloud to the blocked resources.
+                logger.warning(common_utils.format_exception(e))
+                self._blocked_resources.add(
+                    resources_lib.Resources(cloud=to_provision.cloud))
             except exceptions.ResourcesUnavailableError as e:
                 if e.no_failover:
                     raise e
@@ -1658,17 +1710,11 @@ class RetryingVmProvisioner(object):
                         'optimize_target=sky.OptimizeTarget.COST)')
                     raise e
 
-                logger.warning(common_utils.format_exception(e))
-            except (exceptions.CloudUserIdentityError,
-                    exceptions.InvalidClusterNameError) as e:
-                # Let failover below handle this (i.e., block this cloud).
                 logger.warning(common_utils.format_exception(e))
             else:
                 # Provisioning succeeded.
                 break
-            logger.warning(f'\n{style.BRIGHT}Provision failed for {num_nodes}x '
-                           f'{to_provision}. Trying other launchable resources '
-                           f'(if any).{style.RESET_ALL}')
+
             if to_provision.zone is None:
                 region_or_zone_str = str(to_provision.region)
             else:
@@ -1952,6 +1998,7 @@ class CloudVmRayBackend(backends.Backend):
 
         self._dag = None
         self._optimize_target = None
+        self._requested_features = set()
 
         # Command for running the setup script. It is only set when the
         # setup needs to be run outside the self._setup() and as part of
@@ -1964,6 +2011,8 @@ class CloudVmRayBackend(backends.Backend):
         self._dag = kwargs.pop('dag', self._dag)
         self._optimize_target = kwargs.pop(
             'optimize_target', self._optimize_target) or OptimizeTarget.COST
+        self._requested_features = kwargs.pop('requested_features',
+                                              self._requested_features)
         assert len(kwargs) == 0, f'Unexpected kwargs: {kwargs}'
 
     def check_resources_fit_cluster(self, handle: ResourceHandle,
@@ -2087,10 +2136,9 @@ class CloudVmRayBackend(backends.Backend):
                 # in if retry_until_up is set, which will kick off new "rounds"
                 # of optimization infinitely.
                 try:
-                    provisioner = RetryingVmProvisioner(self.log_dir, self._dag,
-                                                        self._optimize_target,
-                                                        local_wheel_path,
-                                                        wheel_hash)
+                    provisioner = RetryingVmProvisioner(
+                        self.log_dir, self._dag, self._optimize_target,
+                        self._requested_features, local_wheel_path, wheel_hash)
                     config_dict = provisioner.provision_with_retries(
                         task, to_provision_config, dryrun, stream_logs)
                     break
@@ -3063,6 +3111,27 @@ class CloudVmRayBackend(backends.Backend):
             global_user_state.set_cluster_autostop_value(
                 handle.cluster_name, idle_minutes_to_autostop, down)
 
+    def is_definitely_autostopping(self,
+                                   handle: ResourceHandle,
+                                   stream_logs: bool = True) -> bool:
+        """Check if the cluster is autostopping.
+
+        Returns:
+            True if the cluster is definitely autostopping. It is possible
+            that the cluster is still autostopping when False is returned,
+            due to errors like transient network issues.
+        """
+        code = autostop_lib.AutostopCodeGen.is_autostopping()
+        returncode, stdout, stderr = self.run_on_head(handle,
+                                                      code,
+                                                      require_outputs=True,
+                                                      stream_logs=stream_logs)
+
+        if returncode == 0:
+            return common_utils.decode_payload(stdout)
+        logger.debug(f'Failed to check if cluster is autostopping: {stderr}')
+        return False
+
     # TODO(zhwu): Refactor this to a CommandRunner class, so different backends
     # can support its own command runner.
     @timeline.event
@@ -3123,11 +3192,12 @@ class CloudVmRayBackend(backends.Backend):
                 cluster_exists=True)
         usage_lib.messages.usage.set_new_cluster()
         assert len(task.resources) == 1, task.resources
-        resources = list(task.resources)[0]
-        task_cloud = resources.cloud
         # Use the task_cloud, because the cloud in `to_provision` can be changed
         # later during the retry.
-        backend_utils.check_cluster_name_is_valid(cluster_name, task_cloud)
+        resources = list(task.resources)[0]
+        task_cloud = (resources.cloud
+                      if resources.cloud is not None else clouds.Cloud)
+        task_cloud.check_cluster_name_is_valid(cluster_name)
 
         cloud = to_provision.cloud
         if isinstance(cloud, clouds.Local):
diff --git a/sky/cli.py b/sky/cli.py
index 8d504a0ce..f508c4f14 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -174,6 +174,13 @@ def _interactive_node_cli_command(cli_func):
                                         default=None,
                                         type=str,
                                         help='Instance type to use.')
+    cpus = click.option(
+        '--cpus',
+        default=None,
+        type=str,
+        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.'))
     gpus = click.option('--gpus',
                         default=None,
                         type=str,
@@ -268,6 +275,7 @@ def _interactive_node_cli_command(cli_func):
         region_option,
         zone_option,
         instance_type_option,
+        cpus,
         *([gpus] if cli_func.__name__ == 'gpunode' else []),
         *([tpus] if cli_func.__name__ == 'tpunode' else []),
         spot_option,
@@ -556,6 +564,7 @@ def _parse_override_params(cloud: Optional[str] = None,
                            region: Optional[str] = None,
                            zone: Optional[str] = None,
                            gpus: Optional[str] = None,
+                           cpus: Optional[str] = None,
                            instance_type: Optional[str] = None,
                            use_spot: Optional[bool] = None,
                            image_id: Optional[str] = None,
@@ -582,6 +591,11 @@ def _parse_override_params(cloud: Optional[str] = None,
             override_params['accelerators'] = None
         else:
             override_params['accelerators'] = gpus
+    if cpus is not None:
+        if cpus.lower() == 'none':
+            override_params['cpus'] = None
+        else:
+            override_params['cpus'] = cpus
     if instance_type is not None:
         if instance_type.lower() == 'none':
             override_params['instance_type'] = None
@@ -908,6 +922,7 @@ def _make_task_from_entrypoint_with_overrides(
     region: Optional[str] = None,
     zone: Optional[str] = None,
     gpus: Optional[str] = None,
+    cpus: Optional[str] = None,
     instance_type: Optional[str] = None,
     num_nodes: Optional[int] = None,
     use_spot: Optional[bool] = None,
@@ -949,6 +964,7 @@ def _make_task_from_entrypoint_with_overrides(
                                              region=region,
                                              zone=zone,
                                              gpus=gpus,
+                                             cpus=cpus,
                                              instance_type=instance_type,
                                              use_spot=use_spot,
                                              image_id=image_id,
@@ -1090,6 +1106,13 @@ 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('--disk-size',
               default=None,
               type=int,
@@ -1154,6 +1177,7 @@ def launch(
     region: Optional[str],
     zone: Optional[str],
     gpus: Optional[str],
+    cpus: Optional[str],
     instance_type: Optional[str],
     num_nodes: Optional[int],
     use_spot: Optional[bool],
@@ -1198,6 +1222,7 @@ def launch(
         region=region,
         zone=zone,
         gpus=gpus,
+        cpus=cpus,
         instance_type=instance_type,
         num_nodes=num_nodes,
         use_spot=use_spot,
@@ -1343,6 +1368,7 @@ def exec(
         region=region,
         zone=zone,
         gpus=gpus,
+        cpus=None,
         instance_type=instance_type,
         use_spot=use_spot,
         image_id=image_id,
@@ -2414,11 +2440,11 @@ def _down_or_stop_clusters(
 # pylint: disable=redefined-outer-name
 def gpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
             cloud: Optional[str], region: Optional[str], zone: Optional[str],
-            instance_type: Optional[str], gpus: Optional[str],
-            use_spot: Optional[bool], screen: Optional[bool],
-            tmux: Optional[bool], disk_size: Optional[int],
-            idle_minutes_to_autostop: Optional[int], down: bool,
-            retry_until_up: bool):
+            instance_type: Optional[str], cpus: Optional[str],
+            gpus: Optional[str], use_spot: Optional[bool],
+            screen: Optional[bool], tmux: Optional[bool],
+            disk_size: Optional[int], idle_minutes_to_autostop: Optional[int],
+            down: bool, retry_until_up: bool):
     """Launch or attach to an interactive GPU node.
 
     Examples:
@@ -2457,7 +2483,8 @@ def gpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
 
     user_requested_resources = not (cloud is None and region is None and
                                     zone is None and instance_type is None and
-                                    gpus is None and use_spot is None)
+                                    cpus is None and gpus is None and
+                                    use_spot is None)
     default_resources = _INTERACTIVE_NODE_DEFAULT_RESOURCES['gpunode']
     cloud_provider = clouds.CLOUD_REGISTRY.from_str(cloud)
     if gpus is None and instance_type is None:
@@ -2470,6 +2497,7 @@ def gpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
                               region=region,
                               zone=zone,
                               instance_type=instance_type,
+                              cpus=cpus,
                               accelerators=gpus,
                               use_spot=use_spot,
                               disk_size=disk_size)
@@ -2493,10 +2521,11 @@ def gpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
 # pylint: disable=redefined-outer-name
 def cpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
             cloud: Optional[str], region: Optional[str], zone: Optional[str],
-            instance_type: Optional[str], use_spot: Optional[bool],
-            screen: Optional[bool], tmux: Optional[bool],
-            disk_size: Optional[int], idle_minutes_to_autostop: Optional[int],
-            down: bool, retry_until_up: bool):
+            instance_type: Optional[str], cpus: Optional[str],
+            use_spot: Optional[bool], screen: Optional[bool],
+            tmux: Optional[bool], disk_size: Optional[int],
+            idle_minutes_to_autostop: Optional[int], down: bool,
+            retry_until_up: bool):
     """Launch or attach to an interactive CPU node.
 
     Examples:
@@ -2534,7 +2563,7 @@ def cpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
 
     user_requested_resources = not (cloud is None and region is None and
                                     zone is None and instance_type is None and
-                                    use_spot is None)
+                                    cpus is None and use_spot is None)
     default_resources = _INTERACTIVE_NODE_DEFAULT_RESOURCES['cpunode']
     cloud_provider = clouds.CLOUD_REGISTRY.from_str(cloud)
     if instance_type is None:
@@ -2545,6 +2574,7 @@ def cpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
                               region=region,
                               zone=zone,
                               instance_type=instance_type,
+                              cpus=cpus,
                               use_spot=use_spot,
                               disk_size=disk_size)
 
@@ -2567,11 +2597,12 @@ def cpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
 # pylint: disable=redefined-outer-name
 def tpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
             region: Optional[str], zone: Optional[str],
-            instance_type: Optional[str], tpus: Optional[str],
-            use_spot: Optional[bool], tpu_vm: Optional[bool],
-            screen: Optional[bool], tmux: Optional[bool],
-            disk_size: Optional[int], idle_minutes_to_autostop: Optional[int],
-            down: bool, retry_until_up: bool):
+            instance_type: Optional[str], cpus: Optional[str],
+            tpus: Optional[str], use_spot: Optional[bool],
+            tpu_vm: Optional[bool], screen: Optional[bool],
+            tmux: Optional[bool], disk_size: Optional[int],
+            idle_minutes_to_autostop: Optional[int], down: bool,
+            retry_until_up: bool):
     """Launch or attach to an interactive TPU node.
 
     Examples:
@@ -2608,8 +2639,8 @@ def tpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
         name = _default_interactive_node_name('tpunode')
 
     user_requested_resources = not (region is None and zone is None and
-                                    instance_type is None and tpus is None and
-                                    use_spot is None)
+                                    instance_type is None and cpus is None and
+                                    tpus is None and use_spot is None)
     default_resources = _INTERACTIVE_NODE_DEFAULT_RESOURCES['tpunode']
     accelerator_args = default_resources.accelerator_args
     if tpu_vm:
@@ -2625,6 +2656,7 @@ def tpunode(cluster: str, yes: bool, port_forward: Optional[List[int]],
                               region=region,
                               zone=zone,
                               instance_type=instance_type,
+                              cpus=cpus,
                               accelerators=tpus,
                               accelerator_args=accelerator_args,
                               use_spot=use_spot,
@@ -2967,6 +2999,13 @@ 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('--spot-recovery',
               default=None,
               type=str,
@@ -3009,6 +3048,7 @@ def spot_launch(
     region: Optional[str],
     zone: Optional[str],
     gpus: Optional[str],
+    cpus: Optional[str],
     instance_type: Optional[str],
     num_nodes: Optional[int],
     use_spot: Optional[bool],
@@ -3037,7 +3077,10 @@ def spot_launch(
     if name is None:
         name = backend_utils.generate_cluster_name()
     else:
-        backend_utils.check_cluster_name_is_valid(name)
+        # This does the basic regex check for the cluster name, while the name
+        # length check will be done by the controller when it starts
+        # provisioning the cluster.
+        clouds.Cloud.check_cluster_name_is_valid(name)
 
     task = _make_task_from_entrypoint_with_overrides(
         entrypoint,
@@ -3047,6 +3090,7 @@ def spot_launch(
         region=region,
         zone=zone,
         gpus=gpus,
+        cpus=cpus,
         instance_type=instance_type,
         num_nodes=num_nodes,
         use_spot=use_spot,
diff --git a/sky/clouds/__init__.py b/sky/clouds/__init__.py
index 053cfecc3..ca442dcb5 100644
--- a/sky/clouds/__init__.py
+++ b/sky/clouds/__init__.py
@@ -1,11 +1,13 @@
 """Clouds in Sky."""
 from sky.clouds.cloud import Cloud
 from sky.clouds.cloud import CLOUD_REGISTRY
+from sky.clouds.cloud import CloudImplementationFeatures
 from sky.clouds.cloud import Region
 from sky.clouds.cloud import Zone
 from sky.clouds.aws import AWS
 from sky.clouds.azure import Azure
 from sky.clouds.gcp import GCP
+from sky.clouds.lambda_cloud import Lambda
 from sky.clouds.local import Local
 
 __all__ = [
@@ -13,7 +15,9 @@ __all__ = [
     'Azure',
     'Cloud',
     'GCP',
+    'Lambda',
     'Local',
+    'CloudImplementationFeatures',
     'Region',
     'Zone',
     'CLOUD_REGISTRY',
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index 418415040..acda887f8 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -47,6 +47,14 @@ class AWS(clouds.Cloud):
     """Amazon Web Services."""
 
     _REPR = 'AWS'
+
+    # AWS has a limit of the tag value length to 256 characters.
+    # By testing, the actual limit is 256 - 12 = 244 characters
+    # (ray adds additional `ray-` and `-worker`), due to the
+    # maximum length of DescribeInstances API filter value.
+    # Reference: https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Using_Tags.html # pylint: disable=line-too-long
+    _MAX_CLUSTER_NAME_LEN_LIMIT = 244
+
     _regions: List[clouds.Region] = []
 
     _INDENT_PREFIX = '    '
@@ -58,6 +66,15 @@ class AWS(clouds.Cloud):
         'https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-quickstart.html'  # pylint: disable=line-too-long
     )
 
+    @classmethod
+    def _cloud_unsupported_features(
+            cls) -> Dict[clouds.CloudImplementationFeatures, str]:
+        return dict()
+
+    @classmethod
+    def _max_cluster_name_length(cls) -> Optional[int]:
+        return cls._MAX_CLUSTER_NAME_LEN_LIMIT
+
     @classmethod
     def _sso_credentials_help_str(cls, expired: bool = False) -> str:
         help_str = 'Run the following commands (must use aws v2 CLI):'
@@ -274,10 +291,10 @@ class AWS(clouds.Cloud):
         return isinstance(other, AWS)
 
     @classmethod
-    def get_default_instance_type(cls) -> str:
-        # General-purpose instance with 8 vCPUs and 32 GB RAM.
-        # Intel Ice Lake 8375C
-        return 'm6i.2xlarge'
+    def get_default_instance_type(cls,
+                                  cpus: Optional[str] = None) -> Optional[str]:
+        return service_catalog.get_default_instance_type(cpus=cpus,
+                                                         clouds='aws')
 
     # TODO: factor the following three methods, as they are the same logic
     # between Azure and AWS.
@@ -334,12 +351,11 @@ class AWS(clouds.Cloud):
 
     def get_feasible_launchable_resources(self,
                                           resources: 'resources_lib.Resources'):
-        fuzzy_candidate_list: List[str] = []
         if resources.instance_type is not None:
             assert resources.is_launchable(), resources
             # Treat Resources(AWS, p3.2x, V100) as Resources(AWS, p3.2x).
             resources = resources.copy(accelerators=None)
-            return ([resources], fuzzy_candidate_list)
+            return ([resources], [])
 
         def _make(instance_list):
             resource_list = []
@@ -350,6 +366,7 @@ class AWS(clouds.Cloud):
                     # Setting this to None as AWS doesn't separately bill /
                     # attach the accelerators.  Billed as part of the VM type.
                     accelerators=None,
+                    cpus=None,
                 )
                 resource_list.append(r)
             return resource_list
@@ -357,9 +374,13 @@ class AWS(clouds.Cloud):
         # Currently, handle a filter on accelerators only.
         accelerators = resources.accelerators
         if accelerators is None:
-            # No requirements to filter, so just return a default VM type.
-            return (_make([AWS.get_default_instance_type()]),
-                    fuzzy_candidate_list)
+            # Return a default instance type with the given number of vCPUs.
+            default_instance_type = AWS.get_default_instance_type(
+                cpus=resources.cpus)
+            if default_instance_type is None:
+                return ([], [])
+            else:
+                return (_make([default_instance_type]), [])
 
         assert len(accelerators) == 1, resources
         acc, acc_count = list(accelerators.items())[0]
@@ -368,6 +389,7 @@ class AWS(clouds.Cloud):
             acc,
             acc_count,
             use_spot=resources.use_spot,
+            cpus=resources.cpus,
             region=resources.region,
             zone=resources.zone,
             clouds='aws')
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index ab1da9331..bb6e85079 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -43,8 +43,24 @@ class Azure(clouds.Cloud):
     """Azure."""
 
     _REPR = 'Azure'
+    # Azure has a 90 char limit for resource group; however, SkyPilot adds the
+    # suffix `-<region name>`. Azure also has a 64 char limit for VM names, and
+    # ray adds addtional `ray-`, `-worker`, and `-<9 chars hash>` for the VM
+    # 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
+
     _regions: List[clouds.Region] = []
 
+    @classmethod
+    def _cloud_unsupported_features(
+            cls) -> Dict[clouds.CloudImplementationFeatures, str]:
+        return dict()
+
+    @classmethod
+    def _max_cluster_name_length(cls) -> int:
+        return cls._MAX_CLUSTER_NAME_LEN_LIMIT
+
     def instance_type_to_hourly_cost(self,
                                      instance_type: str,
                                      use_spot: bool,
@@ -94,10 +110,10 @@ class Azure(clouds.Cloud):
         return isinstance(other, Azure)
 
     @classmethod
-    def get_default_instance_type(cls) -> str:
-        # General-purpose instance with 8 vCPUs and 32 GB RAM.
-        # Intel Ice Lake 8370C
-        return 'Standard_D8_v5'
+    def get_default_instance_type(cls,
+                                  cpus: Optional[str] = None) -> Optional[str]:
+        return service_catalog.get_default_instance_type(cpus=cpus,
+                                                         clouds='azure')
 
     def _get_image_config(self, gen_version, instance_type):
         # az vm image list \
@@ -250,12 +266,11 @@ class Azure(clouds.Cloud):
             # TODO(zhwu): our azure subscription offer ID does not support spot.
             # Need to support it.
             return ([], [])
-        fuzzy_candidate_list = []
         if resources.instance_type is not None:
             assert resources.is_launchable(), resources
             # Treat Resources(AWS, p3.2x, V100) as Resources(AWS, p3.2x).
             resources = resources.copy(accelerators=None)
-            return ([resources], fuzzy_candidate_list)
+            return ([resources], [])
 
         def _make(instance_list):
             resource_list = []
@@ -265,16 +280,22 @@ class Azure(clouds.Cloud):
                     instance_type=instance_type,
                     # Setting this to None as Azure doesn't separately bill /
                     # attach the accelerators.  Billed as part of the VM type.
-                    accelerators=None)
+                    accelerators=None,
+                    cpus=None,
+                )
                 resource_list.append(r)
             return resource_list
 
         # Currently, handle a filter on accelerators only.
         accelerators = resources.accelerators
         if accelerators is None:
-            # No requirements to filter, so just return a default VM type.
-            return (_make([Azure.get_default_instance_type()]),
-                    fuzzy_candidate_list)
+            # Return a default instance type with the given number of vCPUs.
+            default_instance_type = Azure.get_default_instance_type(
+                cpus=resources.cpus)
+            if default_instance_type is None:
+                return ([], [])
+            else:
+                return (_make([default_instance_type]), [])
 
         assert len(accelerators) == 1, resources
         acc, acc_count = list(accelerators.items())[0]
@@ -282,6 +303,7 @@ class Azure(clouds.Cloud):
         ) = service_catalog.get_instance_type_for_accelerator(
             acc,
             acc_count,
+            cpus=resources.cpus,
             use_spot=resources.use_spot,
             region=resources.region,
             zone=resources.zone,
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index 19b9865a9..db3512470 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -1,15 +1,33 @@
 """Interfaces: clouds, regions, and zones."""
 import collections
+import enum
+import re
 import typing
-from typing import Dict, Iterator, List, Optional, Tuple, Type
+from typing import Dict, Iterator, List, Optional, Set, Tuple, Type
 
+from sky import exceptions
 from sky.clouds import service_catalog
+from sky.utils import log_utils
 from sky.utils import ux_utils
 
 if typing.TYPE_CHECKING:
     from sky import resources
 
 
+class CloudImplementationFeatures(enum.Enum):
+    """Features that might not be implemented for all clouds.
+
+    Used by Cloud.check_features_are_supported().
+
+    Note: If any new feature is added, please check and update
+    _cloud_unsupported_features in all clouds to make sure the
+    check_features_are_supported() works as expected.
+    """
+    STOP = 'stop'
+    AUTOSTOP = 'autostop'
+    MULTI_NODE = 'multi-node'
+
+
 class Region(collections.namedtuple('Region', ['name'])):
     """A region."""
     name: str
@@ -55,6 +73,31 @@ class Cloud:
 
     _REPR = '<Cloud>'
 
+    @classmethod
+    def _cloud_unsupported_features(
+            cls) -> Dict[CloudImplementationFeatures, str]:
+        """The features not supported by the cloud implementation.
+
+        This method is used by check_features_are_supported() to check if the
+        cloud implementation supports all the requested features.
+
+        Returns:
+            A dict of {feature: reason} for the features not supported by the
+            cloud implementation.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    def _max_cluster_name_length(cls) -> Optional[int]:
+        """Returns the maximum length limit of a cluster name.
+
+        This method is used by check_cluster_name_is_valid() to check if the
+        cluster name is too long.
+
+        None means no limit.
+        """
+        raise NotImplementedError
+
     #### Regions/Zones ####
 
     @classmethod
@@ -180,7 +223,18 @@ class Cloud:
         raise NotImplementedError
 
     @classmethod
-    def get_default_instance_type(cls) -> str:
+    def get_default_instance_type(cls,
+                                  cpus: Optional[str] = None) -> Optional[str]:
+        """Returns the default instance type with the given number of vCPUs.
+
+        For example, if cpus='4', this method returns the default instance type
+        with 4 vCPUs.  If cpus='4+', this method returns the default instance
+        type with 4 or more vCPUs.
+
+        When cpus 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').
+        """
         raise NotImplementedError
 
     @classmethod
@@ -297,5 +351,63 @@ class Cloud:
         del resource
         return False
 
+    @classmethod
+    def check_features_are_supported(
+            cls, requested_features: Set[CloudImplementationFeatures]) -> None:
+        """Errors out if the cloud does not support all requested features.
+
+        For instance, Lambda Cloud does not support autostop, so
+        Lambda.check_features_are_supported({
+            CloudImplementationFeatures.AUTOSTOP
+        }) raises the exception.
+
+        Raises:
+            exceptions.NotSupportedError: If the cloud does not support all the
+            requested features.
+        """
+        unsupported_features2reason = cls._cloud_unsupported_features()
+        unsupported_features = set(unsupported_features2reason.keys())
+        unsupported_features = requested_features.intersection(
+            unsupported_features)
+        if unsupported_features:
+            table = log_utils.create_table(['Feature', 'Reason'])
+            for feature in unsupported_features:
+                table.add_row(
+                    [feature.value, unsupported_features2reason[feature]])
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.NotSupportedError(
+                    f'The following features are not supported by {cls._REPR}:'
+                    '\n\t' + table.get_string().replace('\n', '\n\t'))
+
+    @classmethod
+    def check_cluster_name_is_valid(cls, cluster_name: str) -> None:
+        """Errors out on invalid cluster names not supported by cloud providers.
+
+        Bans (including but not limited to) names that:
+        - are digits-only
+        - contain underscore (_)
+
+        Raises:
+            exceptions.InvalidClusterNameError: If the cluster name is invalid.
+        """
+        if cluster_name is None:
+            return
+        max_cluster_name_len_limit = cls._max_cluster_name_length()
+        valid_regex = '[a-z]([-a-z0-9]*[a-z0-9])?'
+        if re.fullmatch(valid_regex, cluster_name) is None:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.InvalidClusterNameError(
+                    f'Cluster name "{cluster_name}" is invalid; '
+                    'ensure it is fully matched by regex (e.g., '
+                    'only contains lower letters, numbers and dash): '
+                    f'{valid_regex}')
+        if max_cluster_name_len_limit is not None and len(
+                cluster_name) > max_cluster_name_len_limit:
+            with ux_utils.print_exception_no_traceback():
+                raise exceptions.InvalidClusterNameError(
+                    f'Cluster name {cluster_name!r} has {len(cluster_name)} '
+                    'chars; maximum length is '
+                    f'{max_cluster_name_len_limit} chars.')
+
     def __repr__(self):
         return self._REPR
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index 24dc3155a..2754a5791 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -85,9 +85,29 @@ class GCP(clouds.Cloud):
     """Google Cloud Platform."""
 
     _REPR = 'GCP'
+
+    # GCP has a 63 char limit; however, Ray autoscaler adds many
+    # characters. Through testing, this is the maximum length for the Sky
+    # cluster name on GCP.  Ref:
+    # https://cloud.google.com/compute/docs/naming-resources#resource-name-format
+    # NOTE: actually 37 is maximum for a single-node cluster which gets the
+    # suffix '-head', but 35 for a multinode cluster because workers get the
+    # suffix '-worker'. Here we do not distinguish these cases and take the
+    # lower limit.
+    _MAX_CLUSTER_NAME_LEN_LIMIT = 35
+
     _regions: List[clouds.Region] = []
     _zones: List[clouds.Zone] = []
 
+    @classmethod
+    def _cloud_unsupported_features(
+            cls) -> Dict[clouds.CloudImplementationFeatures, str]:
+        return dict()
+
+    @classmethod
+    def _max_cluster_name_length(cls) -> Optional[int]:
+        return cls._MAX_CLUSTER_NAME_LEN_LIMIT
+
     #### Regions/Zones ####
 
     @classmethod
@@ -279,10 +299,10 @@ class GCP(clouds.Cloud):
             raise
 
     @classmethod
-    def get_default_instance_type(cls) -> str:
-        # General-purpose instance with 8 vCPUs and 32 GB RAM.
-        # Intel Ice Lake 8373C or Cascade Lake 6268CL
-        return 'n2-standard-8'
+    def get_default_instance_type(cls,
+                                  cpus: Optional[str] = None) -> Optional[str]:
+        return service_catalog.get_default_instance_type(cpus=cpus,
+                                                         clouds='gcp')
 
     @classmethod
     def _get_default_region(cls) -> clouds.Region:
@@ -377,45 +397,73 @@ class GCP(clouds.Cloud):
         return resources_vars
 
     def get_feasible_launchable_resources(self, resources):
-        fuzzy_candidate_list = []
         if resources.instance_type is not None:
             assert resources.is_launchable(), resources
-            return ([resources], fuzzy_candidate_list)
+            return ([resources], [])
 
-        # No other resources (cpu/mem) to filter for now, so just return a
-        # default VM type.
-        host_vm_type = GCP.get_default_instance_type()
-        acc_dict = None
-        # Find instance candidates to meet user's requirements
-        if resources.accelerators is not None:
-            assert len(resources.accelerators.items(
-            )) == 1, 'cannot handle more than one accelerator candidates.'
-            acc, acc_count = list(resources.accelerators.items())[0]
-            (instance_list, fuzzy_candidate_list
-            ) = service_catalog.get_instance_type_for_accelerator(
-                acc,
-                acc_count,
-                use_spot=resources.use_spot,
-                region=resources.region,
-                zone=resources.zone,
-                clouds='gcp')
-
-            if instance_list is None:
-                return ([], fuzzy_candidate_list)
-            assert len(
-                instance_list
-            ) == 1, f'More than one instance type matched, {instance_list}'
+        if resources.accelerators is None:
+            # Return a default instance type with the given number of vCPUs.
+            host_vm_type = GCP.get_default_instance_type(cpus=resources.cpus)
+            if host_vm_type is None:
+                return ([], [])
+            else:
+                r = resources.copy(
+                    cloud=GCP(),
+                    instance_type=host_vm_type,
+                    accelerators=None,
+                    cpus=None,
+                )
+                return ([r], [])
+
+        use_tpu_vm = False
+        if resources.accelerator_args is not None:
+            use_tpu_vm = resources.accelerator_args.get('tpu_vm', False)
 
+        # Find instance candidates to meet user's requirements
+        assert len(resources.accelerators.items()
+                  ) == 1, 'cannot handle more than one accelerator candidates.'
+        acc, acc_count = list(resources.accelerators.items())[0]
+
+        # For TPU VMs, the instance type is fixed to 'TPU-VM'. However, we still
+        # need to call the below function to get the fuzzy candidate list.
+        (instance_list, fuzzy_candidate_list
+        ) = service_catalog.get_instance_type_for_accelerator(
+            acc,
+            acc_count,
+            cpus=resources.cpus if not use_tpu_vm else None,
+            use_spot=resources.use_spot,
+            region=resources.region,
+            zone=resources.zone,
+            clouds='gcp')
+
+        if instance_list is None:
+            return ([], fuzzy_candidate_list)
+        assert len(
+            instance_list
+        ) == 1, f'More than one instance type matched, {instance_list}'
+
+        if use_tpu_vm:
+            host_vm_type = 'TPU-VM'
+            # FIXME(woosuk): This leverages the fact that TPU VMs have 96 vCPUs.
+            num_cpus_in_tpu_vm = 96
+            if resources.cpus is not None:
+                if resources.cpus.endswith('+'):
+                    cpus = float(resources.cpus[:-1])
+                    if cpus > num_cpus_in_tpu_vm:
+                        return ([], fuzzy_candidate_list)
+                else:
+                    cpus = float(resources.cpus)
+                    if cpus != num_cpus_in_tpu_vm:
+                        return ([], fuzzy_candidate_list)
+        else:
             host_vm_type = instance_list[0]
-            acc_dict = {acc: acc_count}
-            if resources.accelerator_args is not None:
-                use_tpu_vm = resources.accelerator_args.get('tpu_vm', False)
-                if use_tpu_vm:
-                    host_vm_type = 'TPU-VM'
+
+        acc_dict = {acc: acc_count}
         r = resources.copy(
             cloud=GCP(),
             instance_type=host_vm_type,
             accelerators=acc_dict,
+            cpus=None,
         )
         return ([r], fuzzy_candidate_list)
 
diff --git a/sky/clouds/lambda_cloud.py b/sky/clouds/lambda_cloud.py
new file mode 100644
index 000000000..02d90d7e1
--- /dev/null
+++ b/sky/clouds/lambda_cloud.py
@@ -0,0 +1,269 @@
+"""Lambda Cloud."""
+import json
+import typing
+from typing import Dict, Iterator, List, Optional, Tuple
+
+from sky import clouds
+from sky.clouds import service_catalog
+from sky.skylet.providers.lambda_cloud import lambda_utils
+
+if typing.TYPE_CHECKING:
+    # Renaming to avoid shadowing variables.
+    from sky import resources as resources_lib
+
+# Minimum set of files under ~/.lambda_cloud that grant Lambda Cloud access.
+_CREDENTIAL_FILES = [
+    'lambda_keys',
+]
+
+
+@clouds.CLOUD_REGISTRY.register
+class Lambda(clouds.Cloud):
+    """Lambda Labs GPU Cloud."""
+
+    _REPR = 'Lambda'
+
+    # Lamdba has a 64 char limit for cluster name.
+    # Reference: https://cloud.lambdalabs.com/api/v1/docs#operation/launchInstance # pylint: disable=line-too-long
+    _MAX_CLUSTER_NAME_LEN_LIMIT = 64
+    # Currently, none of clouds.CloudImplementationFeatures are implemented
+    # for Lambda Cloud.
+    # STOP/AUTOSTOP: The Lambda cloud provider does not support stopping VMs.
+    # MULTI_NODE: Multi-node is not supported by the implementation yet.
+    _CLOUD_UNSUPPORTED_FEATURES = {
+        clouds.CloudImplementationFeatures.STOP: 'Lambda cloud does not support stopping VMs.',
+        clouds.CloudImplementationFeatures.AUTOSTOP: 'Lambda cloud does not support stopping VMs.',
+        clouds.CloudImplementationFeatures.MULTI_NODE: 'Multi-node is not supported by the Lambda Cloud implementation yet.',
+    }
+
+    _regions: List[clouds.Region] = []
+
+    @classmethod
+    def _cloud_unsupported_features(
+            cls) -> Dict[clouds.CloudImplementationFeatures, str]:
+        return cls._CLOUD_UNSUPPORTED_FEATURES
+
+    @classmethod
+    def _max_cluster_name_length(cls) -> Optional[int]:
+        return cls._MAX_CLUSTER_NAME_LEN_LIMIT
+
+    @classmethod
+    def regions(cls) -> List[clouds.Region]:
+        if not cls._regions:
+            cls._regions = [
+                # Popular US regions
+                clouds.Region('us-east-1'),
+                clouds.Region('us-west-2'),
+                clouds.Region('us-west-1'),
+                clouds.Region('us-south-1'),
+
+                # Everyone else
+                clouds.Region('asia-northeast-1'),
+                clouds.Region('asia-northeast-2'),
+                clouds.Region('asia-south-1'),
+                clouds.Region('australia-southeast-1'),
+                clouds.Region('europe-central-1'),
+                clouds.Region('europe-south-1'),
+                clouds.Region('me-west-1'),
+            ]
+        return cls._regions
+
+    @classmethod
+    def regions_with_offering(cls, instance_type: Optional[str],
+                              accelerators: Optional[Dict[str, int]],
+                              use_spot: bool, region: Optional[str],
+                              zone: Optional[str]) -> List[clouds.Region]:
+        del accelerators, zone  # unused
+        if use_spot:
+            return []
+        if instance_type is None:
+            # Fall back to default regions
+            regions = cls.regions()
+        else:
+            regions = service_catalog.get_region_zones_for_instance_type(
+                instance_type, use_spot, 'lambda')
+
+        if region is not None:
+            regions = [r for r in regions if r.name == region]
+        return regions
+
+    @classmethod
+    def region_zones_provision_loop(
+        cls,
+        *,
+        instance_type: Optional[str] = None,
+        accelerators: Optional[Dict[str, int]] = None,
+        use_spot: bool = False,
+    ) -> Iterator[Tuple[clouds.Region, List[clouds.Zone]]]:
+        regions = cls.regions_with_offering(instance_type,
+                                            accelerators,
+                                            use_spot,
+                                            region=None,
+                                            zone=None)
+        for region in regions:
+            yield region, region.zones
+
+    def instance_type_to_hourly_cost(self,
+                                     instance_type: str,
+                                     use_spot: bool,
+                                     region: Optional[str] = None,
+                                     zone: Optional[str] = None) -> float:
+        return service_catalog.get_hourly_cost(instance_type,
+                                               use_spot=use_spot,
+                                               region=region,
+                                               zone=zone,
+                                               clouds='lambda')
+
+    def accelerators_to_hourly_cost(self,
+                                    accelerators: Dict[str, int],
+                                    use_spot: bool,
+                                    region: Optional[str] = None,
+                                    zone: Optional[str] = None) -> float:
+        del accelerators, use_spot, region, zone  # unused
+        # Lambda includes accelerators as part of the instance type.
+        return 0.0
+
+    def get_egress_cost(self, num_gigabytes: float) -> float:
+        return 0.0
+
+    def __repr__(self):
+        return 'Lambda'
+
+    def is_same_cloud(self, other: clouds.Cloud) -> bool:
+        # Returns true if the two clouds are the same cloud type.
+        return isinstance(other, Lambda)
+
+    @classmethod
+    def get_default_instance_type(cls,
+                                  cpus: Optional[str] = None) -> Optional[str]:
+        return service_catalog.get_default_instance_type(cpus=cpus,
+                                                         clouds='lambda')
+
+    @classmethod
+    def get_accelerators_from_instance_type(
+        cls,
+        instance_type: str,
+    ) -> Optional[Dict[str, int]]:
+        return service_catalog.get_accelerators_from_instance_type(
+            instance_type, clouds='lambda')
+
+    @classmethod
+    def get_vcpus_from_instance_type(
+        cls,
+        instance_type: str,
+    ) -> Optional[float]:
+        return service_catalog.get_vcpus_from_instance_type(instance_type,
+                                                            clouds='lambda')
+
+    @classmethod
+    def get_zone_shell_cmd(cls) -> Optional[str]:
+        return None
+
+    def make_deploy_resources_variables(
+            self, resources: 'resources_lib.Resources',
+            region: Optional['clouds.Region'],
+            zones: Optional[List['clouds.Zone']]) -> Dict[str, Optional[str]]:
+        del zones
+        if region is None:
+            region = self._get_default_region()
+
+        r = resources
+        acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
+        if acc_dict is not None:
+            custom_resources = json.dumps(acc_dict, separators=(',', ':'))
+        else:
+            custom_resources = None
+
+        return {
+            'instance_type': resources.instance_type,
+            'custom_resources': custom_resources,
+            'region': region.name,
+        }
+
+    def get_feasible_launchable_resources(self,
+                                          resources: 'resources_lib.Resources'):
+        if resources.use_spot:
+            return ([], [])
+        if resources.instance_type is not None:
+            assert resources.is_launchable(), resources
+            # Accelerators are part of the instance type in Lambda Cloud
+            resources = resources.copy(accelerators=None)
+            return ([resources], [])
+
+        def _make(instance_list):
+            resource_list = []
+            for instance_type in instance_list:
+                r = resources.copy(
+                    cloud=Lambda(),
+                    instance_type=instance_type,
+                    # Setting this to None as Lambda doesn't separately bill /
+                    # attach the accelerators.  Billed as part of the VM type.
+                    accelerators=None,
+                    cpus=None,
+                )
+                resource_list.append(r)
+            return resource_list
+
+        # Currently, handle a filter on accelerators only.
+        accelerators = resources.accelerators
+        if accelerators is None:
+            # Return a default instance type with the given number of vCPUs.
+            default_instance_type = Lambda.get_default_instance_type(
+                cpus=resources.cpus)
+            if default_instance_type is None:
+                return ([], [])
+            else:
+                return (_make([default_instance_type]), [])
+
+        assert len(accelerators) == 1, resources
+        acc, acc_count = list(accelerators.items())[0]
+        (instance_list, fuzzy_candidate_list
+        ) = service_catalog.get_instance_type_for_accelerator(
+            acc,
+            acc_count,
+            use_spot=resources.use_spot,
+            cpus=resources.cpus,
+            region=resources.region,
+            zone=resources.zone,
+            clouds='lambda')
+        if instance_list is None:
+            return ([], fuzzy_candidate_list)
+        return (_make(instance_list), fuzzy_candidate_list)
+
+    def check_credentials(self) -> Tuple[bool, Optional[str]]:
+        try:
+            lambda_utils.LambdaCloudClient().list_instances()
+        except (AssertionError, KeyError, lambda_utils.LambdaCloudError):
+            return False, ('Failed to access Lambda Cloud with credentials. '
+                           'To configure credentials, go to:\n    '
+                           '  https://cloud.lambdalabs.com/api-keys\n    '
+                           'to generate API key and add the line\n    '
+                           '  api_key = [YOUR API KEY]\n    '
+                           'to ~/.lambda_cloud/lambda_keys')
+        return True, None
+
+    def get_credential_file_mounts(self) -> Dict[str, str]:
+        return {
+            f'~/.lambda_cloud/{filename}': f'~/.lambda_cloud/{filename}'
+            for filename in _CREDENTIAL_FILES
+        }
+
+    def get_current_user_identity(self) -> Optional[str]:
+        # TODO(ewzeng): Implement get_current_user_identity for Lambda
+        return None
+
+    def instance_type_exists(self, instance_type: str) -> bool:
+        return service_catalog.instance_type_exists(instance_type, 'lambda')
+
+    def validate_region_zone(self, region: Optional[str], zone: Optional[str]):
+        return service_catalog.validate_region_zone(region,
+                                                    zone,
+                                                    clouds='lambda')
+
+    def accelerator_in_region_or_zone(self,
+                                      accelerator: str,
+                                      acc_count: int,
+                                      region: Optional[str] = None,
+                                      zone: Optional[str] = None) -> bool:
+        return service_catalog.accelerator_in_region_or_zone(
+            accelerator, acc_count, region, zone, 'lambda')
diff --git a/sky/clouds/local.py b/sky/clouds/local.py
index 627358f7d..438126001 100644
--- a/sky/clouds/local.py
+++ b/sky/clouds/local.py
@@ -36,8 +36,23 @@ class Local(clouds.Cloud):
     """
     _DEFAULT_INSTANCE_TYPE = 'on-prem'
     LOCAL_REGION = clouds.Region('Local')
+    _CLOUD_UNSUPPORTED_FEATURES = {
+        clouds.CloudImplementationFeatures.STOP:
+            ('Local cloud does not support stopping instances.'),
+        clouds.CloudImplementationFeatures.AUTOSTOP:
+            ('Local cloud does not support stopping instances.')
+    }
     _regions: List[clouds.Region] = [LOCAL_REGION]
 
+    @classmethod
+    def _cloud_unsupported_features(
+            cls) -> Dict[clouds.CloudImplementationFeatures, str]:
+        return cls._CLOUD_UNSUPPORTED_FEATURES
+
+    @classmethod
+    def _max_cluster_name_length(cls) -> Optional[int]:
+        return None
+
     @classmethod
     def regions(cls):
         return cls._regions
@@ -94,8 +109,9 @@ class Local(clouds.Cloud):
         return isinstance(other, Local)
 
     @classmethod
-    def get_default_instance_type(cls) -> str:
+    def get_default_instance_type(cls, cpus: Optional[str] = None) -> str:
         # There is only "1" instance type for local cloud: on-prem
+        del cpus  # Unused.
         return Local._DEFAULT_INSTANCE_TYPE
 
     @classmethod
diff --git a/sky/clouds/service_catalog/__init__.py b/sky/clouds/service_catalog/__init__.py
index a722701cc..23d90faf3 100644
--- a/sky/clouds/service_catalog/__init__.py
+++ b/sky/clouds/service_catalog/__init__.py
@@ -15,7 +15,7 @@ if typing.TYPE_CHECKING:
     from sky.clouds.service_catalog import common
 
 CloudFilter = Optional[Union[List[str], str]]
-_ALL_CLOUDS = ('aws', 'azure', 'gcp')
+_ALL_CLOUDS = ('aws', 'azure', 'gcp', 'lambda')
 
 
 def _map_clouds_catalog(clouds: CloudFilter, method_name: str, *args, **kwargs):
@@ -159,6 +159,17 @@ def get_vcpus_from_instance_type(instance_type: str,
                                instance_type)
 
 
+def get_default_instance_type(cpus: Optional[str] = None,
+                              clouds: CloudFilter = None) -> Optional[str]:
+    """Returns the cloud's default instance type for the given number of vCPUs.
+
+    For example, if cpus='4', this method returns the default instance type
+    with 4 vCPUs.  If cpus='4+', this method returns the default instance
+    type with 4 or more vCPUs.
+    """
+    return _map_clouds_catalog(clouds, 'get_default_instance_type', cpus)
+
+
 def get_accelerators_from_instance_type(
         instance_type: str,
         clouds: CloudFilter = None) -> Optional[Dict[str, int]]:
@@ -170,6 +181,7 @@ def get_accelerators_from_instance_type(
 def get_instance_type_for_accelerator(
     acc_name: str,
     acc_count: int,
+    cpus: Optional[str] = None,
     use_spot: bool = False,
     region: Optional[str] = None,
     zone: Optional[str] = None,
@@ -180,7 +192,8 @@ def get_instance_type_for_accelerator(
     accelerators with sorted prices and a list of candidates with fuzzy search.
     """
     return _map_clouds_catalog(clouds, 'get_instance_type_for_accelerator',
-                               acc_name, acc_count, use_spot, region, zone)
+                               acc_name, acc_count, cpus, use_spot, region,
+                               zone)
 
 
 def get_accelerator_hourly_cost(
diff --git a/sky/clouds/service_catalog/aws_catalog.py b/sky/clouds/service_catalog/aws_catalog.py
index 150e45b38..dfcba5084 100644
--- a/sky/clouds/service_catalog/aws_catalog.py
+++ b/sky/clouds/service_catalog/aws_catalog.py
@@ -19,6 +19,12 @@ if typing.TYPE_CHECKING:
 
 logger = sky_logging.init_logger(__name__)
 
+# This is the latest general-purpose instance family as of Jan 2023.
+# CPU: Intel Ice Lake 8375C.
+# Memory: 4 GiB RAM per 1 vCPU.
+_DEFAULT_INSTANCE_FAMILY = 'm6i'
+_DEFAULT_NUM_VCPUS = 8
+
 # Keep it synced with the frequency in
 # skypilot-catalog/.github/workflows/update-aws-catalog.yml
 _PULL_FREQUENCY_HOURS = 7
@@ -94,6 +100,14 @@ def get_vcpus_from_instance_type(instance_type: str) -> Optional[float]:
     return common.get_vcpus_from_instance_type_impl(_df, instance_type)
 
 
+def get_default_instance_type(cpus: Optional[str] = None) -> Optional[str]:
+    if cpus is None:
+        cpus = str(_DEFAULT_NUM_VCPUS)
+    instance_type_prefix = f'{_DEFAULT_INSTANCE_FAMILY}.'
+    df = _df[_df['InstanceType'].str.startswith(instance_type_prefix)]
+    return common.get_instance_type_for_cpus_impl(df, cpus)
+
+
 def get_accelerators_from_instance_type(
         instance_type: str) -> Optional[Dict[str, int]]:
     return common.get_accelerators_from_instance_type_impl(_df, instance_type)
@@ -102,6 +116,7 @@ def get_accelerators_from_instance_type(
 def get_instance_type_for_accelerator(
     acc_name: str,
     acc_count: int,
+    cpus: Optional[str] = None,
     use_spot: bool = False,
     region: Optional[str] = None,
     zone: Optional[str] = None,
@@ -113,6 +128,7 @@ def get_instance_type_for_accelerator(
     return common.get_instance_type_for_accelerator_impl(df=_df,
                                                          acc_name=acc_name,
                                                          acc_count=acc_count,
+                                                         cpus=cpus,
                                                          use_spot=use_spot,
                                                          region=region,
                                                          zone=zone)
diff --git a/sky/clouds/service_catalog/azure_catalog.py b/sky/clouds/service_catalog/azure_catalog.py
index c07f8e8ac..a3428c679 100644
--- a/sky/clouds/service_catalog/azure_catalog.py
+++ b/sky/clouds/service_catalog/azure_catalog.py
@@ -3,6 +3,7 @@
 This module loads the service catalog file and can be used to query
 instance types and pricing information for Azure.
 """
+import re
 from typing import Dict, List, Optional, Tuple
 
 from sky import clouds as cloud_lib
@@ -11,6 +12,12 @@ from sky.utils import ux_utils
 
 _df = common.read_catalog('azure/vms.csv')
 
+# This is the latest general-purpose instance family as of Jan 2023.
+# CPU: Intel Ice Lake 8370C.
+# Memory: 4 GiB RAM per 1 vCPU.
+_DEFAULT_INSTANCE_FAMILY = 'D_v5'
+_DEFAULT_NUM_VCPUS = 8
+
 
 def instance_type_exists(instance_type: str) -> bool:
     return common.instance_type_exists_impl(_df, instance_type)
@@ -53,6 +60,37 @@ def get_vcpus_from_instance_type(instance_type: str) -> Optional[float]:
     return common.get_vcpus_from_instance_type_impl(_df, instance_type)
 
 
+def _get_instance_family(instance_type: str) -> str:
+    if instance_type.startswith('Basic_A'):
+        return 'basic_a'
+
+    assert instance_type.startswith('Standard_')
+    # Remove the 'Standard_' prefix.
+    instance_type = instance_type[len('Standard_'):]
+    # Remove the '_Promo' suffix if exists.
+    if '_Promo' in instance_type:
+        instance_type = instance_type[:-len('_Promo')]
+
+    # TODO(woosuk): Use better regex.
+    if '-' in instance_type:
+        x = re.match(r'([A-Za-z]+)([0-9]+)(-)([0-9]+)(.*)', instance_type)
+        assert x is not None, x
+        instance_family = x.group(1) + '_' + x.group(5)
+    else:
+        x = re.match(r'([A-Za-z]+)([0-9]+)(.*)', instance_type)
+        assert x is not None, x
+        instance_family = x.group(1) + x.group(3)
+    return instance_family
+
+
+def get_default_instance_type(cpus: Optional[str] = None) -> Optional[str]:
+    if cpus is None:
+        cpus = str(_DEFAULT_NUM_VCPUS)
+    df = _df[_df['InstanceType'].apply(_get_instance_family) ==
+             _DEFAULT_INSTANCE_FAMILY]
+    return common.get_instance_type_for_cpus_impl(df, cpus)
+
+
 def get_accelerators_from_instance_type(
         instance_type: str) -> Optional[Dict[str, int]]:
     return common.get_accelerators_from_instance_type_impl(_df, instance_type)
@@ -61,6 +99,7 @@ def get_accelerators_from_instance_type(
 def get_instance_type_for_accelerator(
         acc_name: str,
         acc_count: int,
+        cpus: Optional[str] = None,
         use_spot: bool = False,
         region: Optional[str] = None,
         zone: Optional[str] = None) -> Tuple[Optional[List[str]], List[str]]:
@@ -74,6 +113,7 @@ def get_instance_type_for_accelerator(
     return common.get_instance_type_for_accelerator_impl(df=_df,
                                                          acc_name=acc_name,
                                                          acc_count=acc_count,
+                                                         cpus=cpus,
                                                          use_spot=use_spot,
                                                          region=region,
                                                          zone=zone)
diff --git a/sky/clouds/service_catalog/common.py b/sky/clouds/service_catalog/common.py
index 84b2c67be..d0f83597b 100644
--- a/sky/clouds/service_catalog/common.py
+++ b/sky/clouds/service_catalog/common.py
@@ -270,6 +270,39 @@ def get_vcpus_from_instance_type_impl(
     return float(vcpus)
 
 
+def _filter_with_cpus(df: pd.DataFrame, cpus: Optional[str]) -> pd.DataFrame:
+    if cpus is None:
+        return df
+
+    # The following code is redundant with the code in resources.py::_set_cpus()
+    # but we add it here for safety.
+    if cpus.endswith('+'):
+        num_cpus_str = cpus[:-1]
+    else:
+        num_cpus_str = cpus
+    try:
+        num_cpus = float(num_cpus_str)
+    except ValueError:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(f'The "cpus" field should be either a number or '
+                             f'a string "<number>+". Found: {cpus!r}') from None
+
+    if cpus.endswith('+'):
+        return df[df['vCPUs'] >= num_cpus]
+    else:
+        return df[df['vCPUs'] == num_cpus]
+
+
+def get_instance_type_for_cpus_impl(
+        df: pd.DataFrame, cpus: Optional[str] = None) -> Optional[str]:
+    df = _filter_with_cpus(df, cpus)
+    if df.empty:
+        return None
+    # Sort by the number of vCPUs and then by the price.
+    df = df.sort_values(by=['vCPUs', 'Price'], ascending=True)
+    return df['InstanceType'].iloc[0]
+
+
 def get_accelerators_from_instance_type_impl(
     df: pd.DataFrame,
     instance_type: str,
@@ -289,6 +322,7 @@ def get_instance_type_for_accelerator_impl(
     df: pd.DataFrame,
     acc_name: str,
     acc_count: int,
+    cpus: Optional[str] = None,
     use_spot: bool = False,
     region: Optional[str] = None,
     zone: Optional[str] = None,
@@ -313,6 +347,7 @@ def get_instance_type_for_accelerator_impl(
                                             f'{int(row["AcceleratorCount"])}')
         return (None, fuzzy_candidate_list)
 
+    result = _filter_with_cpus(result, cpus)
     if region is not None:
         result = result[result['Region'] == region]
     if zone is not None:
diff --git a/sky/clouds/service_catalog/gcp_catalog.py b/sky/clouds/service_catalog/gcp_catalog.py
index b2cb8f4b8..01fbea392 100644
--- a/sky/clouds/service_catalog/gcp_catalog.py
+++ b/sky/clouds/service_catalog/gcp_catalog.py
@@ -25,6 +25,13 @@ _TPU_REGIONS = [
     'asia-east1',
 ]
 
+# Default instance family for CPU-only VMs.
+# This is the latest general-purpose instance family as of Jan 2023.
+# CPU: Intel Ice Lake 8373C or Cascade Lake 6268CL.
+# Memory: 4 GiB RAM per 1 vCPU.
+_DEFAULT_INSTANCE_FAMILY = 'n2-standard'
+_DEFAULT_NUM_VCPUS = 8
+
 # This can be switched between n1 and n2.
 # n2 is not allowed for launching GPUs.
 _DEFAULT_HOST_VM_FAMILY = 'n1'
@@ -164,9 +171,19 @@ def get_vcpus_from_instance_type(instance_type: str) -> Optional[float]:
     return common.get_vcpus_from_instance_type_impl(_df, instance_type)
 
 
+def get_default_instance_type(cpus: Optional[str] = None) -> Optional[str]:
+    if cpus is None:
+        cpus = str(_DEFAULT_NUM_VCPUS)
+    instance_type_prefix = f'{_DEFAULT_INSTANCE_FAMILY}-'
+    df = _df[_df['InstanceType'].notna()]
+    df = df[df['InstanceType'].str.startswith(instance_type_prefix)]
+    return common.get_instance_type_for_cpus_impl(df, cpus)
+
+
 def get_instance_type_for_accelerator(
         acc_name: str,
         acc_count: int,
+        cpus: Optional[str] = None,
         use_spot: bool = False,
         region: Optional[str] = None,
         zone: Optional[str] = None) -> Tuple[Optional[List[str]], List[str]]:
@@ -178,20 +195,48 @@ def get_instance_type_for_accelerator(
     """
     (instance_list,
      fuzzy_candidate_list) = common.get_instance_type_for_accelerator_impl(
-         _df, acc_name, acc_count, use_spot, region, zone)
+         _df, acc_name, acc_count, cpus, use_spot, region, zone)
     if instance_list is None:
         return None, fuzzy_candidate_list
 
     if acc_name in _A100_INSTANCE_TYPE_DICTS:
         # If A100 is used, host VM type must be A2.
         # https://cloud.google.com/compute/docs/gpus#a100-gpus
+
+        # FIXME(woosuk): This uses the knowledge that the A2 machines provide
+        # 12 vCPUs per GPU, except for a2-megagpu-16g which has 16 GPUs.
+        if cpus is not None:
+            num_a2_cpus = min(12 * acc_count, 96)
+            if cpus.endswith('+'):
+                if num_a2_cpus < float(cpus[:-1]):
+                    return None, []
+            else:
+                if num_a2_cpus != float(cpus):
+                    return None, []
         return [_A100_INSTANCE_TYPE_DICTS[acc_name][acc_count]], []
+
     if acc_name not in _NUM_ACC_TO_NUM_CPU:
         acc_name = 'DEFAULT'
 
-    num_cpus = _NUM_ACC_TO_NUM_CPU[acc_name].get(acc_count, None)
-    # The (acc_name, acc_count) should be validated in the caller.
-    assert num_cpus is not None, (acc_name, acc_count)
+    assert _DEFAULT_HOST_VM_FAMILY == 'n1'
+    num_cpus = None
+    if cpus is None:
+        num_cpus = _NUM_ACC_TO_NUM_CPU[acc_name].get(acc_count, None)
+    else:
+        # FIXME(woosuk): This uses the knowledge that the N1-highmem machines
+        # have 2, 4, 8, 16, 32, 64, or 96 vCPUs.
+        for num_n1_cpus in [2, 4, 8, 16, 32, 64, 96]:
+            if cpus.endswith('+'):
+                if num_n1_cpus >= float(cpus[:-1]):
+                    num_cpus = num_n1_cpus
+                    break
+            else:
+                if num_n1_cpus == float(cpus):
+                    num_cpus = num_n1_cpus
+                    break
+    if num_cpus is None:
+        return None, []
+
     mem_type = 'highmem'
     # patches for the number of cores per GPU, as some of the combinations
     # are not supported by GCP.
@@ -403,7 +448,7 @@ def check_accelerator_attachable_to_host(instance_type: str,
     acc_name, acc_count = acc[0]
 
     if acc_name.startswith('tpu-'):
-        # TODO(woosuk): Check max vcpus and memory for each TPU type.
+        # TODO(woosuk): Check max vCPUs and memory for each TPU type.
         assert instance_type == 'TPU-VM' or instance_type.startswith('n1-')
         return
 
diff --git a/sky/clouds/service_catalog/lambda_catalog.py b/sky/clouds/service_catalog/lambda_catalog.py
new file mode 100644
index 000000000..21975719e
--- /dev/null
+++ b/sky/clouds/service_catalog/lambda_catalog.py
@@ -0,0 +1,125 @@
+"""Lambda Cloud Catalog.
+
+This module loads the service catalog file and can be used to query
+instance types and pricing information for Lambda.
+"""
+import typing
+from typing import Dict, List, Optional, Tuple
+
+from sky.clouds.service_catalog import common
+from sky.utils import ux_utils
+
+if typing.TYPE_CHECKING:
+    from sky.clouds import cloud
+
+_df = common.read_catalog('lambda/vms.csv')
+
+# Number of vCPUS for gpu_1x_a100_sxm4
+_DEFAULT_NUM_VCPUS = 30
+
+
+def instance_type_exists(instance_type: str) -> bool:
+    return common.instance_type_exists_impl(_df, instance_type)
+
+
+def validate_region_zone(
+        region: Optional[str],
+        zone: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
+    if zone is not None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Lambda Cloud does not support zones.')
+    return common.validate_region_zone_impl('lambda', _df, region, zone)
+
+
+def accelerator_in_region_or_zone(acc_name: str,
+                                  acc_count: int,
+                                  region: Optional[str] = None,
+                                  zone: Optional[str] = None) -> bool:
+    if zone is not None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Lambda Cloud does not support zones.')
+    return common.accelerator_in_region_or_zone_impl(_df, acc_name, acc_count,
+                                                     region, zone)
+
+
+def get_hourly_cost(instance_type: str,
+                    use_spot: bool = False,
+                    region: Optional[str] = None,
+                    zone: Optional[str] = None) -> float:
+    """Returns the cost, or the cheapest cost among all zones for spot."""
+    assert not use_spot, 'Lambda Cloud does not support spot.'
+    if zone is not None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Lambda Cloud does not support zones.')
+    return common.get_hourly_cost_impl(_df, instance_type, use_spot, region,
+                                       zone)
+
+
+def get_vcpus_from_instance_type(instance_type: str) -> Optional[float]:
+    return common.get_vcpus_from_instance_type_impl(_df, instance_type)
+
+
+def get_default_instance_type(cpus: Optional[str] = None) -> Optional[str]:
+    if cpus is None:
+        cpus = str(_DEFAULT_NUM_VCPUS)
+    # Set to gpu_1x_a100_sxm4 to be the default instance type if match vCPU
+    # requirement.
+    df = _df[_df['InstanceType'].eq('gpu_1x_a100_sxm4')]
+    instance = common.get_instance_type_for_cpus_impl(df, cpus)
+    if not instance:
+        instance = common.get_instance_type_for_cpus_impl(_df, cpus)
+    return instance
+
+
+def get_accelerators_from_instance_type(
+        instance_type: str) -> Optional[Dict[str, int]]:
+    return common.get_accelerators_from_instance_type_impl(_df, instance_type)
+
+
+def get_instance_type_for_accelerator(
+        acc_name: str,
+        acc_count: int,
+        cpus: Optional[str] = None,
+        use_spot: bool = False,
+        region: Optional[str] = None,
+        zone: Optional[str] = None) -> Tuple[Optional[List[str]], List[str]]:
+    """
+    Returns a list of instance types satisfying the required count of
+    accelerators with sorted prices and a list of candidates with fuzzy search.
+    """
+    if zone is not None:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Lambda Cloud does not support zones.')
+    return common.get_instance_type_for_accelerator_impl(df=_df,
+                                                         acc_name=acc_name,
+                                                         acc_count=acc_count,
+                                                         cpus=cpus,
+                                                         use_spot=use_spot,
+                                                         region=region,
+                                                         zone=zone)
+
+
+def get_region_zones_for_instance_type(instance_type: str,
+                                       use_spot: bool) -> List['cloud.Region']:
+    df = _df[_df['InstanceType'] == instance_type]
+    region_list = common.get_region_zones(df, use_spot)
+    # Hack: Enforce US regions are always tried first
+    us_region_list = []
+    other_region_list = []
+    for region in region_list:
+        if region.name.startswith('us-'):
+            us_region_list.append(region)
+        else:
+            other_region_list.append(region)
+    return us_region_list + other_region_list
+
+
+def list_accelerators(
+        gpus_only: bool,
+        name_filter: Optional[str],
+        region_filter: Optional[str],
+        case_sensitive: bool = True
+) -> Dict[str, List[common.InstanceTypeInfo]]:
+    """Returns all instance types in Lambda offering GPUs."""
+    return common.list_accelerators_impl('Lambda', _df, gpus_only, name_filter,
+                                         region_filter, case_sensitive)
diff --git a/sky/core.py b/sky/core.py
index 1242c0fa3..529469f4d 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -4,6 +4,7 @@ import getpass
 import sys
 from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
+from sky import clouds
 from sky import dag
 from sky import task
 from sky import backends
@@ -309,6 +310,11 @@ def stop(cluster_name: str, purge: bool = False) -> None:
             f'Stopping cluster {cluster_name!r} with TPU VM Pod '
             'is not supported.')
 
+    # Check cloud supports stopping instances
+    cloud = handle.launched_resources.cloud
+    cloud.check_features_are_supported(
+        {clouds.CloudImplementationFeatures.STOP})
+
     backend = backend_utils.get_backend_from_handle(handle)
     if (isinstance(backend, backends.CloudVmRayBackend) and
             handle.launched_resources.use_spot):
@@ -427,6 +433,12 @@ def autostop(
             f'{operation} cluster {cluster_name!r} with TPU VM Pod '
             'is not supported.')
 
+    # Check autostop is implemented for cloud
+    cloud = handle.launched_resources.cloud
+    if not down and idle_minutes >= 0:
+        cloud.check_features_are_supported(
+            {clouds.CloudImplementationFeatures.AUTOSTOP})
+
     backend = backend_utils.get_backend_from_handle(handle)
     usage_lib.record_cluster_name_for_current_operation(cluster_name)
     backend.set_autostop(handle, idle_minutes, down)
diff --git a/sky/data/storage.py b/sky/data/storage.py
index 513a173cc..f16b73282 100644
--- a/sky/data/storage.py
+++ b/sky/data/storage.py
@@ -87,6 +87,9 @@ def get_storetype_from_cloud(cloud: clouds.Cloud) -> StoreType:
     elif isinstance(cloud, clouds.Azure):
         with ux_utils.print_exception_no_traceback():
             raise ValueError('Azure Blob Storage is not supported yet.')
+    elif isinstance(cloud, clouds.Lambda):
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError('Lambda Cloud does not provide cloud storage.')
     else:
         with ux_utils.print_exception_no_traceback():
             raise ValueError(f'Unknown cloud type: {cloud}')
diff --git a/sky/execution.py b/sky/execution.py
index 8164f107c..6038ba16c 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -23,6 +23,7 @@ import colorama
 
 import sky
 from sky import backends
+from sky import clouds
 from sky import exceptions
 from sky import global_user_state
 from sky import optimizer
@@ -176,9 +177,24 @@ def _execute(
         existing_handle = global_user_state.get_handle_from_cluster_name(
             cluster_name)
         cluster_exists = existing_handle is not None
+    if cluster_exists:
+        assert len(task.resources) == 1
+        task_resources = list(task.resources)[0]
+        if task_resources.cpus is not None:
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(
+                    'Cannot specify CPU when using an existing cluster. '
+                    'CPU is only used for selecting the instance type when '
+                    'creating a new cluster.')
 
     stages = stages if stages is not None else list(Stage)
 
+    # Requested features that some clouds support and others don't.
+    requested_features = set()
+
+    if task.num_nodes > 1:
+        requested_features.add(clouds.CloudImplementationFeatures.MULTI_NODE)
+
     backend = backend if backend is not None else backends.CloudVmRayBackend()
     if isinstance(backend, backends.CloudVmRayBackend):
         if down and idle_minutes_to_autostop is None:
@@ -199,6 +215,11 @@ def _execute(
                             f'{colorama.Style.RESET_ALL}')
                 idle_minutes_to_autostop = 1
             stages.remove(Stage.DOWN)
+
+            if not down:
+                requested_features.add(
+                    clouds.CloudImplementationFeatures.AUTOSTOP)
+
     elif idle_minutes_to_autostop is not None:
         # TODO(zhwu): Autostop is not supported for non-CloudVmRayBackend.
         with ux_utils.print_exception_no_traceback():
@@ -232,7 +253,9 @@ def _execute(
                     task = dag.tasks[0]  # Keep: dag may have been deep-copied.
                     assert task.best_resources is not None, task
 
-    backend.register_info(dag=dag, optimize_target=optimize_target)
+    backend.register_info(dag=dag,
+                          optimize_target=optimize_target,
+                          requested_features=requested_features)
 
     if task.storage_mounts is not None:
         # Optimizer should eventually choose where to store bucket
diff --git a/sky/optimizer.py b/sky/optimizer.py
index 3a9189fa5..7d5d4aede 100644
--- a/sky/optimizer.py
+++ b/sky/optimizer.py
@@ -960,6 +960,10 @@ def _fill_in_launchable_resources(
                                 f'{colorama.Fore.CYAN}'
                                 f'{sorted(all_fuzzy_candidates)}'
                                 f'{colorama.Style.RESET_ALL}')
+                elif resources.cpus is not None:
+                    logger.info('Try specifying a different CPU count, '
+                                'or add "+" to the end of the CPU count '
+                                'to allow for larger instances.')
 
         launchable[resources] = _filter_out_blocked_launchable_resources(
             launchable[resources], blocked_resources)
diff --git a/sky/registry.py b/sky/registry.py
index bd1606700..553464f21 100644
--- a/sky/registry.py
+++ b/sky/registry.py
@@ -13,6 +13,7 @@ _CLOUDS = [
     clouds.AWS(),
     clouds.Azure(),
     clouds.GCP(),
+    clouds.Lambda(),
 ]
 
 
diff --git a/sky/resources.py b/sky/resources.py
index 10baeaae9..f5b960c74 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -41,16 +41,15 @@ class Resources:
         # TODO:
         sky.Resources(requests={'mem': '16g', 'cpu': 8})
     """
-    # If any fields changed:
-    # 1. Increment the version. For backward compatibility.
-    # 2. Change the __setstate__ method to handle the new fields.
-    # 3. Modify the to_config method to handle the new fields.
-    _VERSION = 6
+    # If any fields changed, increment the version. For backward compatibility,
+    # modify the __setstate__ method to handle the old version.
+    _VERSION = 7
 
     def __init__(
         self,
         cloud: Optional[clouds.Cloud] = None,
         instance_type: Optional[str] = None,
+        cpus: Union[None, int, float, str] = None,
         accelerators: Union[None, str, Dict[str, int]] = None,
         accelerator_args: Optional[Dict[str, str]] = None,
         use_spot: Optional[bool] = None,
@@ -96,10 +95,12 @@ class Resources:
                     k.strip(): v.strip() for k, v in image_id.items()
                 }
 
+        self._set_cpus(cpus)
         self._set_accelerators(accelerators, accelerator_args)
 
         self._try_validate_local()
         self._try_validate_instance_type()
+        self._try_validate_cpus()
         self._try_validate_accelerators()
         self._try_validate_spot()
         self._try_validate_image_id()
@@ -135,6 +136,10 @@ class Resources:
             if self.accelerator_args is not None:
                 accelerator_args = f', accelerator_args={self.accelerator_args}'
 
+        cpus = ''
+        if self.cpus is not None:
+            cpus = f', cpus={self.cpus}'
+
         if isinstance(self.cloud, clouds.Local):
             return f'{self.cloud}({self.accelerators})'
 
@@ -160,7 +165,7 @@ class Resources:
 
         hardware_str = (
             f'{instance_type}{use_spot}'
-            f'{accelerators}{accelerator_args}{image_id}{disk_size}')
+            f'{cpus}{accelerators}{accelerator_args}{image_id}{disk_size}')
         # It may have leading ',' (for example, instance_type not set) or empty
         # spaces.  Remove them.
         while hardware_str and hardware_str[0] in (',', ' '):
@@ -188,6 +193,19 @@ class Resources:
     def instance_type(self):
         return self._instance_type
 
+    @property
+    def cpus(self) -> Optional[str]:
+        """Returns the number of vCPUs that each instance must have.
+
+        For example, cpus='4' means each instance must have exactly 4 vCPUs,
+        and cpus='4+' means each instance must have at least 4 vCPUs.
+
+        (Developer note: The cpus field is only used to select the instance type
+        at launch time. Thus, Resources in the backend's ResourceHandle will
+        always have the cpus field set to None.)
+        """
+        return self._cpus
+
     @property
     def accelerators(self) -> Optional[Dict[str, int]]:
         """Returns the accelerators field directly or by inferring.
@@ -227,6 +245,36 @@ class Resources:
     def image_id(self) -> Optional[Dict[str, str]]:
         return self._image_id
 
+    def _set_cpus(
+        self,
+        cpus: Union[None, int, float, str],
+    ) -> None:
+        if cpus is None:
+            self._cpus = None
+            return
+
+        self._cpus = str(cpus)
+        if isinstance(cpus, str):
+            if cpus.endswith('+'):
+                num_cpus_str = cpus[:-1]
+            else:
+                num_cpus_str = cpus
+
+            try:
+                num_cpus = float(num_cpus_str)
+            except ValueError:
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        f'The "cpus" field should be either a number or '
+                        f'a string "<number>+". Found: {cpus!r}') from None
+        else:
+            num_cpus = float(cpus)
+
+        if num_cpus <= 0:
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(
+                    f'The "cpus" field should be positive. Found: {cpus!r}')
+
     def _set_accelerators(
         self,
         accelerators: Union[None, str, Dict[str, int]],
@@ -375,6 +423,30 @@ class Resources:
                 f'inferred from the instance_type {self.instance_type!r}.')
             self._cloud = valid_clouds[0]
 
+    def _try_validate_cpus(self) -> None:
+        if self.cpus is None:
+            return
+        if self.instance_type is not None:
+            # The assertion should be true because we have already executed
+            # _try_validate_instance_type() before this method.
+            # The _try_validate_instance_type() method infers and sets
+            # self.cloud if self.instance_type is not None.
+            assert self.cloud is not None
+            cpus = self.cloud.get_vcpus_from_instance_type(self.instance_type)
+            if self.cpus.endswith('+'):
+                if cpus < float(self.cpus[:-1]):
+                    with ux_utils.print_exception_no_traceback():
+                        raise ValueError(
+                            f'{self.instance_type} does not have enough vCPUs. '
+                            f'{self.instance_type} has {cpus} vCPUs, '
+                            f'but {self.cpus} is requested.')
+            elif cpus != float(self.cpus):
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        f'{self.instance_type} does not have the requested '
+                        f'number of vCPUs. {self.instance_type} has {cpus} '
+                        f'vCPUs, but {self.cpus} is requested.')
+
     def _try_validate_accelerators(self) -> None:
         """Validate accelerators against the instance type and region/zone."""
         acc_requested = self.accelerators
@@ -526,62 +598,6 @@ class Resources:
                 self.accelerators, self.use_spot, self._region, self._zone)
         return hourly_cost * hours
 
-    def is_same_resources(self, other: 'Resources') -> bool:
-        """Returns whether two resources are the same.
-
-        Returns True if they are the same, False if not.
-        """
-        if (self.cloud is None) != (other.cloud is None):
-            # self and other's cloud should be both None or both not None
-            return False
-
-        if self.cloud is not None and not self.cloud.is_same_cloud(other.cloud):
-            return False
-        # self.cloud == other.cloud
-
-        if (self.region is None) != (other.region is None):
-            # self and other's region should be both None or both not None
-            return False
-
-        if self.region is not None and self.region != other.region:
-            return False
-        # self.region <= other.region
-
-        if (self.zone is None) != (other.zone is None):
-            # self and other's zone should be both None or both not None
-            return False
-
-        if self.zone is not None and self.zone != other.zone:
-            return False
-
-        if (self.image_id is None) != (other.image_id is None):
-            # self and other's image id should be both None or both not None
-            return False
-
-        if (self.image_id is not None and self.image_id != other.image_id):
-            return False
-
-        if (self._instance_type is not None and
-                self._instance_type != other.instance_type):
-            return False
-        # self._instance_type == other.instance_type
-
-        other_accelerators = other.accelerators
-        accelerators = self.accelerators
-        if accelerators != other_accelerators:
-            return False
-        # self.accelerators == other.accelerators
-
-        if self.accelerator_args != other.accelerator_args:
-            return False
-        # self.accelerator_args == other.accelerator_args
-
-        if self.use_spot != other.use_spot:
-            return False
-
-        # self == other
-        return True
-
     def less_demanding_than(self,
                             other: Union[List['Resources'], 'Resources'],
                             requested_num_nodes: int = 1) -> bool:
@@ -677,6 +693,7 @@ class Resources:
         return all([
             self.cloud is None,
             self._instance_type is None,
+            self.cpus is None,
             self.accelerators is None,
             self.accelerator_args is None,
             not self._use_spot_specified,
@@ -688,6 +705,7 @@ class Resources:
         resources = Resources(
             cloud=override.pop('cloud', self.cloud),
             instance_type=override.pop('instance_type', self.instance_type),
+            cpus=override.pop('cpus', self.cpus),
             accelerators=override.pop('accelerators', self.accelerators),
             accelerator_args=override.pop('accelerator_args',
                                           self.accelerator_args),
@@ -726,6 +744,8 @@ class Resources:
                 config.pop('cloud'))
         if config.get('instance_type') is not None:
             resources_fields['instance_type'] = config.pop('instance_type')
+        if config.get('cpus') is not None:
+            resources_fields['cpus'] = str(config.pop('cpus'))
         if config.get('accelerators') is not None:
             resources_fields['accelerators'] = config.pop('accelerators')
         if config.get('accelerator_args') is not None:
@@ -759,6 +779,7 @@ class Resources:
 
         add_if_not_none('cloud', str(self.cloud))
         add_if_not_none('instance_type', self.instance_type)
+        add_if_not_none('cpus', self.cpus)
         add_if_not_none('accelerators', self.accelerators)
         add_if_not_none('accelerator_args', self.accelerator_args)
 
@@ -818,6 +839,9 @@ class Resources:
                 }
             state['_accelerators'] = accelerators
 
+        if version < 7:
+            self._cpus = None
+
         image_id = state.get('_image_id', None)
         if isinstance(image_id, str):
             state['_image_id'] = {state.get('_region', None): image_id}
diff --git a/sky/setup_files/MANIFEST.in b/sky/setup_files/MANIFEST.in
index 7981e7aae..f6ba9e298 100644
--- a/sky/setup_files/MANIFEST.in
+++ b/sky/setup_files/MANIFEST.in
@@ -4,6 +4,7 @@ include sky/skylet/providers/aws/*
 include sky/skylet/providers/aws/cloudwatch/*
 include sky/skylet/providers/azure/*
 include sky/skylet/providers/gcp/*
+include sky/skylet/providers/lambda_cloud/*
 include sky/skylet/ray_patches/*.patch
 include sky/templates/*
 include sky/setup_files/*
diff --git a/sky/skylet/autostop_lib.py b/sky/skylet/autostop_lib.py
index 0756ec29e..be97e30a2 100644
--- a/sky/skylet/autostop_lib.py
+++ b/sky/skylet/autostop_lib.py
@@ -7,6 +7,7 @@ from typing import List, Optional
 
 from sky import sky_logging
 from sky.skylet import configs
+from sky.utils import common_utils
 
 logger = sky_logging.init_logger(__name__)
 
@@ -16,6 +17,10 @@ _AUTOSTOP_CONFIG_KEY = 'autostop_config'
 # user-issued commands (this module) and the Skylet process running the
 # AutostopEvent need to access that state.
 _AUTOSTOP_LAST_ACTIVE_TIME = 'autostop_last_active_time'
+# AutostopEvent sets this to the boot time when the autostop of the cluster
+# starts. This is used for checking whether the cluster is in the process
+# of autostopping for the current machine.
+_AUTOSTOP_INDICATOR = 'autostop_indicator'
 
 
 class AutostopConfig:
@@ -57,6 +62,26 @@ def set_autostop(idle_minutes: int, backend: Optional[str], down: bool) -> None:
         set_last_active_time_to_now()
 
 
+def set_autostopping_started() -> None:
+    """Sets the boot time of the machine when autostop starts.
+
+    This function should be called when the cluster is started to autostop,
+    and the boot time of the machine will be stored in the configs database
+    as an autostop indicator, which is used for checking whether the cluster
+    is in the process of autostopping. The indicator is valid only when the
+    machine has the same boot time as the one stored in the indicator.
+    """
+    logger.debug('Setting is_autostopping.')
+    configs.set_config(_AUTOSTOP_INDICATOR, str(psutil.boot_time()))
+
+
+def get_is_autostopping_payload() -> str:
+    """Returns whether the cluster is in the process of autostopping."""
+    result = configs.get_config(_AUTOSTOP_INDICATOR)
+    is_autostopping = (result == str(psutil.boot_time()))
+    return common_utils.encode_payload(is_autostopping)
+
+
 def get_last_active_time() -> float:
     """Returns the last active time, or -1 if none has been set."""
     result = configs.get_config(_AUTOSTOP_LAST_ACTIVE_TIME)
@@ -88,6 +113,11 @@ class AutostopCodeGen:
         ]
         return cls._build(code)
 
+    @classmethod
+    def is_autostopping(cls) -> str:
+        code = ['print(autostop_lib.get_is_autostopping_payload())']
+        return cls._build(code)
+
     @classmethod
     def _build(cls, code: List[str]) -> str:
         code = cls._PREFIX + code
diff --git a/sky/skylet/configs.py b/sky/skylet/configs.py
index fadeed194..b5d177646 100644
--- a/sky/skylet/configs.py
+++ b/sky/skylet/configs.py
@@ -60,6 +60,7 @@ def get_config(key: str) -> Optional[str]:
         rows = cursor.execute('SELECT value FROM config WHERE key = ?', (key,))
         for (value,) in rows:
             return value
+        return None
 
 
 @ensure_table
diff --git a/sky/skylet/events.py b/sky/skylet/events.py
index d3b66dabf..83f2fd4e9 100644
--- a/sky/skylet/events.py
+++ b/sky/skylet/events.py
@@ -128,6 +128,7 @@ class AutostopEvent(SkyletEvent):
     def _stop_cluster(self, autostop_config):
         if (autostop_config.backend ==
                 cloud_vm_ray_backend.CloudVmRayBackend.NAME):
+            autostop_lib.set_autostopping_started()
             self._replace_yaml_for_stopping(self._ray_yaml_path,
                                             autostop_config.down)
 
diff --git a/sky/skylet/providers/gcp/config.py b/sky/skylet/providers/gcp/config.py
index 2bd81ad61..62a10212b 100644
--- a/sky/skylet/providers/gcp/config.py
+++ b/sky/skylet/providers/gcp/config.py
@@ -4,6 +4,7 @@ import logging
 import os
 import time
 from functools import partial
+from typing import Dict, List, Set, Tuple
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
@@ -13,7 +14,7 @@ from google.oauth2.credentials import Credentials as OAuthCredentials
 from googleapiclient import discovery, errors
 
 from sky.skylet.providers.gcp.node import MAX_POLLS, POLL_INTERVAL, GCPNodeType, GCPCompute
-from sky.skylet.providers.gcp.constants import SKYPILOT_VPC_NAME, VPC_TEMPLATE, FIREWALL_RULES_TEMPLATE
+from sky.skylet.providers.gcp.constants import SKYPILOT_VPC_NAME, VPC_TEMPLATE, FIREWALL_RULES_TEMPLATE, FIREWALL_RULES_REQUIRED
 from ray.autoscaler._private.util import check_legacy_fields
 
 logger = logging.getLogger(__name__)
@@ -490,7 +491,7 @@ def _configure_key_pair(config, compute):
 
 def _check_firewall_rules(vpc_name, config, compute):
     """Check if the firewall rules in the VPC are sufficient."""
-    required_rules = FIREWALL_RULES_TEMPLATE.copy()
+    required_rules = FIREWALL_RULES_REQUIRED.copy()
 
     operation = (
         compute.networks().
@@ -504,24 +505,90 @@ def _check_firewall_rules(vpc_name, config, compute):
         return False
     effective_rules = response["firewalls"]
 
-    def _get_refined_rule(rule):
-        KEY_TO_COMPARE = {"sourceRanges", "allowed", "direction"}
-        refined_rule = {}
-        for k in KEY_TO_COMPARE:
-            if k not in rule:
-                continue
-            if k == "allowed":
-                refined_rule[k] = sorted(rule[k], key=lambda x: x["IPProtocol"])
-            else:
-                refined_rule[k] = rule[k]
-        return refined_rule
-
-    required_rules = list(map(_get_refined_rule, required_rules))
-    effective_rules = list(map(_get_refined_rule, effective_rules))
+    def _merge_and_refine_rule(rules):
+        """Returns the reformatted rules from the firewall rules
+
+        The function translates firewall rules fetched from the cloud provider
+        to a format for simple comparison.
+
+        Example of firewall rules from the cloud:
+        [
+            {
+                ...
+                "direction": "INGRESS",
+                "allowed": [
+                    {"IPProtocol": "tcp", "ports": ['80', '443']},
+                    {"IPProtocol": "udp", "ports": ['53']},
+                ],
+                "sourceRanges": ["10.128.0.0/9"],
+            },
+            {
+                ...
+                "direction": "INGRESS",
+                "allowed": [{
+                    "IPProtocol": "tcp",
+                    "ports": ["22"],
+                }],
+                "sourceRanges": ["0.0.0.0/0"],
+            },
+        ]
 
-    for rule in required_rules:
-        if rule not in effective_rules:
+        Returns:
+            source2rules: Dict[(direction, sourceRanges) -> Dict(protocol -> Set[ports])]
+                Example {
+                    ("INGRESS", "10.128.0.0/9"): {"tcp": {80, 443}, "udp": {53}},
+                    ("INGRESS", "0.0.0.0/0"): {"tcp": {22}},
+                }
+        """
+        source2rules : Dict[Tuple[str, str], Dict[str, Set[int]]] = {}
+        source2allowed_list : Dict[Tuple[str, str], List[Dict[str, str]]] = {}
+        for rule in rules:
+            direction = rule.get("direction", "")
+            sources = rule.get("sourceRanges", [])
+            allowed = rule.get("allowed", [])
+            for source in sources:
+                key = (direction, source)
+                source2allowed_list[key] = source2allowed_list.get(key, []) + allowed
+        for direction_source, allowed_list in source2allowed_list.items():
+            source2rules[direction_source] = {}
+            for allowed in allowed_list:
+                # Example of port_list: ["20", "50-60"]
+                # If list is empty, it means all ports
+                port_list = allowed.get("ports", [])
+                port_set = set()
+                if port_list == []:
+                    port_set.update(set(range(1, 65536)))
+                else:
+                    for port_range in port_list:
+                        parse_ports = port_range.split('-')
+                        if len(parse_ports) == 1:
+                            port_set.add(int(parse_ports[0]))
+                        else:
+                            assert (
+                                len(parse_ports) == 2
+                            ), f"Failed to parse the port range: {port_range}"
+                            port_set.update(
+                                set(range(int(parse_ports[0]), int(parse_ports[1]) + 1)))
+                if allowed["IPProtocol"] not in source2rules[direction_source]:
+                    source2rules[direction_source][allowed["IPProtocol"]] = set()
+                source2rules[direction_source][allowed["IPProtocol"]].update(port_set)
+        return source2rules
+
+    effective_rules = _merge_and_refine_rule(effective_rules)
+    required_rules = _merge_and_refine_rule(required_rules)
+
+    for direction_source, allowed_req in required_rules.items():
+        if direction_source not in effective_rules:
             return False
+        allowed_eff = effective_rules[direction_source]
+        # Special case: "all" means allowing all traffic
+        if "all" in allowed_eff:
+            continue
+        # Check if the required ports are a subset of the effective ports
+        for protocol, ports_req in allowed_req.items():
+            ports_eff = allowed_eff.get(protocol, set())
+            if not ports_req.issubset(ports_eff):
+                return False
     return True
 
 
@@ -557,19 +624,32 @@ def get_usable_vpc(config):
     if usable_vpc_name is None:
         logger.info(f"Creating a default VPC network, {SKYPILOT_VPC_NAME}...")
 
-        # Create a default VPC network
         proj_id = config["provider"]["project_id"]
-        body = VPC_TEMPLATE.copy()
-        body["name"] = body["name"].format(VPC_NAME=SKYPILOT_VPC_NAME)
-        body["selfLink"] = body["selfLink"].format(PROJ_ID=proj_id, VPC_NAME=SKYPILOT_VPC_NAME)
-        _create_vpcnet(config, compute, body)
+        # Create a SkyPilot VPC network if it doesn't exist
+        vpc_list = _list_vpcnets(config, compute, filter=f"name={SKYPILOT_VPC_NAME}")
+        if len(vpc_list) == 0:
+            body = VPC_TEMPLATE.copy()
+            body["name"] = body["name"].format(VPC_NAME=SKYPILOT_VPC_NAME)
+            body["selfLink"] = body["selfLink"].format(
+                PROJ_ID=proj_id, VPC_NAME=SKYPILOT_VPC_NAME)
+            _create_vpcnet(config, compute, body)
 
         # Create firewall rules
         for rule in FIREWALL_RULES_TEMPLATE:
+            # Query firewall rule by its name (unique in a project).
+            # If the rule already exists, delete it first.
+            rule_name = rule["name"].format(VPC_NAME=SKYPILOT_VPC_NAME)
+            rule_list = _list_firewall_rules(
+                config, compute, filter=f"(name={rule_name})")
+            if len(rule_list) > 0:
+                _delete_firewall_rule(config, compute, rule_name)
+
             body = rule.copy()
             body["name"] = body["name"].format(VPC_NAME=SKYPILOT_VPC_NAME)
-            body["network"] = body["network"].format(PROJ_ID=proj_id, VPC_NAME=SKYPILOT_VPC_NAME)
-            body["selfLink"] = body["selfLink"].format(PROJ_ID=proj_id, VPC_NAME=SKYPILOT_VPC_NAME)
+            body["network"] = body["network"].format(
+                PROJ_ID=proj_id, VPC_NAME=SKYPILOT_VPC_NAME)
+            body["selfLink"] = body["selfLink"].format(
+                PROJ_ID=proj_id, VPC_NAME=SKYPILOT_VPC_NAME)
             _create_firewall_rule(config, compute, body)
 
         usable_vpc_name = SKYPILOT_VPC_NAME
@@ -639,6 +719,30 @@ def _create_firewall_rule(config, compute, body):
     return response
 
 
+def _delete_firewall_rule(config, compute, name):
+    operation = (
+        compute.firewalls()
+        .delete(project=config["provider"]["project_id"], firewall=name)
+        .execute()
+    )
+    response = wait_for_compute_global_operation(
+        config["provider"]["project_id"], operation, compute
+    )
+    return response
+
+
+def _list_firewall_rules(config, compute, filter=None):
+    response = (
+        compute.firewalls()
+        .list(
+            project=config["provider"]["project_id"],
+            filter=filter,
+        )
+        .execute()
+    )
+    return response["items"] if "items" in response else []
+
+
 def _create_vpcnet(config, compute, body):
     operation = (
         compute.networks()
@@ -651,11 +755,12 @@ def _create_vpcnet(config, compute, body):
     return response
 
 
-def _list_vpcnets(config, compute):
+def _list_vpcnets(config, compute, filter=None):
     response = (
         compute.networks()
         .list(
             project=config["provider"]["project_id"],
+            filter=filter,
         )
         .execute()
     )
diff --git a/sky/skylet/providers/gcp/constants.py b/sky/skylet/providers/gcp/constants.py
index 00efecde8..1fc266249 100644
--- a/sky/skylet/providers/gcp/constants.py
+++ b/sky/skylet/providers/gcp/constants.py
@@ -9,6 +9,28 @@ VPC_TEMPLATE = {
     "mtu": 1460,
     "routingConfig": {"routingMode": "GLOBAL"},
 }
+# Required firewall rules for SkyPilot to work.
+FIREWALL_RULES_REQUIRED = [
+    # Allow internal connections between GCP VMs for Ray multi-node cluster.
+    {
+        "direction": "INGRESS",
+        "allowed": [
+            {'IPProtocol': 'tcp', 'ports': ['0-65535']},
+            {'IPProtocol': 'udp', 'ports': ['0-65535']},
+        ],
+        "sourceRanges": ["10.128.0.0/9"],
+    },
+    # Allow ssh connection from anywhere.
+    {
+        "direction": "INGRESS",
+        "allowed": [{
+            "IPProtocol": "tcp",
+            "ports": ["22"],
+        }],
+        "sourceRanges": ["0.0.0.0/0"],
+    }
+]
+# Template when creating firewall rules for a new VPC.
 FIREWALL_RULES_TEMPLATE = [
     {
         "name": "{VPC_NAME}-allow-custom",
diff --git a/sky/skylet/providers/lambda_cloud/__init__.py b/sky/skylet/providers/lambda_cloud/__init__.py
new file mode 100644
index 000000000..64dac295e
--- /dev/null
+++ b/sky/skylet/providers/lambda_cloud/__init__.py
@@ -0,0 +1,2 @@
+"""Lambda Cloud node provider"""
+from sky.skylet.providers.lambda_cloud.node_provider import LambdaNodeProvider
diff --git a/sky/skylet/providers/lambda_cloud/lambda_utils.py b/sky/skylet/providers/lambda_cloud/lambda_utils.py
new file mode 100644
index 000000000..0ae886070
--- /dev/null
+++ b/sky/skylet/providers/lambda_cloud/lambda_utils.py
@@ -0,0 +1,185 @@
+"""Lambda Cloud helper functions."""
+import os
+import json
+import requests
+from typing import Any, Dict, List
+
+CREDENTIALS_PATH = '~/.lambda_cloud/lambda_keys'
+API_ENDPOINT = 'https://cloud.lambdalabs.com/api/v1'
+
+
+class LambdaCloudError(Exception):
+    pass
+
+
+class Metadata:
+    """Per-cluster metadata file."""
+
+    def __init__(self, path_prefix: str, cluster_name: str) -> None:
+        # TODO(ewzeng): Metadata file is not thread safe. This is fine for
+        # now since SkyPilot uses a per-cluster lock for ray-related
+        # operations. In the future, add a filelock around __getitem__,
+        # __setitem__ and refresh.
+        self.path = os.path.expanduser(f'{path_prefix}-{cluster_name}')
+        # In case parent directory does not exist
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+
+    def __getitem__(self, instance_id: str) -> Dict[str, Any]:
+        assert os.path.exists(self.path), 'Metadata file not found'
+        with open(self.path, 'r') as f:
+            metadata = json.load(f)
+        return metadata.get(instance_id)
+
+    def __setitem__(self, instance_id: str, value: Dict[str, Any]) -> None:
+        # Read from metadata file
+        if os.path.exists(self.path):
+            with open(self.path, 'r') as f:
+                metadata = json.load(f)
+        else:
+            metadata = {}
+        # Update metadata
+        if value is None:
+            if instance_id in metadata:
+                metadata.pop(instance_id) # del entry
+            if len(metadata) == 0:
+                if os.path.exists(self.path):
+                    os.remove(self.path)
+                return
+        else:
+            metadata[instance_id] = value
+        # Write to metadata file
+        with open(self.path, 'w') as f:
+            json.dump(metadata, f)
+
+    def refresh(self, instance_ids: List[str]) -> None:
+        """Remove all tags for instances not in instance_ids."""
+        if not os.path.exists(self.path):
+            return
+        with open(self.path, 'r') as f:
+            metadata = json.load(f)
+        for instance_id in list(metadata.keys()):
+            if instance_id not in instance_ids:
+                del metadata[instance_id]
+        if len(metadata) == 0:
+            os.remove(self.path)
+            return
+        with open(self.path, 'w') as f:
+            json.dump(metadata, f)
+
+
+def raise_lambda_error(response: requests.Response) -> None:
+    """Raise LambdaCloudError if appropriate. """
+    status_code = response.status_code
+    if status_code == 200:
+        return
+    if status_code == 429:
+        # https://docs.lambdalabs.com/cloud/rate-limiting/
+        raise LambdaCloudError('Your API requests are being rate limited.')
+    try:
+        resp_json = response.json()
+        code = resp_json['error']['code']
+        message = resp_json['error']['message']
+    except (KeyError, json.decoder.JSONDecodeError):
+        raise LambdaCloudError(f'Unexpected error. Status code: {status_code}')
+    raise LambdaCloudError(f'{code}: {message}')
+
+
+class LambdaCloudClient:
+    """Wrapper functions for Lambda Cloud API."""
+
+    def __init__(self) -> None:
+        self.credentials = os.path.expanduser(CREDENTIALS_PATH)
+        assert os.path.exists(self.credentials), 'Credentials not found'
+        with open(self.credentials, 'r') as f:
+            lines = [line.strip() for line in f.readlines() if ' = ' in line]
+            self._credentials = {
+                line.split(' = ')[0]: line.split(' = ')[1]
+                for line in lines
+            }
+        self.api_key = self._credentials['api_key']
+        self.ssh_key_name = self._credentials.get('ssh_key_name', None)
+        self.headers = {'Authorization': f'Bearer {self.api_key}'}
+
+    def create_instances(self,
+                         instance_type: str = 'gpu_1x_a100_sxm4',
+                         region: str = 'us-east-1',
+                         quantity: int = 1,
+                         name: str = '') -> Dict[str, Any]:
+        """Launch new instances."""
+        assert self.ssh_key_name is not None
+
+        # Optimization:
+        # Most API requests are rate limited at ~1 request every second but
+        # launch requests are rate limited at ~1 request every 10 seconds.
+        # So don't use launch requests to check availability.
+        # See https://docs.lambdalabs.com/cloud/rate-limiting/ for more.
+        available_regions = self.list_catalog()[instance_type]\
+                ['regions_with_capacity_available']
+        available_regions = [reg['name'] for reg in available_regions]
+        if region not in available_regions:
+            if len(available_regions) > 0:
+                aval_reg = ' '.join(available_regions)
+            else:
+                aval_reg = 'None'
+            raise LambdaCloudError(('instance-operations/launch/'
+                                   'insufficient-capacity: Not enough '
+                                   'capacity to fulfill launch request. '
+                                   'Regions with capacity available: '
+                                   f'{aval_reg}'))
+
+        # Try to launch instance
+        data = json.dumps({
+                    'region_name': region,
+                    'instance_type_name': instance_type,
+                    'ssh_key_names': [
+                        self.ssh_key_name
+                    ],
+                    'quantity': quantity,
+                    'name': name
+                })
+        response = requests.post(f'{API_ENDPOINT}/instance-operations/launch',
+                                 data=data,
+                                 headers=self.headers)
+        raise_lambda_error(response)
+        return response.json().get('data', []).get('instance_ids', [])
+
+    def remove_instances(self, *instance_ids: str) -> Dict[str, Any]:
+        """Terminate instances."""
+        data = json.dumps({
+            'instance_ids': [
+                instance_ids[0] # TODO(ewzeng) don't hardcode
+            ]
+        })
+        response = requests.post(f'{API_ENDPOINT}/instance-operations/terminate',
+                                 data=data,
+                                 headers=self.headers)
+        raise_lambda_error(response)
+        return response.json().get('data', []).get('terminated_instances', [])
+
+    def list_instances(self) -> Dict[str, Any]:
+        """List existing instances."""
+        response = requests.get(f'{API_ENDPOINT}/instances', headers=self.headers)
+        raise_lambda_error(response)
+        return response.json().get('data', [])
+
+    def set_ssh_key(self, name: str, pub_key: str) -> None:
+        """Set ssh key."""
+        data = json.dumps({
+            'name': name,
+            'public_key': pub_key
+        })
+        response = requests.post(f'{API_ENDPOINT}/ssh-keys',
+                                 data=data,
+                                 headers=self.headers)
+        raise_lambda_error(response)
+        self.ssh_key_name = name
+        with open(self.credentials, 'w') as f:
+            f.write(f'api_key = {self.api_key}\n')
+            f.write(f'ssh_key_name = {self.ssh_key_name}\n')
+
+    def list_catalog(self) -> Dict[str, Any]:
+        """List offered instances and their availability."""
+        response = requests.get(f'{API_ENDPOINT}/instance-types',
+                                headers=self.headers)
+        raise_lambda_error(response)
+        return response.json().get('data', [])
diff --git a/sky/skylet/providers/lambda_cloud/node_provider.py b/sky/skylet/providers/lambda_cloud/node_provider.py
new file mode 100644
index 000000000..47ef9b162
--- /dev/null
+++ b/sky/skylet/providers/lambda_cloud/node_provider.py
@@ -0,0 +1,217 @@
+import logging
+import os
+import time
+from threading import RLock
+from typing import Any, Dict, List, Optional
+
+from ray.autoscaler.node_provider import NodeProvider
+from ray.autoscaler.tags import (
+    TAG_RAY_CLUSTER_NAME,
+    TAG_RAY_USER_NODE_TYPE,
+    TAG_RAY_NODE_NAME,
+    TAG_RAY_LAUNCH_CONFIG,
+    TAG_RAY_NODE_STATUS,
+    STATUS_UP_TO_DATE,
+    TAG_RAY_NODE_KIND,
+    NODE_KIND_WORKER,
+    NODE_KIND_HEAD,
+)
+from ray.autoscaler._private.util import hash_launch_conf
+from sky.skylet.providers.lambda_cloud import lambda_utils
+from sky.utils import common_utils
+
+TAG_PATH_PREFIX = '~/.sky/generated/lambda_cloud/metadata'
+REMOTE_RAY_YAML = '~/ray_bootstrap_config.yaml'
+
+logger = logging.getLogger(__name__)
+
+
+def synchronized(f):
+    def wrapper(self, *args, **kwargs):
+        self.lock.acquire()
+        try:
+            return f(self, *args, **kwargs)
+        finally:
+            self.lock.release()
+
+    return wrapper
+
+
+class LambdaNodeProvider(NodeProvider):
+    """Node Provider for Lambda Cloud.
+
+    This provider assumes Lambda Cloud credentials are set.
+    """
+
+    def __init__(self,
+                 provider_config: Dict[str, Any],
+                 cluster_name: str) -> None:
+        NodeProvider.__init__(self, provider_config, cluster_name)
+        self.lock = RLock()
+        self.lambda_client = lambda_utils.LambdaCloudClient()
+        self.cached_nodes = {}
+        self.metadata = lambda_utils.Metadata(TAG_PATH_PREFIX, cluster_name)
+        vms = self._list_instances_in_cluster()
+
+        # The tag file for autodowned clusters is not autoremoved. Hence, if
+        # a previous cluster was autodowned and has the same name as the
+        # current cluster, then self.metadata might load the old tag file.
+        # We prevent this by removing any old vms in the tag file.
+        self.metadata.refresh([node['id'] for node in vms])
+
+        # If tag file does not exist on head, create it and add basic tags.
+        # This is a hack to make sure that ray on head can access some
+        # important tags.
+        # TODO(ewzeng): change when Lambda Cloud adds tag support.
+        ray_yaml_path = os.path.expanduser(REMOTE_RAY_YAML)
+        if os.path.exists(ray_yaml_path) and not os.path.exists(
+                self.metadata.path):
+            config = common_utils.read_yaml(ray_yaml_path)
+            # Ensure correct cluster so sky launch on head node works correctly
+            if config['cluster_name'] != cluster_name:
+                return
+            # Compute launch hash
+            head_node_config = config.get('head_node', {})
+            head_node_type = config.get('head_node_type')
+            if head_node_type:
+                head_config = config['available_node_types'][head_node_type]
+                head_node_config.update(head_config["node_config"])
+            launch_hash = hash_launch_conf(head_node_config, config['auth'])
+            # Populate tags
+            for node in vms:
+                self.metadata[node['id']] = {'tags':
+                    {
+                        TAG_RAY_CLUSTER_NAME: cluster_name,
+                        TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
+                        TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
+                        TAG_RAY_USER_NODE_TYPE: 'ray_head_default',
+                        TAG_RAY_NODE_NAME: f'ray-{cluster_name}-head',
+                        TAG_RAY_LAUNCH_CONFIG: launch_hash,
+                    }}
+
+    def _list_instances_in_cluster(self) -> Dict[str, Any]:
+        """List running instances in cluster."""
+        vms = self.lambda_client.list_instances()
+        return [
+            node for node in vms
+            if node['name'] == self.cluster_name
+        ]
+
+    @synchronized
+    def _get_filtered_nodes(self,
+                            tag_filters: Dict[str, str]) -> Dict[str, Any]:
+
+        def match_tags(vm):
+            vm_info = self.metadata[vm['id']]
+            tags = {} if vm_info is None else vm_info['tags']
+            for k, v in tag_filters.items():
+                if tags.get(k) != v:
+                    return False
+            return True
+
+        vms = self._list_instances_in_cluster()
+        nodes = [self._extract_metadata(vm) for vm in filter(match_tags, vms)]
+        self.cached_nodes = {node['id']: node for node in nodes}
+        return self.cached_nodes
+
+    def _extract_metadata(self, vm: Dict[str, Any]) -> Dict[str, Any]:
+        metadata = {'id': vm['id'], 'status': vm['status'], 'tags': {}}
+        instance_info = self.metadata[vm['id']]
+        if instance_info is not None:
+            metadata['tags'] = instance_info['tags']
+        ip = vm['ip']
+        metadata['external_ip'] = ip
+        # TODO(ewzeng): The internal ip is hard to get, so set it to the
+        # external ip as a hack. This should be changed in the future.
+        #   https://docs.lambdalabs.com/cloud/learn-private-ip-address/
+        metadata['internal_ip'] = ip
+        return metadata
+
+    def non_terminated_nodes(self, tag_filters: Dict[str, str]) -> List[str]:
+        """Return a list of node ids filtered by the specified tags dict.
+
+        This list must not include terminated nodes. For performance reasons,
+        providers are allowed to cache the result of a call to
+        non_terminated_nodes() to serve single-node queries
+        (e.g. is_running(node_id)). This means that non_terminated_nodes() must
+        be called again to refresh results.
+
+        Examples:
+            >>> provider.non_terminated_nodes({TAG_RAY_NODE_KIND: "worker"})
+            ["node-1", "node-2"]
+        """
+        nodes = self._get_filtered_nodes(tag_filters=tag_filters)
+        return [k for k, _ in nodes.items()]
+
+    def is_running(self, node_id: str) -> bool:
+        """Return whether the specified node is running."""
+        return self._get_cached_node(node_id=node_id) is not None
+
+    def is_terminated(self, node_id: str) -> bool:
+        """Return whether the specified node is terminated."""
+        return self._get_cached_node(node_id=node_id) is None
+
+    def node_tags(self, node_id: str) -> Dict[str, str]:
+        """Returns the tags of the given node (string dict)."""
+        return self._get_cached_node(node_id=node_id)['tags']
+
+    def external_ip(self, node_id: str) -> str:
+        """Returns the external ip of the given node."""
+        return self._get_cached_node(node_id=node_id)['external_ip']
+
+    def internal_ip(self, node_id: str) -> str:
+        """Returns the internal ip (Ray ip) of the given node."""
+        return self._get_cached_node(node_id=node_id)['internal_ip']
+
+    def create_node(self,
+                    node_config: Dict[str, Any],
+                    tags: Dict[str, str],
+                    count: int) -> None:
+        """Creates a number of nodes within the namespace."""
+        assert count == 1, count   # Only support 1-node clusters for now
+
+        # get the tags
+        config_tags = node_config.get('tags', {}).copy()
+        config_tags.update(tags)
+        config_tags[TAG_RAY_CLUSTER_NAME] = self.cluster_name
+
+        # create the node
+        ttype = node_config['InstanceType']
+        region = self.provider_config['region']
+        vm_list = self.lambda_client.create_instances(instance_type=ttype,
+                                                      region=region,
+                                                      quantity=1,
+                                                      name=self.cluster_name)
+        assert len(vm_list) == 1, len(vm_list)
+        vm_id = vm_list[0]
+        self.metadata[vm_id] = {'tags': config_tags}
+
+        # Wait for booting to finish
+        # TODO(ewzeng): For multi-node, launch all vms first and then wait.
+        while True:
+            vms = self.lambda_client.list_instances()
+            for vm in vms:
+                if vm['id'] == vm_id and vm['status'] == 'active':
+                    return
+            time.sleep(10)
+
+    @synchronized
+    def set_node_tags(self, node_id: str, tags: Dict[str, str]) -> None:
+        """Sets the tag values (string dict) for the specified node."""
+        node = self._get_node(node_id)
+        node['tags'].update(tags)
+        self.metadata[node_id] = {'tags': node['tags']}
+
+    def terminate_node(self, node_id: str) -> None:
+        """Terminates the specified node."""
+        self.lambda_client.remove_instances(node_id)
+        self.metadata[node_id] = None
+
+    def _get_node(self, node_id: str) -> Optional[Dict[str, Any]]:
+        self._get_filtered_nodes({})  # Side effect: updates cache
+        return self.cached_nodes.get(node_id, None)
+
+    def _get_cached_node(self, node_id: str) -> Optional[Dict[str, Any]]:
+        if node_id in self.cached_nodes:
+            return self.cached_nodes[node_id]
+        return self._get_node(node_id=node_id)
diff --git a/sky/task.py b/sky/task.py
index 613f23edb..d9f464223 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -448,6 +448,7 @@ class Task:
         """
         if isinstance(resources, sky.Resources):
             resources = {resources}
+        # TODO(woosuk): Check if the resources are None.
         self.resources = resources
         return self
 
diff --git a/sky/templates/lambda-ray.yml.j2 b/sky/templates/lambda-ray.yml.j2
new file mode 100644
index 000000000..fb6306fa6
--- /dev/null
+++ b/sky/templates/lambda-ray.yml.j2
@@ -0,0 +1,110 @@
+cluster_name: {{cluster_name}}
+
+# The maximum number of workers nodes to launch in addition to the head node.
+max_workers: {{num_nodes - 1}}
+upscaling_speed: {{num_nodes - 1}}
+idle_timeout_minutes: 60
+
+provider:
+  type: external
+  module: sky.skylet.providers.lambda_cloud.LambdaNodeProvider
+  region: {{region}}
+
+auth:
+  ssh_user: ubuntu
+  ssh_private_key: {{ssh_private_key}}
+
+available_node_types:
+  ray_head_default:
+    resources: {}
+    node_config:
+      InstanceType: {{instance_type}}
+{% if num_nodes > 1 %}
+  ray_worker_default:
+    min_workers: {{num_nodes - 1}}
+    max_workers: {{num_nodes - 1}}
+    resources: {}
+    node_config:
+      InstanceType: {{instance_type}}
+{%- endif %}
+
+head_node_type: ray_head_default
+
+# Format: `REMOTE_PATH : LOCAL_PATH`
+file_mounts: {
+  "{{sky_ray_yaml_remote_path}}": "{{sky_ray_yaml_local_path}}",
+  "{{sky_remote_path}}/{{sky_wheel_hash}}": "{{sky_local_path}}",
+{%- for remote_path, local_path in credentials.items() %}
+  "{{remote_path}}": "{{local_path}}",
+{%- endfor %}
+}
+
+rsync_exclude: []
+
+initialization_commands: []
+
+# List of shell commands to run to set up nodes.
+# NOTE: these are very performance-sensitive. Each new item opens/closes an SSH
+# connection, which is expensive. Try your best to co-locate commands into fewer
+# items!
+#
+# Increment the following for catching performance bugs easier:
+#   current num items (num SSH connections): 1
+setup_commands:
+  # Disable `unattended-upgrades` to prevent apt-get from hanging. It should be called at the beginning before the process started to avoid being blocked. (This is a temporary fix.)
+  # Create ~/.ssh/config file in case the file does not exist in the image.
+  # Line 'rm ..': there is another installation of pip.
+  # Line 'sudo bash ..': set the ulimit as suggested by ray docs for performance. https://docs.ray.io/en/latest/cluster/vms/user-guides/large-cluster-best-practices.html#system-configuration
+  # Line 'sudo grep ..': set the number of threads per process to unlimited to avoid ray job submit stucking issue when the number of running ray jobs increase.
+  # Line 'mkdir -p ..': disable host key check
+  # Line 'python3 -c ..': patch the buggy ray files and enable `-o allow_other` option for `goofys`
+  - sudo systemctl stop unattended-upgrades || true;
+    sudo systemctl disable unattended-upgrades || true;
+    sudo sed -i 's/Unattended-Upgrade "1"/Unattended-Upgrade "0"/g' /etc/apt/apt.conf.d/20auto-upgrades || true;
+    sudo kill -9 `sudo lsof /var/lib/dpkg/lock-frontend | awk '{print $2}' | tail -n 1` || true;
+    sudo pkill -9 apt-get;
+    sudo pkill -9 dpkg;
+    sudo dpkg --configure -a;
+    mkdir -p ~/.ssh; touch ~/.ssh/config;
+    rm ~/.local/bin/pip ~/.local/bin/pip3 ~/.local/bin/pip3.8 ~/.local/bin/pip3.10;
+    (type -a python | grep -q python3) || echo 'alias python=python3' >> ~/.bashrc;
+    (type -a pip | grep -q pip3) || echo 'alias pip=pip3' >> ~/.bashrc;
+    which conda > /dev/null 2>&1 || (wget -nc https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh && bash Miniconda3-latest-Linux-x86_64.sh -b && eval "$(~/miniconda3/bin/conda shell.bash hook)" && conda init && conda config --set auto_activate_base true);
+    source ~/.bashrc;
+    (pip3 list | grep ray | grep {{ray_version}} 2>&1 > /dev/null || pip3 install -U ray[default]=={{ray_version}}) && mkdir -p ~/sky_workdir && mkdir -p ~/.sky/sky_app && touch ~/.sudo_as_admin_successful;
+    (pip3 list | grep skypilot && [ "$(cat {{sky_remote_path}}/current_sky_wheel_hash)" == "{{sky_wheel_hash}}" ]) || (pip3 uninstall skypilot -y; pip3 install "$(echo {{sky_remote_path}}/{{sky_wheel_hash}}/skypilot-{{sky_version}}*.whl)" && echo "{{sky_wheel_hash}}" > {{sky_remote_path}}/current_sky_wheel_hash || exit 1);
+    sudo bash -c 'rm -rf /etc/security/limits.d; echo "* soft nofile 1048576" >> /etc/security/limits.conf; echo "* hard nofile 1048576" >> /etc/security/limits.conf';
+    sudo grep -e '^DefaultTasksMax' /etc/systemd/system.conf || (sudo bash -c 'echo "DefaultTasksMax=infinity" >> /etc/systemd/system.conf'); sudo systemctl set-property user-$(id -u $(whoami)).slice TasksMax=infinity; sudo systemctl daemon-reload;
+    mkdir -p ~/.ssh; (grep -Pzo -q "Host \*\n  StrictHostKeyChecking no" ~/.ssh/config) || printf "Host *\n  StrictHostKeyChecking no\n" >> ~/.ssh/config;
+    python3 -c "from sky.skylet.ray_patches import patch; patch()" || exit 1;
+    [ -f /etc/fuse.conf ] && sudo sed -i 's/#user_allow_other/user_allow_other/g' /etc/fuse.conf || (sudo sh -c 'echo "user_allow_other" > /etc/fuse.conf');
+
+# Command to start ray on the head node. You don't need to change this.
+# NOTE: these are very performance-sensitive. Each new item opens/closes an SSH
+# connection, which is expensive. Try your best to co-locate commands into fewer
+# items! The same comment applies for worker_start_ray_commands.
+#
+# Increment the following for catching performance bugs easier:
+#   current num items (num SSH connections): 2
+head_start_ray_commands:
+  # Start skylet daemon. (Should not place it in the head_setup_commands, otherwise it will run before skypilot is installed.)
+  - ((ps aux | grep -v nohup | grep -v grep | grep -q -- "python3 -m sky.skylet.skylet") || nohup python3 -m sky.skylet.skylet >> ~/.sky/skylet.log 2>&1 &);
+    ray stop; RAY_SCHEDULER_EVENTS=0 ray start --disable-usage-stats --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
+
+{%- if num_nodes > 1 %}
+worker_start_ray_commands:
+  - ray stop; RAY_SCHEDULER_EVENTS=0 ray start --disable-usage-stats --address=$RAY_HEAD_IP:6379 --object-manager-port=8076 {{"--resources='%s'" % custom_resources if custom_resources}} || exit 1;
+    which prlimit && for id in $(pgrep -f raylet/raylet); do sudo prlimit --nofile=1048576:1048576 --pid=$id || true; done;
+{%- else %}
+worker_start_ray_commands: []
+{%- endif %}
+
+head_node: {}
+worker_nodes: {}
+
+# These fields are required for external cloud providers.
+head_setup_commands: []
+worker_setup_commands: []
+cluster_synced_files: []
+file_mounts_sync_continuously: False
diff --git a/sky/utils/common_utils.py b/sky/utils/common_utils.py
index 5ca1d1800..ecb6bfc31 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -16,6 +16,8 @@ from typing import Any, Dict, Optional
 import uuid
 import yaml
 
+import colorama
+
 from sky import sky_logging
 
 _USER_HASH_FILE = os.path.expanduser('~/.sky/user_hash')
@@ -325,9 +327,11 @@ def format_exception(e: Exception, use_bracket: bool = False) -> str:
     Returns:
         A string that represents the exception.
     """
+    bright = colorama.Style.BRIGHT
+    reset = colorama.Style.RESET_ALL
     if use_bracket:
-        return f'[{class_fullname(e.__class__)}]: {e}'
-    return f'{class_fullname(e.__class__)}: {e}'
+        return f'{bright}[{class_fullname(e.__class__)}]:{reset} {e}'
+    return f'{bright}{class_fullname(e.__class__)}:{reset} {e}'
 
 
 def remove_color(s: str):
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index cdd1865aa..c71c83b0e 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -25,6 +25,13 @@ def get_resources_schema():
             'zone': {
                 'type': 'string',
             },
+            'cpus': {
+                'anyOf': [{
+                    'type': 'string',
+                }, {
+                    'type': 'number',
+                }],
+            },
             'accelerators': {
                 'anyOf': [{
                     'type': 'string',


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 4849d9c76..8383a9266 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -12,14 +12,24 @@ from typing import List
 # By default, only run generic tests and cloud-specific tests for GCP and Azure,
 # due to the cloud credit limit for the development account.
 # To only run tests for a specific cloud (as well as generic tests), use
-# --aws, --gcp, or --azure.
+# --aws, --gcp, --azure, or --lambda.
 # To only run tests for managed spot (without generic tests), use --managed-spot.
 # A "generic test" tests a generic functionality (e.g., autostop) that
 # should work on any cloud we support. The cloud used for such a test
 # is controlled by `--generic-cloud` (typically you do not need to set it).
-all_clouds_in_smoke_tests = ['aws', 'gcp', 'azure']
+all_clouds_in_smoke_tests = ['aws', 'gcp', 'azure', 'lambda']
 default_clouds_to_run = ['gcp', 'azure']
 
+# Translate cloud name to pytest keyword. We need this because
+# @pytest.mark.lambda is not allowed, so we use @pytest.mark.lambda_cloud
+# instead.
+cloud_to_pytest_keyword = {
+    'aws': 'aws',
+    'gcp': 'gcp',
+    'azure': 'azure',
+    'lambda': 'lambda_cloud'
+}
+
 
 def pytest_addoption(parser):
     # tests marked as `slow` will be skipped by default, use --runslow to run
@@ -49,8 +59,9 @@ def pytest_addoption(parser):
 def pytest_configure(config):
     config.addinivalue_line('markers', 'slow: mark test as slow to run')
     for cloud in all_clouds_in_smoke_tests:
-        config.addinivalue_line('markers',
-                                f'{cloud}: mark test as {cloud} specific')
+        cloud_keyword = cloud_to_pytest_keyword[cloud]
+        config.addinivalue_line(
+            'markers', f'{cloud_keyword}: mark test as {cloud} specific')
 
 
 def _get_cloud_to_run(config) -> List[str]:
@@ -78,23 +89,56 @@ def pytest_collection_modifyitems(config, items):
         if 'slow' in item.keywords and not config.getoption('--runslow'):
             item.add_marker(skip_marks['slow'])
         for cloud in all_clouds_in_smoke_tests:
-            if cloud in item.keywords and cloud not in cloud_to_run:
+            cloud_keyword = cloud_to_pytest_keyword[cloud]
+            if (f'no_{cloud_keyword}' in item.keywords or
+                (cloud_keyword in item.keywords and cloud not in cloud_to_run)):
                 item.add_marker(skip_marks[cloud])
 
         if (not 'managed_spot'
                 in item.keywords) and config.getoption('--managed-spot'):
             item.add_marker(skip_marks['managed_spot'])
 
+    # We run Lambda Cloud tests serially because Lambda Cloud rate limits its
+    # launch API to one launch every 10 seconds.
+    serial_mark = pytest.mark.xdist_group(name='serial_lambda_cloud')
+    # Handle generic tests
+    if _generic_cloud(config) == 'lambda':
+        for item in items:
+            if (_is_generic_test(item) and
+                    'no_lambda_cloud' not in item.keywords):
+                item.add_marker(serial_mark)
+                # Adding the serial mark does not update the item.nodeid,
+                # but item.nodeid is important for pytest.xdist_group, e.g.
+                #   https://github.com/pytest-dev/pytest-xdist/blob/master/src/xdist/scheduler/loadgroup.py
+                # This is a hack to update item.nodeid
+                item._nodeid = f'{item.nodeid}@serial_lambda_cloud'
+    # Handle Lambda Cloud specific tests
+    for item in items:
+        if 'lambda_cloud' in item.keywords:
+            item.add_marker(serial_mark)
+            item._nodeid = f'{item.nodeid}@serial_lambda_cloud'  # See comment on item.nodeid above
 
-@pytest.fixture
-def generic_cloud(request) -> str:
-    c = request.config.getoption('--generic-cloud')
-    cloud_to_run = _get_cloud_to_run(request.config)
+
+def _is_generic_test(item) -> bool:
+    for cloud in all_clouds_in_smoke_tests:
+        if cloud_to_pytest_keyword[cloud] in item.keywords:
+            return False
+    return True
+
+
+def _generic_cloud(config) -> str:
+    c = config.getoption('--generic-cloud')
+    cloud_to_run = _get_cloud_to_run(config)
     if c not in cloud_to_run:
         c = cloud_to_run[0]
     return c
 
 
+@pytest.fixture
+def generic_cloud(request) -> str:
+    return _generic_cloud(request.config)
+
+
 def pytest_sessionstart(session):
     from sky.clouds.service_catalog import common
     aws_az_mapping_path = common.get_catalog_path('aws/az_mappings.csv')
diff --git a/tests/test_optimizer_dryruns.py b/tests/test_optimizer_dryruns.py
index 079d28232..16fd6c6e9 100644
--- a/tests/test_optimizer_dryruns.py
+++ b/tests/test_optimizer_dryruns.py
@@ -9,6 +9,15 @@ from sky import clouds
 from sky import exceptions
 
 
+def _test_parse_cpus(spec, expected_cpus):
+    with tempfile.NamedTemporaryFile('w') as f:
+        f.write(spec)
+        f.flush()
+        with sky.Dag():
+            task = sky.Task.from_yaml(f.name)
+            assert list(task.resources)[0].cpus == expected_cpus
+
+
 def _test_parse_accelerators(spec, expected_accelerators):
     with tempfile.NamedTemporaryFile('w') as f:
         f.write(spec)
@@ -90,6 +99,12 @@ def test_resources_gcp(monkeypatch):
     _test_resources_launch(monkeypatch, sky.GCP(), 'n1-standard-16')
 
 
+def test_partial_cpus(monkeypatch):
+    _test_resources_launch(monkeypatch, cpus=4)
+    _test_resources_launch(monkeypatch, cpus='4')
+    _test_resources_launch(monkeypatch, cpus='7+')
+
+
 def test_partial_k80(monkeypatch):
     _test_resources_launch(monkeypatch, accelerators='K80')
 
@@ -148,6 +163,41 @@ def test_clouds_not_enabled(monkeypatch):
                                enabled_clouds=[sky.AWS()])
 
 
+def test_instance_type_mismatches_cpus(monkeypatch):
+    bad_instance_and_cpus = [
+        # Actual: 8
+        ('m6i.2xlarge', 4),
+        # Actual: 2
+        ('c6i.large', 4),
+    ]
+    for instance, cpus in bad_instance_and_cpus:
+        with pytest.raises(ValueError) as e:
+            _test_resources_launch(monkeypatch,
+                                   sky.AWS(),
+                                   instance_type=instance,
+                                   cpus=cpus)
+        assert 'does not have the requested number of vCPUs' in str(e.value)
+
+
+def test_instance_type_matches_cpus(monkeypatch):
+    _test_resources_launch(monkeypatch,
+                           sky.AWS(),
+                           instance_type='c6i.8xlarge',
+                           cpus=32)
+    _test_resources_launch(monkeypatch,
+                           sky.Azure(),
+                           instance_type='Standard_E8s_v5',
+                           cpus='8')
+    _test_resources_launch(monkeypatch,
+                           sky.GCP(),
+                           instance_type='n1-standard-8',
+                           cpus='7+')
+    _test_resources_launch(monkeypatch,
+                           sky.AWS(),
+                           instance_type='g4dn.2xlarge',
+                           cpus=8.0)
+
+
 def test_instance_type_mistmatches_accelerators(monkeypatch):
     bad_instance_and_accs = [
         # Actual: V100
@@ -209,6 +259,13 @@ def test_infer_cloud_from_instance_type(monkeypatch):
                     expected_cloud=sky.Azure())
 
 
+def test_invalid_cpus(monkeypatch):
+    for cloud in [sky.AWS(), sky.Azure(), sky.GCP(), None]:
+        with pytest.raises(ValueError) as e:
+            _test_resources(monkeypatch, cloud, cpus='invalid')
+        assert '"cpus" field should be' in str(e.value)
+
+
 def test_invalid_region(monkeypatch):
     for cloud in [sky.AWS(), sky.Azure(), sky.GCP()]:
         with pytest.raises(ValueError) as e:
@@ -287,6 +344,23 @@ def test_valid_image(monkeypatch):
     )
 
 
+def test_parse_cpus_from_yaml():
+    spec = textwrap.dedent("""\
+        resources:
+            cpus: 1""")
+    _test_parse_cpus(spec, '1')
+
+    spec = textwrap.dedent("""\
+        resources:
+            cpus: 1.5""")
+    _test_parse_cpus(spec, '1.5')
+
+    spec = textwrap.dedent("""\
+        resources:
+            cpus: '3+' """)
+    _test_parse_cpus(spec, '3+')
+
+
 def test_parse_accelerators_from_yaml():
     spec = textwrap.dedent("""\
       resources:
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index c065deb2a..a25275d0a 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -35,6 +35,8 @@ _smoke_test_hash = hashlib.md5(
 # id.
 test_id = str(uuid.uuid4())[-2:]
 
+LAMBDA_TYPE = '--cloud lambda --gpus A100'
+
 storage_setup_commands = [
     'touch ~/tmpfile', 'mkdir -p ~/tmp-workdir',
     'touch ~/tmp-workdir/tmp\ file', 'touch ~/tmp-workdir/foo',
@@ -484,6 +486,7 @@ def test_image_no_conda():
 
 
 # ------------ Test stale job ------------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support stopping instances
 def test_stale_job(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -561,6 +564,7 @@ def test_gcp_stale_job_manual_restart():
 
 
 # ---------- Check Sky's environment variables; workdir. ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_env_check(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -575,6 +579,7 @@ def test_env_check(generic_cloud: str):
 
 
 # ---------- file_mounts ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_file_mounts(generic_cloud: str):
     name = _get_cluster_name()
     test_commands = [
@@ -591,6 +596,24 @@ def test_file_mounts(generic_cloud: str):
     run_one_test(test)
 
 
+# TODO(ewzeng): merge this with 'test_file_mounts' when multi-node is supported.
+@pytest.mark.lambda_cloud
+def test_lambda_file_mounts():
+    name = _get_cluster_name()
+    test_commands = [
+        *storage_setup_commands,
+        f'sky launch -y -c {name} {LAMBDA_TYPE} --num-nodes 1 examples/using_file_mounts.yaml',
+        f'sky logs {name} 1 --status',  # Ensure the job succeeded.
+    ]
+    test = Test(
+        'lambda_using_file_mounts',
+        test_commands,
+        f'sky down -y {name}',
+        timeout=20 * 60,  # 20 mins
+    )
+    run_one_test(test)
+
+
 # ---------- storage ----------
 @pytest.mark.aws
 def test_aws_storage_mounts():
@@ -647,6 +670,7 @@ def test_gcp_storage_mounts():
 
 
 # ---------- CLI logs ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_cli_logs(generic_cloud: str):
     name = _get_cluster_name()
     timestamp = time.time()
@@ -668,7 +692,31 @@ def test_cli_logs(generic_cloud: str):
     run_one_test(test)
 
 
+# TODO(ewzeng): merge this with 'test_cli_logs' when multi-node is supported.
+@pytest.mark.lambda_cloud
+def test_lambda_logs():
+    name = _get_cluster_name()
+    timestamp = time.time()
+    test = Test(
+        'lambda_cli_logs',
+        [
+            f'sky launch -y -c {name} {LAMBDA_TYPE} "echo {timestamp} 1"',
+            f'sky exec {name} "echo {timestamp} 2"',
+            f'sky exec {name} "echo {timestamp} 3"',
+            f'sky exec {name} "echo {timestamp} 4"',
+            f'sky logs {name} 2 --status',
+            f'sky logs {name} 3 4 --sync-down',
+            f'sky logs {name} * --sync-down',
+            f'sky logs {name} 1 | grep "{timestamp} 1"',
+            f'sky logs {name} | grep "{timestamp} 4"',
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
 # ---------- Job Queue. ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have K80 gpus
 def test_job_queue(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -695,6 +743,30 @@ def test_job_queue(generic_cloud: str):
     run_one_test(test)
 
 
+@pytest.mark.lambda_cloud
+def test_lambda_job_queue():
+    name = _get_cluster_name()
+    test = Test(
+        'lambda_job_queue',
+        [
+            f'sky launch -y -c {name} {LAMBDA_TYPE} examples/job_queue/cluster.yaml',
+            f'sky exec {name} -n {name}-1 --gpus A100:0.5 -d examples/job_queue/job.yaml',
+            f'sky exec {name} -n {name}-2 --gpus A100:0.5 -d examples/job_queue/job.yaml',
+            f'sky exec {name} -n {name}-3 --gpus A100:0.5 -d examples/job_queue/job.yaml',
+            f'sky queue {name} | grep {name}-1 | grep RUNNING',
+            f'sky queue {name} | grep {name}-2 | grep RUNNING',
+            f'sky queue {name} | grep {name}-3 | grep PENDING',
+            f'sky cancel -y {name} 2',
+            'sleep 5',
+            f'sky queue {name} | grep {name}-3 | grep RUNNING',
+            f'sky cancel -y {name} 3',
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_job_queue_multinode(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -731,6 +803,7 @@ def test_job_queue_multinode(generic_cloud: str):
     run_one_test(test)
 
 
+@pytest.mark.no_lambda_cloud  # No Lambda Cloud VM has 8 CPUs
 def test_large_job_queue(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -751,6 +824,7 @@ def test_large_job_queue(generic_cloud: str):
 
 
 # ---------- Submitting multiple tasks to the same cluster. ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_multi_echo(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -771,6 +845,7 @@ def test_multi_echo(generic_cloud: str):
 
 
 # ---------- Task: 1 node training. ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 instances
 def test_huggingface(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -786,6 +861,22 @@ def test_huggingface(generic_cloud: str):
     run_one_test(test)
 
 
+@pytest.mark.lambda_cloud
+def test_lambda_huggingface(generic_cloud: str):
+    name = _get_cluster_name()
+    test = Test(
+        'lambda_huggingface_glue_imdb_app',
+        [
+            f'sky launch -y -c {name} {LAMBDA_TYPE} examples/huggingface_glue_imdb_app.yaml',
+            f'sky logs {name} 1 --status',  # Ensure the job succeeded.
+            f'sky exec {name} {LAMBDA_TYPE} examples/huggingface_glue_imdb_app.yaml',
+            f'sky logs {name} 2 --status',  # Ensure the job succeeded.
+        ],
+        f'sky down -y {name}',
+    )
+    run_one_test(test)
+
+
 # ---------- TPU. ----------
 @pytest.mark.gcp
 def test_tpu():
@@ -847,6 +938,7 @@ def test_tpu_vm_pod():
 
 
 # ---------- Simple apps. ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_multi_hostname(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -863,6 +955,7 @@ def test_multi_hostname(generic_cloud: str):
 
 
 # ---------- Task: n=2 nodes with setups. ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_distributed_tf(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -926,6 +1019,7 @@ def test_azure_start_stop():
 
 
 # ---------- Testing Autostopping ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support stopping instances
 def test_autostop(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -981,6 +1075,7 @@ def test_autostop(generic_cloud: str):
 
 
 # ---------- Testing Autodowning ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_autodown(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1015,6 +1110,41 @@ def test_autodown(generic_cloud: str):
     run_one_test(test)
 
 
+@pytest.mark.lambda_cloud
+def test_lambda_autodown():
+    name = _get_cluster_name()
+    test = Test(
+        'lambda_autodown',
+        [
+            f'sky launch -y -d -c {name} {LAMBDA_TYPE} tests/test_yamls/minimal.yaml',
+            f'sky autostop -y {name} --down -i 1',
+            # Ensure autostop is set.
+            f'sky status | grep {name} | grep "1m (down)"',
+            # Ensure the cluster is not terminated early.
+            'sleep 45',
+            f'sky status --refresh | grep {name} | grep UP',
+            # Ensure the cluster is terminated.
+            'sleep 200',
+            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && {{ echo "$s" | grep {name} | grep "Autodowned cluster\|terminated on the cloud"; }} || {{ echo "$s" | grep {name} && exit 1 || exit 0; }}',
+            f'sky launch -y -d -c {name} {LAMBDA_TYPE} --down tests/test_yamls/minimal.yaml',
+            f'sky status | grep {name} | grep UP',  # Ensure the cluster is UP.
+            f'sky exec {name} {LAMBDA_TYPE} tests/test_yamls/minimal.yaml',
+            f'sky status | grep {name} | grep "1m (down)"',
+            'sleep 200',
+            # Ensure the cluster is terminated.
+            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && {{ echo "$s" | grep {name} | grep "Autodowned cluster\|terminated on the cloud"; }} || {{ echo "$s" | grep {name} && exit 1 || exit 0; }}',
+            f'sky launch -y -d -c {name} {LAMBDA_TYPE} --down tests/test_yamls/minimal.yaml',
+            f'sky autostop -y {name} --cancel',
+            'sleep 200',
+            # Ensure the cluster is still UP.
+            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && echo "$s" | grep {name} | grep UP',
+        ],
+        f'sky down -y {name}',
+        timeout=25 * 60,
+    )
+    run_one_test(test)
+
+
 def _get_cancel_task_with_cloud(name, cloud, timeout=15 * 60):
     test = Test(
         f'{cloud}-cancel-task',
@@ -1058,7 +1188,7 @@ def test_cancel_azure():
     run_one_test(test)
 
 
-# ---------- Testing `sky cancel` ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_cancel_pytorch(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1082,6 +1212,7 @@ def test_cancel_pytorch(generic_cloud: str):
 
 
 # ---------- Testing use-spot option ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
 def test_use_spot(generic_cloud: str):
     """Test use-spot and sky exec."""
     name = _get_cluster_name()
@@ -1099,6 +1230,7 @@ def test_use_spot(generic_cloud: str):
 
 
 # ---------- Testing managed spot ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot(generic_cloud: str):
     """Test the spot yaml."""
@@ -1124,6 +1256,7 @@ def test_spot(generic_cloud: str):
     run_one_test(test)
 
 
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_failed_setup(generic_cloud: str):
     """Test managed spot job with failed setup."""
@@ -1207,6 +1340,7 @@ def test_spot_recovery_gcp():
     run_one_test(test)
 
 
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_recovery_default_resources(generic_cloud: str):
     """Test managed spot recovery for default resources."""
@@ -1408,6 +1542,7 @@ def test_spot_cancellation_gcp():
 
 
 # ---------- Testing storage for managed spot ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_storage(generic_cloud: str):
     """Test storage with managed spot"""
@@ -1459,6 +1594,7 @@ def test_spot_tpu():
 
 
 # ---------- Testing env for spot ----------
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
 @pytest.mark.managed_spot
 def test_spot_inline_env(generic_cloud: str):
     """Test spot env"""