Repository: skypilot-org/skypilot
Pull request #1594 (2023-01-15 00:23:19)

Add GCP VPC creation if no VPC available
<!-- Describe the changes in this PR -->

Fixes https://github.com/skypilot-org/skypilot/issues/1591

Problem: A fresh GCP project may not have any VPC created which is required for `sky launch`.
~~We add VPC creation to `sky check` if it's not found.~~

~~However, there's a design choice to make. What's the VPC name we should use, if there are many?~~
Currently, Ray just blindly uses the first VPC net in the order, which is obviously a bad idea. what if it's a user created one without sufficient permissions?
https://github.com/skypilot-org/skypilot/blob/3cceb5cf48943782ddfb7683ccad72a8f42d0f6f/sky/skylet/providers/gcp/config.py#L509-L517

We can use `default`, but the risk is the VPC with such name may not have sufficient permissions for skypilot to work.
Now, I try to create a dedicated VPC net with a unique name `skypilot-vpc` if such VPC is not found when running `ray up`.
~~However, the consequence is that after this PR the GCP users need to re-run `sky check` to create the VPC.~~


<!-- 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] Tested `sky launch` with a fresh GCP project without any VPC
- [x] Tested a project with VPC + sufficient permissions
- [x] Tested a project with VPC + insufficient permissions
- [x] Tested TPU Node
- [x] `pytest tests/test_smoke.py --gcp` passed

infwinston: > Q: is there a reason we put these two commands in gcp.py / `sky check` code path, rather than putting them in config.py? E.g., detect whether the VPC exists, if not, create them. That seems like a better place to handle these stack setup issues.

Good point. I've moved the code that creates a VPC net + firewall rules to `ray up`. Also I switched from `gcloud` to restful API implementation, which is better I think. 
infwinston: OK I just realize a possible issue with this change. This will add another IAM permission requirement `compute.networks.create` for creating a VPC network. which is not needed before as Ray just blindly choose the first VPC net as I mentioned.
https://github.com/skypilot-org/skypilot/blob/3cceb5cf48943782ddfb7683ccad72a8f42d0f6f/sky/skylet/providers/gcp/config.py#L509-L517

I'm not entirely sure if this additional permission would create barriers to new users. but in some cases (e.g., no vpc or vpc without sufficient firewall rules) this permission is essential.
One idea is that we can still merge this and in IAM doc we ask those users without such permission to ask their admins to create a VPC net `skypilot-vpc` with essential firewall rules.
Michaelvll: Great findings! Is it possible to do it with best effort, i.e try to create the network and error out hinting the user either adding the permission or manually do it?
infwinston: Thanks @Michaelvll ! Those are great points. I've implemented a best effort search for VPC available. 
We first check if there's a VPC available with sufficient permissions. If so, we'll directly use it. If not, we'll create one with name `SKYPILOT_VPC_NAME`. so it should be backward compatible now.
It's a bit tricky when doing firewall rules comparisons, but I think current implementation should work. I also confirm that with our `skycamp-gcp` project, it'll directly use the `default` VPC rather than creating a new one.

Please take a look, thanks :)
infwinston: @Michaelvll OK. when I was testing the required permissions, I realized I was pretty much wrong..

The `Compute Admin` role actually contains the permission `compute.networks.create` for creating a new VPC net...
so in fact we don't really add more requirements to our assumption. 
ref: https://cloud.google.com/compute/docs/access/iam#compute.admin

Sorry for the mistake. but nonetheless, this PR is still needed and should work.
infwinston: @Michaelvll thanks again for the reviews. I've tested this PR with 
```
pytest tests/test_smoke.py --gcp
```
and fixed a bug! (when launching a TPU VM, a weird firewall rule will be added to the VPC..)
also thanks @Michaelvll for the smoke tests improvement.

Now the GCP smoke tests are all passed. Please let me know if it's good to go.
infwinston: @Michaelvll @concretevitamin I've added the code to ensure there's no resource leakage. Basically we'll reuse the VPC if the VM is already launched.
(by listing all the instances and match them by cluster name)
https://github.com/skypilot-org/skypilot/blob/b4cfd2be60929be3a7ab1ae6d32fdde4ed15ae7c/sky/skylet/providers/gcp/config.py#L535-L544

I've also tested with below commands and confirmed no resource leaked after this change.
```
git checkout master
sky launch -c tmp --cloud gcp
git checkout fix-gcp-vpc
sky launch -c tmp --cloud gcp
```
Please let me know if there's any additional thoughts.

Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 3a5a2cfea..34129868e 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -35,6 +35,9 @@ pytest --lf
 # Run one of the smoke tests
 pytest tests/test_smoke.py::test_minimal
 
+# Only run managed spot tests
+pytest tests/test_smoke.py --managed-spot
+
 # Only run test for AWS + generic tests
 pytest tests/test_smoke.py --aws
 
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index a89973fac..8bcd44a0c 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -371,6 +371,12 @@ class SSHConfigHelper(object):
             proxy = f'ProxyCommand {proxy_command}'
         else:
             proxy = ''
+        # StrictHostKeyChecking=no skips the host key check for the first
+        # time. UserKnownHostsFile=/dev/null and GlobalKnownHostsFile/dev/null
+        # prevent the host key from being added to the known_hosts file and
+        # always return an empty file for known hosts, making the ssh think
+        # this is a first-time connection, and thus skipping the host key
+        # check.
         codegen = textwrap.dedent(f"""\
             {autogen_comment}
             Host {host_name}
@@ -380,6 +386,8 @@ class SSHConfigHelper(object):
               IdentitiesOnly yes
               ForwardAgent yes
               StrictHostKeyChecking no
+              UserKnownHostsFile=/dev/null
+              GlobalKnownHostsFile=/dev/null
               Port 22
               {proxy}
             """.rstrip())
@@ -903,6 +911,11 @@ def write_cluster_config(
             tpu_name = cluster_name
 
         user_file_dir = os.path.expanduser(f'{SKY_USER_FILE_PATH}/')
+
+        from sky.skylet.providers.gcp import config as gcp_config  # pylint: disable=import-outside-toplevel
+        config = common_utils.read_yaml(os.path.expanduser(config_dict['ray']))
+        vpc_name = gcp_config.get_usable_vpc(config)
+
         scripts = tuple(
             fill_template(
                 template_name,
@@ -910,6 +923,7 @@ def write_cluster_config(
                     resources_vars, **{
                         'tpu_name': tpu_name,
                         'gcp_project_id': gcp_project_id,
+                        'vpc_name': vpc_name,
                     }),
                 # Use new names for TPU scripts so that different runs can use
                 # different TPUs.  Put in SKY_USER_FILE_PATH to be consistent
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 67bbdf3f0..6ea6d4e78 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -1100,13 +1100,15 @@ class RetryingVmProvisioner(object):
             # but the optimizer that does the filtering will not be involved
             # until the next region.
             remaining_unblocked_zones = copy.deepcopy(zones)
-            for zone in zones:
-                for blocked_resources in self._blocked_resources:
-                    if to_provision.copy(region=region.name,
-                                         zone=zone.name).should_be_blocked_by(
-                                             blocked_resources):
-                        remaining_unblocked_zones.remove(zone)
-                        break
+            if zones is not None:
+                for zone in zones:
+                    for blocked_resources in self._blocked_resources:
+                        if to_provision.copy(
+                                region=region.name,
+                                zone=zone.name).should_be_blocked_by(
+                                    blocked_resources):
+                            remaining_unblocked_zones.remove(zone)
+                            break
             if zones and not remaining_unblocked_zones:
                 # Skip the region if all zones are blocked.
                 continue
@@ -2371,13 +2373,9 @@ class CloudVmRayBackend(backends.Backend):
                             f'{colorama.Style.RESET_ALL}')
                     return err_msg
 
-                try:
-                    subprocess_utils.handle_returncode(returncode=returncode,
-                                                       command=setup_cmd,
-                                                       error_msg=error_message)
-                except exceptions.CommandError as e:
-                    with ux_utils.print_exception_no_traceback():
-                        raise exceptions.ClusterSetUpError(str(e)) from e
+                subprocess_utils.handle_returncode(returncode=returncode,
+                                                   command=setup_cmd,
+                                                   error_msg=error_message)
 
             num_nodes = len(ip_list)
             plural = 's' if num_nodes > 1 else ''
diff --git a/sky/cli.py b/sky/cli.py
index e26411c07..8d504a0ce 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -2667,8 +2667,20 @@ def check():
               default=None,
               type=str,
               help='Cloud provider to query.')
+@click.option(
+    '--region',
+    required=False,
+    type=str,
+    help=
+    ('The region to use. If not specified, shows accelerators from all regions.'
+    ),
+)
 @usage_lib.entrypoint
-def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pylint: disable=redefined-builtin
+def show_gpus(
+        gpu_name: Optional[str],
+        all: bool,  # pylint: disable=redefined-builtin
+        cloud: Optional[str],
+        region: Optional[str]):
     """Show supported GPU/TPU/accelerators.
 
     The names and counts shown can be set in the ``accelerators`` field in task
@@ -2682,9 +2694,14 @@ def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pyli
     To show all accelerators, including less common ones and their detailed
     information, use ``sky show-gpus --all``.
 
-    NOTE: The price displayed for each instance type is the lowest across all
-    regions for both on-demand and spot instances.
+    NOTE: If region is not specified, the price displayed for each instance type
+    is the lowest across all regions for both on-demand and spot instances.
     """
+    # validation for the --region flag
+    if region is not None and cloud is None:
+        raise click.UsageError(
+            'The --region flag is only valid when the --cloud flag is set.')
+    service_catalog.validate_region_zone(region, None, clouds=cloud)
     show_all = all
     if show_all and gpu_name is not None:
         raise click.UsageError('--all is only allowed without a GPU name.')
@@ -2701,8 +2718,11 @@ def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pyli
             ['OTHER_GPU', 'AVAILABLE_QUANTITIES'])
 
         if gpu_name is None:
-            result = service_catalog.list_accelerator_counts(gpus_only=True,
-                                                             clouds=cloud)
+            result = service_catalog.list_accelerator_counts(
+                gpus_only=True,
+                clouds=cloud,
+                region_filter=region,
+            )
             # NVIDIA GPUs
             for gpu in service_catalog.get_common_gpus():
                 if gpu in result:
@@ -2730,6 +2750,7 @@ def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pyli
         # Show detailed accelerator information
         result = service_catalog.list_accelerators(gpus_only=True,
                                                    name_filter=gpu_name,
+                                                   region_filter=region,
                                                    clouds=cloud)
         if len(result) == 0:
             yield f'Resources \'{gpu_name}\' not found. '
@@ -2742,7 +2763,7 @@ def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pyli
         yield 'the host VM\'s cost is not included.\n\n'
         import pandas as pd  # pylint: disable=import-outside-toplevel
         for i, (gpu, items) in enumerate(result.items()):
-            accelerator_table = log_utils.create_table([
+            accelerator_table_headers = [
                 'GPU',
                 'QTY',
                 'CLOUD',
@@ -2751,7 +2772,11 @@ def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pyli
                 'HOST_MEMORY',
                 'HOURLY_PRICE',
                 'HOURLY_SPOT_PRICE',
-            ])
+            ]
+            if not show_all:
+                accelerator_table_headers.append('REGION')
+            accelerator_table = log_utils.create_table(
+                accelerator_table_headers)
             for item in items:
                 instance_type_str = item.instance_type if not pd.isna(
                     item.instance_type) else '(attachable)'
@@ -2769,11 +2794,20 @@ def show_gpus(gpu_name: Optional[str], all: bool, cloud: Optional[str]):  # pyli
                     item.price) else '-'
                 spot_price_str = f'$ {item.spot_price:.3f}' if not pd.isna(
                     item.spot_price) else '-'
-                accelerator_table.add_row([
-                    item.accelerator_name, item.accelerator_count, item.cloud,
-                    instance_type_str, cpu_str, mem_str, price_str,
-                    spot_price_str
-                ])
+                region_str = item.region if not pd.isna(item.region) else '-'
+                accelerator_table_vals = [
+                    item.accelerator_name,
+                    item.accelerator_count,
+                    item.cloud,
+                    instance_type_str,
+                    cpu_str,
+                    mem_str,
+                    price_str,
+                    spot_price_str,
+                ]
+                if not show_all:
+                    accelerator_table_vals.append(region_str)
+                accelerator_table.add_row(accelerator_table_vals)
 
             if i != 0:
                 yield '\n\n'
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index 0fdbd5dfd..418415040 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -275,7 +275,8 @@ class AWS(clouds.Cloud):
 
     @classmethod
     def get_default_instance_type(cls) -> str:
-        # 8 vCpus, 32 GB RAM. 3rd generation Intel Xeon. General Purpose.
+        # General-purpose instance with 8 vCPUs and 32 GB RAM.
+        # Intel Ice Lake 8375C
         return 'm6i.2xlarge'
 
     # TODO: factor the following three methods, as they are the same logic
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index 0b32c2c90..ab1da9331 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -95,8 +95,9 @@ class Azure(clouds.Cloud):
 
     @classmethod
     def get_default_instance_type(cls) -> str:
-        # 8 vCpus, 32 GB RAM.  Prev-gen (as of 2021) general purpose.
-        return 'Standard_D8_v4'
+        # General-purpose instance with 8 vCPUs and 32 GB RAM.
+        # Intel Ice Lake 8370C
+        return 'Standard_D8_v5'
 
     def _get_image_config(self, gen_version, instance_type):
         # az vm image list \
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index 6267d4d08..24dc3155a 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -280,8 +280,9 @@ class GCP(clouds.Cloud):
 
     @classmethod
     def get_default_instance_type(cls) -> str:
-        # 8 vCpus, 52 GB RAM.  First-gen general purpose.
-        return 'n1-highmem-8'
+        # General-purpose instance with 8 vCPUs and 32 GB RAM.
+        # Intel Ice Lake 8373C or Cascade Lake 6268CL
+        return 'n2-standard-8'
 
     @classmethod
     def _get_default_region(cls) -> clouds.Region:
diff --git a/sky/clouds/service_catalog/__init__.py b/sky/clouds/service_catalog/__init__.py
index 2acd61d52..a722701cc 100644
--- a/sky/clouds/service_catalog/__init__.py
+++ b/sky/clouds/service_catalog/__init__.py
@@ -49,6 +49,7 @@ def _map_clouds_catalog(clouds: CloudFilter, method_name: str, *args, **kwargs):
 def list_accelerators(
     gpus_only: bool = True,
     name_filter: Optional[str] = None,
+    region_filter: Optional[str] = None,
     clouds: CloudFilter = None,
     case_sensitive: bool = True,
 ) -> 'Dict[str, List[common.InstanceTypeInfo]]':
@@ -58,7 +59,7 @@ def list_accelerators(
     of instance type offerings. See usage in cli.py.
     """
     results = _map_clouds_catalog(clouds, 'list_accelerators', gpus_only,
-                                  name_filter, case_sensitive)
+                                  name_filter, region_filter, case_sensitive)
     if not isinstance(results, list):
         results = [results]
     ret: Dict[str,
@@ -72,6 +73,7 @@ def list_accelerators(
 def list_accelerator_counts(
     gpus_only: bool = True,
     name_filter: Optional[str] = None,
+    region_filter: Optional[str] = None,
     clouds: CloudFilter = None,
 ) -> Dict[str, List[int]]:
     """List all accelerators offered by Sky and available counts.
@@ -80,7 +82,7 @@ def list_accelerator_counts(
     of available counts. See usage in cli.py.
     """
     results = _map_clouds_catalog(clouds, 'list_accelerators', gpus_only,
-                                  name_filter)
+                                  name_filter, region_filter, False)
     if not isinstance(results, list):
         results = [results]
     accelerator_counts: Dict[str, Set[int]] = collections.defaultdict(set)
diff --git a/sky/clouds/service_catalog/aws_catalog.py b/sky/clouds/service_catalog/aws_catalog.py
index 9fccd495b..150e45b38 100644
--- a/sky/clouds/service_catalog/aws_catalog.py
+++ b/sky/clouds/service_catalog/aws_catalog.py
@@ -71,7 +71,7 @@ def instance_type_exists(instance_type: str) -> bool:
 def validate_region_zone(
         region: Optional[str],
         zone: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
-    return common.validate_region_zone_impl(_df, region, zone)
+    return common.validate_region_zone_impl('aws', _df, region, zone)
 
 
 def accelerator_in_region_or_zone(acc_name: str,
@@ -134,13 +134,15 @@ def get_region_zones_for_instance_type(instance_type: str,
     return us_region_list + other_region_list
 
 
-def list_accelerators(gpus_only: bool,
-                      name_filter: Optional[str],
-                      case_sensitive: bool = True
-                     ) -> Dict[str, List[common.InstanceTypeInfo]]:
+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 AWS offering accelerators."""
     return common.list_accelerators_impl('AWS', _df, gpus_only, name_filter,
-                                         case_sensitive)
+                                         region_filter, case_sensitive)
 
 
 def get_image_id_from_tag(tag: str, region: Optional[str]) -> Optional[str]:
diff --git a/sky/clouds/service_catalog/azure_catalog.py b/sky/clouds/service_catalog/azure_catalog.py
index fc74fc795..c07f8e8ac 100644
--- a/sky/clouds/service_catalog/azure_catalog.py
+++ b/sky/clouds/service_catalog/azure_catalog.py
@@ -22,7 +22,7 @@ def validate_region_zone(
     if zone is not None:
         with ux_utils.print_exception_no_traceback():
             raise ValueError('Azure does not support zones.')
-    return common.validate_region_zone_impl(_df, region, zone)
+    return common.validate_region_zone_impl('azure', _df, region, zone)
 
 
 def accelerator_in_region_or_zone(acc_name: str,
@@ -89,10 +89,12 @@ def get_gen_version_from_instance_type(instance_type: str) -> Optional[int]:
     return _df[_df['InstanceType'] == instance_type]['Generation'].iloc[0]
 
 
-def list_accelerators(gpus_only: bool,
-                      name_filter: Optional[str],
-                      case_sensitive: bool = True
-                     ) -> Dict[str, List[common.InstanceTypeInfo]]:
+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 Azure offering GPUs."""
     return common.list_accelerators_impl('Azure', _df, gpus_only, name_filter,
-                                         case_sensitive)
+                                         region_filter, case_sensitive)
diff --git a/sky/clouds/service_catalog/common.py b/sky/clouds/service_catalog/common.py
index d64667311..84b2c67be 100644
--- a/sky/clouds/service_catalog/common.py
+++ b/sky/clouds/service_catalog/common.py
@@ -34,6 +34,7 @@ class InstanceTypeInfo(NamedTuple):
     - memory: Instance memory in GiB.
     - price: Regular instance price per hour (cheapest across all regions).
     - spot_price: Spot instance price per hour (cheapest across all regions).
+    - region: Region where this instance type belongs to.
     """
     cloud: str
     instance_type: Optional[str]
@@ -43,6 +44,7 @@ class InstanceTypeInfo(NamedTuple):
     memory: Optional[float]
     price: float
     spot_price: float
+    region: str
 
 
 def get_catalog_path(filename: str) -> str:
@@ -161,7 +163,7 @@ def instance_type_exists_impl(df: pd.DataFrame, instance_type: str) -> bool:
 
 
 def validate_region_zone_impl(
-        df: pd.DataFrame, region: Optional[str],
+        cloud_name: str, df: pd.DataFrame, region: Optional[str],
         zone: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
     """Validates whether region and zone exist in the catalog."""
 
@@ -174,6 +176,11 @@ def validate_region_zone_impl(
             candidate_strs = f'\nDid you mean one of these: {candidate_strs!r}?'
         return candidate_strs
 
+    def _get_all_supported_regions_str() -> str:
+        all_regions: List[str] = sorted(df['Region'].unique().tolist())
+        return \
+        f'\nList of supported {cloud_name} regions: {", ".join(all_regions)!r}'
+
     validated_region, validated_zone = region, zone
 
     filter_df = df
@@ -182,7 +189,12 @@ def validate_region_zone_impl(
         if len(filter_df) == 0:
             with ux_utils.print_exception_no_traceback():
                 error_msg = (f'Invalid region {region!r}')
-                error_msg += _get_candidate_str(region, df['Region'].unique())
+                candidate_strs = _get_candidate_str(region,
+                                                    df['Region'].unique())
+                if not candidate_strs:
+                    error_msg += _get_all_supported_regions_str()
+                    raise ValueError(error_msg)
+                error_msg += candidate_strs
                 raise ValueError(error_msg)
 
     if zone is not None:
@@ -321,6 +333,7 @@ def list_accelerators_impl(
     df: pd.DataFrame,
     gpus_only: bool,
     name_filter: Optional[str],
+    region_filter: Optional[str],
     case_sensitive: bool = True,
 ) -> Dict[str, List[InstanceTypeInfo]]:
     """Lists accelerators offered in a cloud service catalog.
@@ -341,11 +354,16 @@ def list_accelerators_impl(
         'MemoryGiB',
         'Price',
         'SpotPrice',
+        'Region',
     ]].dropna(subset=['AcceleratorName']).drop_duplicates()
     if name_filter is not None:
         df = df[df['AcceleratorName'].str.contains(name_filter,
                                                    case=case_sensitive,
                                                    regex=True)]
+    if region_filter is not None:
+        df = df[df['Region'].str.contains(region_filter,
+                                          case=case_sensitive,
+                                          regex=True)]
     df['AcceleratorCount'] = df['AcceleratorCount'].astype(int)
     grouped = df.groupby('AcceleratorName')
 
@@ -366,6 +384,7 @@ def list_accelerators_impl(
                 row['MemoryGiB'],
                 row['Price'],
                 row['SpotPrice'],
+                row['Region'],
             ),
             axis='columns',
         ).tolist()
diff --git a/sky/clouds/service_catalog/gcp_catalog.py b/sky/clouds/service_catalog/gcp_catalog.py
index 39e3eecb9..b2cb8f4b8 100644
--- a/sky/clouds/service_catalog/gcp_catalog.py
+++ b/sky/clouds/service_catalog/gcp_catalog.py
@@ -210,7 +210,7 @@ def get_instance_type_for_accelerator(
 def validate_region_zone(
         region: Optional[str],
         zone: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
-    return common.validate_region_zone_impl(_df, region, zone)
+    return common.validate_region_zone_impl('gcp', _df, region, zone)
 
 
 def accelerator_in_region_or_zone(acc_name: str,
@@ -273,11 +273,12 @@ def get_accelerator_hourly_cost(accelerator: str,
 def list_accelerators(
     gpus_only: bool,
     name_filter: Optional[str] = None,
+    region_filter: Optional[str] = None,
     case_sensitive: bool = True,
 ) -> Dict[str, List[common.InstanceTypeInfo]]:
     """Returns all instance types in GCP offering GPUs."""
     results = common.list_accelerators_impl('GCP', _df, gpus_only, name_filter,
-                                            case_sensitive)
+                                            region_filter, case_sensitive)
 
     a100_infos = results.get('A100', []) + results.get('A100-80GB', [])
     if not a100_infos:
diff --git a/sky/global_user_state.py b/sky/global_user_state.py
index e91dfce52..5c763e856 100644
--- a/sky/global_user_state.py
+++ b/sky/global_user_state.py
@@ -185,7 +185,6 @@ def add_or_update_cluster(cluster_name: str,
 
     # first time a cluster is being launched
     if not usage_intervals:
-        assert requested_resources is not None
         usage_intervals = []
 
     # if this is the cluster init or we are starting after a stop
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index 629094406..360abab05 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -79,6 +79,8 @@ class JobStatus(enum.Enum):
     # The `job_id` has been generated, but the generated ray program has
     # not started yet. skylet can transit the state from INIT to FAILED
     # directly, if the ray program fails to start.
+    # In the 'jobs' table, the `submitted_at` column will be set to the current
+    # time, when the job is firstly created (in the INIT state).
     INIT = 'INIT'
     # Running the user's setup script (only in effect if --detach-setup is
     # set). Our update_job_status() can temporarily (for a short period) set
@@ -90,6 +92,8 @@ class JobStatus(enum.Enum):
     # by the placement constraints.)
     PENDING = 'PENDING'
     # The job is running.
+    # In the 'jobs' table, the `start_at` column will be set to the current
+    # time, when the job is firstly transitioned to RUNNING.
     RUNNING = 'RUNNING'
     # 3 terminal states below: once reached, they do not transition.
     # The job finished successfully.
@@ -290,12 +294,26 @@ def get_latest_job_id() -> Optional[int]:
         return job_id
 
 
-def get_job_time_payload(job_id: int, is_end: bool) -> Optional[int]:
-    field = 'end_at' if is_end else 'start_at'
+def get_job_submitted_or_ended_timestamp_payload(job_id: int,
+                                                 get_ended_time: bool) -> str:
+    """Get the job submitted/ended timestamp.
+
+    This function should only be called by the spot controller,
+    which is ok to use `submitted_at` instead of `start_at`,
+    because the spot job duration need to include both setup
+    and running time and the job will not stay in PENDING
+    state.
+
+    The normal job duration will use `start_at` instead of
+    `submitted_at` (in `format_job_queue()`), because the job
+    may stay in PENDING if the cluster is busy.
+    """
+    field = 'end_at' if get_ended_time else 'submitted_at'
     rows = _CURSOR.execute(f'SELECT {field} FROM jobs WHERE job_id=(?)',
                            (job_id,))
     for (timestamp,) in rows:
         return common_utils.encode_payload(timestamp)
+    return common_utils.encode_payload(None)
 
 
 def _get_records_from_rows(rows) -> List[Dict[str, Any]]:
@@ -362,17 +380,14 @@ def _get_jobs_by_ids(job_ids: List[int]) -> List[Dict[str, Any]]:
 def update_job_status(job_owner: str,
                       job_ids: List[int],
                       silent: bool = False) -> List[JobStatus]:
-    """Updates and returns the job statuses matching our `JobStatus` semantics
+    """Updates and returns the job statuses matching our `JobStatus` semantics.
 
-    "True" statuses: this function queries `ray job status` and processes
-    those results to match our semantics.
+    This function queries `ray job status` and processes those results to match
+    our semantics.
 
-    This function queries `ray job status` and processes those results to
-    match our semantics.
-
-    Though we update job status actively in ray program and job cancelling,
-    we still need this to handle staleness problem, caused by instance
-    restarting and other corner cases (if any).
+    Though we update job status actively in the generated ray program and
+    during job cancelling, we still need this to handle the staleness problem,
+    caused by instance restarting and other corner cases (if any).
 
     This function should only be run on the remote instance with ray==2.0.1.
     """
@@ -413,7 +428,8 @@ def update_job_status(job_owner: str,
             if status is None:
                 original_status = get_status_no_lock(job_id)
                 status = original_status
-                if not original_status.is_terminal():
+                if (original_status is not None and
+                        not original_status.is_terminal()):
                     # The job may be stale, when the instance is restarted
                     # (the ray redis is volatile). We need to reset the
                     # status of the task to FAILED if its original status
@@ -665,14 +681,12 @@ class JobLibCodeGen:
                   job_id: Optional[int],
                   spot_job_id: Optional[int],
                   follow: bool = True) -> str:
+        # pylint: disable=line-too-long
         code = [
-            f'job_id = {job_id} if {job_id} is not None '
-            'else job_lib.get_latest_job_id()',
+            f'job_id = {job_id} if {job_id} is not None else job_lib.get_latest_job_id()',
             'run_timestamp = job_lib.get_run_timestamp(job_id)',
-            (f'log_dir = os.path.join({constants.SKY_LOGS_DIRECTORY!r}, '
-             'run_timestamp)'),
-            (f'log_lib.tail_logs({job_owner!r},'
-             f'job_id, log_dir, {spot_job_id!r}, follow={follow})'),
+            f'log_dir = None if run_timestamp is None else os.path.join({constants.SKY_LOGS_DIRECTORY!r}, run_timestamp)',
+            f'log_lib.tail_logs({job_owner!r}, job_id, log_dir, {spot_job_id!r}, follow={follow})',
         ]
         return cls._build(code)
 
@@ -688,13 +702,16 @@ class JobLibCodeGen:
         return cls._build(code)
 
     @classmethod
-    def get_job_time_payload(cls,
-                             job_id: Optional[int] = None,
-                             is_end: bool = False) -> str:
+    def get_job_submitted_or_ended_timestamp_payload(
+            cls,
+            job_id: Optional[int] = None,
+            get_ended_time: bool = False) -> str:
         code = [
             f'job_id = {job_id} if {job_id} is not None '
             'else job_lib.get_latest_job_id()',
-            f'job_time = job_lib.get_job_time_payload(job_id, {is_end})',
+            'job_time = '
+            'job_lib.get_job_submitted_or_ended_timestamp_payload('
+            f'job_id, {get_ended_time})',
             'print(job_time, flush=True)',
         ]
         return cls._build(code)
diff --git a/sky/skylet/log_lib.py b/sky/skylet/log_lib.py
index 6ab16dcf0..afc822093 100644
--- a/sky/skylet/log_lib.py
+++ b/sky/skylet/log_lib.py
@@ -383,14 +383,14 @@ def tail_logs(job_owner: str,
     job_str = f'job {job_id}'
     if spot_job_id is not None:
         job_str = f'spot job {spot_job_id}'
-    logger.debug(f'Tailing logs for job, real job_id {job_id}, spot_job_id '
-                 f'{spot_job_id}.')
-    logger.info(f'{colorama.Fore.YELLOW}Start streaming logs for {job_str}.'
-                f'{colorama.Style.RESET_ALL}')
     if log_dir is None:
         print(f'{job_str.capitalize()} not found (see `sky queue`).',
               file=sys.stderr)
         return
+    logger.debug(f'Tailing logs for job, real job_id {job_id}, spot_job_id '
+                 f'{spot_job_id}.')
+    logger.info(f'{colorama.Fore.YELLOW}Start streaming logs for {job_str}.'
+                f'{colorama.Style.RESET_ALL}')
     log_path = os.path.join(log_dir, 'run.log')
     log_path = os.path.expanduser(log_path)
 
diff --git a/sky/skylet/providers/gcp/config.py b/sky/skylet/providers/gcp/config.py
index 42ab7caf5..2bd81ad61 100644
--- a/sky/skylet/providers/gcp/config.py
+++ b/sky/skylet/providers/gcp/config.py
@@ -12,7 +12,8 @@ from google.oauth2 import service_account
 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
+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 ray.autoscaler._private.util import check_legacy_fields
 
 logger = logging.getLogger(__name__)
@@ -46,7 +47,6 @@ HAS_TPU_PROVIDER_FIELD = "_has_tpus"
 # NOTE: iam.serviceAccountUser allows the Head Node to create worker nodes
 # with ServiceAccounts.
 
-
 def get_node_type(node: dict) -> GCPNodeType:
     """Returns node type based on the keys in ``node``.
 
@@ -488,6 +488,96 @@ def _configure_key_pair(config, compute):
     return config
 
 
+def _check_firewall_rules(vpc_name, config, compute):
+    """Check if the firewall rules in the VPC are sufficient."""
+    required_rules = FIREWALL_RULES_TEMPLATE.copy()
+
+    operation = (
+        compute.networks().
+        getEffectiveFirewalls(
+            project=config["provider"]["project_id"],
+            network=vpc_name
+        )
+    )
+    response = operation.execute()
+    if len(response) == 0:
+        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))
+
+    for rule in required_rules:
+        if rule not in effective_rules:
+            return False
+    return True
+
+
+def get_usable_vpc(config):
+    """Return a usable VPC.
+
+    If not found, create a new one with sufficient firewall rules.
+    """
+    _, _, compute, _ = construct_clients_from_provider_config(config["provider"])
+
+    # For backward compatibility, reuse the VPC if the VM is launched.
+    resource = GCPCompute(
+        compute,
+        config["provider"]["project_id"],
+        config["provider"]["availability_zone"],
+        config["cluster_name"],
+    )
+    node = resource._list_instances(label_filters=None, status_filter=None)
+    if len(node) > 0:
+        netInterfaces = node[0].get("networkInterfaces", [])
+        if len(netInterfaces) > 0:
+            vpc_name = netInterfaces[0]["network"].split("/")[-1]
+            return vpc_name
+
+    vpcnets_all = _list_vpcnets(config, compute)
+
+    usable_vpc_name = None
+    for vpc in vpcnets_all:
+        if _check_firewall_rules(vpc["name"], config, compute):
+            usable_vpc_name = vpc["name"]
+            break
+
+    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 firewall rules
+        for rule in FIREWALL_RULES_TEMPLATE:
+            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)
+            _create_firewall_rule(config, compute, body)
+
+        usable_vpc_name = SKYPILOT_VPC_NAME
+        logger.info(f"A VPC network {SKYPILOT_VPC_NAME} created.")
+
+    return usable_vpc_name
+
+
 def _configure_subnet(config, compute):
     """Pick a reasonable subnet if not specified by the config."""
     config = copy.deepcopy(config)
@@ -506,14 +596,9 @@ def _configure_subnet(config, compute):
     ):
         return config
 
-    subnets = _list_subnets(config, compute)
-
-    if not subnets:
-        raise NotImplementedError("Should be able to create subnet.")
-
-    # TODO: make sure that we have usable subnet. Maybe call
-    # compute.subnetworks().listUsable? For some reason it didn't
-    # work out-of-the-box
+    # SkyPilot: make sure there's a usable VPC
+    usable_vpc_name = get_usable_vpc(config)
+    subnets = _list_subnets(config, compute, filter=f"(name=\"{usable_vpc_name}\")")
     default_subnet = subnets[0]
 
     default_interfaces = [
@@ -542,17 +627,54 @@ def _configure_subnet(config, compute):
     return config
 
 
-def _list_subnets(config, compute):
+def _create_firewall_rule(config, compute, body):
+    operation = (
+        compute.firewalls()
+        .insert(project=config["provider"]["project_id"], body=body)
+        .execute()
+    )
+    response = wait_for_compute_global_operation(
+        config["provider"]["project_id"], operation, compute
+    )
+    return response
+
+
+def _create_vpcnet(config, compute, body):
+    operation = (
+        compute.networks()
+        .insert(project=config["provider"]["project_id"], body=body)
+        .execute()
+    )
+    response = wait_for_compute_global_operation(
+        config["provider"]["project_id"], operation, compute
+    )
+    return response
+
+
+def _list_vpcnets(config, compute):
+    response = (
+        compute.networks()
+        .list(
+            project=config["provider"]["project_id"],
+        )
+        .execute()
+    )
+
+    return response["items"] if "items" in response else []
+
+
+def _list_subnets(config, compute, filter=None):
     response = (
         compute.subnetworks()
         .list(
             project=config["provider"]["project_id"],
             region=config["provider"]["region"],
+            filter=filter,
         )
         .execute()
     )
 
-    return response["items"]
+    return response["items"] if "items" in response else []
 
 
 def _get_subnet(config, subnet_id, compute):
diff --git a/sky/skylet/providers/gcp/constants.py b/sky/skylet/providers/gcp/constants.py
new file mode 100644
index 000000000..00efecde8
--- /dev/null
+++ b/sky/skylet/providers/gcp/constants.py
@@ -0,0 +1,52 @@
+SKYPILOT_VPC_NAME = "skypilot-vpc"
+
+# Below parameters are from the default VPC on GCP.
+# https://cloud.google.com/vpc/docs/firewalls#more_rules_default_vpc
+VPC_TEMPLATE = {
+    "name": "{VPC_NAME}",
+    "selfLink": "projects/{PROJ_ID}/global/networks/{VPC_NAME}",
+    "autoCreateSubnetworks": True,
+    "mtu": 1460,
+    "routingConfig": {"routingMode": "GLOBAL"},
+}
+FIREWALL_RULES_TEMPLATE = [
+    {
+        "name": "{VPC_NAME}-allow-custom",
+        "description": "Allows connection from any source to any instance on the network using custom protocols.",
+        "network": "projects/{PROJ_ID}/global/networks/{VPC_NAME}",
+        "selfLink": "projects/{PROJ_ID}/global/firewalls/{VPC_NAME}-allow-custom",
+        "direction": "INGRESS",
+        "priority": 65534,
+        "allowed": [
+            {'IPProtocol': 'tcp', 'ports': ['0-65535']},
+            {'IPProtocol': 'udp', 'ports': ['0-65535']},
+            {'IPProtocol': 'icmp'}
+        ],
+        "sourceRanges": ["10.128.0.0/9"],
+    },
+    {
+        "name": "{VPC_NAME}-allow-ssh",
+        "description": "Allows TCP connections from any source to any instance on the network using port 22.",
+        "network": "projects/{PROJ_ID}/global/networks/{VPC_NAME}",
+        "selfLink": "projects/{PROJ_ID}/global/firewalls/{VPC_NAME}-allow-ssh",
+        "direction": "INGRESS",
+        "priority": 65534,
+        "allowed": [{
+            "IPProtocol": "tcp",
+            "ports": ["22"],
+        }],
+        "sourceRanges": ["0.0.0.0/0"],
+    },
+    {
+        "name": "{VPC_NAME}-allow-icmp",
+        "description": "Allows ICMP connections from any source to any instance on the network.",
+        "network": "projects/{PROJ_ID}/global/networks/{VPC_NAME}",
+        "selfLink": "projects/{PROJ_ID}/global/firewalls/{VPC_NAME}-allow-icmp",
+        "direction": "INGRESS",
+        "priority": 65534,
+        "allowed": [{
+            "IPProtocol": "icmp",
+        }],
+        "sourceRanges": ["0.0.0.0/0"],
+    },
+]
diff --git a/sky/skylet/providers/gcp/node.py b/sky/skylet/providers/gcp/node.py
index 5dec7451c..da112a7e3 100644
--- a/sky/skylet/providers/gcp/node.py
+++ b/sky/skylet/providers/gcp/node.py
@@ -378,7 +378,8 @@ class GCPCompute(GCPResource):
         return self._list_instances(label_filters, non_terminated_status)
 
     def _list_instances(self, label_filters: Optional[dict],
-                        status_filter: List[str]) -> List[GCPComputeNode]:
+                        status_filter: Optional[List[str]]
+                        ) -> List[GCPComputeNode]:
         label_filters = label_filters or {}
 
         if label_filters:
@@ -395,16 +396,19 @@ class GCPCompute(GCPResource):
         else:
             label_filter_expr = ""
 
-        instance_state_filter_expr = (
-            "("
-            + " OR ".join(
-                [
-                    "(status = {status})".format(status=status)
-                    for status in status_filter
-                ]
+        if status_filter:
+            instance_state_filter_expr = (
+                "("
+                + " OR ".join(
+                    [
+                        "(status = {status})".format(status=status)
+                        for status in status_filter
+                    ]
+                )
+                + ")"
             )
-            + ")"
-        )
+        else:
+            instance_state_filter_expr = ""
 
         cluster_name_filter_expr = "(labels.{key} = {value})".format(
             key=TAG_RAY_CLUSTER_NAME, value=self.cluster_name
diff --git a/sky/spot/controller.py b/sky/spot/controller.py
index 1b766f3f3..df8d9511c 100644
--- a/sky/spot/controller.py
+++ b/sky/spot/controller.py
@@ -64,9 +64,9 @@ class SpotController:
         logger.info(f'Started monitoring spot task {self._task_name} '
                     f'(id: {self._job_id})')
         spot_state.set_starting(self._job_id)
-        start_at = self._strategy_executor.launch()
+        job_submitted_at = self._strategy_executor.launch()
 
-        spot_state.set_started(self._job_id, start_time=start_at)
+        spot_state.set_started(self._job_id, start_time=job_submitted_at)
         while True:
             time.sleep(spot_utils.JOB_STATUS_CHECK_GAP_SECONDS)
 
@@ -120,7 +120,9 @@ class SpotController:
                 if job_status is not None and not job_status.is_terminal():
                     # The multi-node job is still running, continue monitoring.
                     continue
-                elif job_status == job_lib.JobStatus.FAILED:
+                elif job_status in [
+                        job_lib.JobStatus.FAILED, job_lib.JobStatus.FAILED_SETUP
+                ]:
                     # The user code has probably crashed, fail immediately.
                     end_time = spot_utils.get_job_timestamp(self._backend,
                                                             self._cluster_name,
@@ -132,10 +134,12 @@ class SpotController:
                                             None,
                                             spot_job_id=self._job_id)
                     logger.info(f'\n== End of logs (ID: {self._job_id}) ==')
-                    spot_state.set_failed(
-                        self._job_id,
-                        failure_type=spot_state.SpotStatus.FAILED,
-                        end_time=end_time)
+                    status_to_set = spot_state.SpotStatus.FAILED
+                    if job_status == job_lib.JobStatus.FAILED_SETUP:
+                        status_to_set = spot_state.SpotStatus.FAILED_SETUP
+                    spot_state.set_failed(self._job_id,
+                                          failure_type=status_to_set,
+                                          end_time=end_time)
                     break
                 # Although the cluster is healthy, we fail to access the
                 # job status. Try to recover the job (will not restart the
@@ -176,11 +180,6 @@ class SpotController:
             spot_state.set_failed(
                 self._job_id,
                 failure_type=spot_state.SpotStatus.FAILED_NO_RESOURCE)
-        except exceptions.ClusterSetUpError as e:
-            logger.error(f'{common_utils.class_fullname(e.__class__)}: '
-                         f'{colorama.Fore.RED}{e}{colorama.Style.RESET_ALL}')
-            spot_state.set_failed(self._job_id,
-                                  failure_type=spot_state.SpotStatus.FAILED)
         except (Exception, SystemExit) as e:  # pylint: disable=broad-except
             logger.error(traceback.format_exc())
             logger.error('Unexpected error occurred: '
diff --git a/sky/spot/recovery_strategy.py b/sky/spot/recovery_strategy.py
index 4b1284aa4..f08d91137 100644
--- a/sky/spot/recovery_strategy.py
+++ b/sky/spot/recovery_strategy.py
@@ -80,7 +80,7 @@ class StrategyExecutor:
         It can fail if resource is not available. Need to check the cluster
         status, after calling.
 
-        Returns: The job's start timestamp, or None if failed to start.
+        Returns: The job's submit timestamp, or None if failed.
         """
         if self.retry_until_up:
             return self._launch(max_retry=None)
@@ -140,8 +140,9 @@ class StrategyExecutor:
             raise_on_failure: Whether to raise an exception if the launch fails.
 
         Returns:
-            The job's start timestamp, or None if failed to start and
-            raise_on_failure is False.
+            The job's submit timestamp, or None if failed to submit the job
+            (either provisioning fails or any error happens in job submission)
+            and raise_on_failure is False.
         """
         # TODO(zhwu): handle the failure during `preparing sky runtime`.
         retry_cnt = 0
@@ -152,19 +153,17 @@ class StrategyExecutor:
             exception = None
             try:
                 usage_lib.messages.usage.set_internal()
+                # Detach setup, so that the setup failure can be detected
+                # by the controller process (job_status -> FAILED_SETUP).
                 sky.launch(self.dag,
                            cluster_name=self.cluster_name,
+                           detach_setup=True,
                            detach_run=True,
                            _is_launched_by_spot_controller=True)
                 logger.info('Spot cluster launched.')
             except exceptions.InvalidClusterNameError as e:
                 # The cluster name is too long.
                 raise exceptions.ResourcesUnavailableError(str(e)) from e
-            except exceptions.ClusterSetUpError as e:
-                # When setup fails, do not retry, as it will likely fail again.
-                # The exception will be handled by the caller and set the job
-                # state to FAILED.
-                raise
             except Exception as e:  # pylint: disable=broad-except
                 # If the launch fails, it will be recovered by the following
                 # code.
@@ -207,7 +206,7 @@ class StrategyExecutor:
                     # The cluster can be preempted before the job is launched.
                     # Break to let the retry launch kick in.
                     logger.info('The cluster is preempted before the job '
-                                'starts.')
+                                'is submitted.')
                     # TODO(zhwu): we should recover the preemption with the
                     # recovery strategy instead of the current while loop.
                     retry_launch = True
@@ -228,11 +227,11 @@ class StrategyExecutor:
                     continue
 
                 # Check the job status until it is not in initialized status
-                if status is not None and job_lib.JobStatus.PENDING < status:
+                if status is not None and status > job_lib.JobStatus.INIT:
                     try:
-                        launch_time = spot_utils.get_job_timestamp(
+                        job_submitted_at = spot_utils.get_job_timestamp(
                             self.backend, self.cluster_name, get_end_time=False)
-                        return launch_time
+                        return job_submitted_at
                     except Exception as e:  # pylint: disable=broad-except
                         # If we failed to get the job timestamp, we will retry
                         # job checking loop.
@@ -276,8 +275,8 @@ class FailoverStrategyExecutor(StrategyExecutor, name='FAILOVER', default=True):
                                                     'sky.clouds.Region']] = None
 
     def _launch(self, max_retry=3, raise_on_failure=True) -> Optional[float]:
-        launch_time = super()._launch(max_retry, raise_on_failure)
-        if launch_time is not None:
+        job_submitted_at = super()._launch(max_retry, raise_on_failure)
+        if job_submitted_at is not None:
             # Only record the cloud/region if the launch is successful.
             handle = global_user_state.get_handle_from_cluster_name(
                 self.cluster_name)
@@ -285,7 +284,7 @@ class FailoverStrategyExecutor(StrategyExecutor, name='FAILOVER', default=True):
             launched_resources = handle.launched_resources
             self._launched_cloud_region = (launched_resources.cloud,
                                            launched_resources.region)
-        return launch_time
+        return job_submitted_at
 
     def recover(self) -> float:
         # 1. Cancel the jobs and launch the cluster with the STOPPED status,
@@ -313,11 +312,11 @@ class FailoverStrategyExecutor(StrategyExecutor, name='FAILOVER', default=True):
                                                region=launched_region)
                 task.set_resources({new_resources})
                 # Not using self.launch to avoid the retry until up logic.
-                launched_time = self._launch(raise_on_failure=False)
+                job_submitted_at = self._launch(raise_on_failure=False)
                 # Restore the original dag, i.e. reset the region constraint.
                 task.set_resources({original_resources})
-                if launched_time is not None:
-                    return launched_time
+                if job_submitted_at is not None:
+                    return job_submitted_at
 
             # Step 2
             logger.debug('Terminating unhealthy spot cluster and '
@@ -329,9 +328,9 @@ class FailoverStrategyExecutor(StrategyExecutor, name='FAILOVER', default=True):
             logger.debug('Relaunch the cluster  without constraining to prior '
                          'cloud/region.')
             # Not using self.launch to avoid the retry until up logic.
-            launched_time = self._launch(max_retry=self._MAX_RETRY_CNT,
-                                         raise_on_failure=False)
-            if launched_time is None:
+            job_submitted_at = self._launch(max_retry=self._MAX_RETRY_CNT,
+                                            raise_on_failure=False)
+            if job_submitted_at is None:
                 # Failed to launch the cluster.
                 if self.retry_until_up:
                     gap_seconds = self.RETRY_INIT_GAP_SECONDS
@@ -344,4 +343,4 @@ class FailoverStrategyExecutor(StrategyExecutor, name='FAILOVER', default=True):
                         f'Failed to recover the spot cluster after retrying '
                         f'{self._MAX_RETRY_CNT} times.')
 
-            return launched_time
+            return job_submitted_at
diff --git a/sky/spot/spot_state.py b/sky/spot/spot_state.py
index 42b8ff3b0..edf8b89c0 100644
--- a/sky/spot/spot_state.py
+++ b/sky/spot/spot_state.py
@@ -35,8 +35,9 @@ _CURSOR.execute("""\
     recovery_count INTEGER DEFAULT 0,
     job_duration FLOAT DEFAULT 0)""")
 
-# job_duration is the time a job actually runs before last_recover,
-# excluding the provision and recovery time.
+# job_duration is the time a job actually runs (including the
+# setup duration) before last_recover, excluding the provision
+# and recovery time.
 # If the job is not finished:
 # total_job_duration = now() - last_recovered_at + job_duration
 # If the job is not finished:
@@ -51,17 +52,66 @@ columns = [
 
 
 class SpotStatus(enum.Enum):
-    """Spot job status, designed to be in serverless style"""
+    """Spot job status, designed to be in serverless style.
+
+    The SpotStatus is a higher level status than the JobStatus.
+    Each spot job submitted to the spot cluster, will have a JobStatus
+    on that spot cluster:
+        JobStatus = [INIT, SETTING_UP, PENDING, RUNNING, ...]
+    Whenever the spot cluster is preempted and recovered, the JobStatus
+    will go through the statuses above again.
+    That means during the lifetime of a spot job, its JobsStatus could be
+    reset to INIT or SETTING_UP multiple times (depending on the preemptions).
+
+    However, a spot job only has one SpotStatus on the spot controller.
+        SpotStatus = [PENDING, SUBMITTED, STARTING, RUNNING, ...]
+    Mapping from JobStatus to SpotStatus:
+        INIT            ->  STARTING/RECOVERING
+        SETTING_UP      ->  RUNNING
+        PENDING         ->  RUNNING
+        RUNNING         ->  RUNNING
+        SUCCEEDED       ->  SUCCEEDED
+        FAILED          ->  FAILED
+        FAILED_SETUP    ->  FAILED_SETUP
+    Note that the JobStatus will not be stuck in PENDING, because each spot
+    cluster is dedicated to a spot job, i.e. there should always be enough
+    resource to run the job and the job will be immediately transitioned to
+    RUNNING.
+    """
+    # PENDING: Waiting for the spot controller to have a slot to run the
+    # controller process.
+    # The submitted_at timestamp of the spot job in the 'spot' table will be
+    # set to the time when the job is firstly submitted by the user (set to
+    # PENDING).
     PENDING = 'PENDING'
+    # SUBMITTED: The spot controller starts the controller process.
     SUBMITTED = 'SUBMITTED'
+    # STARTING: The controller process is launching the spot cluster for
+    # the spot job.
     STARTING = 'STARTING'
+    # RUNNING: The job is submitted to the spot cluster, and is setting up
+    # or running.
+    # The start_at timestamp of the spot job in the 'spot' table will be set
+    # to the time when the job is firstly transitioned to RUNNING.
     RUNNING = 'RUNNING'
+    # RECOVERING: The spot cluster is preempted, and the controller process
+    # is recovering the spot cluster (relaunching/failover).
     RECOVERING = 'RECOVERING'
     # Terminal statuses
+    # SUCCEEDED: The job is finished successfully.
     SUCCEEDED = 'SUCCEEDED'
+    # FAILED: The job is finished with failure from the user's program.
     FAILED = 'FAILED'
+    # FAILED_SETUP: The job is finished with failure from the user's setup
+    # script.
+    FAILED_SETUP = 'FAILED_SETUP'
+    # FAILED_NO_RESOURCE: The job is finished with failure because there is no
+    # resource available in the cloud provider(s) to launch the spot cluster.
     FAILED_NO_RESOURCE = 'FAILED_NO_RESOURCE'
+    # FAILED_CONTROLLER: The job is finished with failure because of unexpected
+    # error in the controller process.
     FAILED_CONTROLLER = 'FAILED_CONTROLLER'
+    # CANCELLED: The job is cancelled by the user.
     CANCELLED = 'CANCELLED'
 
     def is_terminal(self) -> bool:
@@ -83,7 +133,10 @@ class SpotStatus(enum.Enum):
 
     @classmethod
     def failure_statuses(cls) -> List['SpotStatus']:
-        return [cls.FAILED, cls.FAILED_NO_RESOURCE, cls.FAILED_CONTROLLER]
+        return [
+            cls.FAILED, cls.FAILED_SETUP, cls.FAILED_NO_RESOURCE,
+            cls.FAILED_CONTROLLER
+        ]
 
 
 _SPOT_STATUS_TO_COLOR = {
@@ -94,6 +147,7 @@ _SPOT_STATUS_TO_COLOR = {
     SpotStatus.RECOVERING: colorama.Fore.CYAN,
     SpotStatus.SUCCEEDED: colorama.Fore.GREEN,
     SpotStatus.FAILED: colorama.Fore.RED,
+    SpotStatus.FAILED_SETUP: colorama.Fore.RED,
     SpotStatus.FAILED_NO_RESOURCE: colorama.Fore.RED,
     SpotStatus.FAILED_CONTROLLER: colorama.Fore.RED,
     SpotStatus.CANCELLED: colorama.Fore.YELLOW,
diff --git a/sky/spot/spot_utils.py b/sky/spot/spot_utils.py
index 3fc0f6dd8..3940185e0 100644
--- a/sky/spot/spot_utils.py
+++ b/sky/spot/spot_utils.py
@@ -43,6 +43,12 @@ _JOB_WAITING_STATUS_MESSAGE = ('[bold cyan]Waiting for the job to start'
 _JOB_CANCELLED_MESSAGE = ('[bold cyan]Waiting for the job status to be updated.'
                           '[/] It may take a minute.')
 
+# The maximum time to wait for the spot job status to transition to terminal
+# state, after the job finished. This is a safeguard to avoid the case where
+# the spot job status fails to be updated and keep the `sky spot logs` blocking
+# for a long time.
+_FINAL_SPOT_STATUS_WAIT_TIMEOUT_SECONDS = 10
+
 
 class UserSignal(enum.Enum):
     """The signal to be sent to the user."""
@@ -119,9 +125,9 @@ def update_spot_job_status(job_id: Optional[int] = None):
 
 def get_job_timestamp(backend: 'backends.CloudVmRayBackend', cluster_name: str,
                       get_end_time: bool) -> float:
-    """Get the started/ended time of the job."""
-    code = job_lib.JobLibCodeGen.get_job_time_payload(job_id=None,
-                                                      is_end=get_end_time)
+    """Get the submitted/ended time of the job."""
+    code = job_lib.JobLibCodeGen.get_job_submitted_or_ended_timestamp_payload(
+        job_id=None, get_ended_time=get_end_time)
     handle = global_user_state.get_handle_from_cluster_name(cluster_name)
     returncode, stdout, stderr = backend.run_on_head(handle,
                                                      code,
@@ -317,10 +323,13 @@ def stream_logs_by_id(job_id: int, follow: bool = True) -> str:
     # The spot_status may not be in terminal status yet, since the controllerhas
     # not updated the spot state yet. We wait for a while, until the spot state
     # is updated.
+    wait_seconds = 0
     spot_status = spot_state.get_status(job_id)
     assert spot_status is not None, job_id
-    while not spot_status.is_terminal() and follow:
+    while (not spot_status.is_terminal() and follow and
+           wait_seconds < _FINAL_SPOT_STATUS_WAIT_TIMEOUT_SECONDS):
         time.sleep(1)
+        wait_seconds += 1
         spot_status = spot_state.get_status(job_id)
         assert spot_status is not None, job_id
 
@@ -351,12 +360,12 @@ def dump_spot_job_queue() -> str:
         if end_at is None:
             end_at = time.time()
 
-        job_start_at = job['last_recovered_at'] - job['job_duration']
+        job_submitted_at = job['last_recovered_at'] - job['job_duration']
         if job['status'] == spot_state.SpotStatus.RECOVERING:
             # When job is recovering, the duration is exact job['job_duration']
             job_duration = job['job_duration']
-        elif job_start_at > 0:
-            job_duration = end_at - job_start_at
+        elif job_submitted_at > 0:
+            job_duration = end_at - job_submitted_at
         else:
             # When job_start_at <= 0, that means the last_recovered_at is not
             # set yet, i.e. the job is not started.
diff --git a/sky/templates/gcp-tpu-create.sh.j2 b/sky/templates/gcp-tpu-create.sh.j2
index 8e977cb67..eab462bbb 100644
--- a/sky/templates/gcp-tpu-create.sh.j2
+++ b/sky/templates/gcp-tpu-create.sh.j2
@@ -9,4 +9,5 @@ gcloud config set project $PROJECT_ID
 yes | gcloud compute tpus create {{tpu_name}} \
 --zone={{zones}} \
 --version={{runtime_version}} \
---accelerator-type={{tpu_type}}
+--accelerator-type={{tpu_type}} \
+--network={{vpc_name}}


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index c5354f9e9..4849d9c76 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -13,6 +13,7 @@ from typing import List
 # 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.
+# 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).
@@ -25,12 +26,16 @@ def pytest_addoption(parser):
     parser.addoption('--runslow',
                      action='store_true',
                      default=False,
-                     help='run slow tests')
+                     help='run slow tests.')
     for cloud in all_clouds_in_smoke_tests:
         parser.addoption(f'--{cloud}',
                          action='store_true',
                          default=False,
-                         help=f'Only run {cloud.upper()} tests')
+                         help=f'Only run {cloud.upper()} tests.')
+    parser.addoption('--managed-spot',
+                     action='store_true',
+                     default=False,
+                     help='Only run tests for managed spot.')
     parser.addoption(
         '--generic-cloud',
         type=str,
@@ -59,8 +64,10 @@ def _get_cloud_to_run(config) -> List[str]:
 
 
 def pytest_collection_modifyitems(config, items):
-    skip_slow = pytest.mark.skip(reason='need --runslow option to run')
     skip_marks = {}
+    skip_marks['slow'] = pytest.mark.skip(reason='need --runslow option to run')
+    skip_marks['managed_spot'] = pytest.mark.skip(
+        reason='skipped, because --managed-spot option is set')
     for cloud in all_clouds_in_smoke_tests:
         skip_marks[cloud] = pytest.mark.skip(
             reason=f'tests for {cloud} is skipped, try setting --{cloud}')
@@ -69,11 +76,15 @@ def pytest_collection_modifyitems(config, items):
 
     for item in items:
         if 'slow' in item.keywords and not config.getoption('--runslow'):
-            item.add_marker(skip_slow)
+            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:
                 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'])
+
 
 @pytest.fixture
 def generic_cloud(request) -> str:
diff --git a/tests/test_list_accelerators.py b/tests/test_list_accelerators.py
index 56bde6b4f..c8e0579b2 100644
--- a/tests/test_list_accelerators.py
+++ b/tests/test_list_accelerators.py
@@ -17,6 +17,17 @@ def test_list_ccelerators_all():
     assert 'A100-80GB' in result, result
 
 
-def test_list_accelerators_filters():
+def test_list_accelerators_name_filter():
     result = sky.list_accelerators(gpus_only=False, name_filter='V100')
     assert sorted(result.keys()) == ['V100', 'V100-32GB'], result
+
+
+def test_list_accelerators_region_filter():
+    result = sky.list_accelerators(gpus_only=False,
+                                   clouds="aws",
+                                   region_filter='us-west-1')
+    all_regions = []
+    for res in result.values():
+        for instance in res:
+            all_regions.append(instance.region)
+    assert all([region == 'us-west-1' for region in all_regions])
diff --git a/tests/test_optimizer_random_dag.py b/tests/test_optimizer_random_dag.py
index f4b98fa18..4e04adabf 100644
--- a/tests/test_optimizer_random_dag.py
+++ b/tests/test_optimizer_random_dag.py
@@ -12,7 +12,7 @@ CLOUDS = {
     'Azure': sky.Azure(),
 }
 ALL_INSTANCE_TYPES = sum(sky.list_accelerators(gpus_only=True).values(), [])
-GCP_DEFAULT_INSTANCE_TYPE = sky.GCP.get_default_instance_type()
+GCP_HOST_VM = 'n1-highmem-8'
 
 DUMMY_NODES = [
     sky.optimizer._DUMMY_SOURCE_NAME,
@@ -67,7 +67,7 @@ def generate_random_dag(
             for candidate in candidate_instance_types:
                 instance_type = candidate.instance_type
                 if pd.isna(instance_type):
-                    instance_type = GCP_DEFAULT_INSTANCE_TYPE
+                    instance_type = GCP_HOST_VM
                 resources = sky.Resources(
                     cloud=CLOUDS[candidate.cloud],
                     instance_type=instance_type,
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 29c0eebc8..dcd4c2d33 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -1099,11 +1099,10 @@ def test_use_spot(generic_cloud: str):
 
 
 # ---------- Testing managed spot ----------
+@pytest.mark.managed_spot
 def test_spot(generic_cloud: str):
     """Test the spot yaml."""
     name = _get_cluster_name()
-    cancel_command = (
-        f'sky spot cancel -y -n {name}-1; sky spot cancel -y -n {name}-2')
     test = Test(
         'managed-spot',
         [
@@ -1118,11 +1117,14 @@ def test_spot(generic_cloud: str):
             'sleep 200',
             f'{_SPOT_QUEUE_WAIT}| grep {name}-2 | head -n1 | grep "RUNNING\|SUCCEEDED"',
         ],
-        cancel_command,
+        f'sky spot cancel -y -n {name}-1; sky spot cancel -y -n {name}-2',
+        # Increase timeout since sky spot queue -r can be blocked by other spot tests.
+        timeout=20 * 60,
     )
     run_one_test(test)
 
 
+@pytest.mark.managed_spot
 def test_spot_failed_setup(generic_cloud: str):
     """Test managed spot job with failed setup."""
     name = _get_cluster_name()
@@ -1132,7 +1134,7 @@ def test_spot_failed_setup(generic_cloud: str):
             f'sky spot launch -n {name} --cloud {generic_cloud} -y -d tests/test_yamls/failed_setup.yaml',
             'sleep 200',
             # Make sure the job failed quickly.
-            f's=$(sky spot queue); echo "$s"; echo; echo; echo "$s" | grep {name} | head -n1 | grep "FAILED"',
+            f'{_SPOT_QUEUE_WAIT} | grep {name} | head -n1 | grep "FAILED"',
         ],
         f'sky spot cancel -y -n {name}',
         # Increase timeout since sky spot queue -r can be blocked by other spot tests.
@@ -1143,6 +1145,7 @@ def test_spot_failed_setup(generic_cloud: str):
 
 # ---------- Testing managed spot recovery ----------
 @pytest.mark.aws
+@pytest.mark.managed_spot
 def test_spot_recovery_aws():
     """Test managed spot recovery."""
     name = _get_cluster_name()
@@ -1173,6 +1176,7 @@ def test_spot_recovery_aws():
 
 
 @pytest.mark.gcp
+@pytest.mark.managed_spot
 def test_spot_recovery_gcp():
     """Test managed spot recovery."""
     name = _get_cluster_name()
@@ -1203,6 +1207,7 @@ def test_spot_recovery_gcp():
     run_one_test(test)
 
 
+@pytest.mark.managed_spot
 def test_spot_recovery_default_resources(generic_cloud: str):
     """Test managed spot recovery for default resources."""
     name = _get_cluster_name()
@@ -1220,6 +1225,7 @@ def test_spot_recovery_default_resources(generic_cloud: str):
 
 
 @pytest.mark.aws
+@pytest.mark.managed_spot
 def test_spot_recovery_multi_node_aws():
     """Test managed spot recovery."""
     name = _get_cluster_name()
@@ -1251,6 +1257,7 @@ def test_spot_recovery_multi_node_aws():
 
 
 @pytest.mark.gcp
+@pytest.mark.managed_spot
 def test_spot_recovery_multi_node_gcp():
     """Test managed spot recovery."""
     name = _get_cluster_name()
@@ -1284,6 +1291,7 @@ def test_spot_recovery_multi_node_gcp():
 
 
 @pytest.mark.aws
+@pytest.mark.managed_spot
 def test_spot_cancellation_aws():
     name = _get_cluster_name()
     region = 'us-east-2'
@@ -1344,6 +1352,7 @@ def test_spot_cancellation_aws():
 
 
 @pytest.mark.gcp
+@pytest.mark.managed_spot
 def test_spot_cancellation_gcp():
     name = _get_cluster_name()
     zone = 'us-west3-b'
@@ -1399,6 +1408,7 @@ def test_spot_cancellation_gcp():
 
 
 # ---------- Testing storage for managed spot ----------
+@pytest.mark.managed_spot
 def test_spot_storage(generic_cloud: str):
     """Test storage with managed spot"""
     name = _get_cluster_name()
@@ -1420,12 +1430,15 @@ def test_spot_storage(generic_cloud: str):
                 f'[ $(aws s3api list-buckets --query "Buckets[?contains(Name, \'{storage_name}\')].Name" --output text | wc -l) -eq 0 ]'
             ],
             f'sky spot cancel -y -n {name}',
+            # Increase timeout since sky spot queue -r can be blocked by other spot tests.
+            timeout=20 * 60,
         )
         run_one_test(test)
 
 
 # ---------- Testing spot TPU ----------
 @pytest.mark.gcp
+@pytest.mark.managed_spot
 def test_spot_tpu():
     """Test managed spot on TPU."""
     name = _get_cluster_name()
@@ -1439,39 +1452,44 @@ def test_spot_tpu():
             f'{_SPOT_QUEUE_WAIT}| grep {name} | head -n1 | grep "RUNNING\|SUCCEEDED"',
         ],
         f'sky spot cancel -y -n {name}',
+        # Increase timeout since sky spot queue -r can be blocked by other spot tests.
+        timeout=20 * 60,
     )
     run_one_test(test)
 
 
-# ---------- Testing env ----------
-def test_inline_env(generic_cloud: str):
-    """Test env"""
+# ---------- Testing env for spot ----------
+@pytest.mark.managed_spot
+def test_spot_inline_env(generic_cloud: str):
+    """Test spot env"""
     name = _get_cluster_name()
     test = Test(
-        'test-inline-env',
+        'test-spot-inline-env',
         [
-            f'sky launch -c {name} -y --cloud {generic_cloud} --env TEST_ENV="hello world" -- "([[ ! -z \\"\$TEST_ENV\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_IPS\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_RANK\\" ]]) || exit 1"',
-            f'sky logs {name} 1 --status',
-            f'sky exec {name} --env TEST_ENV2="success" "([[ ! -z \\"\$TEST_ENV2\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_IPS\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_RANK\\" ]]) || exit 1"',
-            f'sky logs {name} 2 --status',
+            f'sky spot launch -n {name} -y --cloud {generic_cloud} --env TEST_ENV="hello world" -- "([[ ! -z \\"\$TEST_ENV\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_IPS\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_RANK\\" ]]) || exit 1"',
+            'sleep 20',
+            f'{_SPOT_QUEUE_WAIT} | grep {name} | grep SUCCEEDED',
         ],
-        f'sky down -y {name}',
+        f'sky spot cancel -y -n {name}',
+        # Increase timeout since sky spot queue -r can be blocked by other spot tests.
+        timeout=20 * 60,
     )
     run_one_test(test)
 
 
-# ---------- Testing env for spot ----------
-def test_inline_spot_env(generic_cloud: str):
+# ---------- Testing env ----------
+def test_inline_env(generic_cloud: str):
     """Test env"""
     name = _get_cluster_name()
     test = Test(
-        'test-inline-spot-env',
+        'test-inline-env',
         [
-            f'sky spot launch -n {name} -y --cloud {generic_cloud} --env TEST_ENV="hello world" -- "([[ ! -z \\"\$TEST_ENV\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_IPS\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_RANK\\" ]]) || exit 1"',
-            'sleep 20',
-            f'{_SPOT_QUEUE_WAIT} | grep {name} | grep SUCCEEDED',
+            f'sky launch -c {name} -y --cloud {generic_cloud} --env TEST_ENV="hello world" -- "([[ ! -z \\"\$TEST_ENV\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_IPS\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_RANK\\" ]]) || exit 1"',
+            f'sky logs {name} 1 --status',
+            f'sky exec {name} --env TEST_ENV2="success" "([[ ! -z \\"\$TEST_ENV2\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_IPS\\" ]] && [[ ! -z \\"\$SKYPILOT_NODE_RANK\\" ]]) || exit 1"',
+            f'sky logs {name} 2 --status',
         ],
-        f'sky spot cancel -y -n {name}',
+        f'sky down -y {name}',
     )
     run_one_test(test)