Send log-event if worker is restarted because of memory pressure (#8617)

distributed/tests/test_nanny.py

@@ -660,6 +660,12 @@ async def test_restart_memory(c, s, n):
     while not s.workers:
         await asyncio.sleep(0.1)
 
+    msgs = s.get_events("worker-restart-memory")
+    assert len(msgs)
+    msg = msgs[0][1]
+    assert isinstance(msg, dict)
+    assert {"worker", "pid", "rss"}.issubset(set(msg))
+
 
 class BlockClose(WorkerPlugin):
     def __init__(self, close_happened):

distributed/worker_memory.py

@@ -417,12 +417,19 @@ def memory_monitor(self, nanny: Nanny) -> None:
             return
 
         if self._last_terminated_pid != process.pid:
-            nanny_logger.warning(
+            msg = (
                 f"Worker {nanny.worker_address} (pid={process.pid}) exceeded "
                 f"{self.memory_terminate_fraction * 100:.0f}% memory budget. "
-                "Restarting...",
+                f"Restarting..."
             )
+            nanny_logger.warning(msg)
             self._last_terminated_pid = process.pid
+            event = {
+                "worker": nanny.worker_address,
+                "pid": process.pid,
+                "rss": memory,
+            }
+            nanny.log_event("worker-restart-memory", event)
             process.terminate()
         else:
             # We already sent SIGTERM to the worker, but the process is still alive