update

python/ray/autoscaler/autoscaler.py

@@ -87,6 +87,7 @@ def __init__(self):
         self.last_used_time_by_ip = {}
         self.static_resources_by_ip = {}
         self.dynamic_resources_by_ip = {}
+        self.local_ip = services.get_node_ip_address()
 
     def update(self, ip, static_resources, dynamic_resources):
         self.static_resources_by_ip[ip] = static_resources
@@ -97,6 +98,7 @@ def update(self, ip, static_resources, dynamic_resources):
 
     def prune_active_ips(self, active_ips):
         active_ips = set(active_ips)
+        active_ips.add(self.local_ip)
 
         def prune(mapping):
             unwanted = set(mapping) - active_ips

python/ray/autoscaler/aws/example.yaml

@@ -73,7 +73,7 @@ head_init_commands:
 #    - pip3 install --user -U https://s3-us-west-2.amazonaws.com/ray-wheels/f5ea44338eca392df3a868035df3901829cc2ca1/ray-0.3.0-cp35-cp35m-manylinux1_x86_64.whl
 #    - pip3 install --user cython
 #    - git clone https://github.com/ericl/ray.git || true
-    - cd ray && git fetch && git checkout db01d3e0
+    - cd ray && git fetch && git checkout 419ca59b
 #    - cd ray/python && python setup.py develop --user
     - yes | ~/anaconda3/bin/conda install boto3=1.4.8  # 1.4.8 adds InstanceMarketOptions
     - ray stop

python/ray/autoscaler/aws/node_provider.py

@@ -12,6 +12,8 @@ class AWSNodeProvider(NodeProvider):
     def __init__(self, provider_config, cluster_name):
         NodeProvider.__init__(self, provider_config, cluster_name)
         self.ec2 = boto3.resource("ec2", region_name=provider_config["region"])
+        self.internal_ip_cache = {}
+        self.external_ip_cache = {}
 
     def nodes(self, tag_filters):
         filters = [
@@ -49,12 +51,22 @@ def node_tags(self, node_id):
         return tags
 
     def external_ip(self, node_id):
+        if node_id in self.external_ip_cache:
+            return self.external_ip_cache[node_id]
         node = self._node(node_id)
-        return node.public_ip_address
+        ip = node.public_ip_address
+        if ip:
+            self.external_ip_cache[node_id] = ip
+        return ip
 
     def internal_ip(self, node_id):
+        if node_id in self.internal_ip_cache:
+            return self.internal_ip_cache[node_id]
         node = self._node(node_id)
-        return node.private_ip_address
+        ip = node.private_ip_address
+        if ip:
+            self.internal_ip_cache[node_id] = ip
+        return ip
 
     def set_node_tags(self, node_id, tags):
         node = self._node(node_id)