Check long cluster names only for GCP; robustify smoke tests. (skypilot-org#966)

concretevitamin · web-flow · commit 3c3182545177 · 2022-07-22T22:09:57.000-07:00
* Check long cluster names only for GCP; robustify smoke tests.

* test_id comment

* Recheck cluster name after cross-cloud retry.

* UX: sky spot cancel -a

* pylint

* Shorten (1) smoke test cluster names (2) gcp cluster name limit.

* Revert back timeout

* Fix merge error

* Make smoke test cluster names a hash; pass into py scripts.

* Remove redundant name gen
diff --git a/examples/multi_echo.py b/examples/multi_echo.py
@@ -2,32 +2,44 @@
 import hashlib
 from multiprocessing import pool
 import socket
+import sys
+from typing import Optional
 
 import sky
 
-# (username, last 4 chars of hash of hostname): for uniquefying users on
-# shared-account cloud providers.
-hostname_hash = hashlib.md5(socket.gethostname().encode()).hexdigest()[-4:]
-_user_and_host = f'{getpass.getuser()}-{hostname_hash}'
-cluster = f'test-multi-echo-{_user_and_host}'
 
-# Create the cluster.
-with sky.Dag() as dag:
-    cluster_resources = sky.Resources(sky.AWS(), accelerators={'K80': 1})
-    task = sky.Task(num_nodes=2).set_resources(cluster_resources)
-# `detach_run` will only detach the `run` command. The provision and `setup` are
-# still blocking.
-sky.launch(dag, cluster_name=cluster, detach_run=True)
+def run(cluster: Optional[str] = None):
+    if cluster is None:
+        # (username, last 4 chars of hash of hostname): for uniquefying users on
+        # shared-account cloud providers.
+        hostname_hash = hashlib.md5(
+            socket.gethostname().encode()).hexdigest()[-4:]
+        _user_and_host = f'{getpass.getuser()}-{hostname_hash}'
+        cluster = f'test-multi-echo-{_user_and_host}'
 
-
-# Submit multiple tasks in parallel to trigger queueing behaviors.
-def _exec(i):
+    # Create the cluster.
     with sky.Dag() as dag:
-        task = sky.Task(run=f'echo {i}; sleep 5')
-        resources = sky.Resources(accelerators={'K80': 0.5})
-        task.set_resources(resources)
-    sky.exec(dag, cluster_name=cluster, detach_run=True)
+        cluster_resources = sky.Resources(sky.AWS(), accelerators={'K80': 1})
+        task = sky.Task(num_nodes=2).set_resources(cluster_resources)
+    # `detach_run` will only detach the `run` command. The provision and
+    # `setup` are still blocking.
+    sky.launch(dag, cluster_name=cluster, detach_run=True)
+
+    # Submit multiple tasks in parallel to trigger queueing behaviors.
+    def _exec(i):
+        with sky.Dag() as dag:
+            task = sky.Task(run=f'echo {i}; sleep 5')
+            resources = sky.Resources(accelerators={'K80': 0.5})
+            task.set_resources(resources)
+        sky.exec(dag, cluster_name=cluster, detach_run=True)
+
+    with pool.ThreadPool(8) as p:
+        list(p.imap(_exec, range(32)))
 
 
-with pool.ThreadPool(8) as p:
-    list(p.imap(_exec, range(32)))
+if __name__ == '__main__':
+    cluster = None
+    if len(sys.argv) > 1:
+        # For smoke test passing in a cluster name.
+        cluster = sys.argv[1]
+    run(cluster)
diff --git a/examples/resnet_distributed_tf_app.py b/examples/resnet_distributed_tf_app.py
@@ -1,76 +1,88 @@
 import getpass
 import hashlib
 import socket
-import subprocess
+import sys
 from typing import List, Optional
 
 import sky
 
-# (username, last 4 chars of hash of hostname): for uniquefying users on
-# shared-account cloud providers.
-hostname_hash = hashlib.md5(socket.gethostname().encode()).hexdigest()[-4:]
-_user_and_host = f'{getpass.getuser()}-{hostname_hash}'
-cluster = f'test-distributed-tf-{_user_and_host}'
 
-with sky.Dag() as dag:
-    # Total Nodes, INCLUDING Head Node
-    num_nodes = 2
+def run(cluster: Optional[str] = None):
+    if cluster is None:
+        # (username, last 4 chars of hash of hostname): for uniquefying users on
+        # shared-account cloud providers.
+        hostname_hash = hashlib.md5(
+            socket.gethostname().encode()).hexdigest()[-4:]
+        _user_and_host = f'{getpass.getuser()}-{hostname_hash}'
+        cluster = f'test-distributed-tf-{_user_and_host}'
 
-    # The setup command.  Will be run under the working directory.
-    setup = """
-            git clone https://github.com/concretevitamin/tpu || true
-            cd tpu && git checkout 9459fee
-            conda create -n resnet python=3.7 -y
-            conda activate resnet
-            conda install cudatoolkit=11.0 -y
-            pip install tensorflow==2.4.0 pyyaml
-            pip install protobuf==3.20
-            cd models && pip install -e .
-            """
+    with sky.Dag() as dag:
+        # Total Nodes, INCLUDING Head Node
+        num_nodes = 2
 
-    # The command to run.  Will be run under the working directory.
-    # If a str, run the same command on all nodes.
-    # Generates per-node commands.  Must be a self-contained lambda
-    # that doesn't refer to any external variables.
-    #
-    # Will be run under the working directory.
-    def run_fn(node_rank: int, ip_list: List[str]) -> Optional[str]:
-        import json
-        tf_config = {
-            'cluster': {
-                'worker': [ip + ':8008' for ip in ip_list]
-            },
-            'task': {
-                'type': 'worker',
-                'index': node_rank
+        # The setup command.  Will be run under the working directory.
+        setup = """
+                git clone https://github.com/concretevitamin/tpu || true
+                cd tpu && git checkout 9459fee
+                conda create -n resnet python=3.7 -y
+                conda activate resnet
+                conda install cudatoolkit=11.0 -y
+                pip install tensorflow==2.4.0 pyyaml
+                pip install protobuf==3.20
+                cd models && pip install -e .
+                """
+
+        # The command to run.  Will be run under the working directory.
+        # If a str, run the same command on all nodes.
+        # Generates per-node commands.  Must be a self-contained lambda
+        # that doesn't refer to any external variables.
+        #
+        # Will be run under the working directory.
+        def run_fn(node_rank: int, ip_list: List[str]) -> Optional[str]:
+            import json
+            tf_config = {
+                'cluster': {
+                    'worker': [ip + ':8008' for ip in ip_list]
+                },
+                'task': {
+                    'type': 'worker',
+                    'index': node_rank
+                }
             }
-        }
-        str_tf_config = json.dumps(tf_config)
-        print(f'{str_tf_config!r}')
-        run = f"""
-            cd tpu
-            conda activate resnet
-            rm -rf resnet_model-dir
-            export TF_CONFIG={str_tf_config!r}
-            export XLA_FLAGS='--xla_gpu_cuda_data_dir=/usr/local/cuda/'
-            python models/official/resnet/resnet_main.py --use_tpu=False \
-                --mode=train --train_batch_size=256 --train_steps=500 \
-                --iterations_per_loop=125 \
-                --data_dir=gs://cloud-tpu-test-datasets/fake_imagenet \
-                --model_dir=resnet-model-dir \
-                --amp --xla --loss_scale=128"""
-        return run
+            str_tf_config = json.dumps(tf_config)
+            print(f'{str_tf_config!r}')
+            run = f"""
+                cd tpu
+                conda activate resnet
+                rm -rf resnet_model-dir
+                export TF_CONFIG={str_tf_config!r}
+                export XLA_FLAGS='--xla_gpu_cuda_data_dir=/usr/local/cuda/'
+                python models/official/resnet/resnet_main.py --use_tpu=False \
+                    --mode=train --train_batch_size=256 --train_steps=500 \
+                    --iterations_per_loop=125 \
+                    --data_dir=gs://cloud-tpu-test-datasets/fake_imagenet \
+                    --model_dir=resnet-model-dir \
+                    --amp --xla --loss_scale=128"""
+            return run
+
+        train = sky.Task(
+            'train',
+            setup=setup,
+            num_nodes=num_nodes,
+            run=run_fn,
+        )
+
+        train.set_inputs('gs://cloud-tpu-test-datasets/fake_imagenet',
+                         estimated_size_gigabytes=70)
+        train.set_outputs('resnet-model-dir', estimated_size_gigabytes=0.1)
+        train.set_resources(sky.Resources(sky.AWS(), accelerators='V100'))
 
-    train = sky.Task(
-        'train',
-        setup=setup,
-        num_nodes=num_nodes,
-        run=run_fn,
-    )
+    sky.launch(dag, cluster_name=cluster)
 
-    train.set_inputs('gs://cloud-tpu-test-datasets/fake_imagenet',
-                     estimated_size_gigabytes=70)
-    train.set_outputs('resnet-model-dir', estimated_size_gigabytes=0.1)
-    train.set_resources(sky.Resources(sky.AWS(), accelerators='V100'))
 
-sky.launch(dag, cluster_name=cluster)
+if __name__ == '__main__':
+    cluster = None
+    if len(sys.argv) > 1:
+        # For smoke test passing in a cluster name.
+        cluster = sys.argv[1]
+    run(cluster)
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
@@ -89,10 +89,13 @@
 _TEST_IP = 'https://8.8.8.8'
 
 # GCP has a 63 char limit; however, Ray autoscaler adds many
-# characters. Through testing, 37 chars is the maximum length for the Sky
-# cluster name on GCP.  Ref:
+# characters. Through testing, this is the maximum length for the Sky cluster
+# name on GCP.  Ref:
 # https://cloud.google.com/compute/docs/naming-resources#resource-name-format
-_MAX_CLUSTER_NAME_LEN = 37
+# NOTE: actually 37 is maximum for a single-node cluster which gets the suffix
+# '-head', but 35 for a multinode cluster because workers get the suffix
+# '-worker'. Here we do not distinguish these cases and take the lower limit.
+_MAX_CLUSTER_NAME_LEN_FOR_GCP = 35
 
 # Allow each CPU thread take 2 tasks.
 # Note: This value cannot be too small, otherwise OOM issue may occur.
@@ -796,7 +799,9 @@ def wait_until_ray_cluster_ready(
                   nodes_so_far != num_nodes):
                 worker_status.stop()
                 logger.error(
-                    'Timed out when waiting for workers to be provisioned.')
+                    'Timed out: waited for workers to be provisioned '
+                    f'for more than {nodes_launching_progress_timeout} seconds.'
+                )
                 return False  # failed
 
             if '(no pending nodes)' in output and '(no failures)' in output:
@@ -1555,7 +1560,8 @@ def get_task_resources_str(task: 'task_lib.Task') -> str:
     return resources_str
 
 
-def check_cluster_name_is_valid(cluster_name: str) -> None:
+def check_cluster_name_is_valid(cluster_name: str,
+                                cloud: Optional[clouds.Cloud] = None) -> None:
     """Errors out on invalid cluster names not supported by cloud providers.
 
     Bans (including but not limited to) names that:
@@ -1564,19 +1570,23 @@ def check_cluster_name_is_valid(cluster_name: str) -> None:
     """
     if cluster_name is None:
         return
-    # GCP errors return this exact regex.  An informal description is also at:
+    # GCP errors return this exact regex.  An informal description is at:
     # https://cloud.google.com/compute/docs/naming-resources#resource-name-format
     valid_regex = '[a-z]([-a-z0-9]{0,61}[a-z0-9])?'
     if re.fullmatch(valid_regex, cluster_name) is None:
         with ux_utils.print_exception_no_traceback():
             raise ValueError(
                 f'Cluster name "{cluster_name}" is invalid; '
                 f'ensure it is fully matched by regex: {valid_regex}')
-    if len(cluster_name) > _MAX_CLUSTER_NAME_LEN:
-        with ux_utils.print_exception_no_traceback():
-            raise ValueError(
-                f'Cluster name {cluster_name!r} has {len(cluster_name)}'
-                f' chars; maximum length is {_MAX_CLUSTER_NAME_LEN} chars.')
+    if isinstance(cloud, clouds.GCP):
+        # GCP has too restrictive of a length limit. Don't check for other
+        # clouds.
+        if len(cluster_name) > _MAX_CLUSTER_NAME_LEN_FOR_GCP:
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(
+                    f'Cluster name {cluster_name!r} has {len(cluster_name)}'
+                    f' chars; maximum length is {_MAX_CLUSTER_NAME_LEN_FOR_GCP}'
+                    ' chars.')
 
 
 def check_cluster_name_not_reserved(
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
@@ -350,7 +350,7 @@ def add_ray_task(
         if script is not None:
             sky_env_vars_dict['SKY_NODE_RANK'] = {gang_scheduling_id!r}
             sky_env_vars_dict['SKY_JOB_ID'] = {self.job_id}
-            
+
             futures.append(run_bash_command_with_log \\
                     .options({name_str}{cpu_str}{resources_str}{num_gpus_str}) \\
                     .remote(
@@ -1187,6 +1187,15 @@ def provision_with_retries(
         while provision_failed:
             provision_failed = False
             try:
+                try:
+                    # Recheck cluster name as the 'except:' block below may
+                    # change the cloud assignment.
+                    backend_utils.check_cluster_name_is_valid(
+                        cluster_name, to_provision.cloud)
+                except ValueError as value_error:
+                    # Let failover below handle this (i.e., block this cloud).
+                    raise exceptions.ResourcesUnavailableError(
+                    ) from value_error
                 config_dict = self._retry_region_zones(
                     to_provision,
                     num_nodes,
@@ -1449,26 +1458,25 @@ def _provision(self,
                    cluster_name: str,
                    retry_until_up: bool = False) -> ResourceHandle:
         """Provisions using 'ray up'."""
-        # Try to launch the exiting cluster first
-        backend_utils.check_cluster_name_is_valid(cluster_name)
-        # ray up: the VMs.
         # FIXME: ray up for Azure with different cluster_names will overwrite
         # each other.
-
         lock_path = os.path.expanduser(
             backend_utils.CLUSTER_STATUS_LOCK_PATH.format(cluster_name))
         with timeline.FileLockEvent(lock_path):
             to_provision_config = RetryingVmProvisioner.ToProvisionConfig(
                 cluster_name, to_provision, task.num_nodes)
             prev_cluster_status = None
             if not dryrun:  # dry run doesn't need to check existing cluster.
+                # Try to launch the exiting cluster first
                 to_provision_config = self._check_existing_cluster(
                     task, to_provision, cluster_name)
                 prev_cluster_status, _ = (
                     backend_utils.refresh_cluster_status_handle(
                         cluster_name, acquire_per_cluster_status_lock=False))
             assert to_provision_config.resources is not None, (
                 'to_provision should not be None', to_provision_config)
+            backend_utils.check_cluster_name_is_valid(
+                cluster_name, to_provision_config.resources.cloud)
 
             usage_lib.messages.usage.update_cluster_resources(
                 to_provision_config.num_nodes, to_provision_config.resources)
diff --git a/sky/cli.py b/sky/cli.py
@@ -2591,14 +2591,14 @@ def spot_cancel(name: Optional[str], job_ids: Tuple[int], all: bool, yes: bool):
             f'Provided {argument_str!r}.')
 
     if not yes:
-        job_identity_str = f'with IDs {job_id_str}' if job_ids else repr(name)
+        job_identity_str = (f'managed spot jobs with IDs {job_id_str}'
+                            if job_ids else repr(name))
         if all:
             job_identity_str = 'all managed spot jobs'
-        click.confirm(
-            f'Cancelling managed spot job {job_identity_str}. Proceed?',
-            default=True,
-            abort=True,
-            show_default=True)
+        click.confirm(f'Cancelling {job_identity_str}. Proceed?',
+                      default=True,
+                      abort=True,
+                      show_default=True)
 
     backend = backend_utils.get_backend_from_handle(handle)
     assert isinstance(backend, backends.CloudVmRayBackend)
diff --git a/sky/usage/usage_lib.py b/sky/usage/usage_lib.py
@@ -1,4 +1,4 @@
-"""Logging events to Grafana Loki"""
+"""Logging events to Grafana Loki."""
 
 import enum
 import click
diff --git a/tests/test_smoke.py b/tests/test_smoke.py

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-"""Logging events to Grafana Loki"""`
	`1`	`+"""Logging events to Grafana Loki."""`
`2`	`2`
`3`	`3`	`import enum`
`4`	`4`	`import click`