dask-kubernetes
dask-kubernetes copied to clipboard
dask
TOCTOU Bug while scaling down workers
The currently implemented logic for deletion can be summarized into:
- The operator asks the scheduler to retire n of the workers
- The scheduler retires them (process exits) and returns to the operator names of retired workers
- 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.
Is it possible to allow the scheduler to delete the deployment directly? maybe with something like a plugin or so?
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.
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
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 ;-(
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.
Not sure if this is the same issue you experienced, but I noticed that running worker pods with --nworkers > 1 fails in this way. The individual worker processes are shut down by the scheduler one at a time but the deployment is never deleted because no worker name matches a deployment name.
The operator logs errors like kr8s._exceptions.ServerError: deployments.apps "simple-default-worker-bad5ba722a-4" not found
The operator as currently implemented seems to require only 1 process per worker pod.
@amanfredi that definitely sounds like a bug, it may well be the same bug as this issue.