Pass host through LocalCluster to workers (dask#5427)

Previously the `host` parameter to `LocalCluster` would only be
forwarded to `Scheduler` instances and not `Worker`/`Nanny` instances,
leading to workers listening on non-localhost in some configurations.
This fixes that and adds a test.

Co-authored-by: James Bourbeau <[email protected]>

Author: 2 people

distributed/deploy/local.py

@@ -198,6 +198,7 @@ def __init__(
 
         worker_kwargs.update(
             {
+                "host": host,
                 "nthreads": threads_per_worker,
                 "services": worker_services,
                 "dashboard_address": worker_dashboard_address,

distributed/deploy/tests/test_local.py

@@ -7,6 +7,7 @@
 from distutils.version import LooseVersion
 from threading import Lock
 from time import sleep
+from urllib.parse import urlparse
 
 import pytest
 import tornado
@@ -1100,3 +1101,20 @@ async def test_cluster_info_sync():
 
         info = cluster.scheduler.get_metadata(keys=["cluster-manager-info"])
         assert info["foo"] == "bar"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("host", [None, "127.0.0.1"])
+@pytest.mark.parametrize("use_nanny", [True, False])
+async def test_cluster_host_used_throughout_cluster(host, use_nanny):
+    """Ensure that the `host` kwarg is propagated through scheduler, nanny, and workers"""
+    async with LocalCluster(host=host, asynchronous=True) as cluster:
+        url = urlparse(cluster.scheduler_address)
+        assert url.hostname == "127.0.0.1"
+        for worker in cluster.workers.values():
+            url = urlparse(worker.address)
+            assert url.hostname == "127.0.0.1"
+
+            if use_nanny:
+                url = urlparse(worker.process.worker_address)
+                assert url.hostname == "127.0.0.1"