[autoscaler] Change the get behavior of node providers' _get_node (#4132)

* Change the get behavior of GCPNodeProvider._get_node

* Add lock around the GCPNodeProvider._get_node call

* rename nodes

* lint

* Update GCPNodeProvider._get_node to match aws implementation

* assert

* log

* log highest heartbeats

* rename

* bringup to connected

* prune heartbeat times

* fix bringup

python/ray/autoscaler/autoscaler.py

@@ -155,6 +155,8 @@ class LoadMetrics(object):
         self.last_heartbeat_time_by_ip[ip] = now
 
     def mark_active(self, ip):
+        assert ip is not None, "IP should be known at this time"
+        logger.info("Node {} is newly setup, treating as active".format(ip))
         self.last_heartbeat_time_by_ip[ip] = time.time()
 
     def prune_active_ips(self, active_ips):
@@ -177,10 +179,14 @@ class LoadMetrics(object):
         prune(self.last_used_time_by_ip)
         prune(self.static_resources_by_ip)
         prune(self.dynamic_resources_by_ip)
+        prune(self.last_heartbeat_time_by_ip)
 
     def approx_workers_used(self):
         return self._info()["NumNodesUsed"]
 
+    def num_workers_connected(self):
+        return self._info()["NumNodesConnected"]
+
     def info_string(self):
         return ", ".join(
             ["{}={}".format(k, v) for k, v in sorted(self._info().items())])
@@ -210,6 +216,13 @@ class LoadMetrics(object):
         heartbeat_times = [
             now - t for t in self.last_heartbeat_time_by_ip.values()
         ]
+        most_delayed_heartbeats = sorted(
+            list(self.last_heartbeat_time_by_ip.items()),
+            key=lambda pair: pair[1])[:5]
+        most_delayed_heartbeats = {
+            ip: (now - t)
+            for ip, t in most_delayed_heartbeats
+        }
         return {
             "ResourceUsage": ", ".join([
                 "{}/{} {}".format(
@@ -227,6 +240,7 @@ class LoadMetrics(object):
                 int(np.min(heartbeat_times)) if heartbeat_times else -1,
                 int(np.mean(heartbeat_times)) if heartbeat_times else -1,
                 int(np.max(heartbeat_times)) if heartbeat_times else -1),
+            "MostDelayedHeartbeats": most_delayed_heartbeats,
         }
 
 
@@ -239,7 +253,7 @@ class NodeLauncher(threading.Thread):
 
     def _launch_node(self, config, count):
         tag_filters = {TAG_RAY_NODE_TYPE: "worker"}
-        before = self.provider.nodes(tag_filters=tag_filters)
+        before = self.provider.non_terminated_nodes(tag_filters=tag_filters)
         launch_hash = hash_launch_conf(config["worker_nodes"], config["auth"])
         self.provider.create_node(
             config["worker_nodes"], {
@@ -249,7 +263,7 @@ class NodeLauncher(threading.Thread):
                 TAG_RAY_NODE_STATUS: "uninitialized",
                 TAG_RAY_LAUNCH_CONFIG: launch_hash,
             }, count)
-        after = self.provider.nodes(tag_filters=tag_filters)
+        after = self.provider.non_terminated_nodes(tag_filters=tag_filters)
         if set(after).issubset(before):
             logger.error("NodeLauncher: "
                          "No new nodes reported after node creation")
@@ -430,7 +444,8 @@ class StandardAutoscaler(object):
             self.launch_new_node(num_launches)
             nodes = self.workers()
             self.log_info_string(nodes)
-        else:
+        elif self.load_metrics.num_workers_connected() >= target_workers:
+            logger.info("Ending bringup phase")
             self.bringup = False
 
         # Process any completed updates
@@ -492,15 +507,15 @@ class StandardAutoscaler(object):
                                  "Error parsing config.")
 
     def target_num_workers(self):
-        initial_workers = self.config["initial_workers"]
+        target_frac = self.config["target_utilization_fraction"]
+        cur_used = self.load_metrics.approx_workers_used()
+        ideal_num_nodes = int(np.ceil(cur_used / float(target_frac)))
+        ideal_num_workers = ideal_num_nodes - 1  # subtract 1 for head node
 
         if self.bringup:
-            ideal_num_workers = initial_workers
-        else:
-            target_frac = self.config["target_utilization_fraction"]
-            cur_used = self.load_metrics.approx_workers_used()
-            ideal_num_nodes = int(np.ceil(cur_used / float(target_frac)))
-            ideal_num_workers = ideal_num_nodes - 1  # subtract 1 for head node
+            ideal_num_workers = max(ideal_num_workers,
+                                    self.config["initial_workers"])
+
         return min(self.config["max_workers"],
                    max(self.config["min_workers"], ideal_num_workers))
 
@@ -603,7 +618,8 @@ class StandardAutoscaler(object):
         self.launch_queue.put((config, count))
 
     def workers(self):
-        return self.provider.nodes(tag_filters={TAG_RAY_NODE_TYPE: "worker"})
+        return self.provider.non_terminated_nodes(
+            tag_filters={TAG_RAY_NODE_TYPE: "worker"})
 
     def log_info_string(self, nodes):
         logger.info("StandardAutoscaler: {}".format(self.info_string(nodes)))