Check worker connectivity before sending work

[jacobtomlinson]

What happened:

On some compute platforms workers are able to start and connect to the scheduler before they are set up with inbound networking themselves. This results in a race condition where a worker can register with the scheduler, but the scheduler and other workers are unable to connect to the worker to send work resulting in errors.

One concrete example of this would be when using Readiness Probes on Kubernetes. If probes are configured Kubernetes will not allow network connections into a pod until the probes have passed N times. This may mean that network into a worker is not enabled until a couple of seconds after it begins listening on its TCP address. But during those seconds it will register with the scheduler and the scheduler may try and send work within a few milliseconds.

What you expected to happen:

When the scheduler receives a new worker registration it should check the connectivity back is available before putting that worker into service.

Minimal Complete Verifiable Example:

This is a little hard to create an MCVE for.

Anything else we need to know?:

This bug was highlighted by @stephan-erb-by and @philipp-sontag-by when discussing the new operator in dask/dask-kubernetes#256.

[stephan-erb-by]

Maybe just one clarification here: The problem also surfaces in worker-to-worker communication (e.g. trying to move data to the newly registered worker). In any case, it would probably be sufficient as a safe guard if the scheduler considers a worker as "ready" only once it managed to open a new fresh connection to the newly registered worker.

[fjetter]

The worker only connects to the scheduler once its listener is up. At least on our application layer, it should actually be able to receive incoming requests before the scheduler knows the worker exists.

The scheduler very, very rarely opens a direct connection to a worker. That typically only happens if the client is involved with calls like fetch profile, gather data, etc. Otherwise the scheduler reuses the initial comm that was used to register the worker. We're using this comm for a bidirectional message stream which is used to control task assignments, task results, heartbeats, etc.

If other workers cannot connect to a peer, there are multiple retries in place such that this should not impact any actual workload

1. The connect itself tries for distributed.comm.timeouts.connects (default 30s) to connect to the remove before raising, see here.
2. If application level retries are implemented (default off, distributed.comm.retry), we will retry the entire get_data attempt
3. If all of this fails, an exception is raised issuing a log like Worker stream died during communication . This will escalate to the scheduler and may reschedule a key

Is 3) the issue being observed here or what exactly is going wrong?

The K8s operator ticket mentions readiness probes to be an issue. Are we actually talking about the internal RPC framework or does this concern the HTTP server?

[jacobtomlinson]

Is 3) the issue being observed here or what exactly is going wrong?

Perhaps @stephan-erb-by and @philipp-sontag-by have thoughts here?

The K8s operator ticket mentions readiness probes to be an issue. Are we actually talking about the internal RPC framework or does this concern the HTTP server?

This is related to the RPC. The probe checks the TCP port is available.

[stephan-erb-by]

The worker only connects to the scheduler once its listener is up. At least on our application layer, it should actually be able to receive incoming requests before the scheduler knows the worker exists.

I think purely from the application layer this works as expected, but there are problems if you throw Kubernetes and Istio into the mix: Kubernetes probes in a configurable interval if your processes are up. Only once these probes have passed successfully it will allow inbound HTTP and RPC traffic flow to the process, i.e. our workers.

There is now a race condition that lasts for multiple tens of seconds:

1. We have a scheduler and a running worker w0
2. New worker w1 starts up
3. Worker w1 connects to scheduler
4. The scheduler assigns some quick'n easy work to w1
5. For whatever reason, the scheduler decides w0 should fetch results from w1.
6. Worker w0 tries to connect to worker w1 but fails. It retries multiple times. It might give up eventually or succeed, depending on when we reach step 7.
7. After an eternity of ~5-15s, the Kubernetes startup/liveness/readiness probe comes along and sees that worker w1 is up and running. Only now it allows ingress traffic to flow that worker.
8. If w0 is still there and trying to connect the connection can now be established successfully.

In additional complication here is that the setup looks like this:

[worker w0] -> [istio proxy w0] -------------> [istio proxy w1] -> [worker w1]

So from a Dask worker w0 perspective it can always establish a TCP connection. But this connection is only towards its local Istio proxy w0. The proxy w0 then fails to reach out to w1 as the inbound request is denied (as the probes have not passed yet). This seems to bypass some the desired retry logic in Dask.

I am now blurry on the details, but IIRC the scheduler will increase the "suspicious count" of the task that was assigned to the worker w1. If this happens too often for a single task, the scheduler will inform the distributed client, which will raise a KilledWorker exception.

I know, this really sounds esoteric, but it was a real significant issues for us. Our current workaround is to run without any of probes but this has other downsides.

[fjetter]

So from a Dask worker w0 perspective it can always establish a TCP connection. But this connection is only towards its local Istio proxy w0. The proxy w0 then fails to reach out to w1 as the inbound request is denied (as the probes have not passed yet). This seems to bypass some the desired retry logic in Dask.

Just opening a TCP connection to a proxy is not sufficient for dask to consider a connect to be successful. On every connection attempt we are performing a handshake. This handshake is not covered by the retry but it is included in the distributed.comm.timeouts.connect.
If you have a successful TCP connection but are failing the handshake, you should see Timed out during handshake while connecting to [...] messages (or if it fails hard, the actual exception, of course)

What is istio doing when the TCP connection is up and it receives an incoming message before the peer is up? If it just holds on to the messages for Xs until the peer is up, you should be good as long as the connect timeout is sufficiently large.

I am now blurry on the details, but IIRC the scheduler will increase the "suspicious count" of the task that was assigned to the worker w1. If this happens too often for a single task, the scheduler will inform the distributed client, which will raise a KilledWorker exception.

The suspicious count is increased if the scheduler considers a worker to be dead. This typically happens if the connection that has been used originally to register to the scheduler (i.e. Worker initiated a connection to Scheduler) is closed. If this connection is not interrupted, the suspicious counts should not go up.

I know, this really sounds esoteric, but it was a real significant issues for us. Our current workaround is to run without any of probes but this has other downsides.

I understand. Your previous setup had similar quirks and we eventually figured it out :) On this issue I'm still a bit confused about what is actually going on and I'm hesitant to open additional connections on scheduler side if it can be avoided. I could see this become disruptive for large clusters