Repository: dask/distributed

Issue:
Scheduler crashes in SSHCluster in 2023.3.2 but not in 2023.3.1
**Describe the issue**: Attempting to use the SSHCluster does not work in 2023.3.2 because the scheduler exits early with an exit code of 1

```
INFO:distributed.deploy.ssh:2023-03-29 18:21:07,199 - distributed.http.proxy - INFO - To route to workers diagnostics web server please install jupyter-server-proxy: python -m pip install jupyter-server-proxy
2023-03-29 18:21:07,204 - distributed.deploy.ssh - INFO - 2023-03-29 18:21:07,204 - distributed.scheduler - INFO - State start
INFO:distributed.deploy.ssh:2023-03-29 18:21:07,204 - distributed.scheduler - INFO - State start
2023-03-29 18:21:07,207 - distributed.deploy.ssh - INFO - 2023-03-29 18:21:07,206 - distributed.scheduler - DEBUG - Clear task state
INFO:distributed.deploy.ssh:2023-03-29 18:21:07,206 - distributed.scheduler - DEBUG - Clear task state
2023-03-29 18:21:07,207 - distributed.deploy.ssh - INFO - 2023-03-29 18:21:07,207 - distributed.scheduler - INFO -   Scheduler at:   tcp://10.15.40.68:36143
INFO:distributed.deploy.ssh:2023-03-29 18:21:07,207 - distributed.scheduler - INFO -   Scheduler at:   tcp://10.15.40.68:36143
INFO:asyncssh:[conn=0, chan=1] Received exit status 1
INFO:asyncssh:[conn=0, chan=1] Received channel close
INFO:asyncssh:[conn=0, chan=1] Channel closed
INFO:asyncssh:[conn=0, chan=1] Sending KILL signal
```

When rolling back to 2023.3.1 the scheduler starts sucessfully:

```
2023-03-29 18:23:33,874 - distributed.deploy.ssh - INFO - 2023-03-29 18:23:33,873 - distributed.http.proxy - INFO - To route to workers diagnostics web server please install jupyter-server-proxy: python -m pip install jupyter-server-proxy
INFO:distributed.deploy.ssh:2023-03-29 18:23:33,873 - distributed.http.proxy - INFO - To route to workers diagnostics web server please install jupyter-server-proxy: python -m pip install jupyter-server-proxy
2023-03-29 18:23:33,878 - distributed.deploy.ssh - INFO - 2023-03-29 18:23:33,878 - distributed.scheduler - INFO - State start
INFO:distributed.deploy.ssh:2023-03-29 18:23:33,878 - distributed.scheduler - INFO - State start
2023-03-29 18:23:33,882 - distributed.deploy.ssh - INFO - 2023-03-29 18:23:33,881 - distributed.scheduler - DEBUG - Clear task state
INFO:distributed.deploy.ssh:2023-03-29 18:23:33,881 - distributed.scheduler - DEBUG - Clear task state
2023-03-29 18:23:33,883 - distributed.deploy.ssh - INFO - 2023-03-29 18:23:33,882 - distributed.scheduler - INFO -   Scheduler at:   tcp://10.15.40.68:40305
INFO:distributed.deploy.ssh:2023-03-29 18:23:33,882 - distributed.scheduler - INFO -   Scheduler at:   tcp://10.15.40.68:40305
INFO:asyncssh:Opening SSH connection to localhost, port 22
INFO:asyncssh:[conn=1] Connected to SSH server at localhost, port 22
```

**Minimal Complete Verifiable Example**:

```python
import logging
logging.basicConfig(level=logging.DEBUG)

from distributed import SSHCluster
cluster = SSHCluster(["localhost", "localhost"])
```

**Anything else we need to know?**: Full repro here:
```
git clone https://github.com/jabbera/distributed-bug.git
cd distributed-bug
python3.10 -m venv .venv
source .venv/bin/activate
pip install -r requirements-bug.txt
python demo.py
```

**Environment**:

- Dask version: 2023.3.2
- Python version: 3.10.5
- Operating System: Ubuntu 20.04.5 LTS
- Install method (conda, pip, source): pip


Discussion:
jacobtomlinson: Awesome thanks @milesgranger. Looks like black is unhappy now though.

Patch:
diff --git a/distributed/compatibility.py b/distributed/compatibility.py
index 30498da6..851a860b 100644
--- a/distributed/compatibility.py
+++ b/distributed/compatibility.py
@@ -42,17 +42,10 @@ else:
 if sys.version_info >= (3, 9):
     from random import randbytes
 else:
-    try:
-        import numpy
+    from random import getrandbits
 
-        def randbytes(size):
-            return numpy.random.randint(255, size=size, dtype="u8").tobytes()
-
-    except ImportError:
-        import secrets
-
-        def randbytes(size):
-            return secrets.token_bytes(size)
+    def randbytes(size):
+        return getrandbits(size * 8).to_bytes(size, "little")
 
 
 if tornado.version_info >= (6, 2, 0, 0):
diff --git a/distributed/scheduler.py b/distributed/scheduler.py
index 376dd368..1f9e0951 100644
--- a/distributed/scheduler.py
+++ b/distributed/scheduler.py
@@ -3872,7 +3872,15 @@ class Scheduler(SchedulerState, ServerNode):
         for name, server in self.services.items():
             if name == "dashboard":
                 addr = get_address_host(listener.contact_address)
-                link = format_dashboard_link(addr, server.port)
+                try:
+                    link = format_dashboard_link(addr, server.port)
+                # formatting dashboard link can fail if distributed.dashboard.link
+                # refers to non-existant env vars.
+                except KeyError as e:
+                    logger.warning(
+                        f"Failed to format dashboard link, unknown value: {e}"
+                    )
+                    link = f":{server.port}"
             else:
                 link = f"{listen_ip}:{server.port}"
             logger.info("%11s at:  %25s", name, link)
diff --git a/distributed/utils.py b/distributed/utils.py
index 4c1a6642..52a2cc11 100644
--- a/distributed/utils.py
+++ b/distributed/utils.py
@@ -67,7 +67,7 @@ from dask.utils import parse_timedelta as _parse_timedelta
 from dask.widgets import get_template
 
 from distributed.compatibility import WINDOWS
-from distributed.metrics import monotonic, time
+from distributed.metrics import context_meter, monotonic, time
 
 try:
     from dask.context import thread_state
@@ -1460,7 +1460,8 @@ def offload(
     run_in_executor_with_context
     https://bugs.python.org/issue34014
     """
-    return run_in_executor_with_context(_offload_executor, func, *args, **kwargs)
+    with context_meter.meter("offload"):
+        return run_in_executor_with_context(_offload_executor, func, *args, **kwargs)
 
 
 class EmptyContext:
diff --git a/distributed/worker.py b/distributed/worker.py
index 1513a045..f443237c 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -2269,20 +2269,29 @@ class Worker(BaseWorker, ServerNode):
                         self.scheduler_delay,
                     )
                 elif "ThreadPoolExecutor" in str(type(e)):
-                    result = await run_in_executor_with_context(
-                        e,
-                        apply_function,
-                        function,
-                        args2,
-                        kwargs2,
-                        self.execution_state,
-                        key,
-                        self.active_threads,
-                        self.active_threads_lock,
-                        self.scheduler_delay,
-                    )
+                    # The 'executor' time metric should be almost zero most of the time,
+                    # e.g. thread synchronization overhead only, since thread-noncpu and
+                    # thread-cpu inside the thread detract from it. However, it may
+                    # become substantial in case of misalignment between the size of the
+                    # thread pool and the number of running tasks in the worker state
+                    # machine (e.g. https://github.com/dask/distributed/issues/5882)
+                    with context_meter.meter("executor"):
+                        result = await run_in_executor_with_context(
+                            e,
+                            apply_function,
+                            function,
+                            args2,
+                            kwargs2,
+                            self.execution_state,
+                            key,
+                            self.active_threads,
+                            self.active_threads_lock,
+                            self.scheduler_delay,
+                        )
                 else:
-                    # Can't capture contextvars across processes
+                    # Can't capture contextvars across processes. If this is a
+                    # ProcessPoolExecutor, the 'executor' time metric will show the
+                    # whole runtime inside the executor.
                     with context_meter.meter("executor"):
                         result = await self.loop.run_in_executor(
                             e,


Test patch:
diff --git a/distributed/tests/test_compatibility.py b/distributed/tests/test_compatibility.py
new file mode 100644
index 00000000..2786f942
--- /dev/null
+++ b/distributed/tests/test_compatibility.py
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import random
+from collections import Counter
+
+from distributed.compatibility import randbytes
+
+
+def test_randbytes():
+    x = randbytes(256_000)
+    assert isinstance(x, bytes)
+    assert len(x) == 256_000
+    c = Counter(x)
+    for i in range(256):
+        assert 800 < c[i] < 1200, (i, c[i])
+
+
+def test_randbytes_seed():
+    state = random.getstate()
+    try:
+        random.seed(0)
+        assert randbytes(8) == b"\xcd\x07,\xd8\xbeo\x9fb"
+    finally:
+        random.setstate(state)
diff --git a/distributed/tests/test_scheduler.py b/distributed/tests/test_scheduler.py
index a52f22b7..abd5bc61 100644
--- a/distributed/tests/test_scheduler.py
+++ b/distributed/tests/test_scheduler.py
@@ -2829,28 +2829,44 @@ async def test_too_many_groups(c, s, a, b):
 
 
 @gen_test()
-async def test_multiple_listeners():
-    with captured_logger("distributed.scheduler") as log:
-        async with Scheduler(dashboard_address=":0", protocol=["inproc", "tcp"]) as s:
-            async with Worker(s.listeners[0].contact_address) as a:
-                async with Worker(s.listeners[1].contact_address) as b:
-                    assert a.address.startswith("inproc")
-                    assert a.scheduler.address.startswith("inproc")
-                    assert b.address.startswith("tcp")
-                    assert b.scheduler.address.startswith("tcp")
-
-                    async with Client(s.address, asynchronous=True) as c:
-                        futures = c.map(inc, range(20))
-                        await wait(futures)
-
-                        # Force inter-worker communication both ways
-                        await c.submit(sum, futures, workers=[a.address])
-                        await c.submit(len, futures, workers=[b.address])
+@pytest.mark.parametrize(
+    "dashboard_link_template,expected_dashboard_link",
+    (
+        ("{scheme}://{host}:{port}/status", r"dashboard at:\s*http://"),
+        ("{ENV_VAR_MISSING}", r"dashboard at:\s*:\d*"),
+    ),
+)
+async def test_multiple_listeners(dashboard_link_template, expected_dashboard_link):
+    with dask.config.set({"distributed.dashboard.link": dashboard_link_template}):
+        with captured_logger("distributed.scheduler") as log:
+            async with Scheduler(
+                dashboard_address=":0", protocol=["inproc", "tcp"]
+            ) as s:
+                async with Worker(s.listeners[0].contact_address) as a:
+                    async with Worker(s.listeners[1].contact_address) as b:
+                        assert a.address.startswith("inproc")
+                        assert a.scheduler.address.startswith("inproc")
+                        assert b.address.startswith("tcp")
+                        assert b.scheduler.address.startswith("tcp")
+
+                        async with Client(s.address, asynchronous=True) as c:
+                            futures = c.map(inc, range(20))
+                            await wait(futures)
+
+                            # Force inter-worker communication both ways
+                            await c.submit(sum, futures, workers=[a.address])
+                            await c.submit(len, futures, workers=[b.address])
 
     log = log.getvalue()
     assert re.search(r"Scheduler at:\s*tcp://", log)
     assert re.search(r"Scheduler at:\s*inproc://", log)
-    assert re.search(r"dashboard at:\s*http://", log)
+
+    # Dashboard link formatting can fail if template contains env vars which aren't
+    # present. Don't kill scheduler, but revert to outputting the port and helpful msg
+    assert re.search(expected_dashboard_link, log)
+    if "ENV_VAR_MISSING" in dashboard_link_template:
+        msg = r"Failed to format dashboard link, unknown value: 'ENV_VAR_MISSING'"
+        assert re.search(msg, log)
 
 
 @gen_cluster(nthreads=[("127.0.0.1", 1)])
diff --git a/distributed/tests/test_worker_metrics.py b/distributed/tests/test_worker_metrics.py
index 881ee21f..2eb7c44f 100644
--- a/distributed/tests/test_worker_metrics.py
+++ b/distributed/tests/test_worker_metrics.py
@@ -1,9 +1,8 @@
 from __future__ import annotations
 
 import asyncio
-import secrets
 import time
-from collections.abc import Hashable
+from collections.abc import Collection, Hashable
 from concurrent.futures import ProcessPoolExecutor
 from time import sleep
 
@@ -11,8 +10,8 @@ import pytest
 
 import dask
 
+import distributed
 from distributed import Event, Worker, wait
-from distributed.comm.utils import OFFLOAD_THRESHOLD
 from distributed.compatibility import WINDOWS
 from distributed.metrics import context_meter, meter
 from distributed.utils_test import (
@@ -25,14 +24,16 @@ from distributed.utils_test import (
 )
 
 
-def get_digests(w: Worker, allow: str | None = None) -> dict[Hashable, float]:
+def get_digests(w: Worker, allow: str | Collection[str] = ()) -> dict[Hashable, float]:
     # import pprint; pprint.pprint(dict(w.digests_total))
+    if isinstance(allow, str):
+        allow = (allow,)
     digests = {
         k: v
         for k, v in w.digests_total.items()
         if k
         not in {"latency", "tick-duration", "transfer-bandwidth", "transfer-duration"}
-        and (allow is None or allow in k)
+        and (any(a in k for a in allow) or not allow)
     }
     assert all(v >= 0 for v in digests.values()), digests
     return digests
@@ -80,6 +81,7 @@ async def test_task_lifecycle(c, s, a, b):
         # -> Run in thread
         ("execute", "z", "thread-cpu", "seconds"),
         ("execute", "z", "thread-noncpu", "seconds"),
+        ("execute", "z", "executor", "seconds"),
         # Spill output; added by _transition_to_memory
         ("execute", "z", "serialize", "seconds"),
         ("execute", "z", "compress", "seconds"),
@@ -150,15 +152,23 @@ async def test_run_spec_deserialization(c, s, a):
 
 
 @gen_cluster(client=True)
-async def test_offload(c, s, a, b):
+async def test_offload(c, s, a, b, monkeypatch):
     """Test that functions wrapped by offload() are metered"""
-    nbytes = int(OFFLOAD_THRESHOLD * 1.1)
-    x = c.submit(secrets.token_bytes, nbytes, key="x", workers=[a.address])
+    monkeypatch.setattr(distributed.comm.utils, "OFFLOAD_THRESHOLD", 1)
+    monkeypatch.setattr(distributed.worker, "OFFLOAD_THRESHOLD", 1)
+
+    x = c.submit(inc, 1, key="x", workers=[a.address])
     y = c.submit(lambda x: None, x, key="y", workers=[b.address])
     await y
 
-    assert 0 < a.digests_total["get-data", "serialize", "seconds"] < 1
-    assert 0 < b.digests_total["gather-dep", "deserialize", "seconds"] < 1
+    assert list(get_digests(b, {"offload", "serialize", "deserialize"})) == [
+        ("gather-dep", "offload", "seconds"),
+        ("gather-dep", "deserialize", "seconds"),
+        ("execute", "y", "offload", "seconds"),
+        ("execute", "y", "deserialize", "seconds"),
+        ("get-data", "offload", "seconds"),
+        ("get-data", "serialize", "seconds"),
+    ]
 
 
 @gen_cluster(client=True, nthreads=[("", 1)])
@@ -315,11 +325,7 @@ async def test_memory_monitor(c, s, a):
     x = c.submit(inc, 1, key="x")
     await async_poll_for(lambda: a.data.disk, timeout=5)
 
-    assert list(get_digests(a)) == [
-        ("execute", "x", "deserialize", "seconds"),
-        ("execute", "x", "thread-cpu", "seconds"),
-        ("execute", "x", "thread-noncpu", "seconds"),
-        ("execute", "x", "other", "seconds"),
+    assert list(get_digests(a, "memory-monitor")) == [
         ("memory-monitor", "serialize", "seconds"),
         ("memory-monitor", "compress", "seconds"),
         ("memory-monitor", "disk-write", "seconds"),
@@ -344,11 +350,13 @@ async def test_user_metrics_sync(c, s, a):
         ("execute", "x", "I/O", "seconds"),
         ("execute", "x", "thread-cpu", "seconds"),
         ("execute", "x", "thread-noncpu", "seconds"),
+        ("execute", "x", "executor", "seconds"),
         ("execute", "x", "other", "seconds"),
     ]
     assert get_digests(a)["execute", "x", "I/O", "seconds"] == 5
     assert get_digests(a)["execute", "x", "thread-cpu", "seconds"] == 0
     assert get_digests(a)["execute", "x", "thread-noncpu", "seconds"] == 0
+    assert get_digests(a)["execute", "x", "executor", "seconds"] == 0
     assert get_digests(a)["execute", "x", "other", "seconds"] == 0