do not initialize logging on import

distributed/config.py

@@ -4,6 +4,7 @@
 import logging.config
 import os
 import sys
+import threading
 from collections.abc import Callable
 from typing import Any
 
@@ -212,4 +213,13 @@ def get_loop_factory() -> Callable[[], asyncio.AbstractEventLoop] | None:
     )
 
 
-initialize_logging(dask.config.config)
+_LOGGING_CONFIGURED = False
+_LOG_CONFIGURE_LOCK = threading.Lock()
+
+
+def ensure_logging_configured() -> None:
+    global _LOGGING_CONFIGURED
+    with _LOG_CONFIGURE_LOCK:
+        if not _LOGGING_CONFIGURED:
+            _LOGGING_CONFIGURED = True
+            initialize_logging(dask.config.config)

distributed/core.py

@@ -45,6 +45,7 @@
 )
 from distributed.comm.core import Listener
 from distributed.compatibility import PeriodicCallback
+from distributed.config import ensure_logging_configured
 from distributed.counter import Counter
 from distributed.diskutils import WorkDir, WorkSpace
 from distributed.metrics import context_meter, time
@@ -647,6 +648,7 @@ async def start_unsafe(self):
 
     @final
     async def start(self):
+        ensure_logging_configured()
         async with self._startup_lock:
             if self.status == Status.failed:
                 assert self.__startup_exc is not None

distributed/deploy/cluster.py

@@ -16,6 +16,7 @@
 from dask.widgets import get_template
 
 from distributed.compatibility import PeriodicCallback
+from distributed.config import ensure_logging_configured
 from distributed.core import Status
 from distributed.deploy.adaptive import Adaptive
 from distributed.metrics import time
@@ -67,6 +68,7 @@ def __init__(
         name=None,
         scheduler_sync_interval=1,
     ):
+        ensure_logging_configured()
         self._loop_runner = LoopRunner(loop=loop, asynchronous=asynchronous)
         self.__asynchronous = asynchronous