Repository: skypilot-org/skypilot
Pull request #3501 (2024-05-01 16:11:30)

[Core][AWS] Allow specification of Security Groups for resources.
Similar to https://github.com/skypilot-org/skypilot/pull/3488 but for SGs
This PR:
 * Allows specification of SGs for controllers vs workers.
 * **Currently removes the port requirement to be able to allow SG specification in serve.**


Addresses: https://github.com/skypilot-org/skypilot/issues/3473

In `~/.sky/config.yaml`:
 * When setting for a controller wildcard and all other resources ("*")
```yaml
aws:
  security_group_name:
    - sky-serve-controller-*: skypilot-controller-fake-sg
    - "*": skypilot-default-fake-sg 
```

 * When setting for everything by specifying a default ("*")
```yaml
aws:
  security_group_name:
    - "*": skypilot-default-fake-sg
```

 * When setting for everything by specifying just the `security_group_name` via string
```yaml
aws:
  security_group_name: skypilot-default-fake-sg
```

Tested (run the relevant ones):

- [x] Code formatting: `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
  * tested running in all 3 configs in AWS
    - Nothing set
    - setting different controller and default values (controller vs replica in serve received different remote_identities)
    - setting default only
- [ ] All smoke tests: `pytest tests/test_smoke.py` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
- [ ] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh`

 
EDITED:
* refactored to match the IAM format.

JGSweets: @Michaelvll @concretevitamin  do you have any suggestions for the concerns I raised? Thanks!
JGSweets: @Michaelvll @cblmemo I've updated the PR with the `ClusterName` refactor.

Do we want to further refactor `provisioners`? If so, should we do it in this PR or a subsequent PR?

JGSweets: @Michaelvll Could you please re-review this PR? Thanks!
Michaelvll: Seems we have missed one place below:
https://github.com/JGSweets/skypilot/blob/22221f7ec7a4270c6308c256b10d2068c06a14be/sky/execution.py#L58-L59

It should be the following instead?
```
    resources_utils.ClusterName(clone_disk_from,
            handle.cluster_name_on_cloud)
```


JGSweets: @Michaelvll Good find! Not sure how I missed that. I've fixed it.
JGSweets: Updated tests passed as well.
Michaelvll: Sorry for the delay @JGSweets! The tests are all passed. Merging it now. Thank you for your contribution!

Patch:
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index cbcc85a4f..bd64c7c05 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -158,11 +158,27 @@ Available fields and semantics:
 
     # Security group (optional).
     #
-    # The name of the security group to use for all instances. If not specified,
+    # Security group name to use for AWS instances. If not specified,
     # 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.
+    #
+    # Some example use cases are shown below. All fields are optional.
+    # - <string>: apply the service account with the specified name to all instances.
+    #    Example:
+    #       security_group_name: my-security-group
+    # - <list of single-element dict>: A list of single-element dict mapping from the cluster name (pattern)
+    #   to the security group name to use. The matching of the cluster name is done in the same order
+    #   as the list.
+    #   NOTE: If none of the wildcard expressions in the dict match the cluster name, SkyPilot will use the default
+    #   security group name as mentioned above:  sky-sg-<hash>
+    #   To specify your default, use "*" as the wildcard expression.
+    #   Example:
+    #       security_group_name:
+    #         - my-cluster-name: my-security-group-1
+    #         - sky-serve-controller-*: my-security-group-2
+    #         - "*": my-default-security-group
     security_group_name: my-security-group
 
     # Identity to use for AWS instances (optional).
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index b80cf6674..65c32293b 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -155,6 +155,8 @@ _RAY_YAML_KEYS_TO_RESTORE_EXCEPTIONS = [
     # we need to take this field from the new yaml.
     ('provider', 'tpu_node'),
     ('provider', 'security_group', 'GroupName'),
+    ('available_node_types', 'ray.head.default', 'node_config',
+     'IamInstanceProfile'),
     ('available_node_types', 'ray.head.default', 'node_config', 'UserData'),
     ('available_node_types', 'ray.worker.default', 'node_config', 'UserData'),
 ]
@@ -793,8 +795,11 @@ def write_cluster_config(
     # move the check out of this function, i.e. the caller should be responsible
     # for the validation.
     # TODO(tian): Move more cloud agnostic vars to resources.py.
-    resources_vars = to_provision.make_deploy_variables(cluster_name_on_cloud,
-                                                        region, zones, dryrun)
+    resources_vars = to_provision.make_deploy_variables(
+        resources_utils.ClusterName(
+            cluster_name,
+            cluster_name_on_cloud,
+        ), region, zones, dryrun)
     config_dict = {}
 
     specific_reservations = set(
@@ -803,11 +808,13 @@ def write_cluster_config(
 
     assert cluster_name is not None
     excluded_clouds = []
-    remote_identity = skypilot_config.get_nested(
-        (str(cloud).lower(), 'remote_identity'),
-        schemas.get_default_remote_identity(str(cloud).lower()))
-    if remote_identity is not None and not isinstance(remote_identity, str):
-        for profile in remote_identity:
+    remote_identity_config = skypilot_config.get_nested(
+        (str(cloud).lower(), 'remote_identity'), None)
+    remote_identity = schemas.get_default_remote_identity(str(cloud).lower())
+    if isinstance(remote_identity_config, str):
+        remote_identity = remote_identity_config
+    if isinstance(remote_identity_config, list):
+        for profile in remote_identity_config:
             if fnmatch.fnmatchcase(cluster_name, list(profile.keys())[0]):
                 remote_identity = list(profile.values())[0]
                 break
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 6fe4211f1..63a198bbf 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -1566,8 +1566,8 @@ class RetryingVmProvisioner(object):
                         to_provision.cloud,
                         region,
                         zones,
-                        provisioner.ClusterName(cluster_name,
-                                                handle.cluster_name_on_cloud),
+                        resources_utils.ClusterName(
+                            cluster_name, handle.cluster_name_on_cloud),
                         num_nodes=num_nodes,
                         cluster_yaml=handle.cluster_yaml,
                         prev_cluster_ever_up=prev_cluster_ever_up,
@@ -1577,8 +1577,10 @@ class RetryingVmProvisioner(object):
                     # caller.
                     resources_vars = (
                         to_provision.cloud.make_deploy_resources_variables(
-                            to_provision, handle.cluster_name_on_cloud, region,
-                            zones))
+                            to_provision,
+                            resources_utils.ClusterName(
+                                cluster_name, handle.cluster_name_on_cloud),
+                            region, zones))
                     config_dict['provision_record'] = provision_record
                     config_dict['resources_vars'] = resources_vars
                     config_dict['handle'] = handle
@@ -2898,8 +2900,8 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
                 # 4. Starting ray cluster and skylet.
                 cluster_info = provisioner.post_provision_runtime_setup(
                     repr(handle.launched_resources.cloud),
-                    provisioner.ClusterName(handle.cluster_name,
-                                            handle.cluster_name_on_cloud),
+                    resources_utils.ClusterName(handle.cluster_name,
+                                                handle.cluster_name_on_cloud),
                     handle.cluster_yaml,
                     provision_record=provision_record,
                     custom_resource=resources_vars.get('custom_resources'),
@@ -3877,7 +3879,7 @@ class CloudVmRayBackend(backends.Backend['CloudVmRayResourceHandle']):
 
             try:
                 provisioner.teardown_cluster(repr(cloud),
-                                             provisioner.ClusterName(
+                                             resources_utils.ClusterName(
                                                  cluster_name,
                                                  cluster_name_on_cloud),
                                              terminate=terminate,
diff --git a/sky/cli.py b/sky/cli.py
index db5291d94..3717138f8 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -3868,7 +3868,7 @@ def _generate_task_with_service(
     env: List[Tuple[str, str]],
     gpus: Optional[str],
     instance_type: Optional[str],
-    ports: Tuple[str],
+    ports: Optional[Tuple[str]],
     cpus: Optional[str],
     memory: Optional[str],
     disk_size: Optional[int],
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index abfb82c05..cb09a3c6b 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -1,5 +1,6 @@
 """Amazon Web Services."""
 import enum
+import fnmatch
 import functools
 import json
 import os
@@ -370,12 +371,13 @@ class AWS(clouds.Cloud):
         return service_catalog.get_vcpus_mem_from_instance_type(instance_type,
                                                                 clouds='aws')
 
-    def make_deploy_resources_variables(self,
-                                        resources: 'resources_lib.Resources',
-                                        cluster_name_on_cloud: str,
-                                        region: 'clouds.Region',
-                                        zones: Optional[List['clouds.Zone']],
-                                        dryrun: bool = False) -> Dict[str, Any]:
+    def make_deploy_resources_variables(
+            self,
+            resources: 'resources_lib.Resources',
+            cluster_name: resources_utils.ClusterName,
+            region: 'clouds.Region',
+            zones: Optional[List['clouds.Zone']],
+            dryrun: bool = False) -> Dict[str, Any]:
         del dryrun  # unused
         assert zones is not None, (region, zones)
 
@@ -397,18 +399,32 @@ class AWS(clouds.Cloud):
         image_id = self._get_image_id(image_id_to_use, region_name,
                                       r.instance_type)
 
-        user_security_group = skypilot_config.get_nested(
+        user_security_group_config = skypilot_config.get_nested(
             ('aws', 'security_group_name'), None)
-        if resources.ports is not None:
-            # Already checked in Resources._try_validate_ports
-            assert user_security_group is None
-            security_group = USER_PORTS_SECURITY_GROUP_NAME.format(
-                cluster_name_on_cloud)
-        elif user_security_group is not None:
-            assert resources.ports is None
-            security_group = user_security_group
-        else:
+        user_security_group = None
+        if isinstance(user_security_group_config, str):
+            user_security_group = user_security_group_config
+        elif isinstance(user_security_group_config, list):
+            for profile in user_security_group_config:
+                if fnmatch.fnmatchcase(cluster_name.display_name,
+                                       list(profile.keys())[0]):
+                    user_security_group = list(profile.values())[0]
+                    break
+        security_group = user_security_group
+        if security_group is None:
             security_group = DEFAULT_SECURITY_GROUP_NAME
+            if resources.ports is not None:
+                # Already checked in Resources._try_validate_ports
+                security_group = USER_PORTS_SECURITY_GROUP_NAME.format(
+                    cluster_name.display_name)
+        elif resources.ports is not None:
+            with ux_utils.print_exception_no_traceback():
+                logger.warning(
+                    f'Skip opening ports {resources.ports} for cluster {cluster_name!r}, '
+                    'as `aws.security_group_name` in `~/.sky/config.yaml` is specified as '
+                    f' {security_group!r}. Please make sure the specified security group '
+                    'has requested ports setup; or, leave out `aws.security_group_name` '
+                    'in `~/.sky/config.yaml`.')
 
         return {
             'instance_type': r.instance_type,
@@ -840,22 +856,24 @@ class AWS(clouds.Cloud):
         assert False, 'This code path should not be used.'
 
     @classmethod
-    def create_image_from_cluster(cls, cluster_name: str,
-                                  cluster_name_on_cloud: str,
+    def create_image_from_cluster(cls,
+                                  cluster_name: resources_utils.ClusterName,
                                   region: Optional[str],
                                   zone: Optional[str]) -> str:
-        assert region is not None, (cluster_name, cluster_name_on_cloud, region)
+        assert region is not None, (cluster_name.display_name,
+                                    cluster_name.name_on_cloud, region)
         del zone  # unused
 
-        image_name = f'skypilot-{cluster_name}-{int(time.time())}'
+        image_name = f'skypilot-{cluster_name.display_name}-{int(time.time())}'
 
-        status = provision_lib.query_instances('AWS', cluster_name_on_cloud,
+        status = provision_lib.query_instances('AWS',
+                                               cluster_name.name_on_cloud,
                                                {'region': region})
         instance_ids = list(status.keys())
         if not instance_ids:
             with ux_utils.print_exception_no_traceback():
                 raise RuntimeError(
-                    f'Failed to find the source cluster {cluster_name!r} on '
+                    f'Failed to find the source cluster {cluster_name.display_name!r} on '
                     'AWS.')
 
         if len(instance_ids) != 1:
@@ -882,7 +900,7 @@ class AWS(clouds.Cloud):
             stream_logs=True)
 
         rich_utils.force_update_status(
-            f'Waiting for the source image {cluster_name!r} from {region} to be available on AWS.'
+            f'Waiting for the source image {cluster_name.display_name!r} from {region} to be available on AWS.'
         )
         # Wait for the image to be available
         wait_image_cmd = (
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index 916a1c01c..c2a3f3eb0 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -269,12 +269,13 @@ class Azure(clouds.Cloud):
     def get_zone_shell_cmd(cls) -> Optional[str]:
         return None
 
-    def make_deploy_resources_variables(self,
-                                        resources: 'resources.Resources',
-                                        cluster_name_on_cloud: str,
-                                        region: 'clouds.Region',
-                                        zones: Optional[List['clouds.Zone']],
-                                        dryrun: bool = False) -> Dict[str, Any]:
+    def make_deploy_resources_variables(
+            self,
+            resources: 'resources.Resources',
+            cluster_name: resources_utils.ClusterName,
+            region: 'clouds.Region',
+            zones: Optional[List['clouds.Zone']],
+            dryrun: bool = False) -> Dict[str, Any]:
         assert zones is None, ('Azure does not support zones', zones)
 
         region_name = region.name
@@ -374,7 +375,7 @@ class Azure(clouds.Cloud):
             'disk_tier': Azure._get_disk_type(_failover_disk_tier()),
             'cloud_init_setup_commands': cloud_init_setup_commands,
             'azure_subscription_id': self.get_project_id(dryrun),
-            'resource_group': f'{cluster_name_on_cloud}-{region_name}',
+            'resource_group': f'{cluster_name.name_on_cloud}-{region_name}',
         }
 
     def _get_feasible_launchable_resources(
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index c5ff78e1c..93048a84e 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -253,7 +253,7 @@ class Cloud:
     def make_deploy_resources_variables(
         self,
         resources: 'resources_lib.Resources',
-        cluster_name_on_cloud: str,
+        cluster_name: resources_utils.ClusterName,
         region: 'Region',
         zones: Optional[List['Zone']],
         dryrun: bool = False,
@@ -726,8 +726,8 @@ class Cloud:
     # cloud._cloud_unsupported_features().
 
     @classmethod
-    def create_image_from_cluster(cls, cluster_name: str,
-                                  cluster_name_on_cloud: str,
+    def create_image_from_cluster(cls,
+                                  cluster_name: resources_utils.ClusterName,
                                   region: Optional[str],
                                   zone: Optional[str]) -> str:
         """Creates an image from the cluster.
diff --git a/sky/clouds/cudo.py b/sky/clouds/cudo.py
index 3ad663065..8f7d4eaf9 100644
--- a/sky/clouds/cudo.py
+++ b/sky/clouds/cudo.py
@@ -194,12 +194,12 @@ class Cudo(clouds.Cloud):
     def make_deploy_resources_variables(
         self,
         resources: 'resources_lib.Resources',
-        cluster_name_on_cloud: str,
+        cluster_name: resources_utils.ClusterName,
         region: 'clouds.Region',
         zones: Optional[List['clouds.Zone']],
         dryrun: bool = False,
     ) -> Dict[str, Optional[str]]:
-        del zones
+        del zones, cluster_name  # unused
         r = resources
         acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
         if acc_dict is not None:
diff --git a/sky/clouds/fluidstack.py b/sky/clouds/fluidstack.py
index d7921a3f5..c4f15a0e5 100644
--- a/sky/clouds/fluidstack.py
+++ b/sky/clouds/fluidstack.py
@@ -10,6 +10,7 @@ from sky import clouds
 from sky import status_lib
 from sky.clouds import service_catalog
 from sky.provision.fluidstack import fluidstack_utils
+from sky.utils import resources_utils
 from sky.utils.resources_utils import DiskTier
 
 _CREDENTIAL_FILES = [
@@ -174,7 +175,7 @@ class Fluidstack(clouds.Cloud):
     def make_deploy_resources_variables(
         self,
         resources: 'resources_lib.Resources',
-        cluster_name_on_cloud: str,
+        cluster_name: resources_utils.ClusterName,
         region: clouds.Region,
         zones: Optional[List[clouds.Zone]],
         dryrun: bool = False,
@@ -189,7 +190,7 @@ class Fluidstack(clouds.Cloud):
         else:
             custom_resources = None
         cuda_installation_commands = """
-        sudo wget https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2004/x86_64/cuda-keyring_1.1-1_all.deb -O /usr/local/cuda-keyring_1.1-1_all.deb; 
+        sudo wget https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2004/x86_64/cuda-keyring_1.1-1_all.deb -O /usr/local/cuda-keyring_1.1-1_all.deb;
         sudo dpkg -i /usr/local/cuda-keyring_1.1-1_all.deb;
         sudo apt-get update;
         sudo apt-get -y install cuda-toolkit-12-3;
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index 86e9a90fa..050fda07f 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -404,7 +404,7 @@ class GCP(clouds.Cloud):
     def make_deploy_resources_variables(
             self,
             resources: 'resources.Resources',
-            cluster_name_on_cloud: str,
+            cluster_name: resources_utils.ClusterName,
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             dryrun: bool = False) -> Dict[str, Optional[str]]:
@@ -495,15 +495,15 @@ class GCP(clouds.Cloud):
 
         firewall_rule = None
         if resources.ports is not None:
-            firewall_rule = (
-                USER_PORTS_FIREWALL_RULE_NAME.format(cluster_name_on_cloud))
+            firewall_rule = (USER_PORTS_FIREWALL_RULE_NAME.format(
+                cluster_name.name_on_cloud))
         resources_vars['firewall_rule'] = firewall_rule
 
         # For TPU nodes. TPU VMs do not need TPU_NAME.
         tpu_node_name = resources_vars.get('tpu_node_name')
         if gcp_utils.is_tpu(resources) and not gcp_utils.is_tpu_vm(resources):
             if tpu_node_name is None:
-                tpu_node_name = cluster_name_on_cloud
+                tpu_node_name = cluster_name.name_on_cloud
 
         resources_vars['tpu_node_name'] = tpu_node_name
 
@@ -1005,8 +1005,8 @@ class GCP(clouds.Cloud):
         assert False, 'This code path should not be used.'
 
     @classmethod
-    def create_image_from_cluster(cls, cluster_name: str,
-                                  cluster_name_on_cloud: str,
+    def create_image_from_cluster(cls,
+                                  cluster_name: resources_utils.ClusterName,
                                   region: Optional[str],
                                   zone: Optional[str]) -> str:
         del region  # unused
@@ -1015,7 +1015,7 @@ class GCP(clouds.Cloud):
         # `ray-cluster-name` tag, which is guaranteed by the current `ray`
         # backend. Once the `provision.query_instances` is implemented for GCP,
         # we should be able to get rid of this assumption.
-        tag_filters = {'ray-cluster-name': cluster_name_on_cloud}
+        tag_filters = {'ray-cluster-name': cluster_name.name_on_cloud}
         label_filter_str = cls._label_filter_str(tag_filters)
         instance_name_cmd = ('gcloud compute instances list '
                              f'--filter="({label_filter_str})" '
@@ -1027,7 +1027,8 @@ class GCP(clouds.Cloud):
         subprocess_utils.handle_returncode(
             returncode,
             instance_name_cmd,
-            error_msg=f'Failed to get instance name for {cluster_name!r}',
+            error_msg=
+            f'Failed to get instance name for {cluster_name.display_name!r}',
             stderr=stderr,
             stream_logs=True)
         instance_names = json.loads(stdout)
@@ -1038,7 +1039,7 @@ class GCP(clouds.Cloud):
                     f'instance, but got: {instance_names}')
         instance_name = instance_names[0]['name']
 
-        image_name = f'skypilot-{cluster_name}-{int(time.time())}'
+        image_name = f'skypilot-{cluster_name.display_name}-{int(time.time())}'
         create_image_cmd = (f'gcloud compute images create {image_name} '
                             f'--source-disk  {instance_name} '
                             f'--source-disk-zone {zone}')
@@ -1050,7 +1051,8 @@ class GCP(clouds.Cloud):
         subprocess_utils.handle_returncode(
             returncode,
             create_image_cmd,
-            error_msg=f'Failed to create image for {cluster_name!r}',
+            error_msg=
+            f'Failed to create image for {cluster_name.display_name!r}',
             stderr=stderr,
             stream_logs=True)
 
@@ -1064,7 +1066,8 @@ class GCP(clouds.Cloud):
         subprocess_utils.handle_returncode(
             returncode,
             image_uri_cmd,
-            error_msg=f'Failed to get image uri for {cluster_name!r}',
+            error_msg=
+            f'Failed to get image uri for {cluster_name.display_name!r}',
             stderr=stderr,
             stream_logs=True)
 
diff --git a/sky/clouds/ibm.py b/sky/clouds/ibm.py
index 86e325a43..e468fecf0 100644
--- a/sky/clouds/ibm.py
+++ b/sky/clouds/ibm.py
@@ -168,7 +168,7 @@ class IBM(clouds.Cloud):
     def make_deploy_resources_variables(
         self,
         resources: 'resources_lib.Resources',
-        cluster_name_on_cloud: str,
+        cluster_name: resources_utils.ClusterName,
         region: 'clouds.Region',
         zones: Optional[List['clouds.Zone']],
         dryrun: bool = False,
@@ -184,7 +184,7 @@ class IBM(clouds.Cloud):
         Returns:
           A dictionary of cloud-specific node type variables.
         """
-        del cluster_name_on_cloud, dryrun  # Unused.
+        del cluster_name, dryrun  # Unused.
 
         def _get_profile_resources(instance_profile):
             """returns a dict representing the
diff --git a/sky/clouds/kubernetes.py b/sky/clouds/kubernetes.py
index 78471e0de..113774142 100644
--- a/sky/clouds/kubernetes.py
+++ b/sky/clouds/kubernetes.py
@@ -224,11 +224,11 @@ class Kubernetes(clouds.Cloud):
     def make_deploy_resources_variables(
             self,
             resources: 'resources_lib.Resources',
-            cluster_name_on_cloud: str,
+            cluster_name: resources_utils.ClusterName,
             region: Optional['clouds.Region'],
             zones: Optional[List['clouds.Zone']],
             dryrun: bool = False) -> Dict[str, Optional[str]]:
-        del cluster_name_on_cloud, zones, dryrun  # Unused.
+        del cluster_name, zones, dryrun  # Unused.
         if region is None:
             region = self._regions[0]
 
diff --git a/sky/clouds/lambda_cloud.py b/sky/clouds/lambda_cloud.py
index 979b48333..036f5a239 100644
--- a/sky/clouds/lambda_cloud.py
+++ b/sky/clouds/lambda_cloud.py
@@ -156,11 +156,11 @@ class Lambda(clouds.Cloud):
     def make_deploy_resources_variables(
             self,
             resources: 'resources_lib.Resources',
-            cluster_name_on_cloud: str,
+            cluster_name: resources_utils.ClusterName,
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             dryrun: bool = False) -> Dict[str, Optional[str]]:
-        del cluster_name_on_cloud, dryrun  # Unused.
+        del cluster_name, dryrun  # Unused.
         assert zones is None, 'Lambda does not support zones.'
 
         r = resources
diff --git a/sky/clouds/oci.py b/sky/clouds/oci.py
index 5fb0111bf..a911c3f38 100644
--- a/sky/clouds/oci.py
+++ b/sky/clouds/oci.py
@@ -187,11 +187,11 @@ class OCI(clouds.Cloud):
     def make_deploy_resources_variables(
             self,
             resources: 'resources_lib.Resources',
-            cluster_name_on_cloud: str,
+            cluster_name: resources_utils.ClusterName,
             region: Optional['clouds.Region'],
             zones: Optional[List['clouds.Zone']],
             dryrun: bool = False) -> Dict[str, Optional[str]]:
-        del cluster_name_on_cloud, dryrun  # Unused.
+        del cluster_name, dryrun  # Unused.
         assert region is not None, resources
 
         acc_dict = self.get_accelerators_from_instance_type(
diff --git a/sky/clouds/paperspace.py b/sky/clouds/paperspace.py
index f67a9a271..efa1afee7 100644
--- a/sky/clouds/paperspace.py
+++ b/sky/clouds/paperspace.py
@@ -173,11 +173,11 @@ class Paperspace(clouds.Cloud):
     def make_deploy_resources_variables(
             self,
             resources: 'resources_lib.Resources',
-            cluster_name_on_cloud: str,
+            cluster_name: resources_utils.ClusterName,
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             dryrun: bool = False) -> Dict[str, Optional[str]]:
-        del zones, dryrun
+        del zones, dryrun, cluster_name
 
         r = resources
         acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
diff --git a/sky/clouds/runpod.py b/sky/clouds/runpod.py
index c7a24e274..3486330b8 100644
--- a/sky/clouds/runpod.py
+++ b/sky/clouds/runpod.py
@@ -166,11 +166,11 @@ class RunPod(clouds.Cloud):
     def make_deploy_resources_variables(
             self,
             resources: 'resources_lib.Resources',
-            cluster_name_on_cloud: str,
+            cluster_name: resources_utils.ClusterName,
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             dryrun: bool = False) -> Dict[str, Optional[str]]:
-        del zones, dryrun  # unused
+        del zones, dryrun, cluster_name  # unused
 
         r = resources
         acc_dict = self.get_accelerators_from_instance_type(r.instance_type)
diff --git a/sky/clouds/scp.py b/sky/clouds/scp.py
index 6a3daf271..da45a7e14 100644
--- a/sky/clouds/scp.py
+++ b/sky/clouds/scp.py
@@ -179,11 +179,11 @@ class SCP(clouds.Cloud):
     def make_deploy_resources_variables(
             self,
             resources: 'resources_lib.Resources',
-            cluster_name_on_cloud: str,
+            cluster_name: resources_utils.ClusterName,
             region: 'clouds.Region',
             zones: Optional[List['clouds.Zone']],
             dryrun: bool = False) -> Dict[str, Optional[str]]:
-        del cluster_name_on_cloud, dryrun  # Unused.
+        del cluster_name, dryrun  # Unused.
         assert zones is None, 'SCP does not support zones.'
 
         r = resources
diff --git a/sky/clouds/vsphere.py b/sky/clouds/vsphere.py
index 872b8df9d..968368ff0 100644
--- a/sky/clouds/vsphere.py
+++ b/sky/clouds/vsphere.py
@@ -171,13 +171,13 @@ class Vsphere(clouds.Cloud):
     def make_deploy_resources_variables(
         self,
         resources: 'resources_lib.Resources',
-        cluster_name_on_cloud: str,
+        cluster_name: resources_utils.ClusterName,
         region: 'clouds.Region',
         zones: Optional[List['clouds.Zone']],
         dryrun: bool = False,
     ) -> Dict[str, Optional[str]]:
         # TODO get image id here.
-        del cluster_name_on_cloud, dryrun  # unused
+        del cluster_name, dryrun  # unused
         assert zones is not None, (region, zones)
         zone_names = [zone.name for zone in zones]
         r = resources
diff --git a/sky/execution.py b/sky/execution.py
index b1fb4ec41..1f6bd09f9 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -19,6 +19,7 @@ from sky.usage import usage_lib
 from sky.utils import controller_utils
 from sky.utils import dag_utils
 from sky.utils import env_options
+from sky.utils import resources_utils
 from sky.utils import rich_utils
 from sky.utils import subprocess_utils
 from sky.utils import timeline
@@ -55,8 +56,9 @@ def _maybe_clone_disk_from_cluster(clone_disk_from: Optional[str],
     with rich_utils.safe_status('Creating image from source cluster '
                                 f'{clone_disk_from!r}'):
         image_id = original_cloud.create_image_from_cluster(
-            clone_disk_from,
-            handle.cluster_name_on_cloud,
+            cluster_name=resources_utils.ClusterName(
+                display_name=clone_disk_from,
+                name_on_cloud=handle.cluster_name_on_cloud),
             region=handle.launched_resources.region,
             zone=handle.launched_resources.zone,
         )
diff --git a/sky/provision/provisioner.py b/sky/provision/provisioner.py
index df9a9fcc5..6e3886828 100644
--- a/sky/provision/provisioner.py
+++ b/sky/provision/provisioner.py
@@ -25,6 +25,7 @@ from sky.provision import logging as provision_logging
 from sky.provision import metadata_utils
 from sky.skylet import constants
 from sky.utils import common_utils
+from sky.utils import resources_utils
 from sky.utils import rich_utils
 from sky.utils import ux_utils
 
@@ -38,23 +39,11 @@ _MAX_RETRY = 3
 _TITLE = '\n\n' + '=' * 20 + ' {} ' + '=' * 20 + '\n'
 
 
-@dataclasses.dataclass
-class ClusterName:
-    display_name: str
-    name_on_cloud: str
-
-    def __repr__(self) -> str:
-        return repr(self.display_name)
-
-    def __str__(self) -> str:
-        return self.display_name
-
-
 def _bulk_provision(
     cloud: clouds.Cloud,
     region: clouds.Region,
     zones: Optional[List[clouds.Zone]],
-    cluster_name: ClusterName,
+    cluster_name: resources_utils.ClusterName,
     bootstrap_config: provision_common.ProvisionConfig,
 ) -> provision_common.ProvisionRecord:
     provider_name = repr(cloud)
@@ -135,7 +124,7 @@ def bulk_provision(
     cloud: clouds.Cloud,
     region: clouds.Region,
     zones: Optional[List[clouds.Zone]],
-    cluster_name: ClusterName,
+    cluster_name: resources_utils.ClusterName,
     num_nodes: int,
     cluster_yaml: str,
     prev_cluster_ever_up: bool,
@@ -225,7 +214,7 @@ def bulk_provision(
             raise
 
 
-def teardown_cluster(cloud_name: str, cluster_name: ClusterName,
+def teardown_cluster(cloud_name: str, cluster_name: resources_utils.ClusterName,
                      terminate: bool, provider_config: Dict) -> None:
     """Deleting or stopping a cluster.
 
@@ -411,8 +400,8 @@ def wait_for_ssh(cluster_info: provision_common.ClusterInfo,
 
 
 def _post_provision_setup(
-        cloud_name: str, cluster_name: ClusterName, cluster_yaml: str,
-        provision_record: provision_common.ProvisionRecord,
+        cloud_name: str, cluster_name: resources_utils.ClusterName,
+        cluster_yaml: str, provision_record: provision_common.ProvisionRecord,
         custom_resource: Optional[str]) -> provision_common.ClusterInfo:
     config_from_yaml = common_utils.read_yaml(cluster_yaml)
     provider_config = config_from_yaml.get('provider')
@@ -563,8 +552,8 @@ def _post_provision_setup(
 
 
 def post_provision_runtime_setup(
-        cloud_name: str, cluster_name: ClusterName, cluster_yaml: str,
-        provision_record: provision_common.ProvisionRecord,
+        cloud_name: str, cluster_name: resources_utils.ClusterName,
+        cluster_yaml: str, provision_record: provision_common.ProvisionRecord,
         custom_resource: Optional[str],
         log_dir: str) -> provision_common.ClusterInfo:
     """Run internal setup commands after provisioning and before user setup.
diff --git a/sky/resources.py b/sky/resources.py
index 38f7a9784..f0cb1abda 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -929,12 +929,6 @@ class Resources:
         """
         if self.ports is None:
             return
-        if skypilot_config.get_nested(('aws', 'security_group_name'),
-                                      None) is not None:
-            with ux_utils.print_exception_no_traceback():
-                raise ValueError(
-                    'Cannot specify ports when AWS security group name is '
-                    'specified.')
         if self.cloud is not None:
             self.cloud.check_features_are_supported(
                 self, {clouds.CloudImplementationFeatures.OPEN_PORTS})
@@ -1009,7 +1003,7 @@ class Resources:
     def get_spot_str(self) -> str:
         return '[Spot]' if self.use_spot else ''
 
-    def make_deploy_variables(self, cluster_name_on_cloud: str,
+    def make_deploy_variables(self, cluster_name: resources_utils.ClusterName,
                               region: clouds.Region,
                               zones: Optional[List[clouds.Zone]],
                               dryrun: bool) -> Dict[str, Optional[str]]:
@@ -1047,7 +1041,7 @@ class Resources:
 
         # Cloud specific variables
         cloud_specific_variables = self.cloud.make_deploy_resources_variables(
-            self, cluster_name_on_cloud, region, zones, dryrun)
+            self, cluster_name, region, zones, dryrun)
         return dict(
             cloud_specific_variables,
             **{
diff --git a/sky/utils/resources_utils.py b/sky/utils/resources_utils.py
index e2357b9ee..87a62dab9 100644
--- a/sky/utils/resources_utils.py
+++ b/sky/utils/resources_utils.py
@@ -1,4 +1,5 @@
 """Utility functions for resources."""
+import dataclasses
 import enum
 import itertools
 import re
@@ -43,6 +44,18 @@ class DiskTier(enum.Enum):
         return types.index(self) <= types.index(other)
 
 
+@dataclasses.dataclass
+class ClusterName:
+    display_name: str
+    name_on_cloud: str
+
+    def __repr__(self) -> str:
+        return repr(self.display_name)
+
+    def __str__(self) -> str:
+        return self.display_name
+
+
 def check_port_str(port: str) -> None:
     if not port.isdigit():
         with ux_utils.print_exception_no_traceback():
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index a529d61f2..a7eb148c5 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -114,6 +114,8 @@ def _get_single_resources_schema():
                             'type': 'integer',
                         }]
                     }
+                }, {
+                    'type': 'null',
                 }],
             },
             'labels': {
@@ -610,6 +612,32 @@ _LABELS_SCHEMA = {
     }
 }
 
+_PRORPERTY_NAME_OR_CLUSTER_NAME_TO_PROPERTY = {
+    'oneOf': [
+        {
+            'type': 'string'
+        },
+        {
+            # A list of single-element dict to pretain the
+            # order.
+            # Example:
+            #  property_name:
+            #    - my-cluster1-*: my-property-1
+            #    - my-cluster2-*: my-property-2
+            #    - "*"": my-property-3
+            'type': 'array',
+            'items': {
+                'type': 'object',
+                'additionalProperties': {
+                    'type': 'string'
+                },
+                'maxProperties': 1,
+                'minProperties': 1,
+            },
+        }
+    ]
+}
+
 
 class RemoteIdentityOptions(enum.Enum):
     """Enum for remote identity types.
@@ -638,33 +666,6 @@ _REMOTE_IDENTITY_SCHEMA = {
     }
 }
 
-_REMOTE_IDENTITY_SCHEMA_AWS = {
-    'remote_identity': {
-        'oneOf': [
-            {
-                'type': 'string'
-            },
-            {
-                # A list of single-element dict to pretain the order.
-                # Example:
-                #  remote_identity:
-                #    - my-cluster1-*: my-iam-role-1
-                #    - my-cluster2-*: my-iam-role-2
-                #    - "*"": my-iam-role-3
-                'type': 'array',
-                'items': {
-                    'type': 'object',
-                    'additionalProperties': {
-                        'type': 'string'
-                    },
-                    'maxProperties': 1,
-                    'minProperties': 1,
-                },
-            }
-        ]
-    }
-}
-
 _REMOTE_IDENTITY_SCHEMA_KUBERNETES = {
     'remote_identity': {
         'type': 'string'
@@ -705,9 +706,8 @@ def get_config_schema():
             'required': [],
             'additionalProperties': False,
             'properties': {
-                'security_group_name': {
-                    'type': 'string'
-                },
+                'security_group_name':
+                    (_PRORPERTY_NAME_OR_CLUSTER_NAME_TO_PROPERTY),
                 **_LABELS_SCHEMA,
                 **_NETWORK_CONFIG_SCHEMA,
             },
@@ -866,7 +866,9 @@ def get_config_schema():
 
     for cloud, config in cloud_configs.items():
         if cloud == 'aws':
-            config['properties'].update(_REMOTE_IDENTITY_SCHEMA_AWS)
+            config['properties'].update({
+                'remote_identity': _PRORPERTY_NAME_OR_CLUSTER_NAME_TO_PROPERTY
+            })
         elif cloud == 'kubernetes':
             config['properties'].update(_REMOTE_IDENTITY_SCHEMA_KUBERNETES)
         else:


Test patch:
diff --git a/tests/test_yamls/test_aws_config.yaml b/tests/test_yamls/test_aws_config.yaml
new file mode 100644
index 000000000..047334703
--- /dev/null
+++ b/tests/test_yamls/test_aws_config.yaml
@@ -0,0 +1,9 @@
+aws:
+  vpc_name: fake-vpc
+  remote_identity:
+    - sky-serve-fake1-*: fake1-skypilot-role
+    - sky-serve-fake2-*: fake2-skypilot-role
+
+  security_group_name:
+    - sky-serve-fake1-*: fake-1-sg
+    - sky-serve-fake2-*: fake-2-sg
diff --git a/tests/unit_tests/test_backend_utils.py b/tests/unit_tests/test_backend_utils.py
new file mode 100644
index 000000000..cb1b83f19
--- /dev/null
+++ b/tests/unit_tests/test_backend_utils.py
@@ -0,0 +1,123 @@
+import pathlib
+from typing import Dict
+from unittest.mock import Mock
+from unittest.mock import patch
+
+import pytest
+
+from sky import clouds
+from sky import skypilot_config
+from sky.backends import backend_utils
+from sky.resources import Resources
+from sky.resources import resources_utils
+
+
+@patch.object(skypilot_config, 'CONFIG_PATH',
+              './tests/test_yamls/test_aws_config.yaml')
+@patch.object(skypilot_config, '_dict', None)
+@patch.object(skypilot_config, '_loaded_config_path', None)
+@patch('sky.clouds.service_catalog.instance_type_exists', return_value=True)
+@patch('sky.clouds.service_catalog.get_accelerators_from_instance_type',
+       return_value={'fake-acc': 2})
+@patch('sky.clouds.service_catalog.get_image_id_from_tag',
+       return_value='fake-image')
+@patch.object(clouds.aws, 'DEFAULT_SECURITY_GROUP_NAME', 'fake-default-sg')
+@patch('sky.check.get_cloud_credential_file_mounts',
+       return_value='~/.aws/credentials')
+@patch('sky.backends.backend_utils._get_yaml_path_from_cluster_name',
+       return_value='/tmp/fake/path')
+@patch('sky.utils.common_utils.fill_template')
+def test_write_cluster_config_w_remote_identity(mock_fill_template,
+                                                *mocks) -> None:
+    skypilot_config._try_load_config()
+
+    cloud = clouds.AWS()
+
+    region = clouds.Region(name='fake-region')
+    zones = [clouds.Zone(name='fake-zone')]
+    resource = Resources(cloud=cloud, instance_type='fake-type: 3')
+
+    cluster_config_template = 'aws-ray.yml.j2'
+
+    # test default
+    backend_utils.write_cluster_config(
+        to_provision=resource,
+        num_nodes=2,
+        cluster_config_template=cluster_config_template,
+        cluster_name="display",
+        local_wheel_path=pathlib.Path('/tmp/fake'),
+        wheel_hash='b1bd84059bc0342f7843fcbe04ab563e',
+        region=region,
+        zones=zones,
+        dryrun=True,
+        keep_launch_fields_in_existing_config=True)
+
+    expected_subset = {
+        'instance_type': 'fake-type: 3',
+        'custom_resources': '{"fake-acc":2}',
+        'region': 'fake-region',
+        'zones': 'fake-zone',
+        'image_id': 'fake-image',
+        'security_group': 'fake-default-sg',
+        'security_group_managed_by_skypilot': 'true',
+        'vpc_name': 'fake-vpc',
+        'remote_identity': 'LOCAL_CREDENTIALS',  # remote identity
+        'sky_local_path': '/tmp/fake',
+        'sky_wheel_hash': 'b1bd84059bc0342f7843fcbe04ab563e',
+    }
+
+    mock_fill_template.assert_called_once()
+    assert mock_fill_template.call_args[0][
+        0] == cluster_config_template, "config template incorrect"
+    assert mock_fill_template.call_args[0][1].items() >= expected_subset.items(
+    ), "config fill values incorrect"
+
+    # test using cluster matches regex, top
+    mock_fill_template.reset_mock()
+    expected_subset.update({
+        'security_group': 'fake-1-sg',
+        'security_group_managed_by_skypilot': 'false',
+        'remote_identity': 'fake1-skypilot-role'
+    })
+    backend_utils.write_cluster_config(
+        to_provision=resource,
+        num_nodes=2,
+        cluster_config_template=cluster_config_template,
+        cluster_name="sky-serve-fake1-1234",
+        local_wheel_path=pathlib.Path('/tmp/fake'),
+        wheel_hash='b1bd84059bc0342f7843fcbe04ab563e',
+        region=region,
+        zones=zones,
+        dryrun=True,
+        keep_launch_fields_in_existing_config=True)
+
+    mock_fill_template.assert_called_once()
+    assert (mock_fill_template.call_args[0][0] == cluster_config_template,
+            "config template incorrect")
+    assert (mock_fill_template.call_args[0][1].items() >=
+            expected_subset.items(), "config fill values incorrect")
+
+    # test using cluster matches regex, middle
+    mock_fill_template.reset_mock()
+    expected_subset.update({
+        'security_group': 'fake-2-sg',
+        'security_group_managed_by_skypilot': 'false',
+        'remote_identity': 'fake2-skypilot-role'
+    })
+    backend_utils.write_cluster_config(
+        to_provision=resource,
+        num_nodes=2,
+        cluster_config_template=cluster_config_template,
+        cluster_name="sky-serve-fake2-1234",
+        local_wheel_path=pathlib.Path('/tmp/fake'),
+        wheel_hash='b1bd84059bc0342f7843fcbe04ab563e',
+        region=region,
+        zones=zones,
+        dryrun=True,
+        keep_launch_fields_in_existing_config=True)
+
+    mock_fill_template.assert_called_once()
+    assert (mock_fill_template.call_args[0][0] == cluster_config_template,
+            "config template incorrect")
+    assert (mock_fill_template.call_args[0][1].items() >=
+            expected_subset.items(), "config fill values incorrect")
diff --git a/tests/unit_tests/test_resources.py b/tests/unit_tests/test_resources.py
index f9e3ad516..450ca692f 100644
--- a/tests/unit_tests/test_resources.py
+++ b/tests/unit_tests/test_resources.py
@@ -1,10 +1,13 @@
 from typing import Dict
 from unittest.mock import Mock
+from unittest.mock import patch
 
 import pytest
 
 from sky import clouds
+from sky import skypilot_config
 from sky.resources import Resources
+from sky.utils import resources_utils
 
 GLOBAL_VALID_LABELS = {
     'plaintext': 'plainvalue',
@@ -86,3 +89,85 @@ def test_kubernetes_labels_resources():
     }
     cloud = clouds.Kubernetes()
     _run_label_test(allowed_labels, invalid_labels, cloud)
+
+
+@patch.object(skypilot_config, 'CONFIG_PATH',
+              './tests/test_yamls/test_aws_config.yaml')
+@patch.object(skypilot_config, '_dict', None)
+@patch.object(skypilot_config, '_loaded_config_path', None)
+@patch('sky.clouds.service_catalog.instance_type_exists', return_value=True)
+@patch('sky.clouds.service_catalog.get_accelerators_from_instance_type',
+       return_value={'fake-acc': 2})
+@patch('sky.clouds.service_catalog.get_image_id_from_tag',
+       return_value='fake-image')
+@patch.object(clouds.aws, 'DEFAULT_SECURITY_GROUP_NAME', 'fake-default-sg')
+def test_aws_make_deploy_variables(*mocks) -> None:
+    skypilot_config._try_load_config()
+
+    cloud = clouds.AWS()
+    cluster_name = resources_utils.ClusterName(display_name='display',
+                                               name_on_cloud='cloud')
+    region = clouds.Region(name='fake-region')
+    zones = [clouds.Zone(name='fake-zone')]
+    resource = Resources(cloud=cloud, instance_type='fake-type: 3')
+    config = resource.make_deploy_variables(cluster_name,
+                                            region,
+                                            zones,
+                                            dryrun=True)
+
+    expected_config_base = {
+        'instance_type': resource.instance_type,
+        'custom_resources': '{"fake-acc":2}',
+        'use_spot': False,
+        'region': 'fake-region',
+        'image_id': 'fake-image',
+        'disk_tier': 'gp3',
+        'disk_throughput': 218,
+        'disk_iops': 3500,
+        'custom_disk_perf': True,
+        'docker_image': None,
+        'docker_container_name': 'sky_container',
+        'docker_login_config': None,
+        'docker_run_options': [],
+        'initial_setup_commands': [],
+        'zones': 'fake-zone'
+    }
+
+    # test using defaults
+    expected_config = expected_config_base.copy()
+    expected_config.update({
+        'security_group': 'fake-default-sg',
+        'security_group_managed_by_skypilot': 'true'
+    })
+    assert config == expected_config, ('unexpected resource '
+                                       'variables generated')
+
+    # test using cluster matches regex, top
+    cluster_name = resources_utils.ClusterName(
+        display_name='sky-serve-fake1-1234', name_on_cloud='name-on-cloud')
+    expected_config = expected_config_base.copy()
+    expected_config.update({
+        'security_group': 'fake-1-sg',
+        'security_group_managed_by_skypilot': 'false'
+    })
+    config = resource.make_deploy_variables(cluster_name,
+                                            region,
+                                            zones,
+                                            dryrun=True)
+    assert config == expected_config, ('unexpected resource '
+                                       'variables generated')
+
+    # test using cluster matches regex, middle
+    cluster_name = resources_utils.ClusterName(
+        display_name='sky-serve-fake2-1234', name_on_cloud='name-on-cloud')
+    expected_config = expected_config_base.copy()
+    expected_config.update({
+        'security_group': 'fake-2-sg',
+        'security_group_managed_by_skypilot': 'false'
+    })
+    config = resource.make_deploy_variables(cluster_name,
+                                            region,
+                                            zones,
+                                            dryrun=True)
+    assert config == expected_config, ('unexpected resource '
+                                       'variables generated')