Configure Dask workers to contact scheduler on a specific address

[etejedor]

At CERN we have a Jupyter notebook service that we are now integrating with HTCondor resources, and we would like to use those resources via Dask.

The setup is the following: users log in to the notebook service and get a user session, which runs in a Docker container. Inside their session, users should be able to create a Dask HTCondorCluster to deploy Dask workers on our HTCondor pool. The problem we have is that the address that the scheduler binds to can’t be the same as the address workers use to contact the scheduler. The scheduler runs inside the container, and should listen on an address:port of the private network of the container. However, the workers (which are running in another network in the HTCondor pool) should contact the scheduler on an address:port of the node that hosts the user container, for which we would setup port forwarding to reach the container.

It looks like there currently no way for the workers to receive a different scheduler address than the address the scheduler binds to. We found dask/distributed#2963, but that only allows to specify a different address for the client to contact the scheduler (i.e. the scheduler must still bind to the same address that the workers receive).

Would it be interesting to support a use case like the one I just described? How could it be implemented? Perhaps via a new option for the scheduler to specify what address should workers use to connect to it. The naming should be clear to avoid confusion with the already existing external_address (added in dask/distributed#2963).

Pinging @oshadura as she had a proposal for such a patch.

(Previously discussed in:
https://dask.discourse.group/t/dask-scheduler-in-a-docker-container-workers-as-htcondor-jobs)

[jacobtomlinson]

The problem we have is that the address that the scheduler binds to can’t be the same as the address workers use to contact the scheduler.

I'm not sure I fully understand this. What kind of errors are you seeing when you try and do this?

We added external_address to handle situations where the client and workers are on different networks and the client needs to connect via some other address, such as through a load balancer or externally assigned IP. This should be exactly the same for a Docker scenario where the workers use the Docker network and the client uses the node network via exposed ports.

[etejedor]

Thank you for the reply @jacobtomlinson , the kind of errors we get in the workers are:

Waiting to connect to:      tcp://172.17.0.4:5060

172.17.0.4 is an address of the private Docker network. This means the workers are trying to connect to the scheduler on an address they can't reach.

This happens because, in our setup, both the client and the scheduler run inside the user Docker container, whereas the workers run on the HTCondor pool, on a different network. The scheduler listens on e.g. 172.17.0.4:5060, where the client can connect, but not the workers. And the scheduler can't listen on the address of the node that hosts the container.

Therefore, for the workers to be able to connect to the scheduler, they need to use a different address: that of the node that hosts the container. Then via port mapping host->container we can forward the connection of the workers to the scheduler.

As far as I understand, the current external_address parameter does not help us here, since it's not a client -> scheduler connection problem, it's worker -> scheduler. The workers still use the same address to which the scheduler binds to connect to it - I believe this happens here.

[jacobtomlinson]

Ah I see. external_address is for when the client connects on a different address to the workers, which is useful for situations where the scheduler and workers are in one network and the client is in another. Dask Jobqueue doesn't run the scheduler as a remote job like most other cluster managers, it runs it wherever you invoke the Cluster (xref #186).

It sounds like you need some way to explicitly state what the scheduler address will be when workers/clients connect to it, rather than setting multiple addresses. @guillaumeeb @lesteve do you have thoughts on this?

[lesteve]

Maybe related to #490 I don't fully recall the details I have to admit ...

There is the scheduler_options parameter that may help, e.g. as in #490 (comment). I have to admit I don't know whether this is similar to SpecCluster.external_address.

Full disclosure: I don't have too much time to devote to Dask-related things unfortunately these days ...

[etejedor]

Maybe related to #490 I don't fully recall the details I have to admit ...

Yes, it's not the same setup as what I described, but we are both asking for the same functionality.

There is the scheduler_options parameter that may help, e.g. as in #490 (comment).

Using scheduler_options={"host": ip} doesn't help because the scheduler will still bind to the same address that is used by the workers to connect to the scheduler, and what we need is to decouple those two.

If there was a way to run the scheduler as a remote job, then both the scheduler and the workers would belong to the same network (the one of the HTCondor pool in my case) and that would be a solution, but that's currently not possible as @jacobtomlinson mentioned. Therefore, it might be easier to add an option to specify the scheduler address to pass to the workers. The closest thing I've found is this patch of distributed by @oshadura :

https://github.com/CoffeaTeam/coffea-casa/blob/master/docker/coffea-casa-cc7/distributed/0004-Add-possibility-to-setup-external_adress-for-schedul.patch

Note that the external_address option in that patch is not the same as the external_address that already exists upstream. We would need to find another name for the new option.

[guillaumeeb]

Hi everyone, sorry it took me a while to look more deeply into this.

There are several ways to solve this, you've already listed two:

• New constructor argument to SpecCluster, something close to the line of @oshadura's patch. But I would only modify spec.py file. Definitly there would be something to change here: https://github.com/dask/distributed/blob/main/distributed/deploy/spec.py#L345.
• Adding the possibility to start a Scheduler as a remote job. Not a direct fix, but very good workaround that would enhance dask-jobqueue a lot.

The third one would be modifying dask-jobqueue itself to introduce a way to customize this address:

• Introducing some attribute that can be customized and use it here: https://github.com/dask/dask-jobqueue/blob/main/dask_jobqueue/core.py#L255.
• We could also push forward Template job scripts with Jinja #370, and make the Jinja script configurable so that we would specify a scheduler address directly into the template.

I think the easyest is still the second bullet. From my point of view it would be much better than the 3rd, as this can be more general than a dask-jobqueue problem (but I may be wrong). Advancing thourgh 1st or 4th bullet points would be also much appreciated 😀.

Anyway, any PR would be great to solve this problem!

[etejedor]

Thank you for the explanations @guillaumeeb , my replies below.

New constructor argument to SpecCluster, something close to the line of @oshadura's patch. But I would only modify spec.py file

If we only modify spec.py, this means the user will need to define their own subclass of Scheduler that accepts this new XXX_address parameter, and provide that class as the scheduler_cls argument of the cluster constructor. Is this correct?

Adding the possibility to start a Scheduler as a remote job. Not a direct fix, but very good workaround that would enhance dask-jobqueue a lot.

We'd be also interested in that possibility, not strictly to solve the problem I reported (or not only), but also to explore that setup - if the scheduler is allocated on the HTCondor cluster side, would the client be able to reconnect to it in case the connection is lost (e.g. my Jupyter container was destructed and I recreated it, and I'd like to reconnect to the scheduler)?

I don't know how many changes this option would require, though.

We tried option 1 with a patched distributed and it seems to work fine, so I think we'll make a PR for that one unless there's some objection.

[guillaumeeb]

If we only modify spec.py, this means the user will need to define their own subclass of Scheduler that accepts this new XXX_address parameter, and provide that class as the scheduler_cls argument of the cluster constructor. Is this correct?

I'm not sure of this, please do propose a PR (if not already done) so this can be discussed, never any objection on proposing a PR!!

I was under the impression that something really close to dask/distributed#2963 could be done, only modifying ProcessInterface or SpecCluster constructor, and changing something here https://github.com/dask/distributed/blob/main/distributed/deploy/spec.py#L345.

if the scheduler is allocated on the HTCondor cluster side, would the client be able to reconnect to it in case the connection is lost (e.g. my Jupyter container was destructed and I recreated it, and I'd like to reconnect to the scheduler)?

Theoretically yes. Loosing the SpecCluster object should not have any effect on the job queuing system. So both the job running the Scheduler and those running workers should continue running. But there won't be any helper to find the Scheduler from a fresh Python kernel. You'll just need to have the Scheduler address saved somewhere. Similarly, currently there is no possibility of recreating a JobQueueCluster object from an existing Scheduler object nor address. But it should be, and that might be part of the effort to start a Scheduler on a remote job (see also #419).

[etejedor]

I'm not sure of this, please do propose a PR (if not already done) so this can be discussed, never any objection on proposing a PR!!

PR is up! dask/distributed#5944

Theoretically yes. Loosing the SpecCluster object should not have any effect on the job queuing system. So both the job running the Scheduler and those running workers should continue running. But there won't be any helper to find the Scheduler from a fresh Python kernel. You'll just need to have the Scheduler address saved somewhere. Similarly, currently there is no possibility of recreating a JobQueueCluster object from an existing Scheduler object nor address. But it should be, and that might be part of the effort to start a Scheduler on a remote job (see also #419).

I see, thanks. But even if I stored the address of the scheduler somewhere and I use it to reconnect with a fresh client, what I guess is not supported is to reconnect to some ongoing computation? E.g. I run client.submit, then the client dies but the job is still running on the cluster side; there's no way I can recreate the future result when the client is back.

[guillaumeeb]

But even if I stored the address of the scheduler somewhere and I use it to reconnect with a fresh client, what I guess is not supported is to reconnect to some ongoing computation? E.g. I run client.submit, then the client dies but the job is still running on the cluster side; there's no way I can recreate the future result when the client is back.

That is an excellent question. If the future object on client side is lost, I don't know what references keeps the Scheduler... Maybe this question could be asked on Discourse.

I'm looking at your PR right now.

[guillaumeeb]

Closing this one as an option has been added to distributed directly. @etejedor feel free to reopen if I missed something.