dask / dask/dask-cloudprovider
AWS - ECS/Fargate Cluster - When providing a `scheduler_address`, calling `cluster.close()` terminates the remote scheduler via tcp
- 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
Assessment
This issue has not been assessed yet.