dask / dask/dask-cloudprovider

AWS - ECS/Fargate Cluster - When providing a `scheduler_address`, calling `cluster.close()` terminates the remote scheduler via tcp

Open
#375 1 comment 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
147
Forks
119
PR merge metrics
No merged PRs in 30d

Description

**What happened**:
Calling `cluster.close` from a FargateCluster instance terminates the remote scheduler process.

```
2022-09-01 17:35:21,106 - distributed.scheduler - INFO - Scheduler closing...
2022-09-01 17:35:21,107 - distributed.scheduler - INFO - Scheduler closing all comms
2022-09-01 17:35:21,108 - distributed.scheduler - INFO - Stopped scheduler at 'tcp://172.27.23.183:8786'
2022-09-01 17:35:21,108 - distributed.scheduler - INFO - End scheduler
```

**What you expected to happen**:
The FargateCluster initiate resources should clean up.

**Minimal Complete Verifiable Example**:

```python
def main():
async def run():
logger.info('-' * 47)
cluster = FargateCluster(asynchronous=True,
image=image,
fargate_spot=True,
scheduler_address=scheduler_address,
cluster_arn=cluster_arn,
execution_role_arn=execution_role_arn,
task_role_arn=task_role_arn,
cloudwatch_logs_group=cloudwatch_logs_group,
vpc=vpc,
subnets=subnets,
security_groups=security_groups,
fargate_use_private_ip=True)
cluster.adapt(minimum=1, maximum=5)
await cluster._start()
logger.info('-' * 47)
await asyncio.sleep(120)
await cluster.close()
try:
asyncio.run(run())
finally:
logger.info('Adaptive scaler has ended')
```

**Anything else we need to know?**:
This was a bugaboo to debug. Some breadcrumbs:

dask.distributed Scheduler [close](https://github.com/dask/distributed/blob/main/distributed/scheduler.py#L3695)
distributed SpecCluster [_close](https://github.com/dask/distributed/blob/main/distributed/deploy/spec.py#L426)
distributed SpecCluster [_start](https://github.com/dask/distributed/blob/main/distributed/deploy/spec.py#L310)

It looks to me like `scheduler_address` in CloudProvider is setting `scheduler_comm` on the SpecCluster; thus when `cluster.close()` gets called, `._close()` gets invoked and sends the [terminate](https://github.com/dask/distributed/blob/bfc5cfea80450954dba5b87a5858cb2e3bac1833/distributed/process.py#L275) command via rpc.

(EDIT: I tested with `cluster._close` as well and observed the same behavior.)

**Environment**:

- Dask version: multiple (2022.7.0, 2022.8.1)
- Python version: multiple (3.9, 3.10)
- Operating System: multiple (linux, mac)
- Install method (conda, pip, source): multiple (conda, pip)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.