Log worker_client event

distributed/tests/test_worker_client.py

@@ -363,3 +363,31 @@ def long_running():
     assert len(res) == 2
     assert res[a.address] > 25
     assert res[b.address] > 25
+
+
+@gen_cluster(client=True, nthreads=[("", 1)])
+async def test_log_event(c, s, a):
+    # Run a task that spawns a worker client
+    def f(x):
+        with worker_client(timeout=10, separate_thread=True) as wc:
+            x = wc.submit(inc, x)
+            y = wc.submit(double, x)
+            result = x.result() + y.result()
+            return result
+
+    future = c.submit(f, 1)
+    result = await future
+    assert result == 6
+
+    # Ensure a corresponding event is logged
+    events = [
+        event
+        for _, event in s.get_events(a.address)
+        if event["action"] == "worker-client"
+    ]
+    assert len(events) == 1
+    assert events[0] == {
+        "action": "worker-client",
+        "timeout": 10,
+        "separate_thread": True,
+    }

distributed/worker_client.py

@@ -53,6 +53,14 @@ def worker_client(timeout=None, separate_thread=True):
 
     worker = get_worker()
     client = get_client(timeout=timeout)
+    worker.log_event(
+        worker.address,
+        {
+            "action": "worker-client",
+            "timeout": timeout,
+            "separate_thread": separate_thread,
+        },
+    )
     with contextlib.ExitStack() as stack:
         if separate_thread:
             try: