TOCTOU Bug while scaling down workers

[BitTheByte]

The currently implemented logic for deletion can be summarized into:

1. The operator asks the scheduler to retire n of the workers
2. The scheduler retires them (process exits) and returns to the operator names of retired workers
3. Operator deletes worker deployment sequentially

Ref: https://github.com/dask/dask-kubernetes/blob/main/dask_kubernetes/operator/controller/controller.py#L600-L611

However, between 2 and 3 the Kubernetes API may interfere and restart the worker deployment so a new pod will be created and join the cluster for some time before the operator deletes the deployment effectively interpreting the pod mid-run.

[BitTheByte]

Is it possible to allow the scheduler to delete the deployment directly? maybe with something like a plugin or so?

[jacobtomlinson]

Are you seeing this in practice? Or is this a hypothetical race condition?

Each worker should have a unique ID, and when the scheduler retires it then a worker with that ID cannot reconnect in the future. So even if Kubernetes restarts the Pod in the time before the delete call happens and the Pod is cascade deleted the new worker should just repeatedly fail to connect to the scheduler.

Is it possible to allow the scheduler to delete the deployment directly?

I don't think we should be giving the scheduler the ability to interact with the Kubernetes API as that would require us to give permissions to the scheduler Pod which can execute arbitrary user code.

[BitTheByte]

Are you seeing this in practice? Or is this a hypothetical race condition?

Yes, I'm seeing this during large-scale cluster scaling e.g. 100-200 workers

[briceruzand]

I'am having that behavior each time I try to scale down even with 2 workers (see #856), so my cluster never scale down.
I need to delete the cluster and to create a new one with less replicas ;-(

[briceruzand]

Hi,
Here are the logs of my behavior.

Scheduler logs :

[2024-02-06 11:02:24] INFO     distributed/scheduler.py:1685            : State start
[2024-02-06 11:02:24] INFO     distributed/scheduler.py:3998            :   Scheduler at:    tcp://100.64.5.23:8786
[2024-02-06 11:02:24] INFO     distributed/scheduler.py:4013            :   dashboard at:  http://100.64.5.23:8787/status
[2024-02-06 11:02:24] INFO     distributed/scheduler.py:7526            : Registering Worker plugin shuffle
[2024-02-06 11:02:51] INFO     distributed/scheduler.py:5494            : Receive client connection: Client-e38267f6-c4d6-11ee-8018-8249c5728274
[2024-02-06 11:02:51] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.4.90:41308
[2024-02-06 11:02:51] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.14.99:33331', status: init, memory: 0, processing: 0>
[2024-02-06 11:02:51] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.14.99:33331
[2024-02-06 11:02:51] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.14.99:53914
[2024-02-06 11:02:52] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.14.185:34011', status: init, memory: 0, processing: 0>
[2024-02-06 11:02:52] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.14.185:34011
[2024-02-06 11:02:52] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.14.185:44712
[2024-02-06 11:03:27] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.15.153:35425', status: init, memory: 0, processing: 0>
[2024-02-06 11:03:27] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.15.153:35425
[2024-02-06 11:03:27] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.15.153:51712
[2024-02-06 11:03:57] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.16.185:34141', status: init, memory: 0, processing: 0>
[2024-02-06 11:03:57] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.16.185:34141
[2024-02-06 11:03:57] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.16.185:48954
[2024-02-06 11:04:28] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.16.72:42101', status: init, memory: 0, processing: 0>
[2024-02-06 11:04:28] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.16.72:42101
[2024-02-06 11:04:28] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.16.72:45816
[2024-02-06 11:04:49] INFO     distributed/scheduler.py:5538            : Remove client Client-e38267f6-c4d6-11ee-8018-8249c5728274
[2024-02-06 11:04:49] INFO     distributed/core.py:993                  : Received 'close-stream' from tcp://100.64.4.90:41308; closing.
[2024-02-06 11:04:49] INFO     distributed/scheduler.py:5538            : Remove client Client-e38267f6-c4d6-11ee-8018-8249c5728274
[2024-02-06 11:04:49] INFO     distributed/scheduler.py:5530            : Close client connection: Client-e38267f6-c4d6-11ee-8018-8249c5728274
[2024-02-06 11:04:52] INFO     distributed/scheduler.py:6978            : Retire worker names ('tcp://100.64.16.72:42101',)
[2024-02-06 11:04:52] INFO     distributed/scheduler.py:7007            : Retiring worker tcp://100.64.16.72:42101
[2024-02-06 11:04:52] INFO     distributed/active_memory_manager.py:712 : Retiring worker tcp://100.64.16.72:42101; no unique keys need to be moved away.
[2024-02-06 11:04:52] INFO     distributed/scheduler.py:5040            : Remove worker <WorkerState 'tcp://100.64.16.72:42101', status: closing_gracefully, memory: 0, processing: 0> (stimulus_id='retire-workers-1707213892.1971838')
[2024-02-06 11:04:52] INFO     distributed/scheduler.py:7094            : Retired worker tcp://100.64.16.72:42101
[2024-02-06 11:04:52] WARNING  distributed/scheduler.py:4140            : Received heartbeat from unregistered worker 'tcp://100.64.16.72:42101'.
[2024-02-06 11:04:52] INFO     distributed/core.py:993                  : Received 'close-stream' from tcp://100.64.16.72:45816; closing.
[2024-02-06 11:04:58] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.15.85:36381', status: init, memory: 0, processing: 0>
[2024-02-06 11:04:58] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.15.85:36381
[2024-02-06 11:04:58] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.15.85:41112
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:6978            : Retire worker names ('tcp://100.64.16.185:34141', 'tcp://100.64.15.85:36381', 'tcp://100.64.15.153:35425', 'tcp://100.64.14.99:33331')
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7007            : Retiring worker tcp://100.64.15.85:36381
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7007            : Retiring worker tcp://100.64.14.99:33331
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7007            : Retiring worker tcp://100.64.16.185:34141
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7007            : Retiring worker tcp://100.64.15.153:35425
[2024-02-06 11:05:52] INFO     distributed/active_memory_manager.py:712 : Retiring worker tcp://100.64.16.185:34141; no unique keys need to be moved away.
[2024-02-06 11:05:52] INFO     distributed/active_memory_manager.py:712 : Retiring worker tcp://100.64.14.99:33331; no unique keys need to be moved away.
[2024-02-06 11:05:52] INFO     distributed/active_memory_manager.py:712 : Retiring worker tcp://100.64.15.85:36381; no unique keys need to be moved away.
[2024-02-06 11:05:52] INFO     distributed/active_memory_manager.py:712 : Retiring worker tcp://100.64.15.153:35425; no unique keys need to be moved away.
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:5040            : Remove worker <WorkerState 'tcp://100.64.15.85:36381', status: closing_gracefully, memory: 0, processing: 0> (stimulus_id='retire-workers-1707213952.5686617')
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7094            : Retired worker tcp://100.64.15.85:36381
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:5040            : Remove worker <WorkerState 'tcp://100.64.14.99:33331', status: closing_gracefully, memory: 0, processing: 0> (stimulus_id='retire-workers-1707213952.5686617')
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7094            : Retired worker tcp://100.64.14.99:33331
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:5040            : Remove worker <WorkerState 'tcp://100.64.16.185:34141', status: closing_gracefully, memory: 0, processing: 0> (stimulus_id='retire-workers-1707213952.5686617')
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7094            : Retired worker tcp://100.64.16.185:34141
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:5040            : Remove worker <WorkerState 'tcp://100.64.15.153:35425', status: closing_gracefully, memory: 0, processing: 0> (stimulus_id='retire-workers-1707213952.5686617')
[2024-02-06 11:05:52] INFO     distributed/scheduler.py:7094            : Retired worker tcp://100.64.15.153:35425
[2024-02-06 11:05:52] WARNING  distributed/scheduler.py:4140            : Received heartbeat from unregistered worker 'tcp://100.64.15.153:35425'.
[2024-02-06 11:05:52] INFO     distributed/core.py:993                  : Received 'close-stream' from tcp://100.64.15.153:51712; closing.
[2024-02-06 11:05:52] WARNING  distributed/scheduler.py:4140            : Received heartbeat from unregistered worker 'tcp://100.64.14.99:33331'.
[2024-02-06 11:05:52] INFO     distributed/core.py:993                  : Received 'close-stream' from tcp://100.64.14.99:53914; closing.
[2024-02-06 11:05:52] WARNING  distributed/scheduler.py:4140            : Received heartbeat from unregistered worker 'tcp://100.64.15.85:36381'.
[2024-02-06 11:05:52] INFO     distributed/core.py:993                  : Received 'close-stream' from tcp://100.64.15.85:41112; closing.
[2024-02-06 11:05:52] WARNING  distributed/scheduler.py:4140            : Received heartbeat from unregistered worker 'tcp://100.64.16.185:34141'.
[2024-02-06 11:05:52] INFO     distributed/core.py:993                  : Received 'close-stream' from tcp://100.64.16.185:48954; closing.
[2024-02-06 11:06:52] INFO     distributed/scheduler.py:6978            : Retire worker names ('tcp://100.64.14.185:34011',)
[2024-02-06 11:06:52] INFO     distributed/scheduler.py:7007            : Retiring worker tcp://100.64.14.185:34011
[2024-02-06 11:06:52] INFO     distributed/active_memory_manager.py:712 : Retiring worker tcp://100.64.14.185:34011; no unique keys need to be moved away.
[2024-02-06 11:06:52] INFO     distributed/scheduler.py:5040            : Remove worker <WorkerState 'tcp://100.64.14.185:34011', status: closing_gracefully, memory: 0, processing: 0> (stimulus_id='retire-workers-1707214012.8366055')
[2024-02-06 11:06:52] INFO     distributed/scheduler.py:5138            : Lost all workers
[2024-02-06 11:06:52] INFO     distributed/scheduler.py:7094            : Retired worker tcp://100.64.14.185:34011
[2024-02-06 11:06:53] WARNING  distributed/scheduler.py:4140            : Received heartbeat from unregistered worker 'tcp://100.64.14.185:34011'.
[2024-02-06 11:06:53] INFO     distributed/core.py:993                  : Received 'close-stream' from tcp://100.64.14.185:44712; closing.
[2024-02-06 11:08:22] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.16.72:46229', status: init, memory: 0, processing: 0>
[2024-02-06 11:08:22] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.16.72:46229
[2024-02-06 11:08:22] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.16.72:37656
[2024-02-06 11:09:23] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.15.153:44645', status: init, memory: 0, processing: 0>
[2024-02-06 11:09:23] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.15.153:44645
[2024-02-06 11:09:23] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.15.153:33564
[2024-02-06 11:09:23] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.14.99:36425', status: init, memory: 0, processing: 0>
[2024-02-06 11:09:23] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.14.99:36425
[2024-02-06 11:09:23] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.14.99:41472
[2024-02-06 11:09:23] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.16.185:35563', status: init, memory: 0, processing: 0>
[2024-02-06 11:09:23] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.16.185:35563
[2024-02-06 11:09:23] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.16.185:42640
[2024-02-06 11:09:24] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.15.85:38805', status: init, memory: 0, processing: 0>
[2024-02-06 11:09:24] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.15.85:38805
[2024-02-06 11:09:24] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.15.85:47140
[2024-02-06 11:10:23] INFO     distributed/scheduler.py:4349            : Register worker <WorkerState 'tcp://100.64.14.185:39287', status: init, memory: 0, processing: 0>
[2024-02-06 11:10:23] INFO     distributed/scheduler.py:5736            : Starting worker compute stream, tcp://100.64.14.185:39287
[2024-02-06 11:10:23] INFO     distributed/core.py:968                  : Starting established connection to tcp://100.64.14.185:43730

One worker logs:

[2024-02-06 11:05:52] ERROR    distributed/worker.py:1278               : Scheduler was unaware of this worker 'tcp://100.64.15.85:38805'. Shutting down.
[2024-02-06 11:05:52] INFO     distributed/worker.py:1547               : Stopping worker at tcp://100.64.15.85:38805. Reason: worker-close
[2024-02-06 11:05:52] INFO     distributed/core.py:978                  : Connection to tcp://dask-cluster-scheduler.dask.svc.cluster.local:8786 has been closed.

Operator logs :

[2024-02-06 11:04:52,131] kopf.objects         [INFO    ] [dask/dask-cluster-default] Scaled worker group dask-cluster-default up to 5 workers.
[2024-02-06 11:04:52,153] httpx                [INFO    ] HTTP Request: GET https://10.32.0.1/api/v1/namespaces/dask/services "HTTP/1.1 200 OK"
[2024-02-06 11:04:52,179] httpx                [INFO    ] HTTP Request: GET https://10.32.0.1/api/v1/namespaces/dask/services "HTTP/1.1 200 OK"
[2024-02-06 11:04:52,202] kopf.objects         [INFO    ] [dask/dask-cluster-default] Workers to close: ('tcp://100.64.16.72:42101',)
[2024-02-06 11:04:52,208] httpx                [INFO    ] HTTP Request: DELETE https://10.32.0.1/apis/apps/v1/namespaces/dask/deployments/tcp://100.64.16.72:42101 "HTTP/1.1 404 Not Found"
[2024-02-06 11:04:52,208] kopf.objects         [ERROR   ] [dask/dask-cluster-default] Handler 'daskworkergroup_replica_update/spec.worker.replicas' failed with an exception. Will retry.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/kr8s/_objects.py", line 247, in delete
    async with self.api.call_api(
  File "/usr/local/lib/python3.10/contextlib.py", line 199, in __aenter__
    return await anext(self.gen)
  File "/usr/local/lib/python3.10/site-packages/kr8s/_api.py", line 134, in call_api
    response.raise_for_status()
  File "/usr/local/lib/python3.10/site-packages/httpx/_models.py", line 759, in raise_for_status
    raise HTTPStatusError(message, request=request, response=self)
httpx.HTTPStatusError: Client error '404 Not Found' for url 'https://10.32.0.1/apis/apps/v1/namespaces/dask/deployments/tcp://100.64.16.72:42101'
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404

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

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/kopf/_core/actions/execution.py", line 276, in execute_handler_once
    result = await invoke_handler(
  File "/usr/local/lib/python3.10/site-packages/kopf/_core/actions/execution.py", line 371, in invoke_handler
    result = await invocation.invoke(
  File "/usr/local/lib/python3.10/site-packages/kopf/_core/actions/invocation.py", line 116, in invoke
    result = await fn(**kwargs)  # type: ignore
  File "/usr/local/lib/python3.10/site-packages/dask_kubernetes/operator/controller/controller.py", line 610, in daskworkergroup_replica_update
    await worker_deployment.delete()
  File "/usr/local/lib/python3.10/site-packages/kr8s/_objects.py", line 257, in delete
    raise NotFoundError(f"Object {self.name} does not exist") from e
kr8s._exceptions.NotFoundError: Object tcp://100.64.16.72:42101 does not exist
[2024-02-06 11:04:52,241] kopf.objects         [WARNING ] [dask/dask-cluster-default] Patching failed with inconsistencies: (('remove', ('status', 'kopf'), {'progress': {'daskworkergroup_replica_update/spec.worker.replicas': {'started': '2024-02-06T10:04:52.069702', 'stopped': None, 'delayed': '2024-02-06T10:05:52.209109', 'purpose': 'update', 'retries': 1, 'success': False, 'failure': False, 'message': 'Object tcp://100.64.16.72:42101 does not exist', 'subrefs': None}}}, None),)
[2024-02-06 11:04:54,921] kopf.objects         [INFO    ] [dask/dask-cluster] Timer 'daskcluster_autoshutdown' succeeded.

My workers deployment are never deleted and my worker group never scale down, my worker restart immediately when the worker stop.