Repository: skypilot-org/skypilot
Pull request: [Core] Add option to avoid uploading credentials and custom labels for GCP
<!-- Describe the changes in this PR -->

This is to avoid uploading credentials of the current cloud to be launched, as we can use the service account assigned to the instance to access the buckets and cloud APIs.

Fixes #2820

To enable this feature, a user should specify the following config in the `~/.sky/config.yaml`, and this only supports AWS and GCP for now.

```yaml
aws:
  remote_identity: SERVICE_ACCOUNT

gcp:
  remote_identity: SERVICE_ACCOUNT
```

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [ ] Code formatting: `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
  - [x] `sky launch --cloud gcp --cpus 2 -i 0 echo hi`
  - [x] `sky launch --cloud gcp --cpus 2 -i 0 echo hi` with the following config.yaml
    ```yaml
    gcp:
      remote_identity: SERVICE_ACCOUNT
      instance_tags:
        owner: zhwu
    ```
  - [x] `sky launch --cloud aws --cpus 2 -i 0 echo hi` with the following config.yaml
    ```yaml
    aws:
      remote_identity: SERVICE_ACCOUNT
      instance_tags:
        owner: zhwu
    ```
  - [x] `sky launch --cloud gcp --cpus 2 -i 0 --num-nodes 2 echo hi`
- [ ] All smoke tests: `pytest tests/test_smoke.py --gcp` 
- [ ] All smoke tests: `pytest tests/test_smoke.py --aws` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
  - [x] `pytest tests/test_smoke.py::test_file_mounts --aws` with two private buckets (1 aws, 1 gcp) enabled in `using_file_mounts.yaml`.
  - [x] `pytest tests/test_smoke.py::test_file_mounts --gcp` with two private buckets (1 aws, 1 gcp) enabled in `using_file_mounts.yaml`.
- [ ] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh`


Problem statement:
Skypilot copies `.aws/credentials`
Hi!

We've started using skypilot recently. Thanks for the great tool. :)

We noticed that when we have a `.aws/credentials` files locally on our machine, Skypilot seems to copy it to the remote.
At the same time, the remote also receive permissions from the `skypilot-v1` role.
We end up having situations where some commands fails because of using `.aws/credentials` instead of the `skypilot-v1` role.

Is this an expected behavior?
Is there a way to change this behavior? Do I want to do it?
Why is this the current design? It seems counterintuitive to me but I feel I am missing something?

Hints from issue discussion:
Thanks for the report @Wirg. We should definitely fix it. For your local machine, are you using any AWS profile (`AWS_PROFILE=xxx`)? What's the credential type you expect to use for local machine and subsequently any SkyPilot launched instances?

Hi @concretevitamin 

On my local machine, I am using access token for the moment, but I am going to switch to a custom SSOs in a few months.
I am setting AWS_PROFILE to select a specific one locally. On the server, I am expecting to have enough permission on the ec2 based on the ec2 role.

Copy is probably not the right term as I don't think the whole file end up on the ec2, just the selected profile.
Thanks @Wirg, we'll investigate how to properly solve this. cc @Michaelvll too.

What were the failures for situations where some commands fail?

One possible workaround to try in the meantime: In the beginning of `run` commands, remove `~/.aws/credentials` so that the VM only has the role. You could confirm that by `ssh mycluster` and then running `aws configure list`, which should not show `shared-credentials-file` anymore.
Hi @concretevitamin ,

Thanks for the workaround, I think it's what we are doing right now.
Copying the .aws seems to work in most case (unless the EC2 has some permissions the user don't have in this profile which was our case). 
On the other hand, I find it weird to both set a role and override it by pushing the .aws of the user. This also might be a security risk.
Indeed. The main reason for uploading the credentials is to allow a non-AWS controller (spot or serve) to launch AWS clusters.  That said, it should be feasible to avoid uploading cloud X's credential file when launching a cluster on cloud X. That may involve some changes in our backend. Stay tuned.

Commit message: address comments

Patch to review:
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index 87dafef38db..20a2228dc58 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -103,15 +103,42 @@ Available fields and semantics:
     # Security group (optional).
     #
     # The name of the security group to use for all instances. If not specified,
-    # SkyPilot will use the default name for the security group: sky-sg-<hash>
+    # SkyPilot will use the default name for the security group: sky-sg-<hash>  
     # Note: please ensure the security group name specified exists in the
     # regions the instances are going to be launched or the AWS account has the
     # permission to create a security group.
     security_group_name: my-security-group
 
+    # The identity to use for AWS on the remote instance (optional).
+    #
+    # If 'LOCAL_CREDENTIALS' is specified, the user's credentials will be
+    # uploaded and used for accessing the cloud on the instance created by
+    # SkyPilot.
+    #
+    # If 'SERVICE_ACCOUNT' is specified, the service account will be used to
+    # authenticate on the remote instance. Note that this will only take
+    # effect if the remote instance is created on AWS, and LOCAL_CREDENTIALS 
+    # will still be uploaded to the instances created on other clouds for
+    # accessing resources on AWS. Also, if `SERVICE_ACCOUNT` is specified, it
+    # will cause the managed spot jobs, or the service replicas fail to access
+    # any resources on AWS, when the SkyPilot controller is running on AWS.
+    #
+    # Default: 'LOCAL_CREDENTIALS'. 
+    remote_identity: LOCAL_CREDENTIALS
+
   # Advanced GCP configurations (optional).
   # Apply to all new instances but not existing ones.
   gcp:
+    # Labels to assign to all instances launched by SkyPilot (optional).
+    #
+    # Example use case: cost tracking by user/team/project.
+    #
+    # Users should guarantee that these key-values are valid GCP labels, otherwise
+    # errors from the cloud provider will be surfaced.
+    instance_tags:
+      Owner: user-unique-name
+      my-tag: my-value
+
     # VPC to use (optional).
     #
     # Default: null, which implies the following behavior. First, all existing
@@ -165,6 +192,23 @@ Available fields and semantics:
       - projects/my-project/reservations/my-reservation1
       - projects/my-project/reservations/my-reservation2
 
+
+    # The identity to use for GCP on the remote instance (optional).
+    #
+    # If 'LOCAL_CREDENTIALS' is specified, the user's credentials will be
+    # uploaded and used for accessing the cloud on the instance created by
+    # SkyPilot.
+    # If 'SERVICE_ACCOUNT' is specified, the service account will be used to
+    # authenticate on the remote instance. Note that this will only take
+    # effect if the remote instance is created on GCP, and LOCAL_CREDENTIALS 
+    # will still be uploaded to the instances created on other clouds for
+    # accessing resources on GCP. Also, if `SERVICE_ACCOUNT` is specified, it
+    # will cause the managed spot jobs, or the service replicas fail to access
+    # any resources on GCP, when the SkyPilot controller is running on GCP.
+    #
+    # Default: 'LOCAL_CREDENTIALS'. 
+    remote_identity: LOCAL_CREDENTIALS
+
   # Advanced Kubernetes configurations (optional).
   kubernetes:
     # The networking mode for accessing SSH jump pod (optional).
diff --git a/examples/using_file_mounts.yaml b/examples/using_file_mounts.yaml
index bd77ba88430..efdf9791f61 100644
--- a/examples/using_file_mounts.yaml
+++ b/examples/using_file_mounts.yaml
@@ -11,6 +11,7 @@
 
 resources:
   cloud: aws
+  cpus: 2+
 
 workdir: .
 
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 248ddb1dc17..5a07d4dd1f9 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -756,6 +756,10 @@ def write_cluster_config(
             not appear in the catalog, or an ssh_proxy_command is specified but
             not for the given region, or GPUs are requested in a Kubernetes
             cluster but the cluster does not have nodes labeled with GPU types.
+        exceptions.InvalidConfigs: if the user specifies some config for the
+            cloud that is not valid, e.g. remote_identity: SERVICE_ACCOUNT
+            for a cloud that does not support it, the caller should skip the
+            cloud in this case.
     """
     # task.best_resources may not be equal to to_provision if the user
     # is running a job with less resources than the cluster has.
@@ -786,7 +790,18 @@ def write_cluster_config(
             (str(to_provision.cloud).lower(), 'specific_reservations'), set()))
 
     assert cluster_name is not None
-    credentials = sky_check.get_cloud_credential_file_mounts()
+    excluded_clouds = []
+    remote_identity = skypilot_config.get_nested(
+        (str(cloud).lower(), 'remote_identity'), 'LOCAL_CREDENTIALS')
+    if remote_identity == 'SERVICE_ACCOUNT':
+        if not cloud.supports_service_account_on_remote():
+            raise exceptions.InvalidCloudConfigs(
+                'remote_identity: SERVICE_ACCOUNT is specified in '
+                f'{skypilot_config.loaded_config_path!r} for {cloud}, but it '
+                'is not supported by this cloud. Remove the config or set: '
+                '`remote_identity: LOCAL_CREDENTIALS`.')
+        excluded_clouds = [cloud]
+    credentials = sky_check.get_cloud_credential_file_mounts(excluded_clouds)
 
     ip_list = None
     auth_config = {'ssh_private_key': auth.PRIVATE_SSH_KEY_PATH}
@@ -828,10 +843,9 @@ def write_cluster_config(
     instance_tags = {}
     instance_tags = skypilot_config.get_nested(
         (str(cloud).lower(), 'instance_tags'), {})
-    if not isinstance(instance_tags, dict):
-        with ux_utils.print_exception_no_traceback():
-            raise ValueError('Custom instance_tags in config.yaml should '
-                             f'be a dict, but received {type(instance_tags)}.')
+    # instance_tags is a dict, which is guaranteed by the type check in
+    # schemas.py
+    assert isinstance(instance_tags, dict), instance_tags
 
     # Dump the Ray ports to a file for Ray job submission
     dump_port_command = (
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 7c5a94681b4..0dd7038f398 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -1509,6 +1509,17 @@ def _retry_zones(
                 # does not have nodes labeled with GPU types.
                 logger.info(f'{e}')
                 continue
+            except exceptions.InvalidCloudConfigs as e:
+                # Failed due to invalid user input, e.g. invalid configs in
+                # ~/.sky/config.yaml.
+                logger.info(f'{common_utils.format_exception(e)}')
+                # We should block the entire cloud if the user input is invalid.
+                _add_to_blocked_resources(
+                    self._blocked_resources,
+                    to_provision.copy(region=None, zone=None))
+                raise exceptions.ResourcesUnavailableError(
+                    f'Failed to provision on cloud {to_provision.cloud} due to '
+                    f'invalid user input: {common_utils.format_exception(e)}')
             if dryrun:
                 return config_dict
             cluster_config_file = config_dict['ray']
diff --git a/sky/check.py b/sky/check.py
index 2ad2c891bab..3a3f76bd477 100644
--- a/sky/check.py
+++ b/sky/check.py
@@ -1,5 +1,5 @@
 """Credential checks: check cloud credentials and enable clouds."""
-from typing import Dict
+from typing import Dict, Iterable, Optional
 
 import click
 
@@ -72,7 +72,8 @@ def check(quiet: bool = False, verbose: bool = False) -> None:
     global_user_state.set_enabled_clouds(enabled_clouds)
 
 
-def get_cloud_credential_file_mounts() -> Dict[str, str]:
+def get_cloud_credential_file_mounts(
+        excluded_clouds: Optional[Iterable[clouds.Cloud]]) -> Dict[str, str]:
     """Returns the files necessary to access all enabled clouds.
 
     Returns a dictionary that will be added to a task's file mounts
@@ -81,6 +82,8 @@ def get_cloud_credential_file_mounts() -> Dict[str, str]:
     enabled_clouds = global_user_state.get_enabled_clouds()
     file_mounts = {}
     for cloud in enabled_clouds:
+        if excluded_clouds is not None and cloud.in_cloud_list(excluded_clouds):
+            continue
         cloud_file_mounts = cloud.get_credential_file_mounts()
         file_mounts.update(cloud_file_mounts)
     # Currently, get_enabled_clouds() does not support r2
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index fef065db071..4527a4acb88 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -102,6 +102,8 @@ class AWS(clouds.Cloud):
     # Reference: https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Using_Tags.html # pylint: disable=line-too-long
     _MAX_CLUSTER_NAME_LEN_LIMIT = 248
 
+    _SUPPORTS_SERVICE_ACCOUNT_ON_REMOTE = True
+
     _regions: List[clouds.Region] = []
 
     _INDENT_PREFIX = '    '
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index 966a44d6c2c..ac6d390e12c 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -11,7 +11,7 @@
 import enum
 import re
 import typing
-from typing import Dict, Iterator, List, Optional, Set, Tuple
+from typing import Dict, Iterable, Iterator, List, Optional, Set, Tuple
 
 from sky import exceptions
 from sky import skypilot_config
@@ -99,6 +99,7 @@ class Cloud:
     _DEFAULT_DISK_TIER = resources_utils.DiskTier.MEDIUM
     _BEST_DISK_TIER = resources_utils.DiskTier.HIGH
     _SUPPORTED_DISK_TIERS = {resources_utils.DiskTier.BEST}
+    _SUPPORTS_SERVICE_ACCOUNT_ON_REMOTE = False
 
     # The version of provisioner and status query. This is used to determine
     # the code path to use for each cloud in the backend.
@@ -117,6 +118,21 @@ def max_cluster_name_length(cls) -> Optional[int]:
         """
         return None
 
+    @classmethod
+    def supports_service_account_on_remote(cls) -> bool:
+        """Returns whether the cloud supports service account on remote cluster.
+
+        This method is used by backend_utils.write_cluster_config() to decide
+        whether to upload user's local cloud credential files to the remote
+        cluster.
+
+        If a cloud supports service account on remote cluster, the user's local
+        cloud credential files are not needed to be uploaded to the remote
+        instance, as the remote instance can be assigned with a service account
+        that has the necessary permissions to access the cloud resources.
+        """
+        return cls._SUPPORTS_SERVICE_ACCOUNT_ON_REMOTE
+
     #### Regions/Zones ####
 
     @classmethod
@@ -230,9 +246,13 @@ def get_egress_cost(self, num_gigabytes: float):
         """
         raise NotImplementedError
 
-    def is_same_cloud(self, other):
+    def is_same_cloud(self, other: 'Cloud'):
         raise NotImplementedError
 
+    def in_cloud_list(self, cloud_list: Iterable['Cloud']) -> bool:
+        """Returns whether this cloud is in the given cloud list."""
+        return any(self.is_same_cloud(cloud) for cloud in cloud_list)
+
     def make_deploy_resources_variables(
         self,
         resources: 'resources_lib.Resources',
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index 0573e98394d..4a056f01664 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -134,6 +134,8 @@ class GCP(clouds.Cloud):
     # lower limit.
     _MAX_CLUSTER_NAME_LEN_LIMIT = 35
 
+    _SUPPORTS_SERVICE_ACCOUNT_ON_REMOTE = True
+
     _INDENT_PREFIX = '    '
     _DEPENDENCY_HINT = (
         'GCP tools are not installed. Run the following commands:\n'
diff --git a/sky/exceptions.py b/sky/exceptions.py
index ae717b23bfb..08206b3305e 100644
--- a/sky/exceptions.py
+++ b/sky/exceptions.py
@@ -46,6 +46,11 @@ def with_failover_history(
         return self
 
 
+class InvalidCloudConfigs(Exception):
+    """Raised when invalid configurations are provided for a given cloud."""
+    pass
+
+
 class ProvisionPrechecksError(Exception):
     """Raised when a spot job fails prechecks before provision.
     Developer note: For now this should only be used by managed
diff --git a/sky/optimizer.py b/sky/optimizer.py
index 77d0f8c854a..8d70ee11043 100644
--- a/sky/optimizer.py
+++ b/sky/optimizer.py
@@ -335,7 +335,7 @@ def _estimate_nodes_cost_or_time(
                 # mention "kubernetes cluster" and/instead of "catalog"
                 # in the error message.
                 enabled_clouds = global_user_state.get_enabled_clouds()
-                if _cloud_in_list(clouds.Kubernetes(), enabled_clouds):
+                if clouds.Kubernetes().in_cloud_list(enabled_clouds):
                     if any(orig_resources.cloud is None
                            for orig_resources in node.resources):
                         source_hint = 'catalog and kubernetes cluster'
@@ -1083,10 +1083,6 @@ class DummyCloud(clouds.Cloud):
     pass
 
 
-def _cloud_in_list(cloud: clouds.Cloud, lst: Iterable[clouds.Cloud]) -> bool:
-    return any(cloud.is_same_cloud(c) for c in lst)
-
-
 def _make_launchables_for_valid_region_zones(
     launchable_resources: resources_lib.Resources
 ) -> List[resources_lib.Resources]:
@@ -1170,8 +1166,8 @@ def _fill_in_launchable_resources(
     if blocked_resources is None:
         blocked_resources = []
     for resources in task.resources:
-        if resources.cloud is not None and not _cloud_in_list(
-                resources.cloud, enabled_clouds):
+        if (resources.cloud is not None and
+                not resources.cloud.in_cloud_list(enabled_clouds)):
             if try_fix_with_sky_check:
                 # Explicitly check again to update the enabled cloud list.
                 check.check(quiet=True)
diff --git a/sky/provision/slurm/__init__.py b/sky/provision/slurm/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/sky/provision/slurm/instance.py b/sky/provision/slurm/instance.py
new file mode 100644
index 00000000000..3ae99dae8d5
--- /dev/null
+++ b/sky/provision/slurm/instance.py
@@ -0,0 +1,208 @@
+"""RunPod instance provisioning."""
+import time
+from typing import Any, Dict, List, Optional
+
+from sky import sky_logging
+from sky import status_lib
+from sky.provision import common
+from sky.provision.runpod import utils
+from sky.utils import common_utils
+from sky.utils import ux_utils
+
+POLL_INTERVAL = 5
+
+logger = sky_logging.init_logger(__name__)
+
+
+def _filter_instances(cluster_name_on_cloud: str,
+                      status_filters: Optional[List[str]]) -> Dict[str, Any]:
+
+    instances = utils.list_instances()
+    possible_names = [
+        f'{cluster_name_on_cloud}-head', f'{cluster_name_on_cloud}-worker'
+    ]
+
+    filtered_instances = {}
+    for instance_id, instance in instances.items():
+        if (status_filters is not None and
+                instance['status'] not in status_filters):
+            continue
+        if instance.get('name') in possible_names:
+            filtered_instances[instance_id] = instance
+    return filtered_instances
+
+
+def _get_head_instance_id(instances: Dict[str, Any]) -> Optional[str]:
+    head_instance_id = None
+    for inst_id, inst in instances.items():
+        if inst['name'].endswith('-head'):
+            head_instance_id = inst_id
+            break
+    return head_instance_id
+
+
+def run_instances(region: str, cluster_name_on_cloud: str,
+                  config: common.ProvisionConfig) -> common.ProvisionRecord:
+    """Runs instances for the given cluster."""
+
+    pending_status = ['CREATED', 'RESTARTING']
+
+    while True:
+        instances = _filter_instances(cluster_name_on_cloud, pending_status)
+        if not instances:
+            break
+        logger.info(f'Waiting for {len(instances)} instances to be ready.')
+        time.sleep(POLL_INTERVAL)
+    exist_instances = _filter_instances(cluster_name_on_cloud, ['RUNNING'])
+    head_instance_id = _get_head_instance_id(exist_instances)
+
+    to_start_count = config.count - len(exist_instances)
+    if to_start_count < 0:
+        raise RuntimeError(
+            f'Cluster {cluster_name_on_cloud} already has '
+            f'{len(exist_instances)} nodes, but {config.count} are required.')
+    if to_start_count == 0:
+        if head_instance_id is None:
+            raise RuntimeError(
+                f'Cluster {cluster_name_on_cloud} has no head node.')
+        logger.info(f'Cluster {cluster_name_on_cloud} already has '
+                    f'{len(exist_instances)} nodes, no need to start more.')
+        return common.ProvisionRecord(provider_name='runpod',
+                                      cluster_name=cluster_name_on_cloud,
+                                      region=region,
+                                      zone=None,
+                                      head_instance_id=head_instance_id,
+                                      resumed_instance_ids=[],
+                                      created_instance_ids=[])
+
+    created_instance_ids = []
+    for _ in range(to_start_count):
+        node_type = 'head' if head_instance_id is None else 'worker'
+        try:
+            instance_id = utils.launch(
+                name=f'{cluster_name_on_cloud}-{node_type}',
+                instance_type=config.node_config['InstanceType'],
+                region=region,
+                disk_size=config.node_config['DiskSize'])
+        except Exception as e:  # pylint: disable=broad-except
+            logger.warning(f'run_instances error: {e}')
+            raise
+        logger.info(f'Launched instance {instance_id}.')
+        created_instance_ids.append(instance_id)
+        if head_instance_id is None:
+            head_instance_id = instance_id
+
+    # Wait for instances to be ready.
+    while True:
+        instances = _filter_instances(cluster_name_on_cloud, ['RUNNING'])
+        ready_instance_cnt = 0
+        for instance_id, instance in instances.items():
+            if instance.get('ssh_port') is not None:
+                ready_instance_cnt += 1
+        logger.info('Waiting for instances to be ready: '
+                    f'({ready_instance_cnt}/{config.count}).')
+        if ready_instance_cnt == config.count:
+            break
+
+        time.sleep(POLL_INTERVAL)
+    assert head_instance_id is not None, 'head_instance_id should not be None'
+    return common.ProvisionRecord(provider_name='runpod',
+                                  cluster_name=cluster_name_on_cloud,
+                                  region=region,
+                                  zone=None,
+                                  head_instance_id=head_instance_id,
+                                  resumed_instance_ids=[],
+                                  created_instance_ids=created_instance_ids)
+
+
+def wait_instances(region: str, cluster_name_on_cloud: str,
+                   state: Optional[status_lib.ClusterStatus]) -> None:
+    del region, cluster_name_on_cloud, state
+
+
+def stop_instances(
+    cluster_name_on_cloud: str,
+    provider_config: Optional[Dict[str, Any]] = None,
+    worker_only: bool = False,
+) -> None:
+    raise NotImplementedError()
+
+
+def terminate_instances(
+    cluster_name_on_cloud: str,
+    provider_config: Optional[Dict[str, Any]] = None,
+    worker_only: bool = False,
+) -> None:
+    """See sky/provision/__init__.py"""
+    del provider_config  # unused
+    instances = _filter_instances(cluster_name_on_cloud, None)
+    for inst_id, inst in instances.items():
+        logger.debug(f'Terminating instance {inst_id}: {inst}')
+        if worker_only and inst['name'].endswith('-head'):
+            continue
+        try:
+            utils.remove(inst_id)
+        except Exception as e:  # pylint: disable=broad-except
+            with ux_utils.print_exception_no_traceback():
+                raise RuntimeError(
+                    f'Failed to terminate instance {inst_id}: '
+                    f'{common_utils.format_exception(e, use_bracket=False)}'
+                ) from e
+
+
+def get_cluster_info(
+        region: str,
+        cluster_name_on_cloud: str,
+        provider_config: Optional[Dict[str, Any]] = None) -> common.ClusterInfo:
+    del region, provider_config  # unused
+    running_instances = _filter_instances(cluster_name_on_cloud, ['RUNNING'])
+    instances: Dict[str, List[common.InstanceInfo]] = {}
+    head_instance_id = None
+    for instance_id, instance_info in running_instances.items():
+        instances[instance_id] = [
+            common.InstanceInfo(
+                instance_id=instance_id,
+                internal_ip=instance_info['internal_ip'],
+                external_ip=instance_info['external_ip'],
+                ssh_port=instance_info['ssh_port'],
+                tags={},
+            )
+        ]
+        if instance_info['name'].endswith('-head'):
+            head_instance_id = instance_id
+
+    return common.ClusterInfo(
+        instances=instances,
+        head_instance_id=head_instance_id,
+    )
+
+
+def query_instances(
+    cluster_name_on_cloud: str,
+    provider_config: Optional[Dict[str, Any]] = None,
+    non_terminated_only: bool = True,
+) -> Dict[str, Optional[status_lib.ClusterStatus]]:
+    """See sky/provision/__init__.py"""
+    assert provider_config is not None, (cluster_name_on_cloud, provider_config)
+    instances = _filter_instances(cluster_name_on_cloud, None)
+
+    status_map = {
+        'CREATED': status_lib.ClusterStatus.INIT,
+        'RESTARTING': status_lib.ClusterStatus.INIT,
+        'PAUSED': status_lib.ClusterStatus.INIT,
+        'RUNNING': status_lib.ClusterStatus.UP,
+    }
+    statuses: Dict[str, Optional[status_lib.ClusterStatus]] = {}
+    for inst_id, inst in instances.items():
+        status = status_map[inst['status']]
+        if non_terminated_only and status is None:
+            continue
+        statuses[inst_id] = status
+    return statuses
+
+
+def cleanup_ports(
+    cluster_name_on_cloud: str,
+    provider_config: Optional[Dict[str, Any]] = None,
+) -> None:
+    del cluster_name_on_cloud, provider_config
diff --git a/sky/skypilot_config.py b/sky/skypilot_config.py
index 78fe4cf6ca5..267bf1897b6 100644
--- a/sky/skypilot_config.py
+++ b/sky/skypilot_config.py
@@ -44,7 +44,7 @@
 import copy
 import os
 import pprint
-from typing import Any, Dict, Iterable
+from typing import Any, Dict, Iterable, Optional
 
 import yaml
 
@@ -153,6 +153,11 @@ def _try_load_config() -> None:
         logger.debug('Config syntax check passed.')
 
 
+def loaded_config_path() -> Optional[str]:
+    """Returns the path to the loaded config file."""
+    return _loaded_config_path
+
+
 # Load on import.
 _try_load_config()
 
diff --git a/sky/templates/gcp-ray.yml.j2 b/sky/templates/gcp-ray.yml.j2
index 5268b4cbf77..76c818b8aef 100644
--- a/sky/templates/gcp-ray.yml.j2
+++ b/sky/templates/gcp-ray.yml.j2
@@ -76,6 +76,9 @@ available_node_types:
     node_config:
       labels:
         skypilot-user: {{ user }}
+      {%- for tag_key, tag_value in instance_tags.items() %}
+        {{ tag_key }}: {{ tag_value }}
+      {%- endfor %}
       {%- if specific_reservations %}
       reservationAffinity:
         consumeReservationType: SPECIFIC_RESERVATION
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index 7d435d32a50..a6613d118d1 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -507,6 +507,23 @@ def get_cluster_schema():
     },
 }
 
+_INSTANCE_TAGS_SCHEMA = {
+    'instance_tags': {
+        'type': 'object',
+        'required': [],
+        'additionalProperties': {
+            'type': 'string',
+        },
+    },
+}
+
+_REMOTE_AUTH_SCHEMA = {
+    'remote_identity': {
+        'type': 'string',
+        'case_insensitive_enum': ['LOCAL_CREDENTIALS', 'SERVICE_ACCOUNT'],
+    }
+}
+
 
 def get_config_schema():
     # pylint: disable=import-outside-toplevel
@@ -534,97 +551,98 @@ def get_config_schema():
             },
         }
     }
-
-    return {
-        '$schema': 'https://json-schema.org/draft/2020-12/schema',
-        'type': 'object',
-        'required': [],
-        'additionalProperties': False,
-        'properties': {
-            'spot': controller_resources_schema,
-            'serve': controller_resources_schema,
-            'aws': {
-                'type': 'object',
-                'required': [],
-                'additionalProperties': False,
-                'properties': {
-                    'instance_tags': {
-                        'type': 'object',
-                        'required': [],
-                        'additionalProperties': {
-                            'type': 'string',
-                        },
-                    },
-                    'security_group_name': {
+    cloud_configs = {
+        'aws': {
+            'type': 'object',
+            'required': [],
+            'additionalProperties': False,
+            'properties': {
+                'security_group_name': {
+                    'type': 'string',
+                },
+                **_INSTANCE_TAGS_SCHEMA,
+                **_NETWORK_CONFIG_SCHEMA,
+            }
+        },
+        'gcp': {
+            'type': 'object',
+            'required': [],
+            'additionalProperties': False,
+            'properties': {
+                'specific_reservations': {
+                    'type': 'array',
+                    'items': {
                         'type': 'string',
                     },
-                    **_NETWORK_CONFIG_SCHEMA,
+                },
+                **_INSTANCE_TAGS_SCHEMA,
+                **_NETWORK_CONFIG_SCHEMA,
+            }
+        },
+        'kubernetes': {
+            'type': 'object',
+            'required': [],
+            'additionalProperties': False,
+            'properties': {
+                'networking': {
+                    'type': 'string',
+                    'case_insensitive_enum': [
+                        type.value
+                        for type in kubernetes_enums.KubernetesNetworkingMode
+                    ]
+                },
+                'ports': {
+                    'type': 'string',
+                    'case_insensitive_enum': [
+                        type.value
+                        for type in kubernetes_enums.KubernetesPortMode
+                    ]
+                },
+                'pod_config': {
+                    'type': 'object',
+                    'required': [],
+                    # Allow arbitrary keys since validating pod spec is hard
+                    'additionalProperties': True,
                 }
-            },
-            'gcp': {
+            }
+        },
+        'oci': {
+            'type': 'object',
+            'required': [],
+            'properties': {},
+            # Properties are either 'default' or a region name.
+            'additionalProperties': {
                 'type': 'object',
                 'required': [],
                 'additionalProperties': False,
                 'properties': {
-                    'specific_reservations': {
-                        'type': 'array',
-                        'items': {
-                            'type': 'string',
-                        },
+                    'compartment_ocid': {
+                        'type': 'string',
                     },
-                    **_NETWORK_CONFIG_SCHEMA,
-                }
-            },
-            'kubernetes': {
-                'type': 'object',
-                'required': [],
-                'additionalProperties': False,
-                'properties': {
-                    'networking': {
+                    'image_tag_general': {
                         'type': 'string',
-                        'case_insensitive_enum': [
-                            type.value for type in
-                            kubernetes_enums.KubernetesNetworkingMode
-                        ]
                     },
-                    'ports': {
+                    'image_tag_gpu': {
+                        'type': 'string',
+                    },
+                    'vcn_subnet': {
                         'type': 'string',
-                        'case_insensitive_enum': [
-                            type.value
-                            for type in kubernetes_enums.KubernetesPortMode
-                        ]
                     },
-                    'pod_config': {
-                        'type': 'object',
-                        'required': [],
-                        # Allow arbitrary keys since validating pod spec is hard
-                        'additionalProperties': True,
-                    }
                 }
             },
-            'oci': {
-                'type': 'object',
-                'required': [],
-                # Properties are either 'default' or a region name.
-                'additionalProperties': {
-                    'type': 'object',
-                    'required': [],
-                    'additionalProperties': False,
-                    'properties': {
-                        'compartment_ocid': {
-                            'type': 'string',
-                        },
-                        'image_tag_general': {
-                            'type': 'string',
-                        },
-                        'image_tag_gpu': {
-                            'type': 'string',
-                        },
-                        'vcn_subnet': {
-                            'type': 'string',
-                        },
-                    }
-                },
-            },
+        },
+    }
+
+    for config in cloud_configs.values():
+        config['properties'].update(_REMOTE_AUTH_SCHEMA)
+    return {
+        '$schema': 'https://json-schema.org/draft/2020-12/schema',
+        'type': 'object',
+        'required': [],
+        'additionalProperties': False,
+        'properties': {
+            'spot': controller_resources_schema,
+            'serve': controller_resources_schema,
+            **cloud_configs,
         }
     }


Code review comments:
diff_hunk: @@ -1509,6 +1509,17 @@ def _retry_zones(
                 # does not have nodes labeled with GPU types.
                 logger.info(f'{e}')
                 continue
+            except exceptions.InvalidCloudConfigs as e:
+                # Failed due to invalid user input, e.g. invalid configs in
+                # ~/.sky/config.yaml.
+                logger.info(f'{common_utils.format_exception(e)}')
+                # We should block the entire cloud if the user input is invalid.
+                _add_to_blocked_resources(
+                    self._blocked_resources,
+                    to_provision.copy(region=None, zone=None))
+                raise exceptions.ResourcesUnavailableError(
+                    f'Failed to provision on cloud {to_provision.cloud} due to '
+                    f'invalid user input: {common_utils.format_exception(e)}')
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
```suggestion
                    f'Failed to provision on cloud {to_provision.cloud} due to '
                    f'invalid cloud config: {common_utils.format_exception(e)}')
```

@author:
Good poitn! Fixed. Thanks!

path: sky/provision/slurm/__init__.py
text: @user1:
Nice! Accidental?

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

diff_hunk: @@ -756,6 +756,10 @@ def write_cluster_config(
             not appear in the catalog, or an ssh_proxy_command is specified but
             not for the given region, or GPUs are requested in a Kubernetes
             cluster but the cluster does not have nodes labeled with GPU types.
+        exceptions.InvalidConfigs: if the user specifies some config for the
path: sky/backends/backend_utils.py
text: @user1:
```suggestion
        exceptions.InvalidCloudConfigs: if the user specifies some config for the
```

diff_hunk: @@ -1509,6 +1509,17 @@ def _retry_zones(
                 # does not have nodes labeled with GPU types.
                 logger.info(f'{e}')
                 continue
+            except exceptions.InvalidCloudConfigs as e:
+                # Failed due to invalid user input, e.g. invalid configs in
+                # ~/.sky/config.yaml.
+                logger.info(f'{common_utils.format_exception(e)}')
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
Should this be warn?

@author:
Good point! Changed.

diff_hunk: @@ -103,15 +103,42 @@ Available fields and semantics:
     # Security group (optional).
     #
     # The name of the security group to use for all instances. If not specified,
-    # SkyPilot will use the default name for the security group: sky-sg-<hash>
+    # SkyPilot will use the default name for the security group: sky-sg-<hash>  
     # Note: please ensure the security group name specified exists in the
     # regions the instances are going to be launched or the AWS account has the
     # permission to create a security group.
     security_group_name: my-security-group
 
+    # The identity to use for AWS on the remote instance (optional).
+    #
+    # If 'LOCAL_CREDENTIALS' is specified, the user's credentials will be
+    # uploaded and used for accessing the cloud on the instance created by
+    # SkyPilot.
+    #
+    # If 'SERVICE_ACCOUNT' is specified, the service account will be used to
+    # authenticate on the remote instance. Note that this will only take
+    # effect if the remote instance is created on AWS, and LOCAL_CREDENTIALS 
+    # will still be uploaded to the instances created on other clouds for
+    # accessing resources on AWS. Also, if `SERVICE_ACCOUNT` is specified, it
+    # will cause the managed spot jobs, or the service replicas fail to access
+    # any resources on AWS, when the SkyPilot controller is running on AWS.
+    #
+    # Default: 'LOCAL_CREDENTIALS'. 
path: docs/source/reference/config.rst
text: @user1:
```suggestion
    # Identity to use for all AWS instances (optional).
    #
    # LOCAL_CREDENTIALS: The user's local credential files will be uploaded to
    # AWS instances created by SkyPilot. They are used for accessing cloud
    # resources (e.g., private buckets) or launching new instances (e.g., for
    # spot/serve controllers).
    #
    # SERVICE_ACCOUNT: Local credential files are not uploaded to AWS
    # instances. SkyPilot will auto-create and reuse a service account (IAM
    # role) for AWS instances.
    #
    # Two caveats of SERVICE_ACCOUNT for multicloud users:
    #
    # - This only affect AWS instances. Local AWS credentials will still be
    #   uploaded to non-AWS instances (since those instances may need to access
    #   AWS resources).
    # - If the SkyPilot spot/serve controller is on AWS, this setting will make
    #   non-AWS managed spot jobs / non-AWS service replicas fail to access any
    #   resources on AWS (since the controllers don't have AWS credential
    #   files to assign to these non-AWS instances).
    #
    # Default: 'LOCAL_CREDENTIALS'.
```

Similarly for GCP?

@user1:
Is the `SkyPilot will auto-create and reuse a service account` part accurate?

@author:
Nice! Thanks for rewriting this! Yes, the auto-create and reuse part is accurate.

diff_hunk: @@ -1509,6 +1509,17 @@ def _retry_zones(
                 # does not have nodes labeled with GPU types.
                 logger.info(f'{e}')
                 continue
+            except exceptions.InvalidCloudConfigs as e:
+                # Failed due to invalid user input, e.g. invalid configs in
+                # ~/.sky/config.yaml.
+                logger.info(f'{common_utils.format_exception(e)}')
+                # We should block the entire cloud if the user input is invalid.
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
```suggestion
                # We should block the entire cloud if the user config is invalid.
```