[tune] Fast Node Recovery (#5053)

python/ray/autoscaler/autoscaler.py

@@ -196,6 +196,7 @@ class LoadMetrics(object):
                     "LoadMetrics: "
                     "Removed {} stale ip mappings: {} not in {}".format(
                         len(unwanted), unwanted, active_ips))
+            assert not (unwanted & set(mapping))
 
         prune(self.last_used_time_by_ip)
         prune(self.static_resources_by_ip)
@@ -266,16 +267,18 @@ class LoadMetrics(object):
 
 
 class NodeLauncher(threading.Thread):
-    def __init__(self, provider, queue, pending, *args, **kwargs):
+    def __init__(self, provider, queue, pending, index=None, *args, **kwargs):
         self.queue = queue
         self.pending = pending
         self.provider = provider
+        self.index = str(index) if index is not None else ""
         super(NodeLauncher, self).__init__(*args, **kwargs)
 
     def _launch_node(self, config, count):
-        tag_filters = {TAG_RAY_NODE_TYPE: "worker"}
-        before = self.provider.non_terminated_nodes(tag_filters=tag_filters)
+        worker_filter = {TAG_RAY_NODE_TYPE: "worker"}
+        before = self.provider.non_terminated_nodes(tag_filters=worker_filter)
         launch_hash = hash_launch_conf(config["worker_nodes"], config["auth"])
+        self.log("Launching {} nodes.".format(count))
         self.provider.create_node(
             config["worker_nodes"], {
                 TAG_RAY_NODE_NAME: "ray-{}-worker".format(
@@ -284,19 +287,25 @@ class NodeLauncher(threading.Thread):
                 TAG_RAY_NODE_STATUS: "uninitialized",
                 TAG_RAY_LAUNCH_CONFIG: launch_hash,
             }, count)
-        after = self.provider.non_terminated_nodes(tag_filters=tag_filters)
+        after = self.provider.non_terminated_nodes(tag_filters=worker_filter)
         if set(after).issubset(before):
-            logger.error("NodeLauncher: "
-                         "No new nodes reported after node creation")
+            self.log("No new nodes reported after node creation.")
 
     def run(self):
         while True:
             config, count = self.queue.get()
+            self.log("Got {} nodes to launch.".format(count))
             try:
                 self._launch_node(config, count)
+            except Exception:
+                logger.exception("Launch failed")
             finally:
                 self.pending.dec(count)
 
+    def log(self, statement):
+        prefix = "NodeLauncher{}:".format(self.index)
+        logger.info(prefix + " {}".format(statement))
+
 
 class ConcurrentCounter():
     def __init__(self):
@@ -375,6 +384,7 @@ class StandardAutoscaler(object):
             node_launcher = NodeLauncher(
                 provider=self.provider,
                 queue=self.launch_queue,
+                index=i,
                 pending=self.num_launches_pending)
             node_launcher.daemon = True
             node_launcher.start()
@@ -633,8 +643,8 @@ class StandardAutoscaler(object):
         return True
 
     def launch_new_node(self, count):
-        logger.info("StandardAutoscaler: "
-                    "Launching {} new nodes".format(count))
+        logger.info(
+            "StandardAutoscaler: Queue {} new nodes for launch".format(count))
         self.num_launches_pending.inc(count)
         config = copy.deepcopy(self.config)
         self.launch_queue.put((config, count))

python/ray/autoscaler/aws/node_provider.py

@@ -5,8 +5,10 @@ from __future__ import print_function
 import random
 import threading
 from collections import defaultdict
+import logging
 
 import boto3
+import botocore
 from botocore.config import Config
 
 from ray.autoscaler.node_provider import NodeProvider
@@ -14,7 +16,6 @@ from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME, TAG_RAY_NODE_NAME
 from ray.ray_constants import BOTO_MAX_RETRIES
 from ray.autoscaler.log_timer import LogTimer
 
-import logging
 logger = logging.getLogger(__name__)
 
 
@@ -207,23 +208,36 @@ class AWSNodeProvider(NodeProvider):
         # SubnetIds is not a real config key: we must resolve to a
         # single SubnetId before invoking the AWS API.
         subnet_ids = conf.pop("SubnetIds")
-        subnet_id = subnet_ids[self.subnet_idx % len(subnet_ids)]
-        self.subnet_idx += 1
-        conf.update({
-            "MinCount": 1,
-            "MaxCount": count,
-            "SubnetId": subnet_id,
-            "TagSpecifications": tag_specs
-        })
 
-        logger.info(
-            "NodeProvider: Calling create_instances (count={}).".format(count))
-        L = self.ec2.create_instances(**conf)
-        for x in L:
-            logger.info("NodeProvider: Created instance "
-                        "[id={}, name={}, info={}]".format(
-                            x.instance_id, x.state["Name"],
-                            x.state_reason["Message"]))
+        max_retries = 5
+        for attempt in range(1, max_retries + 1):
+            try:
+                subnet_id = subnet_ids[self.subnet_idx % len(subnet_ids)]
+                logger.info("NodeProvider: calling create_instances "
+                            "with {} (count={}).".format(subnet_id, count))
+                self.subnet_idx += 1
+                conf.update({
+                    "MinCount": 1,
+                    "MaxCount": count,
+                    "SubnetId": subnet_id,
+                    "TagSpecifications": tag_specs
+                })
+                created = self.ec2.create_instances(**conf)
+                for instance in created:
+                    logger.info("NodeProvider: Created instance "
+                                "[id={}, name={}, info={}]".format(
+                                    instance.instance_id,
+                                    instance.state["Name"],
+                                    instance.state_reason["Message"]))
+                break
+            except botocore.exceptions.ClientError as exc:
+                if attempt == max_retries:
+                    logger.error(
+                        "create_instances: Max attempts ({}) exceeded.".format(
+                            max_retries))
+                    raise exc
+                else:
+                    logger.error(exc)
 
     def terminate_node(self, node_id):
         node = self._get_cached_node(node_id)

python/ray/state.py

@@ -397,8 +397,12 @@ class GlobalState(object):
             Information about the Ray clients in the cluster.
         """
         self._check_connected()
+        client_table = _parse_client_table(self.redis_client)
 
-        return _parse_client_table(self.redis_client)
+        for client in client_table:
+            # These are equivalent and is better for application developers.
+            client["alive"] = client["IsInsertion"]
+        return client_table
 
     def _job_table(self, job_id):
         """Fetch and parse the job table information for a single job ID.

python/ray/tune/ray_trial_executor.py

@@ -288,7 +288,29 @@ class RayTrialExecutor(TrialExecutor):
 
         return list(self._running.values())
 
+    def get_alive_node_ips(self):
+        nodes = ray.state.nodes()
+        ip_addresses = set()
+        for node in nodes:
+            if node["alive"]:
+                ip_addresses.add(node["NodeManagerAddress"])
+        return ip_addresses
+
+    def get_current_trial_ips(self):
+        return {t.node_ip for t in self.get_running_trials()}
+
     def get_next_available_trial(self):
+        if ray.worker._mode() != ray.worker.LOCAL_MODE:
+            live_cluster_ips = self.get_alive_node_ips()
+            if live_cluster_ips - self.get_current_trial_ips():
+                for trial in self.get_running_trials():
+                    if trial.node_ip and trial.node_ip not in live_cluster_ips:
+                        logger.warning(
+                            "{} (ip: {}) detected as stale. This is likely "
+                            "because the node was lost. Processing this "
+                            "trial first.".format(trial, trial.node_ip))
+                        return trial
+
         shuffled_results = list(self._running.keys())
         random.shuffle(shuffled_results)
         # Note: We shuffle the results because `ray.wait` by default returns
@@ -541,8 +563,15 @@ class RayTrialExecutor(TrialExecutor):
                 assert type(value) != Checkpoint, type(value)
                 trial.runner.restore_from_object.remote(value)
             else:
-                worker_ip = ray.get(trial.runner.current_ip.remote())
-                trial.sync_logger_to_new_location(worker_ip)
+                # TODO: Somehow, the call to get the current IP on the
+                # remote actor can be very slow - a better fix would
+                # be to use an actor table to detect the IP of the Trainable
+                # and rsync the files there.
+                # See https://github.com/ray-project/ray/issues/5168
+                with warn_if_slow("get_current_ip"):
+                    worker_ip = ray.get(trial.runner.current_ip.remote())
+                with warn_if_slow("sync_to_new_location"):
+                    trial.sync_logger_to_new_location(worker_ip)
                 with warn_if_slow("restore_from_disk"):
                     ray.get(trial.runner.restore.remote(value))
             trial.last_result = checkpoint.last_result

python/ray/tune/trainable.py

@@ -114,6 +114,7 @@ class Trainable(object):
         return ""
 
     def current_ip(self):
+        logger.warning("Getting current IP.")
         self._local_ip = ray.services.get_node_ip_address()
         return self._local_ip
 

python/ray/tune/trial.py

@@ -559,6 +559,10 @@ class Trial(object):
     def is_finished(self):
         return self.status in [Trial.TERMINATED, Trial.ERROR]
 
+    @property
+    def node_ip(self):
+        return self.last_result.get("node_ip")
+
     def __repr__(self):
         return str(self)
 

python/ray/tune/trial_runner.py

@@ -599,9 +599,20 @@ class TrialRunner(object):
 
         This does not notify the SearchAlgorithm because the function
         evaluation is still in progress.
+
         """
         self._scheduler_alg.on_trial_error(self, trial)
         self.trial_executor.set_status(trial, Trial.PENDING)
+
+        # TODO(rliaw): Right now, this pushes the trial to the end of queue
+        # because restoration can be expensive. However, this is not
+        # ideal since it just hides the issue - a better fix would
+        # be to use an actor table to detect the IP of the Trainable
+        # and rsync the files there.
+        # See https://github.com/ray-project/ray/issues/5168
+        self._trials.pop(self._trials.index(trial))
+        self._trials.append(trial)
+
         with warn_if_slow("scheduler.on_trial_add"):
             self._scheduler_alg.on_trial_add(self, trial)