Add synced dict between cluster and scheduler to store cluster info

distributed/deploy/cluster.py

@@ -1,4 +1,5 @@
 import asyncio
+import copy
 import datetime
 import logging
 import threading
@@ -14,7 +15,15 @@
 
 from ..core import Status
 from ..objects import SchedulerInfo
-from ..utils import Log, Logs, format_dashboard_link, log_errors, sync, thread_state
+from ..utils import (
+    Log,
+    Logs,
+    format_dashboard_link,
+    log_errors,
+    sync,
+    thread_state,
+    typename,
+)
 from .adaptive import Adaptive
 
 logger = logging.getLogger(__name__)
@@ -46,7 +55,7 @@ class Cluster:
     _supports_scaling = True
     name = None
 
-    def __init__(self, asynchronous, quiet=False, name=None):
+    def __init__(self, asynchronous, quiet=False, name=None, scheduler_sync_interval=1):
         self.scheduler_info = {"workers": {}}
         self.periodic_callbacks = {}
         self._asynchronous = asynchronous
@@ -56,13 +65,24 @@ def __init__(self, asynchronous, quiet=False, name=None):
         self.quiet = quiet
         self.scheduler_comm = None
         self._adaptive = None
+        self._sync_interval = parse_timedelta(
+            scheduler_sync_interval, default="seconds"
+        )
+
+        if name is None:
+            name = str(uuid.uuid4())[:8]
 
-        if name is not None:
-            self.name = name
-        elif self.name is None:
-            self.name = str(uuid.uuid4())[:8]
+        self.cluster_info = {"name": name, "type": typename(type(self))}
         self.status = Status.created
 
+    @property
+    def name(self):
+        return self.cluster_info["name"]
+
+    @name.setter
+    def name(self, name):
+        self.cluster_info["name"] = name
+
     async def _start(self):
         comm = await self.scheduler_comm.live_comm()
         await comm.write({"op": "subscribe_worker_status"})
@@ -71,8 +91,25 @@ async def _start(self):
         self._watch_worker_status_task = asyncio.ensure_future(
             self._watch_worker_status(comm)
         )
+
+        info = await self.scheduler_comm.get_metadata(
+            keys=["cluster-manager-info"], default={}
+        )
+        self.cluster_info.update(info)
+
+        self.periodic_callbacks["sync-cluster-info"] = PeriodicCallback(
+            self._sync_cluster_info, self._sync_interval * 1000
+        )
+        for pc in self.periodic_callbacks.values():
+            pc.start()
         self.status = Status.running
 
+    async def _sync_cluster_info(self):
+        await self.scheduler_comm.set_metadata(
+            keys=["cluster-manager-info"],
+            value=copy.copy(self.cluster_info),
+        )
+
     async def _close(self):
         if self.status == Status.closed:
             return
@@ -85,12 +122,12 @@ async def _close(self):
         if self._watch_worker_status_task:
             await self._watch_worker_status_task
 
-        for pc in self.periodic_callbacks.values():
-            pc.stop()
-
         if self.scheduler_comm:
             await self.scheduler_comm.close_rpc()
 
+        for pc in self.periodic_callbacks.values():
+            pc.stop()
+
         self.status = Status.closed
 
     def close(self, timeout=None):

distributed/deploy/local.py

@@ -121,6 +121,7 @@ def __init__(
         interface=None,
         worker_class=None,
         scheduler_kwargs=None,
+        scheduler_sync_interval=1,
         **worker_kwargs,
     ):
         if ip is not None:
@@ -241,6 +242,7 @@ def __init__(
             asynchronous=asynchronous,
             silence_logs=silence_logs,
             security=security,
+            scheduler_sync_interval=scheduler_sync_interval,
         )
 
     def start_worker(self, *args, **kwargs):

distributed/deploy/spec.py

@@ -287,6 +287,7 @@ def __init__(
         silence_logs=False,
         name=None,
         shutdown_on_close=True,
+        scheduler_sync_interval=1,
     ):
         self._created = weakref.WeakSet()
 
@@ -316,6 +317,7 @@ def __init__(
         super().__init__(
             asynchronous=asynchronous,
             name=name,
+            scheduler_sync_interval=scheduler_sync_interval,
         )
 
         if not self.asynchronous:

distributed/deploy/tests/test_local.py

@@ -1073,3 +1073,20 @@ async def test_local_cluster_redundant_kwarg(nanny):
             async with Client(cluster) as c:
                 f = c.submit(sleep, 0)
                 await f
+
+
+@pytest.mark.asyncio
+async def test_cluster_info_sync():
+    async with LocalCluster(
+        processes=False, asynchronous=True, scheduler_sync_interval="1ms"
+    ) as cluster:
+        assert cluster.cluster_info["name"] == cluster.name
+        await asyncio.sleep(0.01)
+        info = await cluster.scheduler_comm.get_metadata(keys=["cluster-manager-info"])
+        assert info["name"] == cluster.name
+        info = cluster.scheduler.get_metadata(keys=["cluster-manager-info"])
+        assert info["name"] == cluster.name
+
+        cluster.cluster_info["foo"] = "bar"
+        await asyncio.sleep(0.01)
+        assert "foo" in cluster.scheduler.get_metadata(keys=["cluster-manager-info"])