do_preprocess fails with error: "RuntimeError: Nanny failed to start."

[derekgrove]

Hi, I'm working through some examples of the DataDiscovery tools in coffea. While working on coffeacasa v 2024.8.1 I tried this simple code to generate a preprocess file:

from coffea.dataset_tools import rucio_utils
from coffea.dataset_tools.dataset_query import print_dataset_query
from rich.console import Console
from coffea.dataset_tools.dataset_query import DataDiscoveryCLI

ddc = DataDiscoveryCLI()

ddc.do_query("/SlepSnu*NanoAOD*v3*")
ddc.do_select("all")
ddc.do_regex_sites(r"T[12]_(US)_\w+")
ddc.do_replicas("first", "all")
fileset_total = ddc.do_preprocess(output_file="fileset",
                  step_size=10000,  #chunk size for files splitting
                  align_to_clusters=False,
                 scheduler_url=None)

and in the very last step, the do_preprocess method, I get this runtime error:

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
File /usr/local/lib/python3.10/site-packages/distributed/core.py:525, in Server.start(self)
    524 try:
--> 525     await wait_for(self.start_unsafe(), timeout=timeout)
    526 except asyncio.TimeoutError as exc:

File /usr/local/lib/python3.10/site-packages/distributed/utils.py:1957, in wait_for(fut, timeout)
   1956 async def wait_for(fut: Awaitable[T], timeout: float) -> T:
-> 1957     return await asyncio.wait_for(fut, timeout)

File /usr/local/lib/python3.10/asyncio/tasks.py:408, in wait_for(fut, timeout)
    407 if timeout is None:
--> 408     return await fut
    410 if timeout <= 0:

File /usr/local/lib/python3.10/site-packages/distributed/nanny.py:337, in Nanny.start_unsafe(self)
    336 try:
--> 337     await self.listen(
    338         start_address, **self.security.get_listen_args("worker")
    339     )
    340 except OSError as e:

File /usr/local/lib/python3.10/site-packages/distributed/core.py:720, in Server.listen(self, port_or_addr, allow_offload, **kwargs)
    719     assert isinstance(addr, str)
--> 720 listener = await listen(
    721     addr,
    722     self.handle_comm,
    723     deserialize=self.deserialize,
    724     allow_offload=allow_offload,
    725     **kwargs,
    726 )
    727 self.listeners.append(listener)

File /usr/local/lib/python3.10/site-packages/distributed/comm/core.py:411, in listen(addr, handle_comm, deserialize, **kwargs)
    409 backend = registry.get_backend(scheme)
--> 411 return backend.get_listener(loc, handle_comm, deserialize, **kwargs)

File /usr/local/lib/python3.10/site-packages/distributed/comm/tcp.py:744, in BaseTCPBackend.get_listener(self, loc, handle_comm, deserialize, **connection_args)
    743 def get_listener(self, loc, handle_comm, deserialize, **connection_args):
--> 744     return self._listener_class(loc, handle_comm, deserialize, **connection_args)

File /usr/local/lib/python3.10/site-packages/distributed/comm/tcp.py:615, in BaseTCPListener.__init__(self, address, comm_handler, deserialize, allow_offload, default_host, default_port, **connection_args)
    614 self.allow_offload = allow_offload
--> 615 self.server_args = self._get_server_args(**connection_args)
    616 self.tcp_server = None

File /usr/local/lib/python3.10/site-packages/distributed/comm/tcp.py:719, in TLSListener._get_server_args(self, **connection_args)
    718 def _get_server_args(self, **connection_args):
--> 719     ctx = _expect_tls_context(connection_args)
    720     return {"ssl_options": ctx}

File /usr/local/lib/python3.10/site-packages/distributed/comm/tcp.py:456, in _expect_tls_context(connection_args)
    455 if not isinstance(ctx, ssl.SSLContext):
--> 456     raise TypeError(
    457         "TLS expects a `ssl_context` argument of type "
    458         "ssl.SSLContext (perhaps check your TLS configuration?)"
    459         f" Instead got {ctx!r}"
    460     )
    461 return ctx

TypeError: TLS expects a `ssl_context` argument of type ssl.SSLContext (perhaps check your TLS configuration?) Instead got None

The above exception was the direct cause of the following exception:

RuntimeError                              Traceback (most recent call last)
Cell In[1], line 12
     10 ddc.do_regex_sites(r"T[12]_(US)_\w+")
     11 ddc.do_replicas("first", "all")
---> 12 fileset_total = ddc.do_preprocess(output_file="fileset",
     13                   step_size=10000,  #chunk size for files splitting
     14                   align_to_clusters=False,
     15                  scheduler_url=None)

File /usr/local/lib/python3.10/site-packages/coffea/dataset_tools/dataset_query.py:568, in DataDiscoveryCLI.do_preprocess(self, output_file, step_size, align_to_clusters, scheduler_url, recalculate_steps, files_per_batch, file_exceptions, save_form, uproot_options, step_size_safety_factor)
    564 # init a local Dask cluster
    565 with self.console.status(
    566     "[red] Preprocessing files to extract available chunks with dask[/]"
    567 ):
--> 568     with Client(scheduler_url) as ddcsched:
    569         self.preprocessed_available, self.preprocessed_total = preprocess(
    570             self.final_output,
    571             step_size=step_size,
   (...)
    580             step_size_safety_factor=step_size_safety_factor,
    581         )
    583 with gzip.open(f"{output_file}_available.json.gz", "wt") as file:

File /usr/local/lib/python3.10/site-packages/distributed/client.py:1235, in Client.__init__(self, address, loop, timeout, set_as_default, scheduler_file, security, asynchronous, name, heartbeat_interval, serializers, deserializers, extensions, direct_to_workers, connection_limit, **kwargs)
   1232 preload_argv = dask.config.get("distributed.client.preload-argv")
   1233 self.preloads = preloading.process_preloads(self, preload, preload_argv)
-> 1235 self.start(timeout=timeout)
   1236 Client._instances.add(self)
   1238 from distributed.recreate_tasks import ReplayTaskClient

File /usr/local/lib/python3.10/site-packages/distributed/client.py:1437, in Client.start(self, **kwargs)
   1435     self._started = asyncio.ensure_future(self._start(**kwargs))
   1436 else:
-> 1437     sync(self.loop, self._start, **kwargs)

File /usr/local/lib/python3.10/site-packages/distributed/utils.py:439, in sync(loop, func, callback_timeout, *args, **kwargs)
    436         wait(10)
    438 if error is not None:
--> 439     raise error
    440 else:
    441     return result

File /usr/local/lib/python3.10/site-packages/distributed/utils.py:413, in sync.<locals>.f()
    411         awaitable = wait_for(awaitable, timeout)
    412     future = asyncio.ensure_future(awaitable)
--> 413     result = yield future
    414 except Exception as exception:
    415     error = exception

File /usr/local/lib/python3.10/site-packages/tornado/gen.py:766, in Runner.run(self)
    764 try:
    765     try:
--> 766         value = future.result()
    767     except Exception as e:
    768         # Save the exception for later. It's important that
    769         # gen.throw() not be called inside this try/except block
    770         # because that makes sys.exc_info behave unexpectedly.
    771         exc: Optional[Exception] = e

File /usr/local/lib/python3.10/site-packages/distributed/client.py:1502, in Client._start(self, timeout, **kwargs)
   1499 elif self._start_arg is None:
   1500     from distributed.deploy import LocalCluster
-> 1502     self.cluster = await LocalCluster(
   1503         loop=self.loop,
   1504         asynchronous=self._asynchronous,
   1505         **self._startup_kwargs,
   1506     )
   1507     address = self.cluster.scheduler_address
   1509 self._gather_semaphore = asyncio.Semaphore(5)

File /usr/local/lib/python3.10/site-packages/distributed/deploy/spec.py:421, in SpecCluster.__await__.<locals>._()
    419     await self._start()
    420 await self.scheduler
--> 421 await self._correct_state()
    422 if self.workers:
    423     await asyncio.wait(
    424         [
    425             asyncio.create_task(_wrap_awaitable(w))
    426             for w in self.workers.values()
    427         ]
    428     )  # maybe there are more

File /usr/local/lib/python3.10/site-packages/distributed/deploy/spec.py:391, in SpecCluster._correct_state_internal(self)
    387     w._cluster = weakref.ref(self)
    388 # Collect exceptions from failed workers. This must happen after all
    389 # *other* workers have finished initialising, so that we can have a
    390 # proper teardown.
--> 391 await asyncio.gather(*worker_futs)

File /usr/local/lib/python3.10/asyncio/tasks.py:650, in _wrap_awaitable(awaitable)
    643 @types.coroutine
    644 def _wrap_awaitable(awaitable):
    645     """Helper for asyncio.ensure_future().
    646 
    647     Wraps awaitable (an object with __await__) into a coroutine
    648     that will later be wrapped in a Task by ensure_future().
    649     """
--> 650     return (yield from awaitable.__await__())

File /usr/local/lib/python3.10/site-packages/distributed/core.py:533, in Server.start(self)
    531 except Exception as exc:
    532     await _close_on_failure(exc)
--> 533     raise RuntimeError(f"{type(self).__name__} failed to start.") from exc
    534 if self.status == Status.init:
    535     self.status = Status.running

RuntimeError: Nanny failed to start.

I don't get this error when running the same code on SWAN, it "just works" on SWAN and generates the preprocess file, so I'm not quite sure what is different about running this on coffeacasa