Workers should use internal IP addresses to communicate with scheduler

dask_cloudprovider/aws/ec2.py

@@ -80,6 +80,8 @@ def __init__(
         self.iam_instance_profile = iam_instance_profile
         self.instance_tags = instance_tags
         self.volume_tags = volume_tags
+        # TODO: we have to use self.config.get("public_ingress", True)
+        #       instead for consistency with Azure and AWS providers
         self.use_private_ip = use_private_ip
         self.enable_detailed_monitoring = enable_detailed_monitoring
         self.spot = spot
@@ -223,7 +225,11 @@ async def configure_vm(self, client):
             await asyncio.sleep(min(backoff, 10) + backoff % 1)
             # Exponential backoff with a cap of 10 seconds and some jitter
             backoff = backoff * 2
-        return self.instance[ip_address_key], None
+
+        if self.use_private_ip:
+            return self.instance["PrivateIpAddress"], None
+
+        return self.instance["PrivateIpAddress"], self.instance["PublicIpAddress"]
 
     async def destroy_vm(self):
         boto_config = botocore.config.Config(retries=dict(max_attempts=10))

dask_cloudprovider/generic/vmcluster.py

@@ -99,6 +99,8 @@ async def start(self):
                 f"{self.cluster.protocol}://{external_ip}:{self.port}"
             )
 
+        self.contact_address = f"{self.cluster.protocol}://{internal_ip}:{self.port}"
+
         await self.wait_for_scheduler()
         await super().start()