Repository: skypilot-org/skypilot
Pull request: [Core] Auto mapping for cluster name
<!-- Describe the changes in this PR -->

Closes #1853 and Closes #963 

The backward compatibility is guaranteed by the restore of ray yaml in the `backend_utils.write_cluster_config`


<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
  - [x] `sky launch -c i-am-a-very-long-cluster-name-with-more-than-35-characters --cloud gcp --cpus 2`
  - [x] `sky status -r`
  - [x] `sky exec  i-am-a-very-long-cluster-name-with-more-than-35-characters echo hi`
  - [x] `sky autostop -i  1 i-am-a-very-long-cluster-name-with-more-than-35-characters`
  - [x] `sky status -r`
- [x] All smoke tests: `pytest tests/test_smoke.py` 
- [x] All smoke tests: `pytest tests/test_smoke.py --aws` 
- [x] Backward compatibility tests: `bash tests/backward_comaptibility_tests.sh`


Problem statement:
Gracefully handle long cluster names & for spot, show FAILED_CONTROLLER instead
Been failing test_inline_spot_env smoke test:
```
» sky spot status | grep test-inline-spot-env-zongheng-4edc-62
428  test-inline-spot-env-zongheng-4edc-62  1x [CPU:0.5]   4 mins ago   2m 27s         -             0            FAILED_NO_RESOURCE

» sky logs sky-spot-controller 428
I 07-10 16:25:40 log_lib.py:340] Start streaming logs for job 428.
SKY INFO: Waiting for task resources on 1 node. This will block if the cluster is full.
SKY INFO: Use Ctrl-C to exit log streaming (task will not be killed).
SKY INFO: All task resources reserved.
SKY INFO: Reserved IPs: ['172.31.70.1']
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:34 controller.py:47] Started monitoring spot task test-inline-spot-env-zongheng-4edc-62 (id: 428)
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:34 spot_state.py:107] Launching the spot cluster...
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:602] == Optimizer ==
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:613] Target: minimizing cost
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:625] Estimated cost: $0.1 / hour
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:625]
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:671] Considered resources (1 node):
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:700] -----------------------------------------------------------------
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:700]  CLOUD   INSTANCE             ACCELERATORS   COST ($)   CHOSEN
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:700] -----------------------------------------------------------------
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:700]  AWS     m6i.2xlarge[Spot]    -              0.08          ✔
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:700]  GCP     n1-highmem-8[Spot]   -              0.1
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:700] -----------------------------------------------------------------
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 optimizer.py:700]
(test-inline-spot-env-zongheng-4edc-62 pid=47430) No existing clusters.
(test-inline-spot-env-zongheng-4edc-62 pid=47430)
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 recovery_strategy.py:83] Failed to launch the spot cluster with error: Cluster name 'test-inline-spot-env-zongheng-4edc-62-428' has 41 chars; maximum length is 37 chars.
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:20:35 recovery_strategy.py:111] Retrying to launch the spot cluster in 56.9 seconds.
(test-inline-spot-env-zongheng-4edc-62 pid=47430) No existing clusters.
(test-inline-spot-env-zongheng-4edc-62 pid=47430)
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:21:33 recovery_strategy.py:83] Failed to launch the spot cluster with error: Cluster name 'test-inline-spot-env-zongheng-4edc-62-428' has 41 chars; maximum length is 37 chars.
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:21:33 recovery_strategy.py:111] Retrying to launch the spot cluster in 88.5 seconds.
(test-inline-spot-env-zongheng-4edc-62 pid=47430) No existing clusters.
(test-inline-spot-env-zongheng-4edc-62 pid=47430)
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:23:02 recovery_strategy.py:83] Failed to launch the spot cluster with error: Cluster name 'test-inline-spot-env-zongheng-4edc-62-428' has 41 chars; maximum length is 37 chars.
(test-inline-spot-env-zongheng-4edc-62 pid=47430) E 07-10 16:23:02 controller.py:131] Resources unavailable: Failed to launch the spot cluster after 3 retries.
(test-inline-spot-env-zongheng-4edc-62 pid=47430) I 07-10 16:23:02 spot_state.py:185] Job failed due to failing to find available resources after retries.

Shared connection to 44.201.29.25 closed.
```

Issues:
1. FAILED_NO_RESOURCE is inaccurate.  We should show FAILED_CONTROLLER instead?
2. We should think about a good way to handle long cluster names.  Some bandaid patches: (1) allow long cluster names for non-GCP; (2) "inline-spot-env" -> "spot-env".  

@Michaelvll by the way, why is test_id needed in smoke?

Users under the same project/root can launch other's VM when same cluster name is used
SkyPilot cannot distinguish the clusters from different user accounts under the same GCP project / AWS root account / Azure subscription. That can cause a problem when the users accidentally specify the same cluster name as the other users.

From user requests.

Hints from issue discussion:
`test_id` was added to avoid the second smoke test reusing the cluster launched in the first smoke test and forgot to be terminated. Also, for the `test_spot_recovery`, we have to have that, to make sure the manual termination with `aws ec2` does not accidentally terminate other spot clusters from the different `spot launch` with the same cluster name but a different job id.
Setting `FAILED_NO_RESOURCE` for the long cluster name could be a bit tricky, as when no clouds are specified, it is possible that AWS failed due to resource unavailable and the GCP failed due to the cluster name length. In that case, the failure reason is vague.
I think one possible way to solve the problem is in spot case, we can check the cluster name length, and if it is longer than the limited length, we can generate a hash from it, and launch the cluster with that hash.
A user recently reported the issue, when running the spot job on AWS:
```
E 01-30 23:15:05 controller.py:174] Cluster name "finetune-x-xxxxx-xxxx-xxx-xxxx-xxx-xxxxx-xxxxx-xx-xxxxxxx-xxxxx-3" is invalid; ensure it is fully matched by regex: [a-z]([-a-z0-9]{0,61}[a-z0-9])?
I 01-30 23:15:05 spot_state.py:216] Job failed due to failing to find available resources after retries.
```

There are several things we need to do, as talked with @concretevitamin:
- [x] Allow longer name for AWS
- [ ] An end-to-end solution for long names: use the hash of the user-provided name as the identifier for the cluster in the cloud provider, and add the full name to the tag of the cluster.
- [x] The `FAILED_NO_RESOURCES` in the spot job table is confusing when the actual problem is the name too long. (This could be solved automatically by the second TODO, since there will be no InvalidClusterName error during the launching of the spot job)
Since we now support much longer names on AWS (244 chars), I think it's okay to close this for now and defer the end to end solution for even longer names. Wdyt @Michaelvll?
I just removed the P0 label for it. I would like to keep it open, as the length limit on GCP is very restrictive and it would still be nice to have an end-to-end solution. :)
Sounds good.

On Mon, Feb 6, 2023 at 11:13 AM Zhanghao Wu ***@***.***>
wrote:

> I just removed the P0 label for it. I would like to keep it open, as the
> length limit on GCP is very restrictive and it would still be nice to have
> an end-to-end solution. :)
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/skypilot-org/skypilot/issues/963#issuecomment-1419609150>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AAEQWHXWR66XPELWLB64ASDWWFENLANCNFSM53FH5Q2Q>
> .
> You are receiving this because you were mentioned.Message ID:
> ***@***.***>
>

This issue is stale because it has been open 120 days with no activity. Remove stale label or comment or this will be closed in 10 days.
This issue was closed because it has been stalled for 10 days with no activity.
Happened to another GCP user

Commit message: Add a readme

Patch to review:
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 2ccc9235198..1c45491c8dc 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -891,6 +891,7 @@ def write_cluster_config(
     # task.best_resources may not be equal to to_provision if the user
     # is running a job with less resources than the cluster has.
     cloud = to_provision.cloud
+    assert cloud is not None, to_provision
     # This can raise a ResourcesUnavailableError, when the region/zones
     # requested does not appear in the catalog. It can be triggered when the
     # user changed the catalog file, while there is a cluster in the removed
@@ -982,10 +983,14 @@ def write_cluster_config(
         f'open(os.path.expanduser("{constants.SKY_REMOTE_RAY_PORT_FILE}"), "w"))\''
     )
 
+    cluster_name_on_cloud = common_utils.make_cluster_name_on_cloud(
+        cluster_name, max_length=cloud.max_cluster_name_length())
+
     # Only using new security group names for clusters with ports specified.
     default_aws_sg_name = f'sky-sg-{common_utils.user_and_hostname_hash()}'
     if ports is not None:
-        default_aws_sg_name += f'-{common_utils.truncate_and_hash_cluster_name(cluster_name)}'
+        default_aws_sg_name = f'sky-sg-{cluster_name_on_cloud}'
+
     # Use a tmp file path to avoid incomplete YAML file being re-used in the
     # future.
     tmp_yaml_path = yaml_path + '.tmp'
@@ -994,7 +999,7 @@ def write_cluster_config(
         dict(
             resources_vars,
             **{
-                'cluster_name': cluster_name,
+                'cluster_name_on_cloud': cluster_name_on_cloud,
                 'num_nodes': num_nodes,
                 'ports': ports,
                 'disk_size': to_provision.disk_size,
@@ -1083,6 +1088,12 @@ def write_cluster_config(
         with open(tmp_yaml_path, 'w') as f:
             f.write(restored_yaml_content)
 
+    # Read the cluster name from the tmp yaml file, to take the backward
+    # compatbility restortion above into account.
+    # TODO: remove this after 2 minor releases, 0.5.0.
+    yaml_config = common_utils.read_yaml(tmp_yaml_path)
+    config_dict['cluster_name_on_cloud'] = yaml_config['cluster_name']
+
     # Optimization: copy the contents of source files in file_mounts to a
     # special dir, and upload that as the only file_mount instead. Delay
     # calling this optimization until now, when all source files have been
@@ -1809,7 +1820,9 @@ def _query_cluster_status_via_cloud_api(
         exceptions.ClusterStatusFetchingError: the cluster status cannot be
           fetched from the cloud provider.
     """
-    cluster_name = handle.cluster_name
+    cluster_name_on_cloud = handle.cluster_name_on_cloud
+    cluster_name_in_hint = common_utils.cluster_name_in_hint(
+        handle.cluster_name, cluster_name_on_cloud)
     # Use region and zone from the cluster config, instead of the
     # handle.launched_resources, because the latter may not be set
     # correctly yet.
@@ -1827,18 +1840,22 @@ def _query_cluster_status_via_cloud_api(
         cloud_name = repr(handle.launched_resources.cloud)
         try:
             node_status_dict = provision_lib.query_instances(
-                cloud_name, cluster_name, provider_config)
-            logger.debug(f'Querying {cloud_name} cluster {cluster_name!r} '
+                cloud_name, cluster_name_on_cloud, provider_config)
+            logger.debug(f'Querying {cloud_name} cluster '
+                         f'{cluster_name_in_hint} '
                          f'status:\n{pprint.pformat(node_status_dict)}')
             node_statuses = list(node_status_dict.values())
         except Exception as e:  # pylint: disable=broad-except
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.ClusterStatusFetchingError(
-                    f'Failed to query {cloud_name} cluster {cluster_name!r} '
-                    f'status: {e}')
+                    f'Failed to query {cloud_name} cluster '
+                    f'{cluster_name_in_hint} '
+                    f'status: {common_utils.format_exception(e, use_bracket=True)}'
+                )
     else:
         node_statuses = handle.launched_resources.cloud.query_status(
-            cluster_name, tag_filter_for_cluster(cluster_name), region, zone,
+            cluster_name_on_cloud,
+            tag_filter_for_cluster(cluster_name_on_cloud), region, zone,
             **kwargs)
     # GCP does not clean up preempted TPU VMs. We remove it ourselves.
     # TODO(wei-lin): handle multi-node cases.
@@ -1846,7 +1863,8 @@ def _query_cluster_status_via_cloud_api(
     # the cluster termination, as the preempted TPU VM should always be
     # removed.
     if kwargs.get('use_tpu_vm', False) and len(node_statuses) == 0:
-        logger.debug(f'Terminating preempted TPU VM cluster {cluster_name}')
+        logger.debug(
+            f'Terminating preempted TPU VM cluster {cluster_name_in_hint}')
         backend = backends.CloudVmRayBackend()
         # Do not use refresh cluster status during teardown, as that will
         # cause infinite recursion by calling cluster status refresh
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index c2ce2770c1b..56d0e18577e 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -45,8 +45,6 @@
 from sky.skylet import constants
 from sky.skylet import job_lib
 from sky.skylet import log_lib
-from sky.skylet.providers.scp.node_provider import SCPError
-from sky.skylet.providers.scp.node_provider import SCPNodeProvider
 from sky.usage import usage_lib
 from sky.utils import command_runner
 from sky.utils import common_utils
@@ -1533,6 +1531,11 @@ def _retry_zones(
             # means a second 'sky launch -c <name>' will attempt to reuse.
             handle = CloudVmRayResourceHandle(
                 cluster_name=cluster_name,
+                # Backward compatibility will be guaranteed by the underlying
+                # backend_utils.write_cluster_config, which gets the cluster
+                # name on cloud from the ray yaml file, if the previous cluster
+                # exists.
+                cluster_name_on_cloud=config_dict['cluster_name_on_cloud'],
                 cluster_yaml=cluster_config_file,
                 launched_nodes=num_nodes,
                 # OK for this to be shown in CLI as status == INIT.
@@ -2179,9 +2182,14 @@ def provision_with_retries(
 
 
 class CloudVmRayResourceHandle(backends.backend.ResourceHandle):
-    """A pickle-able tuple of:
+    """A pickle-able handle to a cluster created by CloudVmRayBackend.
+
+    The handle object will last for the whole lifecycle of the cluster.
 
     - (required) Cluster name.
+    - (required) Cluster name on cloud (different from the cluster name, as we
+        append user hash to avoid conflict b/t multiple users in the same
+        organization/account, and truncate the name for length limit).
     - (required) Path to a cluster.yaml file.
     - (optional) A cached head node public IP.  Filled in after a
         successful provision().
@@ -2192,11 +2200,12 @@ class CloudVmRayResourceHandle(backends.backend.ResourceHandle):
     - (optional) Docker user name
     - (optional) If TPU(s) are managed, a path to a deletion script.
     """
-    _VERSION = 5
+    _VERSION = 6
 
     def __init__(self,
                  *,
                  cluster_name: str,
+                 cluster_name_on_cloud: str,
                  cluster_yaml: str,
                  launched_nodes: int,
                  launched_resources: resources_lib.Resources,
@@ -2207,6 +2216,7 @@ def __init__(self,
                  tpu_delete_script: Optional[str] = None) -> None:
         self._version = self._VERSION
         self.cluster_name = cluster_name
+        self.cluster_name_on_cloud = cluster_name_on_cloud
         self._cluster_yaml = cluster_yaml.replace(os.path.expanduser('~'), '~',
                                                   1)
         # List of (internal_ip, external_ip) tuples for all the nodes
@@ -2223,6 +2233,7 @@ def __init__(self,
     def __repr__(self):
         return (f'ResourceHandle('
                 f'\n\tcluster_name={self.cluster_name},'
+                f'\n\tcluster_name_on_cloud={self.cluster_name_on_cloud},'
                 f'\n\thead_ip={self.head_ip},'
                 '\n\tstable_internal_external_ips='
                 f'{self.stable_internal_external_ips},'
@@ -2518,6 +2529,9 @@ def __setstate__(self, state):
         if version < 5:
             state['docker_user'] = None
 
+        if version < 6:
+            state['cluster_name_on_cloud'] = state['cluster_name']
+
         self.__dict__.update(state)
 
         # Because the update_cluster_ips and update_ssh_ports
@@ -3377,8 +3391,8 @@ def _teardown(self,
                 logger.warning(
                     f'{yellow}Purge (-p/--purge) is set, ignoring the '
                     f'identity mismatch error and removing '
-                    f'the cluser record from cluster table.{reset}\n{yellow}It '
-                    'is the user\'s responsibility to ensure that this '
+                    f'the cluster record from cluster table.{reset}\n{yellow}It'
+                    ' is the user\'s responsibility to ensure that this '
                     f'cluster is actually {verbed} on the cloud.{reset}')
                 is_identity_mismatch_and_purge = True
             else:
@@ -3661,6 +3675,7 @@ def teardown_no_lock(self,
         cloud = handle.launched_resources.cloud
         config = common_utils.read_yaml(handle.cluster_yaml)
         cluster_name = handle.cluster_name
+        cluster_name_on_cloud = handle.cluster_name_on_cloud
 
         # Avoid possibly unbound warnings. Code below must overwrite these vars:
         returncode = 0
@@ -3691,7 +3706,7 @@ def teardown_no_lock(self,
                 operation_fn = provision_lib.terminate_instances
             try:
                 operation_fn(repr(cloud),
-                             cluster_name,
+                             cluster_name_on_cloud,
                              provider_config=config['provider'])
             except Exception as e:  # pylint: disable=broad-except
                 if purge:
@@ -3730,12 +3745,12 @@ def teardown_no_lock(self,
             config_provider = common_utils.read_yaml(
                 handle.cluster_yaml)['provider']
             region = config_provider['region']
-            cluster_name = handle.cluster_name
             search_client = ibm.search_client()
             vpc_found = False
             # pylint: disable=unsubscriptable-object
             vpcs_filtered_by_tags_and_region = search_client.search(
-                query=f'type:vpc AND tags:{cluster_name} AND region:{region}',
+                query=(f'type:vpc AND tags:{cluster_name_on_cloud} '
+                       f'AND region:{region}'),
                 fields=['tags', 'region', 'type'],
                 limit=1000).get_result()['items']
             vpc_id = None
@@ -3752,18 +3767,23 @@ def teardown_no_lock(self,
                 # pylint: disable=line-too-long E1136
                 # Delete VPC and it's associated resources
                 vpc_provider = IBMVPCProvider(
-                    config_provider['resource_group_id'], region, cluster_name)
+                    config_provider['resource_group_id'], region,
+                    cluster_name_on_cloud)
                 vpc_provider.delete_vpc(vpc_id, region)
                 # successfully removed cluster as no exception was raised
                 returncode = 0
 
         elif terminate and isinstance(cloud, clouds.SCP):
+            # pylint: disable=import-outside-toplevel
+            from sky.skylet.providers.scp import node_provider
             config['provider']['cache_stopped_nodes'] = not terminate
-            provider = SCPNodeProvider(config['provider'], handle.cluster_name)
+            provider = node_provider.SCPNodeProvider(config['provider'],
+                                                     cluster_name_on_cloud)
             try:
                 if not os.path.exists(provider.metadata.path):
-                    raise SCPError('SKYPILOT_ERROR_NO_NODES_LAUNCHED: '
-                                   'Metadata file does not exist.')
+                    raise node_provider.SCPError(
+                        'SKYPILOT_ERROR_NO_NODES_LAUNCHED: '
+                        'Metadata file does not exist.')
 
                 with open(provider.metadata.path, 'r') as f:
                     metadata = json.load(f)
@@ -3771,7 +3791,7 @@ def teardown_no_lock(self,
                         'creation', {}).get('virtualServerId', None)
                     provider.terminate_node(node_id)
                 returncode = 0
-            except SCPError as e:
+            except node_provider.SCPError as e:
                 returncode = 1
                 stdout = ''
                 stderr = str(e)
@@ -3791,7 +3811,7 @@ def teardown_no_lock(self,
 
             # 0: All terminated successfully, failed count otherwise
             returncode = oci_query_helper.terminate_instances_by_tags(
-                {TAG_RAY_CLUSTER_NAME: cluster_name}, region)
+                {TAG_RAY_CLUSTER_NAME: cluster_name_on_cloud}, region)
 
             # To avoid undefined local variables error.
             stdout = stderr = ''
@@ -3846,7 +3866,8 @@ def teardown_no_lock(self,
                 raise RuntimeError(
                     _TEARDOWN_FAILURE_MESSAGE.format(
                         extra_reason='',
-                        cluster_name=handle.cluster_name,
+                        cluster_name=common_utils.cluster_name_in_hint(
+                            cluster_name, cluster_name_on_cloud),
                         stdout=stdout,
                         stderr=stderr))
 
@@ -3874,6 +3895,7 @@ def post_teardown_cleanup(self,
         log_path = os.path.join(os.path.expanduser(self.log_dir),
                                 'teardown.log')
         log_abs_path = os.path.abspath(log_path)
+        cluster_name_on_cloud = handle.cluster_name_on_cloud
 
         if (handle.tpu_delete_script is not None and
                 os.path.exists(handle.tpu_delete_script)):
@@ -3896,7 +3918,8 @@ def post_teardown_cleanup(self,
                     raise RuntimeError(
                         _TEARDOWN_FAILURE_MESSAGE.format(
                             extra_reason='It is caused by TPU failure.',
-                            cluster_name=handle.cluster_name,
+                            cluster_name=common_utils.cluster_name_in_hint(
+                                handle.cluster_name, cluster_name_on_cloud),
                             stdout=tpu_stdout,
                             stderr=tpu_stderr))
         if (terminate and handle.launched_resources.is_image_managed is True):
@@ -3926,7 +3949,7 @@ def post_teardown_cleanup(self,
                 # our sky node provider.
                 # TODO(tian): Adding a no-op cleanup_ports API after #2286
                 # merged.
-                provision_lib.cleanup_ports(repr(cloud), handle.cluster_name,
+                provision_lib.cleanup_ports(repr(cloud), cluster_name_on_cloud,
                                             config['provider'])
 
         # The cluster file must exist because the cluster_yaml will only
diff --git a/sky/clouds/aws.py b/sky/clouds/aws.py
index f8b67e58c39..e4c3ba93fe9 100644
--- a/sky/clouds/aws.py
+++ b/sky/clouds/aws.py
@@ -109,7 +109,7 @@ def _cloud_unsupported_features(
         return dict()
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         return cls._MAX_CLUSTER_NAME_LEN_LIMIT
 
     @classmethod
@@ -768,20 +768,22 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
 
     @classmethod
     def create_image_from_cluster(cls, cluster_name: str,
-                                  tag_filters: Dict[str,
-                                                    str], region: Optional[str],
+                                  cluster_name_on_cloud: str,
+                                  region: Optional[str],
                                   zone: Optional[str]) -> str:
-        assert region is not None, (tag_filters, region)
-        del tag_filters, zone  # unused
+        assert region is not None, (cluster_name, cluster_name_on_cloud, region)
+        del zone  # unused
 
         image_name = f'skypilot-{cluster_name}-{int(time.time())}'
 
-        status = provision_lib.query_instances('AWS', cluster_name,
+        status = provision_lib.query_instances('AWS', cluster_name_on_cloud,
                                                {'region': region})
         instance_ids = list(status.keys())
         if not instance_ids:
             with ux_utils.print_exception_no_traceback():
-                raise RuntimeError('Failed to find the source cluster on AWS.')
+                raise RuntimeError(
+                    f'Failed to find the source cluster {cluster_name!r} on '
+                    'AWS.')
 
         if len(instance_ids) != 1:
             with ux_utils.print_exception_no_traceback():
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index fbddf3a5425..85381cef782 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -71,7 +71,7 @@ def _cloud_unsupported_features(
         }
 
     @classmethod
-    def _max_cluster_name_length(cls) -> int:
+    def max_cluster_name_length(cls) -> int:
         return cls._MAX_CLUSTER_NAME_LEN_LIMIT
 
     def instance_type_to_hourly_cost(self,
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index 46f440b63a7..78d0c7dfa92 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -8,6 +8,7 @@
 from sky import exceptions
 from sky import skypilot_config
 from sky.clouds import service_catalog
+from sky.skylet import constants
 from sky.utils import log_utils
 from sky.utils import ux_utils
 
@@ -74,7 +75,7 @@ def _cloud_unsupported_features(
         raise NotImplementedError
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         """Returns the maximum length limit of a cluster name.
 
         This method is used by check_cluster_name_is_valid() to check if the
@@ -478,8 +479,7 @@ def check_cluster_name_is_valid(cls, cluster_name: str) -> None:
         """
         if cluster_name is None:
             return
-        max_cluster_name_len_limit = cls._max_cluster_name_length()
-        valid_regex = '[a-z]([-a-z0-9]*[a-z0-9])?'
+        valid_regex = constants.CLUSTER_NAME_VALID_REGEX
         if re.fullmatch(valid_regex, cluster_name) is None:
             with ux_utils.print_exception_no_traceback():
                 raise exceptions.InvalidClusterNameError(
@@ -487,14 +487,6 @@ def check_cluster_name_is_valid(cls, cluster_name: str) -> None:
                     'ensure it is fully matched by regex (e.g., '
                     'only contains lower letters, numbers and dash): '
                     f'{valid_regex}')
-        if (max_cluster_name_len_limit is not None and
-                len(cluster_name) > max_cluster_name_len_limit):
-            cloud_name = '' if cls is Cloud else f' on {cls._REPR}'
-            with ux_utils.print_exception_no_traceback():
-                raise exceptions.InvalidClusterNameError(
-                    f'Cluster name {cluster_name!r} has {len(cluster_name)} '
-                    'chars; maximum length is '
-                    f'{max_cluster_name_len_limit} chars{cloud_name}.')
 
     @classmethod
     def check_disk_tier_enabled(cls, instance_type: str,
@@ -636,8 +628,8 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
 
     @classmethod
     def create_image_from_cluster(cls, cluster_name: str,
-                                  tag_filters: Dict[str,
-                                                    str], region: Optional[str],
+                                  cluster_name_on_cloud: str,
+                                  region: Optional[str],
                                   zone: Optional[str]) -> str:
         """Creates an image from the cluster.
 
@@ -646,7 +638,7 @@ def create_image_from_cluster(cls, cluster_name: str,
         raise NotImplementedError
 
     @classmethod
-    def maybe_move_image(cls, image_name: str, source_region: str,
+    def maybe_move_image(cls, image_id: str, source_region: str,
                          target_region: str, source_zone: Optional[str],
                          target_zone: Optional[str]) -> str:
         """Move an image if required.
diff --git a/sky/clouds/gcp.py b/sky/clouds/gcp.py
index e81c9793107..bbabef46f2d 100644
--- a/sky/clouds/gcp.py
+++ b/sky/clouds/gcp.py
@@ -219,7 +219,7 @@ def _cloud_unsupported_features(
         return {}
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         return cls._MAX_CLUSTER_NAME_LEN_LIMIT
 
     #### Regions/Zones ####
@@ -1105,11 +1105,16 @@ def query_status(cls, name: str, tag_filters: Dict[str, str],
 
     @classmethod
     def create_image_from_cluster(cls, cluster_name: str,
-                                  tag_filters: Dict[str,
-                                                    str], region: Optional[str],
+                                  cluster_name_on_cloud: str,
+                                  region: Optional[str],
                                   zone: Optional[str]) -> str:
         del region  # unused
         assert zone is not None
+        # TODO(zhwu): This assumes the cluster is created with the
+        # `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}
         label_filter_str = cls._label_filter_str(tag_filters)
         instance_name_cmd = ('gcloud compute instances list '
                              f'--filter="({label_filter_str})" '
diff --git a/sky/clouds/ibm.py b/sky/clouds/ibm.py
index 6809c48f303..94b0327d21f 100644
--- a/sky/clouds/ibm.py
+++ b/sky/clouds/ibm.py
@@ -50,7 +50,7 @@ def _cloud_unsupported_features(
         }
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         return cls._MAX_CLUSTER_NAME_LEN_LIMIT
 
     @classmethod
diff --git a/sky/clouds/lambda_cloud.py b/sky/clouds/lambda_cloud.py
index 487e2328262..4751cbfb4ef 100644
--- a/sky/clouds/lambda_cloud.py
+++ b/sky/clouds/lambda_cloud.py
@@ -53,7 +53,7 @@ def _cloud_unsupported_features(
         return cls._CLOUD_UNSUPPORTED_FEATURES
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         return cls._MAX_CLUSTER_NAME_LEN_LIMIT
 
     @classmethod
diff --git a/sky/clouds/local.py b/sky/clouds/local.py
index bf86d2aec71..41ff3eeed80 100644
--- a/sky/clouds/local.py
+++ b/sky/clouds/local.py
@@ -48,7 +48,7 @@ def _cloud_unsupported_features(
         return cls._CLOUD_UNSUPPORTED_FEATURES
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         return None
 
     @classmethod
diff --git a/sky/clouds/oci.py b/sky/clouds/oci.py
index 3435c3ee80b..e68d6306ca1 100644
--- a/sky/clouds/oci.py
+++ b/sky/clouds/oci.py
@@ -56,7 +56,7 @@ def _cloud_unsupported_features(
         }
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         return cls._MAX_CLUSTER_NAME_LEN_LIMIT
 
     @classmethod
diff --git a/sky/clouds/scp.py b/sky/clouds/scp.py
index 96b321bab65..e9acbdef665 100644
--- a/sky/clouds/scp.py
+++ b/sky/clouds/scp.py
@@ -65,7 +65,7 @@ def _cloud_unsupported_features(
         return cls._CLOUD_UNSUPPORTED_FEATURES
 
     @classmethod
-    def _max_cluster_name_length(cls) -> Optional[int]:
+    def max_cluster_name_length(cls) -> Optional[int]:
         return cls._MAX_CLUSTER_NAME_LEN_LIMIT
 
     @classmethod
diff --git a/sky/design_docs/cluster_name.md b/sky/design_docs/cluster_name.md
new file mode 100644
index 00000000000..a929ba44a5e
--- /dev/null
+++ b/sky/design_docs/cluster_name.md
@@ -0,0 +1,45 @@
+# Cluster Name Mapping
+
+## Background
+
+SkyPilot allows users to assign names to clusters they launched. It is convenient, but
+when the following situation happens, it can be annoying:
+1. The cluster name exceeds clouds' naming length limit.
+2. Multiple users in the same cloud account organization use the same cluster name.
+
+## Solution
+
+After #2403, SkyPilot will automatically truncate cluster names to comply with the
+naming length limit of the cloud, and append a user hash to the end of the name to
+distinguish clusters from different users.
+
+## Details
+
+Cluster names used in different situations:
+- CLI/API/UI: the name users assign to the cluster, i.e., the original cluster name
+    without any modification.
+- Lower-level components: for the components that directly interact with the cloud
+    provider, the newly generated name will be used, including, the underlying
+    `ray up`, `provision` module, etc.
+
+## Examples
+
+For a cluster name less than cloud's naming length limit, the name will only be appended
+with a user hash.
+```
+# Original
+my-cluster-name
+
+# Generated
+my-cluster-name-9eac
+```
+
+For a cluster name longer than cloud's naming length limit, the name will be truncated
+to less than the limit, appended with a cluster name hash, and then appended with a user hash.
+```
+# Original
+my-cluster-name-very-long
+
+# Generated
+my-cluster-name-3m-9eac
+```
diff --git a/sky/execution.py b/sky/execution.py
index 783dbe2adb9..e648cfbe219 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -124,7 +124,7 @@ def _maybe_clone_disk_from_cluster(clone_disk_from: Optional[str],
                                     f'{clone_disk_from!r}'):
         image_id = original_cloud.create_image_from_cluster(
             clone_disk_from,
-            backend_utils.tag_filter_for_cluster(clone_disk_from),
+            handle.cluster_name_on_cloud,
             region=handle.launched_resources.region,
             zone=handle.launched_resources.zone,
         )
diff --git a/sky/provision/aws/instance.py b/sky/provision/aws/instance.py
index aa7f924f027..b288c1c3027 100644
--- a/sky/provision/aws/instance.py
+++ b/sky/provision/aws/instance.py
@@ -5,7 +5,6 @@
 
 from sky import status_lib
 from sky.adaptors import aws
-from sky.utils import common_utils
 
 BOTO_MAX_RETRIES = 12
 # Tag uniquely identifying all nodes of a cluster
@@ -163,10 +162,7 @@ def cleanup_ports(
         'ec2',
         region_name=region,
         config=config.Config(retries={'max_attempts': BOTO_MAX_RETRIES}))
-    # TODO(tian): Add a function to generate SG name for AWS, then replace here
-    # and backend_utils::write_cluster_config
-    sg_name = (f'sky-sg-{common_utils.user_and_hostname_hash()}'
-               f'-{common_utils.truncate_and_hash_cluster_name(cluster_name)}')
+    sg_name = provider_config['security_group']['GroupName']
     # GroupNames will only filter SGs in the default VPC, so we need to use
     # Filters here. Ref:
     # https://boto3.amazonaws.com/v1/documentation/api/1.26.112/reference/services/ec2/service-resource/security_groups.html  # pylint: disable=line-too-long
diff --git a/sky/provision/gcp/instance.py b/sky/provision/gcp/instance.py
index e9215db3063..9d28b799367 100644
--- a/sky/provision/gcp/instance.py
+++ b/sky/provision/gcp/instance.py
@@ -180,8 +180,7 @@ def cleanup_ports(
         # No new ports were opened, so there is nothing to clean up.
         return
     project_id = provider_config['project_id']
-    cluster_name_hash = common_utils.truncate_and_hash_cluster_name(
-        cluster_name)
+    cluster_name_hash = common_utils.make_cluster_name_on_cloud(cluster_name)
     for port in provider_config['ports']:
         rule_name = f'user-ports-{cluster_name_hash}-{port}'
         instance_utils.GCPComputeInstance.delete_firewall_rule(
diff --git a/sky/skylet/constants.py b/sky/skylet/constants.py
index 741199c30bd..6a66e1f98b6 100644
--- a/sky/skylet/constants.py
+++ b/sky/skylet/constants.py
@@ -50,9 +50,9 @@
 # Docker default options
 DEFAULT_DOCKER_CONTAINER_NAME = 'sky_container'
 DEFAULT_DOCKER_PORT = 10022
-DOCKER_USERNAME_ENV_KEY = 'SKYPILOT_DOCKER_USERNAME'
-DOCKER_PASSWORD_ENV_KEY = 'SKYPILOT_DOCKER_PASSWORD'
-DOCKER_SERVER_ENV_KEY = 'SKYPILOT_DOCKER_SERVER'
+DOCKER_USERNAME_ENV_VAR = 'SKYPILOT_DOCKER_USERNAME'
+DOCKER_PASSWORD_ENV_VAR = 'SKYPILOT_DOCKER_PASSWORD'
+DOCKER_SERVER_ENV_VAR = 'SKYPILOT_DOCKER_SERVER'
 
 # Install conda on the remote cluster if it is not already installed.
 # We do not install the latest conda with python 3.11 because ray has not
@@ -70,3 +70,7 @@
     # which is the case for VMs created on GCP, and running `conda init` will
     # cause error and waiting for the error to be reported: #2273.
     'which conda | grep /opt/conda || conda init > /dev/null;')
+
+USER_ID_ENV_VAR = 'SKYPILOT_USER_ID'
+
+CLUSTER_NAME_VALID_REGEX = '[a-z]([-a-z0-9]*[a-z0-9])?'
diff --git a/sky/skylet/providers/command_runner.py b/sky/skylet/providers/command_runner.py
index a9596c78487..894c7d91555 100644
--- a/sky/skylet/providers/command_runner.py
+++ b/sky/skylet/providers/command_runner.py
@@ -22,9 +22,9 @@ class DockerLoginConfig:
     @classmethod
     def from_env_vars(cls, d: Dict[str, str]) -> 'DockerLoginConfig':
         return cls(
-            username=d[constants.DOCKER_USERNAME_ENV_KEY],
-            password=d[constants.DOCKER_PASSWORD_ENV_KEY],
-            server=d[constants.DOCKER_SERVER_ENV_KEY],
+            username=d[constants.DOCKER_USERNAME_ENV_VAR],
+            password=d[constants.DOCKER_PASSWORD_ENV_VAR],
+            server=d[constants.DOCKER_SERVER_ENV_VAR],
         )
 
 
diff --git a/sky/skylet/providers/gcp/config.py b/sky/skylet/providers/gcp/config.py
index 744a689da8c..0b80bf237d3 100644
--- a/sky/skylet/providers/gcp/config.py
+++ b/sky/skylet/providers/gcp/config.py
@@ -807,7 +807,7 @@ def get_usable_vpc(config):
     ports = config["provider"].get("ports", [])
     user_rules = []
     for port in ports:
-        cluster_name_hash = common_utils.truncate_and_hash_cluster_name(
+        cluster_name_hash = common_utils.make_cluster_name_on_cloud(
             config["cluster_name"]
         )
         name = f"user-ports-{cluster_name_hash}-{port}"
diff --git a/sky/spot/__init__.py b/sky/spot/__init__.py
index c262ccdd8f3..eab619dcee9 100644
--- a/sky/spot/__init__.py
+++ b/sky/spot/__init__.py
@@ -1,6 +1,7 @@
 """Modules for managed spot clusters."""
 import pathlib
 
+from sky.spot.constants import SPOT_CLUSTER_NAME_PREFIX_LENGTH
 from sky.spot.constants import SPOT_CONTROLLER_IDLE_MINUTES_TO_AUTOSTOP
 from sky.spot.constants import SPOT_CONTROLLER_TEMPLATE
 from sky.spot.constants import SPOT_CONTROLLER_YAML_PREFIX
diff --git a/sky/spot/constants.py b/sky/spot/constants.py
index 605330691d4..4213e86d525 100644
--- a/sky/spot/constants.py
+++ b/sky/spot/constants.py
@@ -19,3 +19,9 @@
 # for Azure, and n1-standard-8 (8 vCPUs, 32 GB) for GCP.
 # We use 50 GB disk size to reduce the cost.
 CONTROLLER_RESOURCES = {'disk_size': 50}
+
+# Max length of the cluster name for GCP is 35, the user hash to be attached is
+# 4+1 chars, and we assume the maximum length of the job id is 4+1, so the max
+# length of the cluster name prefix is 25 to avoid the cluster name being too
+# long and truncated twice during the cluster creation.
+SPOT_CLUSTER_NAME_PREFIX_LENGTH = 25
diff --git a/sky/spot/spot_utils.py b/sky/spot/spot_utils.py
index 8af33f273c1..a36dd15f054 100644
--- a/sky/spot/spot_utils.py
+++ b/sky/spot/spot_utils.py
@@ -22,6 +22,7 @@
 from sky.skylet import constants
 from sky.skylet import job_lib
 from sky.skylet.log_lib import run_bash_command_with_log
+from sky.spot import constants as spot_constants
 from sky.spot import spot_state
 from sky.utils import common_utils
 from sky.utils import log_utils
@@ -202,7 +203,14 @@ def callback_func(state: str):
 
 def generate_spot_cluster_name(task_name: str, job_id: int) -> str:
     """Generate spot cluster name."""
-    return f'{task_name}-{job_id}'
+    # Truncate the task name to 30 chars to avoid the cluster name being too
+    # long after appending the job id, which will cause another truncation in
+    # the underlying sky.launch, hiding the `job_id` in the cluster name.
+    cluster_name = common_utils.make_cluster_name_on_cloud(
+        task_name,
+        spot_constants.SPOT_CLUSTER_NAME_PREFIX_LENGTH,
+        add_user_hash=False)
+    return f'{cluster_name}-{job_id}'
 
 
 def cancel_jobs_by_id(job_ids: Optional[List[int]]) -> str:
diff --git a/sky/task.py b/sky/task.py
index 9f7451cafc1..995ecd2bcc1 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -114,9 +114,9 @@ def _with_docker_login_config(
     task_envs: Dict[str, str],
 ) -> 'resources_lib.Resources':
     all_keys = {
-        constants.DOCKER_USERNAME_ENV_KEY,
-        constants.DOCKER_PASSWORD_ENV_KEY,
-        constants.DOCKER_SERVER_ENV_KEY,
+        constants.DOCKER_USERNAME_ENV_VAR,
+        constants.DOCKER_PASSWORD_ENV_VAR,
+        constants.DOCKER_SERVER_ENV_VAR,
     }
     existing_keys = all_keys & set(task_envs.keys())
     if not existing_keys:
diff --git a/sky/templates/aws-ray.yml.j2 b/sky/templates/aws-ray.yml.j2
index f91da548bdd..35f4d328bac 100644
--- a/sky/templates/aws-ray.yml.j2
+++ b/sky/templates/aws-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head node.
 max_workers: {{num_nodes - 1}}
diff --git a/sky/templates/azure-ray.yml.j2 b/sky/templates/azure-ray.yml.j2
index f76160c1738..46f3a187f00 100644
--- a/sky/templates/azure-ray.yml.j2
+++ b/sky/templates/azure-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head node.
 max_workers: {{num_nodes - 1}}
@@ -92,6 +92,7 @@ available_node_types:
       node_config:
         tags:
           skypilot-user: {{ user }}
+          skypilot-cluster-name: {{ cluster_name }}
         azure_arm_parameters:
             adminUsername: skypilot:ssh_user
             publicKey: |
diff --git a/sky/templates/gcp-ray.yml.j2 b/sky/templates/gcp-ray.yml.j2
index 08bf8182eb8..da87635f579 100644
--- a/sky/templates/gcp-ray.yml.j2
+++ b/sky/templates/gcp-ray.yml.j2
@@ -1,5 +1,5 @@
 
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head node.
 max_workers: {{num_nodes - 1}}
@@ -63,6 +63,7 @@ available_node_types:
     node_config:
       labels:
         skypilot-user: {{ user }}
+        skypilot-cluster-name: {{ cluster_name }}
       {%- if specific_reservations %}
       reservationAffinity:
         consumeReservationType: SPECIFIC_RESERVATION
diff --git a/sky/templates/ibm-ray.yml.j2 b/sky/templates/ibm-ray.yml.j2
index 6309d1b805b..72686e81f61 100644
--- a/sky/templates/ibm-ray.yml.j2
+++ b/sky/templates/ibm-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head node.
 max_workers: {{num_nodes - 1}}
diff --git a/sky/templates/kubernetes-ray.yml.j2 b/sky/templates/kubernetes-ray.yml.j2
index 9a2e97f551d..e305ca5b2fe 100644
--- a/sky/templates/kubernetes-ray.yml.j2
+++ b/sky/templates/kubernetes-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head
 # node.
@@ -75,12 +75,12 @@ provider:
         metadata:
           labels:
             parent: skypilot
-            skypilot-cluster: {{cluster_name}}
-          name: {{cluster_name}}-ray-head-ssh
+            skypilot-cluster: {{cluster_name_on_cloud}}
+          name: {{cluster_name_on_cloud}}-ray-head-ssh
         spec:
           type: NodePort
           selector:
-            component: {{cluster_name}}-ray-head
+            component: {{cluster_name_on_cloud}}-ray-head
           ports:
             - protocol: TCP
               port: 22
@@ -91,15 +91,15 @@ provider:
         metadata:
             labels:
               parent: skypilot
-              skypilot-cluster: {{cluster_name}}
+              skypilot-cluster: {{cluster_name_on_cloud}}
             # NOTE: If you're running multiple Ray clusters with services
             # on one Kubernetes cluster, they must have unique service
             # names.
-            name: {{cluster_name}}-ray-head
+            name: {{cluster_name_on_cloud}}-ray-head
         spec:
             # This selector must match the head node pod's selector below.
             selector:
-                component: {{cluster_name}}-ray-head
+                component: {{cluster_name_on_cloud}}-ray-head
             ports:
                 - name: client
                   protocol: TCP
@@ -119,13 +119,13 @@ available_node_types:
       apiVersion: v1
       kind: Pod
       metadata:
-        name: {{cluster_name}}-ray-head
+        name: {{cluster_name_on_cloud}}-ray-head
         # Must match the head node service selector above if a head node
         # service is required.
         labels:
             parent: skypilot
-            component: {{cluster_name}}-ray-head
-            skypilot-cluster: {{cluster_name}}
+            component: {{cluster_name_on_cloud}}-ray-head
+            skypilot-cluster: {{cluster_name_on_cloud}}
       spec:
         # Change this if you altered the autoscaler_service_account above
         # or want to provide your own.
@@ -205,9 +205,9 @@ available_node_types:
       metadata:
         labels:
           parent: skypilot
-          skypilot-cluster: {{cluster_name}}
+          skypilot-cluster: {{cluster_name_on_cloud}}
         # Automatically generates a name for the pod with this prefix.
-        generateName: {{cluster_name}}-ray-worker-
+        generateName: {{cluster_name_on_cloud}}-ray-worker-
       spec:
         serviceAccountName: skypilot-service-account
         restartPolicy: Never
diff --git a/sky/templates/lambda-ray.yml.j2 b/sky/templates/lambda-ray.yml.j2
index 984ce3b7809..26cd827387f 100644
--- a/sky/templates/lambda-ray.yml.j2
+++ b/sky/templates/lambda-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head node.
 max_workers: {{num_nodes - 1}}
diff --git a/sky/templates/local-ray.yml.j2 b/sky/templates/local-ray.yml.j2
index e89712c2169..c2f5d09345f 100644
--- a/sky/templates/local-ray.yml.j2
+++ b/sky/templates/local-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 provider:
   type: local
diff --git a/sky/templates/oci-ray.yml.j2 b/sky/templates/oci-ray.yml.j2
index e345594405f..cad00325d14 100644
--- a/sky/templates/oci-ray.yml.j2
+++ b/sky/templates/oci-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head node.
 max_workers: {{num_nodes - 1}}
diff --git a/sky/templates/scp-ray.yml.j2 b/sky/templates/scp-ray.yml.j2
index 75259a2e316..48089c08547 100644
--- a/sky/templates/scp-ray.yml.j2
+++ b/sky/templates/scp-ray.yml.j2
@@ -1,4 +1,4 @@
-cluster_name: {{cluster_name}}
+cluster_name: {{cluster_name_on_cloud}}
 
 # The maximum number of workers nodes to launch in addition to the head node.
 max_workers: {{num_nodes - 1}}
diff --git a/sky/templates/spot-controller.yaml.j2 b/sky/templates/spot-controller.yaml.j2
index 872e43b0d7a..7113fe2f265 100644
--- a/sky/templates/spot-controller.yaml.j2
+++ b/sky/templates/spot-controller.yaml.j2
@@ -38,7 +38,7 @@ run: |
     --job-id $SKYPILOT_INTERNAL_JOB_ID {% if retry_until_up %}--retry-until-up{% endif %}
 
 envs:
-  SKYPILOT_USAGE_USER_ID: {{logging_user_hash}}
+  SKYPILOT_USER_ID: {{logging_user_hash}}
   # skip cloud identity check for spot controller to avoid the overhead.
   SKYPILOT_SKIP_CLOUD_IDENTITY_CHECK: 1
   SKYPILOT_USER: {{user}}
diff --git a/sky/usage/constants.py b/sky/usage/constants.py
index 465c7abda45..c2f8c6d067b 100644
--- a/sky/usage/constants.py
+++ b/sky/usage/constants.py
@@ -15,5 +15,3 @@
 
 USAGE_MESSAGE_REDACT_KEYS = ['setup', 'run', 'envs']
 USAGE_MESSAGE_REDACT_TYPES = {str, dict}
-
-USAGE_USER_ENV = 'SKYPILOT_USAGE_USER_ID'
diff --git a/sky/usage/usage_lib.py b/sky/usage/usage_lib.py
index bd488399d0e..9e4207ecef0 100644
--- a/sky/usage/usage_lib.py
+++ b/sky/usage/usage_lib.py
@@ -32,12 +32,6 @@ def _get_current_timestamp_ns() -> int:
     return int(datetime.datetime.now(datetime.timezone.utc).timestamp() * 1e9)
 
 
-def _get_user_hash():
-    """Returns a unique user-machine specific hash as a user id for logging."""
-    user_id = os.getenv(constants.USAGE_USER_ENV)
-    return common_utils.get_user_hash(default_value=user_id)
-
-
 class MessageType(enum.Enum):
     """Types for messages to be sent to Loki."""
     USAGE = 'usage'
@@ -74,7 +68,7 @@ class UsageMessageToReport(MessageToReport):
     def __init__(self) -> None:
         super().__init__(constants.USAGE_MESSAGE_SCHEMA_VERSION)
         # Message identifier.
-        self.user: str = _get_user_hash()
+        self.user: str = common_utils.get_user_hash()
         self.run_id: str = common_utils.get_usage_run_id()
         self.sky_version: str = sky.__version__
 
diff --git a/sky/utils/common_utils.py b/sky/utils/common_utils.py
index 334ca84b2aa..e146fcb7669 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -19,11 +19,16 @@
 import yaml
 
 from sky import sky_logging
+from sky.skylet import constants
 
 _USER_HASH_FILE = os.path.expanduser('~/.sky/user_hash')
 USER_HASH_LENGTH = 8
+USER_HASH_LENGTH_IN_CLUSTER_NAME = 4
 
-CLUSTER_NAME_HASH_LENGTH = 4
+# We are using base36 to reduce the length of the hash. 2 chars -> 36^2 = 1296
+# possibilities. considering the final cluster name contains the prefix as well,
+# we should be fine with 2 chars.
+CLUSTER_NAME_HASH_LENGTH = 2
 
 _COLOR_PATTERN = re.compile(r'\x1b[^m]*m')
 
@@ -48,7 +53,7 @@ def get_usage_run_id() -> str:
     return _usage_run_id
 
 
-def get_user_hash(default_value: Optional[str] = None) -> str:
+def get_user_hash() -> str:
     """Returns a unique user-machine specific hash as a user id.
 
     We cache the user hash in a file to avoid potential user_name or
@@ -64,7 +69,7 @@ def _is_valid_user_hash(user_hash: Optional[str]) -> bool:
             return False
         return len(user_hash) == USER_HASH_LENGTH
 
-    user_hash = default_value
+    user_hash = os.getenv(constants.USER_ID_ENV_VAR)
     if _is_valid_user_hash(user_hash):
         assert user_hash is not None
         return user_hash
@@ -88,11 +93,67 @@ def _is_valid_user_hash(user_hash: Optional[str]) -> bool:
     return user_hash
 
 
-def truncate_and_hash_cluster_name(cluster_name: str) -> str:
-    if len(cluster_name) < 15:
-        return cluster_name
-    return cluster_name[:10] + hashlib.md5(
-        cluster_name.encode()).hexdigest()[:CLUSTER_NAME_HASH_LENGTH]
+def base36_encode(hex_str: str) -> str:
+    """Converts a hex string to a base36 string."""
+    int_value = int(hex_str, 16)
+
+    def _base36_encode(num: int) -> str:
+        if num == 0:
+            return '0'
+        alphabet = '0123456789abcdefghijklmnopqrstuvwxyz'
+        base36 = ''
+        while num != 0:
+            num, i = divmod(num, 36)
+            base36 = alphabet[i] + base36
+        return base36
+
+    return _base36_encode(int_value)
+
+
+def make_cluster_name_on_cloud(cluster_name: str,
+                               max_length: Optional[int] = 15,
+                               add_user_hash: bool = True) -> str:
+    """Generate valid cluster name on cloud that is unique to the user.
+
+    This is to map the cluster name to a valid length for cloud providers, e.g.
+    GCP limits the length of the cluster name to 35 characters. If the cluster
+    name with user hash is longer than max_length:
+      1. Truncate it to max_length - cluster_hash - user_hash_length.
+      2. Append the hash of the cluster name
+
+    Args:
+        cluster_name: The cluster name to be truncated and hashed.
+        max_length: The maximum length of the cluster name. If None, no
+            truncation is performed.
+        add_user_hash: Whether to append user hash to the cluster name.
+    """
+    user_hash = ''
+    if add_user_hash:
+        user_hash = get_user_hash()[:USER_HASH_LENGTH_IN_CLUSTER_NAME]
+        user_hash = f'-{user_hash}'
+    user_hash_length = len(user_hash)
+
+    if (max_length is None or
+            len(cluster_name) <= max_length - user_hash_length):
+        return f'{cluster_name}{user_hash}'
+    # -1 is for the dash between cluster name and cluster name hash.
+    truncate_cluster_name_length = (max_length - CLUSTER_NAME_HASH_LENGTH - 1 -
+                                    user_hash_length)
+    truncate_cluster_name = cluster_name[:truncate_cluster_name_length]
+    if truncate_cluster_name.endswith('-'):
+        truncate_cluster_name = truncate_cluster_name.rstrip('-')
+    assert truncate_cluster_name_length > 0, (cluster_name, max_length)
+    cluster_name_hash = hashlib.md5(cluster_name.encode()).hexdigest()
+    # Use base36 to reduce the length of the hash.
+    cluster_name_hash = base36_encode(cluster_name_hash)
+    return (f'{truncate_cluster_name}'
+            f'-{cluster_name_hash[:CLUSTER_NAME_HASH_LENGTH]}{user_hash}')
+
+
+def cluster_name_in_hint(cluster_name: str, cluster_name_on_cloud: str) -> str:
+    if cluster_name_on_cloud.startswith(cluster_name):
+        return repr(cluster_name)
+    return f'{cluster_name!r} (name on cloud: {cluster_name_on_cloud!r})'
 
 
 def get_global_job_id(job_timestamp: str,
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index 97031454842..a1be762506d 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -31,6 +31,7 @@ def _mock_cluster_state(self, _mock_db_conn, enable_all_clouds):
 
         handle = backends.CloudVmRayResourceHandle(
             cluster_name='test-cluster1',
+            cluster_name_on_cloud='test-cluster1',
             cluster_yaml='/tmp/cluster1.yaml',
             launched_nodes=2,
             launched_resources=sky.Resources(sky.AWS(),
@@ -45,6 +46,7 @@ def _mock_cluster_state(self, _mock_db_conn, enable_all_clouds):
             ready=True)
         handle = backends.CloudVmRayResourceHandle(
             cluster_name='test-cluster2',
+            cluster_name_on_cloud='test-cluster2',
             cluster_yaml='/tmp/cluster2.yaml',
             launched_nodes=1,
             launched_resources=sky.Resources(sky.GCP(),
@@ -60,6 +62,7 @@ def _mock_cluster_state(self, _mock_db_conn, enable_all_clouds):
             ready=True)
         handle = backends.CloudVmRayResourceHandle(
             cluster_name='test-cluster3',
+            cluster_name_on_cloud='test-cluster3',
             cluster_yaml='/tmp/cluster3.yaml',
             launched_nodes=1,
             launched_resources=sky.Resources(sky.Azure(),
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 25eb5181abb..d60616aa460 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -22,7 +22,6 @@
 # Change cloud for generic tests to aws
 # > pytest tests/test_smoke.py --generic-cloud aws
 
-import hashlib
 import inspect
 import os
 import pathlib
@@ -34,7 +33,6 @@
 from typing import Dict, List, NamedTuple, Optional, Tuple
 import urllib.parse
 import uuid
-import warnings
 
 import colorama
 import jinja2
@@ -42,6 +40,7 @@
 
 import sky
 from sky import global_user_state
+from sky import spot
 from sky.adaptors import cloudflare
 from sky.adaptors import ibm
 from sky.clouds import AWS
@@ -54,11 +53,6 @@
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
 
-# For uniquefying users on shared-account cloud providers. Used as part of the
-# cluster names.
-_smoke_test_hash = hashlib.md5(
-    common_utils.user_and_hostname_hash().encode()).hexdigest()[:4]
-
 # To avoid the second smoke test reusing the cluster launched in the first
 # smoke test. Also required for test_spot_recovery to make sure the manual
 # termination with aws ec2 does not accidentally terminate other spot clusters
@@ -124,10 +118,8 @@ def _get_cluster_name() -> str:
     """
     caller_func_name = inspect.stack()[1][3]
     test_name = caller_func_name.replace('_', '-').replace('test-', 't-')
-    if len(test_name) > 16:
-        test_name = test_name[:16] + hashlib.md5(
-            test_name.encode()).hexdigest()[:3]
-    return f'{test_name}-{_smoke_test_hash}-{test_id}'
+    test_name = common_utils.make_cluster_name_on_cloud(test_name, 24)
+    return f'{test_name}-{test_id}'
 
 
 def run_one_test(test: Test) -> Tuple[int, str, str]:
@@ -662,6 +654,7 @@ def test_stale_job(generic_cloud: str):
 @pytest.mark.aws
 def test_aws_stale_job_manual_restart():
     name = _get_cluster_name()
+    name_on_cloud = sky.AWS.truncate_and_hash_cluster_name(name)
     region = 'us-west-2'
     test = Test(
         'aws_stale_job_manual_restart',
@@ -670,7 +663,7 @@ def test_aws_stale_job_manual_restart():
             f'sky exec {name} -d "echo start; sleep 10000"',
             # Stop the cluster manually.
             f'id=`aws ec2 describe-instances --region {region} --filters '
-            f'Name=tag:ray-cluster-name,Values={name} '
+            f'Name=tag:ray-cluster-name,Values={name_on_cloud} '
             f'--query Reservations[].Instances[].InstanceId '
             '--output text`; '
             f'aws ec2 stop-instances --region {region} '
@@ -691,9 +684,10 @@ def test_aws_stale_job_manual_restart():
 @pytest.mark.gcp
 def test_gcp_stale_job_manual_restart():
     name = _get_cluster_name()
+    name_on_cloud = sky.GCP.truncate_and_hash_cluster_name(name)
     zone = 'us-west2-a'
     query_cmd = (f'gcloud compute instances list --filter='
-                 f'"(labels.ray-cluster-name={name})" '
+                 f'"(labels.ray-cluster-name={name_on_cloud})" '
                  f'--zones={zone} --format="value(name)"')
     stop_cmd = (f'gcloud compute instances stop --zone={zone}'
                 f' --quiet $({query_cmd})')
@@ -1983,6 +1977,8 @@ def test_spot_pipeline_failed_setup(generic_cloud: str):
 def test_spot_recovery_aws(aws_config_region):
     """Test managed spot recovery."""
     name = _get_cluster_name()
+    name_on_cloud = common_utils.truncate_and_hash_cluster_name(
+        name, spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
     region = aws_config_region
     test = Test(
         'spot_recovery_aws',
@@ -1994,7 +1990,7 @@ def test_spot_recovery_aws(aws_config_region):
             # Terminate the cluster manually.
             (f'aws ec2 terminate-instances --region {region} --instance-ids $('
              f'aws ec2 describe-instances --region {region} '
-             f'--filters Name=tag:ray-cluster-name,Values={name}* '
+             f'--filters Name=tag:ray-cluster-name,Values={name_on_cloud}* '
              f'--query Reservations[].Instances[].InstanceId '
              '--output text)'),
             'sleep 100',
@@ -2014,16 +2010,20 @@ def test_spot_recovery_aws(aws_config_region):
 def test_spot_recovery_gcp():
     """Test managed spot recovery."""
     name = _get_cluster_name()
+    name_on_cloud = common_utils.truncate_and_hash_cluster_name(
+        name, spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
     zone = 'us-east4-b'
-    query_cmd = (f'gcloud compute instances list --filter='
-                 f'"(labels.ray-cluster-name:{name})" '
-                 f'--zones={zone} --format="value(name)"')
+    query_cmd = (
+        f'gcloud compute instances list --filter='
+        # `:` means prefix match.
+        f'"(labels.ray-cluster-name:{name_on_cloud})" '
+        f'--zones={zone} --format="value(name)"')
     terminate_cmd = (f'gcloud compute instances delete --zone={zone}'
                      f' --quiet $({query_cmd})')
     test = Test(
         'spot_recovery_gcp',
         [
-            f'sky spot launch --cloud gcp --zone {zone} -n {name} "echo SKYPILOT_TASK_ID: \$SKYPILOT_TASK_ID; sleep 1800"  -y -d',
+            f'sky spot launch --cloud gcp --zone {zone} -n {name} --cpus 2 "echo SKYPILOT_TASK_ID: \$SKYPILOT_TASK_ID; sleep 1800"  -y -d',
             'sleep 360',
             f'{_SPOT_QUEUE_WAIT}| grep {name} | head -n1 | grep "RUNNING"',
             f'RUN_ID=$(sky spot logs -n {name} --no-follow | grep SKYPILOT_TASK_ID | cut -d: -f2); echo "$RUN_ID" | tee /tmp/{name}-run-id',
@@ -2046,6 +2046,8 @@ def test_spot_recovery_gcp():
 def test_spot_pipeline_recovery_aws(aws_config_region):
     """Test managed spot recovery for a pipeline."""
     name = _get_cluster_name()
+    user_hash = common_utils.get_user_hash()
+    user_hash = user_hash[:common_utils.USER_HASH_LENGTH_IN_CLUSTER_NAME]
     region = aws_config_region
     if region != 'us-west-2':
         pytest.skip('Only run spot pipeline recovery test in us-west-2')
@@ -2061,13 +2063,16 @@ def test_spot_pipeline_recovery_aws(aws_config_region):
             # The `cat ...| rev` is to retrieve the job_id from the
             # SKYPILOT_TASK_ID, which gets the second to last field
             # separated by `-`.
-            (f'SPOT_JOB_ID=`cat /tmp/{name}-run-id | rev | '
-             'cut -d\'-\' -f2 | rev`;'
-             f'aws ec2 terminate-instances --region {region} --instance-ids $('
-             f'aws ec2 describe-instances --region {region} '
-             '--filters Name=tag:ray-cluster-name,Values=*-${SPOT_JOB_ID} '
-             f'--query Reservations[].Instances[].InstanceId '
-             '--output text)'),
+            (
+                f'SPOT_JOB_ID=`cat /tmp/{name}-run-id | rev | '
+                'cut -d\'-\' -f2 | rev`;'
+                f'aws ec2 terminate-instances --region {region} --instance-ids $('
+                f'aws ec2 describe-instances --region {region} '
+                # TODO(zhwu): fix the name for spot cluster.
+                '--filters Name=tag:ray-cluster-name,Values=*-${SPOT_JOB_ID}'
+                f'-{user_hash} '
+                f'--query Reservations[].Instances[].InstanceId '
+                '--output text)'),
             'sleep 100',
             f'{_SPOT_QUEUE_WAIT}| grep {name} | head -n1 | grep "RECOVERING"',
             'sleep 200',
@@ -2089,8 +2094,10 @@ def test_spot_pipeline_recovery_gcp():
     """Test managed spot recovery for a pipeline."""
     name = _get_cluster_name()
     zone = 'us-east4-b'
+    user_hash = common_utils.get_user_hash()
+    user_hash = user_hash[:common_utils.USER_HASH_LENGTH_IN_CLUSTER_NAME]
     query_cmd = ('gcloud compute instances list --filter='
-                 '"(labels.ray-cluster-name:*-${SPOT_JOB_ID})" '
+                 f'"(labels.ray-cluster-name:*-${{SPOT_JOB_ID}}-{user_hash})" '
                  f'--zones={zone} --format="value(name)"')
     terminate_cmd = (f'gcloud compute instances delete --zone={zone}'
                      f' --quiet $({query_cmd})')
@@ -2149,6 +2156,8 @@ def test_spot_recovery_default_resources(generic_cloud: str):
 def test_spot_recovery_multi_node_aws(aws_config_region):
     """Test managed spot recovery."""
     name = _get_cluster_name()
+    name_on_cloud = common_utils.make_cluster_name_on_cloud(
+        name, spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
     region = aws_config_region
     test = Test(
         'spot_recovery_multi_node_aws',
@@ -2160,7 +2169,7 @@ def test_spot_recovery_multi_node_aws(aws_config_region):
             # Terminate the worker manually.
             (f'aws ec2 terminate-instances --region {region} --instance-ids $('
              f'aws ec2 describe-instances --region {region} '
-             f'--filters Name=tag:ray-cluster-name,Values={name}* '
+             f'--filters Name=tag:ray-cluster-name,Values={name_on_cloud}* '
              'Name=tag:ray-node-type,Values=worker '
              f'--query Reservations[].Instances[].InstanceId '
              '--output text)'),
@@ -2181,12 +2190,14 @@ def test_spot_recovery_multi_node_aws(aws_config_region):
 def test_spot_recovery_multi_node_gcp():
     """Test managed spot recovery."""
     name = _get_cluster_name()
+    name_on_cloud = common_utils.truncate_and_hash_cluster_name(
+        name, spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
     zone = 'us-west2-a'
     # Use ':' to match as the cluster name will contain the suffix with job id
     query_cmd = (
         f'gcloud compute instances list --filter='
-        f'"(labels.ray-cluster-name:{name} AND labels.ray-node-type=worker)" '
-        f'--zones={zone} --format="value(name)"')
+        f'"(labels.ray-cluster-name:{name_on_cloud} AND '
+        f'labels.ray-node-type=worker)" --zones={zone} --format="value(name)"')
     terminate_cmd = (f'gcloud compute instances delete --zone={zone}'
                      f' --quiet $({query_cmd})')
     test = Test(
@@ -2214,6 +2225,12 @@ def test_spot_recovery_multi_node_gcp():
 @pytest.mark.managed_spot
 def test_spot_cancellation_aws(aws_config_region):
     name = _get_cluster_name()
+    name_on_cloud = common_utils.truncate_and_hash_cluster_name(
+        name, spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
+    name_2_on_cloud = common_utils.truncate_and_hash_cluster_name(
+        f'{name}-2', spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
+    name_3_on_cloud = common_utils.truncate_and_hash_cluster_name(
+        f'{name}-3', spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
     region = aws_config_region
     test = Test(
         'spot_cancellation_aws',
@@ -2228,7 +2245,7 @@ def test_spot_cancellation_aws(aws_config_region):
             'sleep 120',
             f'{_SPOT_QUEUE_WAIT}| grep {name} | head -n1 | grep "CANCELLED"',
             (f's=$(aws ec2 describe-instances --region {region} '
-             f'--filters Name=tag:ray-cluster-name,Values={name}-* '
+             f'--filters Name=tag:ray-cluster-name,Values={name_on_cloud}-* '
              f'--query Reservations[].Instances[].State[].Name '
              '--output text) && echo "$s" && echo; [[ -z "$s" ]] || [[ "$s" = "terminated" ]] || [[ "$s" = "shutting-down" ]]'
             ),
@@ -2241,7 +2258,7 @@ def test_spot_cancellation_aws(aws_config_region):
             'sleep 120',
             f'{_SPOT_QUEUE_WAIT}| grep {name}-2 | head -n1 | grep "CANCELLED"',
             (f's=$(aws ec2 describe-instances --region {region} '
-             f'--filters Name=tag:ray-cluster-name,Values={name}-2-* '
+             f'--filters Name=tag:ray-cluster-name,Values={name_2_on_cloud}-* '
              f'--query Reservations[].Instances[].State[].Name '
              '--output text) && echo "$s" && echo; [[ -z "$s" ]] || [[ "$s" = "terminated" ]] || [[ "$s" = "shutting-down" ]]'
             ),
@@ -2252,7 +2269,7 @@ def test_spot_cancellation_aws(aws_config_region):
             # Terminate the cluster manually.
             (f'aws ec2 terminate-instances --region {region} --instance-ids $('
              f'aws ec2 describe-instances --region {region} '
-             f'--filters Name=tag:ray-cluster-name,Values={name}-3-* '
+             f'--filters Name=tag:ray-cluster-name,Values={name_3_on_cloud}-* '
              f'--query Reservations[].Instances[].InstanceId '
              '--output text)'),
             'sleep 120',
@@ -2265,7 +2282,7 @@ def test_spot_cancellation_aws(aws_config_region):
             # The cluster should be terminated (shutting-down) after cancellation. We don't use the `=` operator here because
             # there can be multiple VM with the same name due to the recovery.
             (f's=$(aws ec2 describe-instances --region {region} '
-             f'--filters Name=tag:ray-cluster-name,Values={name}-3-* '
+             f'--filters Name=tag:ray-cluster-name,Values={name_3_on_cloud}-* '
              f'--query Reservations[].Instances[].State[].Name '
              '--output text) && echo "$s" && echo; [[ -z "$s" ]] || echo "$s" | grep -v -E "pending|running|stopped|stopping"'
             ),
@@ -2278,12 +2295,14 @@ def test_spot_cancellation_aws(aws_config_region):
 @pytest.mark.managed_spot
 def test_spot_cancellation_gcp():
     name = _get_cluster_name()
+    name_on_cloud = common_utils.truncate_and_hash_cluster_name(
+        name, spot.SPOT_CLUSTER_NAME_PREFIX_LENGTH, add_user_hash=False)
     zone = 'us-west3-b'
     query_state_cmd = ('gcloud compute instances list '
-                       f'--filter="(labels.ray-cluster-name:{name})" '
+                       f'--filter="(labels.ray-cluster-name:{name_on_cloud})" '
                        '--format="value(status)"')
     query_cmd = (f'gcloud compute instances list --filter='
-                 f'"(labels.ray-cluster-name:{name})" '
+                 f'"(labels.ray-cluster-name:{name_on_cloud})" '
                  f'--zones={zone} --format="value(name)"')
     terminate_cmd = (f'gcloud compute instances delete --zone={zone}'
                      f' --quiet $({query_cmd})')
@@ -2490,6 +2509,7 @@ def _get_aws_query_command(region, instance_id, field, expected):
     for disk_tier in ['low', 'medium', 'high']:
         specs = AWS._get_disk_specs(disk_tier)
         name = _get_cluster_name() + '-' + disk_tier
+        name_on_cloud = sky.AWS.truncate_and_hash_cluster_name(name)
         region = 'us-west-2'
         test = Test(
             'aws-disk-tier',
@@ -2497,7 +2517,7 @@ def _get_aws_query_command(region, instance_id, field, expected):
                 f'sky launch -y -c {name} --cloud aws --region {region} '
                 f'--disk-tier {disk_tier} echo "hello sky"',
                 f'id=`aws ec2 describe-instances --region {region} --filters '
-                f'Name=tag:ray-cluster-name,Values={name} --query '
+                f'Name=tag:ray-cluster-name,Values={name_on_cloud} --query '
                 f'Reservations[].Instances[].InstanceId --output text`; ' +
                 _get_aws_query_command(region, '$id', 'VolumeType',
                                        specs['disk_tier']) +
@@ -2518,6 +2538,7 @@ def test_gcp_disk_tier():
     for disk_tier in ['low', 'medium', 'high']:
         type = GCP._get_disk_type(disk_tier)
         name = _get_cluster_name() + '-' + disk_tier
+        name_on_cloud = sky.GCP.truncate_and_hash_cluster_name(name)
         region = 'us-west2'
         test = Test(
             'gcp-disk-tier',
@@ -2525,7 +2546,8 @@ def test_gcp_disk_tier():
                 f'sky launch -y -c {name} --cloud gcp --region {region} '
                 f'--disk-tier {disk_tier} echo "hello sky"',
                 f'name=`gcloud compute instances list --filter='
-                f'"labels.ray-cluster-name:{name}" --format="value(name)"`; '
+                f'"labels.ray-cluster-name:{name_on_cloud}" '
+                '--format="value(name)"`; '
                 f'gcloud compute disks list --filter="name=$name" '
                 f'--format="value(type)" | grep {type} '
             ],
@@ -2540,13 +2562,14 @@ def test_azure_disk_tier():
     for disk_tier in ['low', 'medium']:
         type = Azure._get_disk_type(disk_tier)
         name = _get_cluster_name() + '-' + disk_tier
+        name_on_cloud = sky.Azure.truncate_and_hash_cluster_name(name)
         region = 'westus2'
         test = Test(
             'azure-disk-tier',
             [
                 f'sky launch -y -c {name} --cloud azure --region {region} '
                 f'--disk-tier {disk_tier} echo "hello sky"',
-                f'az resource list --tag ray-cluster-name={name} --query '
+                f'az resource list --tag ray-cluster-name={name_on_cloud} --query '
                 f'"[?type==\'Microsoft.Compute/disks\'].sku.name" '
                 f'--output tsv | grep {type}'
             ],
diff --git a/tests/test_spot.py b/tests/test_spot.py
index d0f36dc52a0..e5bb5317012 100644
--- a/tests/test_spot.py
+++ b/tests/test_spot.py
@@ -45,6 +45,7 @@ def _mock_cluster_state(self, _mock_db_conn):
         assert 'state.db' not in global_user_state._DB.db_path
         handle = backends.CloudVmRayResourceHandle(
             cluster_name='test-cluster1',
+            cluster_name_on_cloud='test-cluster1',
             cluster_yaml='/tmp/cluster1.yaml',
             launched_nodes=2,
             launched_resources=sky.Resources(sky.AWS(),
@@ -58,6 +59,7 @@ def _mock_cluster_state(self, _mock_db_conn):
             ready=True)
         handle = backends.CloudVmRayResourceHandle(
             cluster_name='test-cluster2',
+            cluster_name_on_cloud='test-cluster2',
             cluster_yaml='/tmp/cluster2.yaml',
             launched_nodes=1,
             launched_resources=sky.Resources(sky.GCP(),
@@ -72,6 +74,7 @@ def _mock_cluster_state(self, _mock_db_conn):
             ready=True)
         handle = backends.CloudVmRayResourceHandle(
             cluster_name='test-cluster3',
+            cluster_name_on_cloud='test-cluster3',
             cluster_yaml='/tmp/cluster3.yaml',
             launched_nodes=4,
             launched_resources=sky.Resources(sky.Azure(),
@@ -85,6 +88,7 @@ def _mock_cluster_state(self, _mock_db_conn):
             ready=False)
         handle = backends.CloudVmRayResourceHandle(
             cluster_name=spot.SPOT_CONTROLLER_NAME,
+            cluster_name_on_cloud=spot.SPOT_CONTROLLER_NAME,
             cluster_yaml='/tmp/spot_controller.yaml',
             launched_nodes=1,
             launched_resources=sky.Resources(sky.AWS(),


Code review comments:
path: sky/templates/aws-ray.yml.j2
text: @user1:
tag head/workers?

@author:
Removed.

diff_hunk: @@ -63,6 +63,7 @@ available_node_types:
     node_config:
       labels:
         skypilot-user: {{ user }}
+        skypilot-cluster-name: {{ cluster_name }}
path: sky/templates/gcp-ray.yml.j2
text: @user1:
Add to workers too?

@user1:
Just tried launching `sky launch --cpus 2+ -c dbg --use-spot --cloud gcp`. The VM has an empty value for this key.

@author:
Oops, this is a remnant, we decided to remove it from the tag, as the tag has the same character length limit