Repository: skypilot-org/skypilot
Pull request: [Lambda Cloud] Multinode support
This pr extends Lambda Cloud support to multi-node clusters.

Some notes:
- The tag file scheme has been changed (because the head node calls `ray start` to launch the children nodes).
- Generic smoke tests that required multi-node have the `@no_lambda_labs` descriptor removed. 
- We use `ssh` to get `internal_ip` given `external_ip`.
- The default instance has changed from `gpu_1x_a100_sxm4` to `gpu_1x_a10` (cheaper for testing multi-node, but less available regions; can revert back if not desired)

This pr also fixes #1678 by handling the case when there is no (external) ip.

Tested:
- [x] Lambda smoke tests

To try out, run
```
sky launch --cloud lambda --num-nodes 2
```

Problem statement:
[Lambda Cloud] Handle `Ctrl-C` during launch
Running `sky launch --cloud lambda` followed immediately by `Ctrl-C` and `sky down -a` will yield:
```
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042] **** STDERR ****
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042] 2023-02-06 16:45:39,483  INFO util.py:335 -- setting max workers for head node type to 0
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042] 2023-02-06 16:45:39,660  INFO util.py:335 -- setting max workers for head node type to 0
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042] Traceback (most recent call last):
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/bin/ray", line 8, in <module>
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     sys.exit(main())
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/ray/scripts/scripts.py", line 2339, in main
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     return cli()
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/click/core.py", line 1128, in __call__
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     return self.main(*args, **kwargs)
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/click/core.py", line 1053, in main
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     rv = self.invoke(ctx)
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/click/core.py", line 1659, in invoke
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     return _process_result(sub_ctx.command.invoke(sub_ctx))
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/click/core.py", line 1395, in invoke
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     return ctx.invoke(self.callback, **ctx.params)
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/click/core.py", line 754, in invoke
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     return __callback(*args, **kwargs)
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/ray/autoscaler/_private/cli_logger.py", line 852, in wrapper
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     return f(*args, **kwargs)
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/ray/scripts/scripts.py", line 1224, in down
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     cluster_config_file, yes, workers_only, cluster_name, keep_min_workers
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/ray/autoscaler/_private/commands.py", line 489, in teardown_cluster
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     A = remaining_nodes()
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/anaconda3/envs/sky/lib/python3.7/site-packages/ray/autoscaler/_private/commands.py", line 456, in remaining_nodes
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     head = provider.non_terminated_nodes({TAG_RAY_NODE_KIND: NODE_KIND_HEAD})
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/Desktop/ezeng/research/skypilot/sky/skylet/providers/lambda_cloud/node_provider.py", line 143, in non_terminated_nodes
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     nodes = self._get_filtered_nodes(tag_filters=tag_filters)
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/Desktop/ezeng/research/skypilot/sky/skylet/providers/lambda_cloud/node_provider.py", line 33, in wrapper
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     return f(self, *args, **kwargs)
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/Desktop/ezeng/research/skypilot/sky/skylet/providers/lambda_cloud/node_provider.py", line 113, in _get_filtered_nodes
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     nodes = [self._extract_metadata(vm) for vm in filter(match_tags, vms)]
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/Desktop/ezeng/research/skypilot/sky/skylet/providers/lambda_cloud/node_provider.py", line 113, in <listcomp>
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     nodes = [self._extract_metadata(vm) for vm in filter(match_tags, vms)]
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]   File "/home/zeng/Desktop/ezeng/research/skypilot/sky/skylet/providers/lambda_cloud/node_provider.py", line 122, in _extract_metadata
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]     ip = vm['ip']
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042] KeyError: 'ip'
E 02-06 16:45:41 cloud_vm_ray_backend.py:3042]
```
This is because Lambda Cloud assigns ips after a cluster has booted. We should throw a error message to tell users to wait a few minutes before trying `sky down -a` again.

Hints from issue discussion:

Commit message: Increase timeout again and nit

Patch to review:
diff --git a/sky/backends/backend_utils.py b/sky/backends/backend_utils.py
index 2a14c40b287..40781294097 100644
--- a/sky/backends/backend_utils.py
+++ b/sky/backends/backend_utils.py
@@ -1763,11 +1763,13 @@ def _query_status_lambda(
         'terminated': None,
     }
     # TODO(ewzeng): filter by hash_filter_string to be safe
+    status_list = []
     vms = lambda_utils.LambdaCloudClient().list_instances()
+    possible_names = [f'{cluster}-head', f'{cluster}-worker']
     for node in vms:
-        if node['name'] == cluster:
-            return [status_map[node['status']]]
-    return []
+        if node.get('name') in possible_names:
+            status_list.append(status_map[node['status']])
+    return status_list
 
 
 _QUERY_STATUS_FUNCS = {
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index d3959e15539..a08d6d6c5bd 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -66,7 +66,7 @@
 
 # Timeout (seconds) for provision progress: if in this duration no new nodes
 # are launched, abort and failover.
-_NODES_LAUNCHING_PROGRESS_TIMEOUT = 90
+_NODES_LAUNCHING_PROGRESS_TIMEOUT = 120
 
 # Time gap between retries after failing to provision in all possible places.
 # Used only if --retry-until-up is set.
diff --git a/sky/clouds/lambda_cloud.py b/sky/clouds/lambda_cloud.py
index 462bfbbd8d1..424caf65f13 100644
--- a/sky/clouds/lambda_cloud.py
+++ b/sky/clouds/lambda_cloud.py
@@ -25,15 +25,14 @@ class Lambda(clouds.Cloud):
 
     # Lamdba has a 64 char limit for cluster name.
     # Reference: https://cloud.lambdalabs.com/api/v1/docs#operation/launchInstance # pylint: disable=line-too-long
-    _MAX_CLUSTER_NAME_LEN_LIMIT = 64
+    # However, we need to account for the suffixes '-head' and '-worker'
+    _MAX_CLUSTER_NAME_LEN_LIMIT = 57
     # Currently, none of clouds.CloudImplementationFeatures are implemented
     # for Lambda Cloud.
     # STOP/AUTOSTOP: The Lambda cloud provider does not support stopping VMs.
-    # MULTI_NODE: Multi-node is not supported by the implementation yet.
     _CLOUD_UNSUPPORTED_FEATURES = {
         clouds.CloudImplementationFeatures.STOP: 'Lambda cloud does not support stopping VMs.',
         clouds.CloudImplementationFeatures.AUTOSTOP: 'Lambda cloud does not support stopping VMs.',
-        clouds.CloudImplementationFeatures.MULTI_NODE: 'Multi-node is not supported by the Lambda Cloud implementation yet.',
     }
 
     _regions: List[clouds.Region] = []
diff --git a/sky/clouds/service_catalog/lambda_catalog.py b/sky/clouds/service_catalog/lambda_catalog.py
index 21975719ef5..0d566525cfd 100644
--- a/sky/clouds/service_catalog/lambda_catalog.py
+++ b/sky/clouds/service_catalog/lambda_catalog.py
@@ -14,7 +14,7 @@
 
 _df = common.read_catalog('lambda/vms.csv')
 
-# Number of vCPUS for gpu_1x_a100_sxm4
+# Number of vCPUS for gpu_1x_a10
 _DEFAULT_NUM_VCPUS = 30
 
 
@@ -62,13 +62,7 @@ def get_vcpus_from_instance_type(instance_type: str) -> Optional[float]:
 def get_default_instance_type(cpus: Optional[str] = None) -> Optional[str]:
     if cpus is None:
         cpus = str(_DEFAULT_NUM_VCPUS)
-    # Set to gpu_1x_a100_sxm4 to be the default instance type if match vCPU
-    # requirement.
-    df = _df[_df['InstanceType'].eq('gpu_1x_a100_sxm4')]
-    instance = common.get_instance_type_for_cpus_impl(df, cpus)
-    if not instance:
-        instance = common.get_instance_type_for_cpus_impl(_df, cpus)
-    return instance
+    return common.get_instance_type_for_cpus_impl(_df, cpus)
 
 
 def get_accelerators_from_instance_type(
diff --git a/sky/skylet/providers/lambda_cloud/lambda_utils.py b/sky/skylet/providers/lambda_cloud/lambda_utils.py
index dd7c1806059..44915c9fd6e 100644
--- a/sky/skylet/providers/lambda_cloud/lambda_utils.py
+++ b/sky/skylet/providers/lambda_cloud/lambda_utils.py
@@ -66,6 +66,13 @@ def refresh(self, instance_ids: List[str]) -> None:
         with open(self.path, 'w') as f:
             json.dump(metadata, f)
 
+    def exists(self, instance_id: str) -> bool:
+        if not os.path.exists(self.path):
+            return False
+        with open(self.path, 'r') as f:
+            metadata = json.load(f)
+        return instance_id in metadata
+
 
 def raise_lambda_error(response: requests.Response) -> None:
     """Raise LambdaCloudError if appropriate. """
diff --git a/sky/skylet/providers/lambda_cloud/node_provider.py b/sky/skylet/providers/lambda_cloud/node_provider.py
index 23491f2df45..e1cfbc685fe 100644
--- a/sky/skylet/providers/lambda_cloud/node_provider.py
+++ b/sky/skylet/providers/lambda_cloud/node_provider.py
@@ -16,12 +16,14 @@
     NODE_KIND_WORKER,
     NODE_KIND_HEAD,
 )
-from ray.autoscaler._private.util import hash_launch_conf
 from sky.skylet.providers.lambda_cloud import lambda_utils
-from sky.utils import common_utils
+from sky import authentication as auth
+from sky.utils import command_runner
+from sky.utils import subprocess_utils
 
 TAG_PATH_PREFIX = '~/.sky/generated/lambda_cloud/metadata'
-REMOTE_RAY_YAML = '~/ray_bootstrap_config.yaml'
+REMOTE_RAY_SSH_KEY = '~/ray_bootstrap_key.pem'
+GET_INTERNAL_IP_CMD = 'ip -4 -br addr show | grep -Eo "10\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)"'
 
 logger = logging.getLogger(__name__)
 
@@ -51,55 +53,59 @@ def __init__(self, provider_config: Dict[str, Any],
         self.lambda_client = lambda_utils.LambdaCloudClient()
         self.cached_nodes = {}
         self.metadata = lambda_utils.Metadata(TAG_PATH_PREFIX, cluster_name)
-        vms = self._list_instances_in_cluster()
-
-        # The tag file for autodowned clusters is not autoremoved. Hence, if
-        # a previous cluster was autodowned and has the same name as the
-        # current cluster, then self.metadata might load the old tag file.
-        # We prevent this by removing any old vms in the tag file.
-        self.metadata.refresh([node['id'] for node in vms])
-
-        # If tag file does not exist on head, create it and add basic tags.
-        # This is a hack to make sure that ray on head can access some
-        # important tags.
-        # TODO(ewzeng): change when Lambda Cloud adds tag support.
-        ray_yaml_path = os.path.expanduser(REMOTE_RAY_YAML)
-        if os.path.exists(ray_yaml_path) and not os.path.exists(
-                self.metadata.path):
-            config = common_utils.read_yaml(ray_yaml_path)
-            # Ensure correct cluster so sky launch on head node works correctly
-            if config['cluster_name'] != cluster_name:
-                return
-            # Compute launch hash
-            head_node_config = config.get('head_node', {})
-            head_node_type = config.get('head_node_type')
-            if head_node_type:
-                head_config = config['available_node_types'][head_node_type]
-                head_node_config.update(head_config["node_config"])
-            launch_hash = hash_launch_conf(head_node_config, config['auth'])
-            # Populate tags
-            for node in vms:
+        self.ssh_key_path = os.path.expanduser(auth.PRIVATE_SSH_KEY_PATH)
+        remote_ssh_key = os.path.expanduser(REMOTE_RAY_SSH_KEY)
+        if os.path.exists(remote_ssh_key):
+            self.ssh_key_path = remote_ssh_key
+
+    def _guess_and_add_missing_tags(self, vms: Dict[str, Any]) -> None:
+        """Adds missing vms to local tag file and guesses their tags."""
+        for node in vms:
+            if self.metadata.exists(node['id']):
+                pass
+            elif node['name'] == f'{self.cluster_name}-head':
                 self.metadata[node['id']] = {
                     'tags': {
-                        TAG_RAY_CLUSTER_NAME: cluster_name,
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
                         TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
                         TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
                         TAG_RAY_USER_NODE_TYPE: 'ray_head_default',
-                        TAG_RAY_NODE_NAME: f'ray-{cluster_name}-head',
-                        TAG_RAY_LAUNCH_CONFIG: launch_hash,
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-head',
+                    }
+                }
+            elif node['name'] == f'{self.cluster_name}-worker':
+                self.metadata[node['id']] = {
+                    'tags': {
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
+                        TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
+                        TAG_RAY_NODE_KIND: NODE_KIND_WORKER,
+                        TAG_RAY_USER_NODE_TYPE: 'ray_worker_default',
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-worker',
                     }
                 }
 
     def _list_instances_in_cluster(self) -> Dict[str, Any]:
         """List running instances in cluster."""
         vms = self.lambda_client.list_instances()
-        return [node for node in vms if node.get('name') == self.cluster_name]
+        possible_names = [
+            f'{self.cluster_name}-head', f'{self.cluster_name}-worker'
+        ]
+        return [node for node in vms if node.get('name') in possible_names]
 
     @synchronized
     def _get_filtered_nodes(self, tag_filters: Dict[str,
                                                     str]) -> Dict[str, Any]:
 
-        def match_tags(vm):
+        def _extract_metadata(vm: Dict[str, Any]) -> Dict[str, Any]:
+            metadata = {'id': vm['id'], 'status': vm['status'], 'tags': {}}
+            instance_info = self.metadata[vm['id']]
+            if instance_info is not None:
+                metadata['tags'] = instance_info['tags']
+            ip = vm['ip']
+            metadata['external_ip'] = ip
+            return metadata
+
+        def _match_tags(vm: Dict[str, Any]):
             vm_info = self.metadata[vm['id']]
             tags = {} if vm_info is None else vm_info['tags']
             for k, v in tag_filters.items():
@@ -107,24 +113,24 @@ def match_tags(vm):
                     return False
             return True
 
+        def _get_internal_ip(node: Dict[str, Any]):
+            runner = command_runner.SSHCommandRunner(node['external_ip'],
+                                                     'ubuntu',
+                                                     self.ssh_key_path)
+            out = runner.run(GET_INTERNAL_IP_CMD,
+                             require_outputs=True,
+                             stream_logs=False)
+            assert out[0] == 0
+            node['internal_ip'] = out[1].strip()
+
         vms = self._list_instances_in_cluster()
-        nodes = [self._extract_metadata(vm) for vm in filter(match_tags, vms)]
+        self.metadata.refresh([node['id'] for node in vms])
+        self._guess_and_add_missing_tags(vms)
+        nodes = [_extract_metadata(vm) for vm in filter(_match_tags, vms)]
+        subprocess_utils.run_in_parallel(_get_internal_ip, nodes)
         self.cached_nodes = {node['id']: node for node in nodes}
         return self.cached_nodes
 
-    def _extract_metadata(self, vm: Dict[str, Any]) -> Dict[str, Any]:
-        metadata = {'id': vm['id'], 'status': vm['status'], 'tags': {}}
-        instance_info = self.metadata[vm['id']]
-        if instance_info is not None:
-            metadata['tags'] = instance_info['tags']
-        ip = vm['ip']
-        metadata['external_ip'] = ip
-        # TODO(ewzeng): The internal ip is hard to get, so set it to the
-        # external ip as a hack. This should be changed in the future.
-        #   https://docs.lambdalabs.com/cloud/learn-private-ip-address/
-        metadata['internal_ip'] = ip
-        return metadata
-
     def non_terminated_nodes(self, tag_filters: Dict[str, str]) -> List[str]:
         """Return a list of node ids filtered by the specified tags dict.
 
@@ -164,31 +170,39 @@ def internal_ip(self, node_id: str) -> str:
     def create_node(self, node_config: Dict[str, Any], tags: Dict[str, str],
                     count: int) -> None:
         """Creates a number of nodes within the namespace."""
-        assert count == 1, count  # Only support 1-node clusters for now
-
-        # get the tags
+        # Get tags
         config_tags = node_config.get('tags', {}).copy()
         config_tags.update(tags)
         config_tags[TAG_RAY_CLUSTER_NAME] = self.cluster_name
 
-        # create the node
+        # Create nodes
         ttype = node_config['InstanceType']
         region = self.provider_config['region']
-        vm_list = self.lambda_client.create_instances(instance_type=ttype,
-                                                      region=region,
-                                                      quantity=1,
-                                                      name=self.cluster_name)
-        assert len(vm_list) == 1, len(vm_list)
-        vm_id = vm_list[0]
-        self.metadata[vm_id] = {'tags': config_tags}
-
-        # Wait for booting to finish
-        # TODO(ewzeng): For multi-node, launch all vms first and then wait.
+        if config_tags[TAG_RAY_NODE_KIND] == NODE_KIND_HEAD:
+            name = f'{self.cluster_name}-head'
+        else:
+            name = f'{self.cluster_name}-worker'
+        # Lambda launch api only supports launching one node at a time,
+        # so we do a loop. Remove loop when launch api allows quantity > 1
+        booting_list = []
+        for _ in range(count):
+            vm_id = self.lambda_client.create_instances(instance_type=ttype,
+                                                        region=region,
+                                                        quantity=1,
+                                                        name=name)[0]
+            self.metadata[vm_id] = {'tags': config_tags}
+            booting_list.append(vm_id)
+            time.sleep(10)  # Avoid api rate limits
+
+        # Wait for nodes to finish booting
         while True:
-            vms = self.lambda_client.list_instances()
-            for vm in vms:
-                if vm['id'] == vm_id and vm['status'] == 'active':
-                    return
+            vms = self._list_instances_in_cluster()
+            for vm_id in booting_list.copy():
+                for vm in vms:
+                    if vm['id'] == vm_id and vm['status'] == 'active':
+                        booting_list.remove(vm_id)
+            if len(booting_list) == 0:
+                return
             time.sleep(10)
 
     @synchronized
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index edd65e73ca5..0b797d5d858 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -587,7 +587,6 @@ def test_gcp_stale_job_manual_restart():
 
 
 # ---------- Check Sky's environment variables; workdir. ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_env_check(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -602,7 +601,6 @@ def test_env_check(generic_cloud: str):
 
 
 # ---------- file_mounts ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_file_mounts(generic_cloud: str):
     name = _get_cluster_name()
     test_commands = [
@@ -619,24 +617,6 @@ def test_file_mounts(generic_cloud: str):
     run_one_test(test)
 
 
-# TODO(ewzeng): merge this with 'test_file_mounts' when multi-node is supported.
-@pytest.mark.lambda_cloud
-def test_lambda_file_mounts():
-    name = _get_cluster_name()
-    test_commands = [
-        *storage_setup_commands,
-        f'sky launch -y -c {name} {LAMBDA_TYPE} --num-nodes 1 examples/using_file_mounts.yaml',
-        f'sky logs {name} 1 --status',  # Ensure the job succeeded.
-    ]
-    test = Test(
-        'lambda_using_file_mounts',
-        test_commands,
-        f'sky down -y {name}',
-        timeout=20 * 60,  # 20 mins
-    )
-    run_one_test(test)
-
-
 # ---------- storage ----------
 @pytest.mark.aws
 def test_aws_storage_mounts():
@@ -693,7 +673,6 @@ def test_gcp_storage_mounts():
 
 
 # ---------- CLI logs ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_cli_logs(generic_cloud: str):
     name = _get_cluster_name()
     timestamp = time.time()
@@ -715,29 +694,6 @@ def test_cli_logs(generic_cloud: str):
     run_one_test(test)
 
 
-# TODO(ewzeng): merge this with 'test_cli_logs' when multi-node is supported.
-@pytest.mark.lambda_cloud
-def test_lambda_logs():
-    name = _get_cluster_name()
-    timestamp = time.time()
-    test = Test(
-        'lambda_cli_logs',
-        [
-            f'sky launch -y -c {name} {LAMBDA_TYPE} "echo {timestamp} 1"',
-            f'sky exec {name} "echo {timestamp} 2"',
-            f'sky exec {name} "echo {timestamp} 3"',
-            f'sky exec {name} "echo {timestamp} 4"',
-            f'sky logs {name} 2 --status',
-            f'sky logs {name} 3 4 --sync-down',
-            f'sky logs {name} * --sync-down',
-            f'sky logs {name} 1 | grep "{timestamp} 1"',
-            f'sky logs {name} | grep "{timestamp} 4"',
-        ],
-        f'sky down -y {name}',
-    )
-    run_one_test(test)
-
-
 # ---------- Job Queue. ----------
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not have K80 gpus
 def test_job_queue(generic_cloud: str):
@@ -789,7 +745,7 @@ def test_lambda_job_queue():
     run_one_test(test)
 
 
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have T4 gpus
 def test_job_queue_multinode(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -847,7 +803,7 @@ def test_large_job_queue(generic_cloud: str):
 
 
 # ---------- Submitting multiple tasks to the same cluster. ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have K80 gpus
 def test_multi_echo(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -868,7 +824,7 @@ def test_multi_echo(generic_cloud: str):
 
 
 # ---------- Task: 1 node training. ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 instances
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 gpus
 def test_huggingface(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -961,7 +917,6 @@ def test_tpu_vm_pod():
 
 
 # ---------- Simple apps. ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_multi_hostname(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -978,7 +933,7 @@ def test_multi_hostname(generic_cloud: str):
 
 
 # ---------- Task: n=2 nodes with setups. ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 gpus
 def test_distributed_tf(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1098,7 +1053,6 @@ def test_autostop(generic_cloud: str):
 
 
 # ---------- Testing Autodowning ----------
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
 def test_autodown(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(
@@ -1133,41 +1087,6 @@ def test_autodown(generic_cloud: str):
     run_one_test(test)
 
 
-@pytest.mark.lambda_cloud
-def test_lambda_autodown():
-    name = _get_cluster_name()
-    test = Test(
-        'lambda_autodown',
-        [
-            f'sky launch -y -d -c {name} {LAMBDA_TYPE} tests/test_yamls/minimal.yaml',
-            f'sky autostop -y {name} --down -i 1',
-            # Ensure autostop is set.
-            f'sky status | grep {name} | grep "1m (down)"',
-            # Ensure the cluster is not terminated early.
-            'sleep 45',
-            f'sky status --refresh | grep {name} | grep UP',
-            # Ensure the cluster is terminated.
-            'sleep 200',
-            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && {{ echo "$s" | grep {name} | grep "Autodowned cluster\|terminated on the cloud"; }} || {{ echo "$s" | grep {name} && exit 1 || exit 0; }}',
-            f'sky launch -y -d -c {name} {LAMBDA_TYPE} --down tests/test_yamls/minimal.yaml',
-            f'sky status | grep {name} | grep UP',  # Ensure the cluster is UP.
-            f'sky exec {name} {LAMBDA_TYPE} tests/test_yamls/minimal.yaml',
-            f'sky status | grep {name} | grep "1m (down)"',
-            'sleep 200',
-            # Ensure the cluster is terminated.
-            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && {{ echo "$s" | grep {name} | grep "Autodowned cluster\|terminated on the cloud"; }} || {{ echo "$s" | grep {name} && exit 1 || exit 0; }}',
-            f'sky launch -y -d -c {name} {LAMBDA_TYPE} --down tests/test_yamls/minimal.yaml',
-            f'sky autostop -y {name} --cancel',
-            'sleep 200',
-            # Ensure the cluster is still UP.
-            f's=$(SKYPILOT_DEBUG=0 sky status --refresh) && printf "$s" && echo "$s" | grep {name} | grep UP',
-        ],
-        f'sky down -y {name}',
-        timeout=25 * 60,
-    )
-    run_one_test(test)
-
-
 def _get_cancel_task_with_cloud(name, cloud, timeout=15 * 60):
     test = Test(
         f'{cloud}-cancel-task',
@@ -1211,7 +1130,7 @@ def test_cancel_azure():
     run_one_test(test)
 
 
-@pytest.mark.no_lambda_cloud  # Lambda Cloud does not support num_nodes > 1 yet
+@pytest.mark.no_lambda_cloud  # Lambda Cloud does not have V100 gpus
 def test_cancel_pytorch(generic_cloud: str):
     name = _get_cluster_name()
     test = Test(


Code review comments:
diff_hunk: @@ -16,12 +16,14 @@
     NODE_KIND_WORKER,
     NODE_KIND_HEAD,
 )
-from ray.autoscaler._private.util import hash_launch_conf
 from sky.skylet.providers.lambda_cloud import lambda_utils
-from sky.utils import common_utils
+from sky import authentication as auth
+from sky.utils import command_runner
+from sky.utils import subprocess_utils
 
 TAG_PATH_PREFIX = '~/.sky/generated/lambda_cloud/metadata'
-REMOTE_RAY_YAML = '~/ray_bootstrap_config.yaml'
+REMOTE_RAY_SSH_KEY = '~/ray_bootstrap_key.pem'
+GET_INTERNAL_IP_CMD = 'ip -4 -br addr show | grep -Eo "10\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)"'
path: sky/skylet/providers/lambda_cloud/node_provider.py
text: @user1:
Add prefix `_` to these variables as they are internal variables?

@author:
Done.

diff_hunk: @@ -164,31 +170,39 @@ def internal_ip(self, node_id: str) -> str:
     def create_node(self, node_config: Dict[str, Any], tags: Dict[str, str],
                     count: int) -> None:
         """Creates a number of nodes within the namespace."""
-        assert count == 1, count  # Only support 1-node clusters for now
-
-        # get the tags
+        # Get tags
         config_tags = node_config.get('tags', {}).copy()
         config_tags.update(tags)
         config_tags[TAG_RAY_CLUSTER_NAME] = self.cluster_name
 
-        # create the node
+        # Create nodes
         ttype = node_config['InstanceType']
         region = self.provider_config['region']
-        vm_list = self.lambda_client.create_instances(instance_type=ttype,
-                                                      region=region,
-                                                      quantity=1,
-                                                      name=self.cluster_name)
-        assert len(vm_list) == 1, len(vm_list)
-        vm_id = vm_list[0]
-        self.metadata[vm_id] = {'tags': config_tags}
-
-        # Wait for booting to finish
-        # TODO(ewzeng): For multi-node, launch all vms first and then wait.
+        if config_tags[TAG_RAY_NODE_KIND] == NODE_KIND_HEAD:
+            name = f'{self.cluster_name}-head'
+        else:
+            name = f'{self.cluster_name}-worker'
+        # Lambda launch api only supports launching one node at a time,
+        # so we do a loop. Remove loop when launch api allows quantity > 1
+        booting_list = []
+        for _ in range(count):
+            vm_id = self.lambda_client.create_instances(instance_type=ttype,
+                                                        region=region,
+                                                        quantity=1,
+                                                        name=name)[0]
path: sky/skylet/providers/lambda_cloud/node_provider.py
text: @user1:
Can we have some rate limit error handling / retrying for the place we call the lambda API? One possible problem that I could image here is we run `sky launch -c test1 --cloud lambda` and `sky launch -c test2 --cloud lambda` in parallel, and they could fail each other.

I am thinking if we can have the retry logic built-in the `LambdaClient`, i.e. if the API rate limit is reached, retry the function call with some backoff.

@author:
Hmm, let me think about this. In this case, the backoff time would require some randomness (or else the `test1` and `test2` would keep failing each other).

@user1:
You may want to use the the backoff tool here if needed.
https://github.com/skypilot-org/skypilot/blob/4e7d7172f2a3c9196315f5c2e42c68bd5ec967e8/sky/utils/common_utils.py#L98-L123

diff_hunk: @@ -164,31 +170,39 @@ def internal_ip(self, node_id: str) -> str:
     def create_node(self, node_config: Dict[str, Any], tags: Dict[str, str],
                     count: int) -> None:
         """Creates a number of nodes within the namespace."""
-        assert count == 1, count  # Only support 1-node clusters for now
-
-        # get the tags
+        # Get tags
         config_tags = node_config.get('tags', {}).copy()
         config_tags.update(tags)
         config_tags[TAG_RAY_CLUSTER_NAME] = self.cluster_name
 
-        # create the node
+        # Create nodes
         ttype = node_config['InstanceType']
path: sky/skylet/providers/lambda_cloud/node_provider.py
text: @user1:
Would be nice to use fullname for the variable `instance_type`?

@author:
Done.

diff_hunk: @@ -51,80 +53,84 @@ def __init__(self, provider_config: Dict[str, Any],
         self.lambda_client = lambda_utils.LambdaCloudClient()
         self.cached_nodes = {}
         self.metadata = lambda_utils.Metadata(TAG_PATH_PREFIX, cluster_name)
-        vms = self._list_instances_in_cluster()
-
-        # The tag file for autodowned clusters is not autoremoved. Hence, if
-        # a previous cluster was autodowned and has the same name as the
-        # current cluster, then self.metadata might load the old tag file.
-        # We prevent this by removing any old vms in the tag file.
-        self.metadata.refresh([node['id'] for node in vms])
-
-        # If tag file does not exist on head, create it and add basic tags.
-        # This is a hack to make sure that ray on head can access some
-        # important tags.
-        # TODO(ewzeng): change when Lambda Cloud adds tag support.
-        ray_yaml_path = os.path.expanduser(REMOTE_RAY_YAML)
-        if os.path.exists(ray_yaml_path) and not os.path.exists(
-                self.metadata.path):
-            config = common_utils.read_yaml(ray_yaml_path)
-            # Ensure correct cluster so sky launch on head node works correctly
-            if config['cluster_name'] != cluster_name:
-                return
-            # Compute launch hash
-            head_node_config = config.get('head_node', {})
-            head_node_type = config.get('head_node_type')
-            if head_node_type:
-                head_config = config['available_node_types'][head_node_type]
-                head_node_config.update(head_config["node_config"])
-            launch_hash = hash_launch_conf(head_node_config, config['auth'])
-            # Populate tags
-            for node in vms:
+        self.ssh_key_path = os.path.expanduser(auth.PRIVATE_SSH_KEY_PATH)
+        remote_ssh_key = os.path.expanduser(REMOTE_RAY_SSH_KEY)
+        if os.path.exists(remote_ssh_key):
+            self.ssh_key_path = remote_ssh_key
+
+    def _guess_and_add_missing_tags(self, vms: Dict[str, Any]) -> None:
+        """Adds missing vms to local tag file and guesses their tags."""
+        for node in vms:
+            if self.metadata.exists(node['id']):
+                pass
+            elif node['name'] == f'{self.cluster_name}-head':
                 self.metadata[node['id']] = {
                     'tags': {
-                        TAG_RAY_CLUSTER_NAME: cluster_name,
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
                         TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
                         TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
                         TAG_RAY_USER_NODE_TYPE: 'ray_head_default',
-                        TAG_RAY_NODE_NAME: f'ray-{cluster_name}-head',
-                        TAG_RAY_LAUNCH_CONFIG: launch_hash,
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-head',
+                    }
+                }
+            elif node['name'] == f'{self.cluster_name}-worker':
+                self.metadata[node['id']] = {
+                    'tags': {
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
+                        TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
+                        TAG_RAY_NODE_KIND: NODE_KIND_WORKER,
+                        TAG_RAY_USER_NODE_TYPE: 'ray_worker_default',
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-worker',
                     }
                 }
 
     def _list_instances_in_cluster(self) -> Dict[str, Any]:
         """List running instances in cluster."""
         vms = self.lambda_client.list_instances()
-        return [node for node in vms if node.get('name') == self.cluster_name]
+        possible_names = [
+            f'{self.cluster_name}-head', f'{self.cluster_name}-worker'
+        ]
+        return [node for node in vms if node.get('name') in possible_names]
 
     @synchronized
     def _get_filtered_nodes(self, tag_filters: Dict[str,
                                                     str]) -> Dict[str, Any]:
 
-        def match_tags(vm):
+        def _extract_metadata(vm: Dict[str, Any]) -> Dict[str, Any]:
+            metadata = {'id': vm['id'], 'status': vm['status'], 'tags': {}}
+            instance_info = self.metadata[vm['id']]
+            if instance_info is not None:
+                metadata['tags'] = instance_info['tags']
+            ip = vm['ip']
+            metadata['external_ip'] = ip
+            return metadata
+
+        def _match_tags(vm: Dict[str, Any]):
             vm_info = self.metadata[vm['id']]
             tags = {} if vm_info is None else vm_info['tags']
             for k, v in tag_filters.items():
                 if tags.get(k) != v:
                     return False
             return True
 
+        def _get_internal_ip(node: Dict[str, Any]):
+            runner = command_runner.SSHCommandRunner(node['external_ip'],
+                                                     'ubuntu',
+                                                     self.ssh_key_path)
+            out = runner.run(GET_INTERNAL_IP_CMD,
path: sky/skylet/providers/lambda_cloud/node_provider.py
text: @user1:
nit: 
```suggestion
            rc, stdout, stderr = runner.run(GET_INTERNAL_IP_CMD,
```

@author:
Fixed.

diff_hunk: @@ -66,7 +66,7 @@
 
 # Timeout (seconds) for provision progress: if in this duration no new nodes
 # are launched, abort and failover.
-_NODES_LAUNCHING_PROGRESS_TIMEOUT = 90
+_NODES_LAUNCHING_PROGRESS_TIMEOUT = 120
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
Can we only increase this number for Lambda, as that will increase the waiting time for the quota limit error for other clouds?

@author:
Updated! I turned `_NODES_LAUNCHING_PROGRESS_TIMEOUT` into a dictionary.

diff_hunk: @@ -51,80 +53,84 @@ def __init__(self, provider_config: Dict[str, Any],
         self.lambda_client = lambda_utils.LambdaCloudClient()
         self.cached_nodes = {}
         self.metadata = lambda_utils.Metadata(TAG_PATH_PREFIX, cluster_name)
-        vms = self._list_instances_in_cluster()
-
-        # The tag file for autodowned clusters is not autoremoved. Hence, if
-        # a previous cluster was autodowned and has the same name as the
-        # current cluster, then self.metadata might load the old tag file.
-        # We prevent this by removing any old vms in the tag file.
-        self.metadata.refresh([node['id'] for node in vms])
-
-        # If tag file does not exist on head, create it and add basic tags.
-        # This is a hack to make sure that ray on head can access some
-        # important tags.
-        # TODO(ewzeng): change when Lambda Cloud adds tag support.
-        ray_yaml_path = os.path.expanduser(REMOTE_RAY_YAML)
-        if os.path.exists(ray_yaml_path) and not os.path.exists(
-                self.metadata.path):
-            config = common_utils.read_yaml(ray_yaml_path)
-            # Ensure correct cluster so sky launch on head node works correctly
-            if config['cluster_name'] != cluster_name:
-                return
-            # Compute launch hash
-            head_node_config = config.get('head_node', {})
-            head_node_type = config.get('head_node_type')
-            if head_node_type:
-                head_config = config['available_node_types'][head_node_type]
-                head_node_config.update(head_config["node_config"])
-            launch_hash = hash_launch_conf(head_node_config, config['auth'])
-            # Populate tags
-            for node in vms:
+        self.ssh_key_path = os.path.expanduser(auth.PRIVATE_SSH_KEY_PATH)
+        remote_ssh_key = os.path.expanduser(REMOTE_RAY_SSH_KEY)
+        if os.path.exists(remote_ssh_key):
+            self.ssh_key_path = remote_ssh_key
+
+    def _guess_and_add_missing_tags(self, vms: Dict[str, Any]) -> None:
+        """Adds missing vms to local tag file and guesses their tags."""
+        for node in vms:
+            if self.metadata.exists(node['id']):
+                pass
+            elif node['name'] == f'{self.cluster_name}-head':
                 self.metadata[node['id']] = {
                     'tags': {
-                        TAG_RAY_CLUSTER_NAME: cluster_name,
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
                         TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
                         TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
                         TAG_RAY_USER_NODE_TYPE: 'ray_head_default',
-                        TAG_RAY_NODE_NAME: f'ray-{cluster_name}-head',
-                        TAG_RAY_LAUNCH_CONFIG: launch_hash,
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-head',
+                    }
+                }
+            elif node['name'] == f'{self.cluster_name}-worker':
+                self.metadata[node['id']] = {
+                    'tags': {
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
+                        TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
+                        TAG_RAY_NODE_KIND: NODE_KIND_WORKER,
+                        TAG_RAY_USER_NODE_TYPE: 'ray_worker_default',
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-worker',
                     }
                 }
 
     def _list_instances_in_cluster(self) -> Dict[str, Any]:
         """List running instances in cluster."""
         vms = self.lambda_client.list_instances()
-        return [node for node in vms if node.get('name') == self.cluster_name]
+        possible_names = [
+            f'{self.cluster_name}-head', f'{self.cluster_name}-worker'
+        ]
+        return [node for node in vms if node.get('name') in possible_names]
 
     @synchronized
     def _get_filtered_nodes(self, tag_filters: Dict[str,
                                                     str]) -> Dict[str, Any]:
 
-        def match_tags(vm):
+        def _extract_metadata(vm: Dict[str, Any]) -> Dict[str, Any]:
+            metadata = {'id': vm['id'], 'status': vm['status'], 'tags': {}}
+            instance_info = self.metadata[vm['id']]
path: sky/skylet/providers/lambda_cloud/node_provider.py
text: @user1:
Curious, do we need to check the existence with `self.metadata.exists(vm['id'])` first?

@author:
I made `self.metadata[XX]` to return `None` if `XX` doesn't exist, so `self.metadata.exists` is no longer needed.

@user1:
Oh, I thought the `self.metadata` is the dict we store and manage locally. When we call `refresh` in L127, we only removes the instance id that does not exist in `vms` from the `self.metadata` but do not make sure (IIUC) that `self.metadata` contain all the ids in `vms`.

Is it possible that we are calling this function with an id that does not exist in the `self.metadata`?

@user1:
Ahh, nvm. It seems the `self.metadata[id]` will return None if `id` is not in `self.metadata`. 

nit: In that case, will it be better to change the API of `self.metadata.__getitem__` to `self.metadata.get()`, to be more consistent with `dict`'s API. 

@author:
Good idea. Let me do that.

@author:
Updated!

diff_hunk: @@ -51,80 +53,84 @@ def __init__(self, provider_config: Dict[str, Any],
         self.lambda_client = lambda_utils.LambdaCloudClient()
         self.cached_nodes = {}
         self.metadata = lambda_utils.Metadata(TAG_PATH_PREFIX, cluster_name)
-        vms = self._list_instances_in_cluster()
-
-        # The tag file for autodowned clusters is not autoremoved. Hence, if
-        # a previous cluster was autodowned and has the same name as the
-        # current cluster, then self.metadata might load the old tag file.
-        # We prevent this by removing any old vms in the tag file.
-        self.metadata.refresh([node['id'] for node in vms])
-
-        # If tag file does not exist on head, create it and add basic tags.
-        # This is a hack to make sure that ray on head can access some
-        # important tags.
-        # TODO(ewzeng): change when Lambda Cloud adds tag support.
-        ray_yaml_path = os.path.expanduser(REMOTE_RAY_YAML)
-        if os.path.exists(ray_yaml_path) and not os.path.exists(
-                self.metadata.path):
-            config = common_utils.read_yaml(ray_yaml_path)
-            # Ensure correct cluster so sky launch on head node works correctly
-            if config['cluster_name'] != cluster_name:
-                return
-            # Compute launch hash
-            head_node_config = config.get('head_node', {})
-            head_node_type = config.get('head_node_type')
-            if head_node_type:
-                head_config = config['available_node_types'][head_node_type]
-                head_node_config.update(head_config["node_config"])
-            launch_hash = hash_launch_conf(head_node_config, config['auth'])
-            # Populate tags
-            for node in vms:
+        self.ssh_key_path = os.path.expanduser(auth.PRIVATE_SSH_KEY_PATH)
+        remote_ssh_key = os.path.expanduser(REMOTE_RAY_SSH_KEY)
+        if os.path.exists(remote_ssh_key):
+            self.ssh_key_path = remote_ssh_key
+
+    def _guess_and_add_missing_tags(self, vms: Dict[str, Any]) -> None:
+        """Adds missing vms to local tag file and guesses their tags."""
+        for node in vms:
+            if self.metadata.exists(node['id']):
+                pass
+            elif node['name'] == f'{self.cluster_name}-head':
                 self.metadata[node['id']] = {
                     'tags': {
-                        TAG_RAY_CLUSTER_NAME: cluster_name,
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
                         TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
                         TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
                         TAG_RAY_USER_NODE_TYPE: 'ray_head_default',
-                        TAG_RAY_NODE_NAME: f'ray-{cluster_name}-head',
-                        TAG_RAY_LAUNCH_CONFIG: launch_hash,
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-head',
+                    }
+                }
+            elif node['name'] == f'{self.cluster_name}-worker':
+                self.metadata[node['id']] = {
+                    'tags': {
+                        TAG_RAY_CLUSTER_NAME: self.cluster_name,
+                        TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
+                        TAG_RAY_NODE_KIND: NODE_KIND_WORKER,
+                        TAG_RAY_USER_NODE_TYPE: 'ray_worker_default',
+                        TAG_RAY_NODE_NAME: f'ray-{self.cluster_name}-worker',
                     }
                 }
 
     def _list_instances_in_cluster(self) -> Dict[str, Any]:
         """List running instances in cluster."""
         vms = self.lambda_client.list_instances()
-        return [node for node in vms if node.get('name') == self.cluster_name]
+        possible_names = [
+            f'{self.cluster_name}-head', f'{self.cluster_name}-worker'
+        ]
+        return [node for node in vms if node.get('name') in possible_names]
 
     @synchronized
     def _get_filtered_nodes(self, tag_filters: Dict[str,
                                                     str]) -> Dict[str, Any]:
 
-        def match_tags(vm):
+        def _extract_metadata(vm: Dict[str, Any]) -> Dict[str, Any]:
+            metadata = {'id': vm['id'], 'status': vm['status'], 'tags': {}}
+            instance_info = self.metadata[vm['id']]
+            if instance_info is not None:
+                metadata['tags'] = instance_info['tags']
+            ip = vm['ip']
+            metadata['external_ip'] = ip
+            return metadata
+
+        def _match_tags(vm: Dict[str, Any]):
             vm_info = self.metadata[vm['id']]
             tags = {} if vm_info is None else vm_info['tags']
             for k, v in tag_filters.items():
                 if tags.get(k) != v:
                     return False
             return True
 
+        def _get_internal_ip(node: Dict[str, Any]):
+            runner = command_runner.SSHCommandRunner(node['external_ip'],
+                                                     'ubuntu',
+                                                     self.ssh_key_path)
+            out = runner.run(GET_INTERNAL_IP_CMD,
+                             require_outputs=True,
+                             stream_logs=False)
+            assert out[0] == 0
path: sky/skylet/providers/lambda_cloud/node_provider.py
text: @user1:
Do we want to raise an error and have more information in the error message to make it easier to debug in the future? We can use `subprocess_utils.handle_returncode()`

@author:
Good idea.

@author:
Done.