Fix GCP VM leak issue (skypilot-org#1102)

* tmp

* const

* sky start safeguard

* address comments

* style

Author: infwinston

sky/backends/cloud_vm_ray_backend.py

@@ -510,6 +510,9 @@ def _update_blocklist_on_gcp_error(self, region, zones, stdout, stderr):
                     # when VM is preempted during creation.  This seems to be
                     # not documented by GCP.
                     self._blocked_zones.add(zone.name)
+                elif code in ['RESOURCE_NOT_READY']:
+                    # This code is returned when the VM is still STOPPING.
+                    self._blocked_zones.add(zone.name)
                 elif code == 8:
                     # Error code 8 means TPU resources is out of
                     # capacity. Example:

sky/skylet/providers/gcp/node.py

@@ -48,6 +48,8 @@ class inheriting from ``GCPNode``. Those classes are essentially dicts
 # considerably - this probably could be smaller
 # TPU deletion uses MAX_POLLS
 MAX_POLLS_TPU = MAX_POLLS * 8
+# Stopping instances can take several minutes, so we increase the timeout
+MAX_POLLS_STOP =  MAX_POLLS * 8
 POLL_INTERVAL = 5
 
 
@@ -134,6 +136,8 @@ class GCPNode(UserDict, metaclass=abc.ABCMeta):
 
     NON_TERMINATED_STATUSES = None
     RUNNING_STATUSES = None
+    STOPPED_STATUSES = None
+    STOPPING_STATUSES = None
     STATUS_FIELD = None
 
     def __init__(self, base_dict: dict, resource: "GCPResource", **kwargs) -> None:
@@ -147,6 +151,12 @@ def is_running(self) -> bool:
     def is_terminated(self) -> bool:
         return self.get(self.STATUS_FIELD) not in self.NON_TERMINATED_STATUSES
 
+    def is_stopped(self) -> bool:
+        return self.get(self.STATUS_FIELD) in self.STOPPED_STATUSES
+
+    def is_stopping(self) -> bool:
+        return self.get(self.STATUS_FIELD) in self.STOPPING_STATUSES
+
     @property
     def id(self) -> str:
         return self["name"]
@@ -173,6 +183,8 @@ class GCPComputeNode(GCPNode):
     # https://cloud.google.com/compute/docs/instances/instance-life-cycle
     NON_TERMINATED_STATUSES = {"PROVISIONING", "STAGING", "RUNNING"}
     RUNNING_STATUSES = {"RUNNING"}
+    STOPPED_STATUSES = {"TERMINATED"}
+    STOPPING_STATUSES = {"STOPPING"}
     STATUS_FIELD = "status"
 
     def get_labels(self) -> dict:
@@ -196,13 +208,14 @@ class GCPTPUNode(GCPNode):
 
     NON_TERMINATED_STATUSES = {"CREATING", "STARTING", "RESTARTING", "READY"}
     RUNNING_STATUSES = {"READY"}
+    STOPPED_STATUSES = {"STOPPED"}
+    STOPPING_STATUSES = {"STOPPING"}
     STATUS_FIELD = "state"
 
     # SKY: get status of TPU VM for status filtering
     def get_status(self) -> str:
         return self.get(self.STATUS_FIELD)
 
-
     def get_labels(self) -> dict:
         return self.get("labels", {})
 

sky/skylet/providers/gcp/node_provider.py

@@ -29,6 +29,8 @@
     GCPNodeType,
     INSTANCE_NAME_MAX_LEN,
     INSTANCE_NAME_UUID_LEN,
+    MAX_POLLS_STOP,
+    POLL_INTERVAL,
 )
 
 logger = logging.getLogger(__name__)
@@ -190,10 +192,12 @@ def create_node(self, base_config: dict, tags: dict, count: int) -> None:
                 if TAG_RAY_USER_NODE_TYPE in labels:
                     filters[TAG_RAY_USER_NODE_TYPE] = labels[TAG_RAY_USER_NODE_TYPE]
                 # SKY: "TERMINATED" for compute VM, "STOPPED" for TPU VM
+                # "STOPPING" means the VM is being stopped, which needs
+                # to be included to avoid creating a new VM.
                 if isinstance(resource, GCPCompute):
-                    STOPPED_STATUS = ["TERMINATED"]
+                    STOPPED_STATUS = ["TERMINATED", "STOPPING"]
                 else:
-                    STOPPED_STATUS = ["STOPPED"]
+                    STOPPED_STATUS = ["STOPPED", "STOPPING"]
                 reuse_nodes = resource._list_instances(
                     filters, STOPPED_STATUS)[:count]
                 reuse_node_ids = [n.id for n in reuse_nodes]
@@ -229,6 +233,26 @@ def terminate_node(self, node_id: str):
                         ),
                     )
                     result = resource.stop_instance(node_id=node_id)
+
+                    # Check if the instance is actually stopped.
+                    # GCP does not fully stop an instance even after
+                    # the stop operation is finished.
+                    for _ in range(MAX_POLLS_STOP):
+                        instance = resource.get_instance(node_id=node_id)
+                        if instance.is_stopped():
+                            logger.info(f"Instance {node_id} is stopped.")
+                            break
+                        elif instance.is_stopping():
+                            time.sleep(POLL_INTERVAL)
+                        else:
+                            raise RuntimeError(f"Unexpected instance status."
+                                               " Details: {instance}")
+
+                    if instance.is_stopping():
+                        raise RuntimeError(f"Maximum number of polls: "
+                                           f"{MAX_POLLS_STOP} reached. "
+                                           f"Instance {node_id} is still in "
+                                           "STOPPING status.")
                 else:
                     result = resource.delete_instance(
                         node_id=node_id,

sky/utils/cli_utils/status_utils.py

@@ -181,8 +181,6 @@ def show_local_status_table(local_clusters: List[str]):
     cluster_status['launched_at']))
 _get_region = (
     lambda clusters_status: clusters_status['handle'].launched_resources.region)
-_get_zone = (
-    lambda clusters_status: clusters_status['handle'].launched_resources.zone)
 _get_status = (lambda cluster_status: cluster_status['status'].value)
 _get_duration = (lambda cluster_status: log_utils.readable_time_duration(
     cluster_status['launched_at']))
@@ -205,6 +203,13 @@ def _get_resources(cluster_status):
     return resources_str
 
 
+def _get_zone(cluster_status):
+    zone_str = cluster_status['handle'].launched_resources.zone
+    if zone_str is None:
+        zone_str = '-'
+    return zone_str
+
+
 def _get_autostop(cluster_status):
     autostop_str = '-'
     if cluster_status['autostop'] >= 0: