would it make sense to move the implementation to a custom dask-cloudprovider Cluster subclass until the design solidifies and we can adopt it in dask-distributed?

class CloudInfoCluster(Cluster):
    def __init__(self, *args, cloud_info_sync_interval=1, **kwargs):
        self._cloud_info_sync_interval = cloud_info_sync_interval
        self._cloud_info = {}
        super().__init__(self, *args, **kwargs)
        
    async def _start(self):
        await super()._start()
        self._sync_cloud_info_task = asyncio.create_task(self._sync_cloud_info())
        
    async def _sync_cloud_info(self):
        err_count = 0
        warn_at = 5
        max_interval = 10 * self._cloud_info_sync_interval
        # Loop until the cluster is shutting down. We shouldn't really need
        # this check (the `CancelledError` should be enough), but something
        # deep in the comms code is silencing `CancelledError`s _some_ of the
        # time, resulting in a cancellation not always bubbling back up to
        # here. Relying on the status is fine though, not worth changing.
        while self.status == Status.running:
            try:
                await self.scheduler_comm.set_metadata(
                    keys=["cloud-info"],
                    value=self._cloud_info.copy(),
                )
                err_count = 0
            except asyncio.CancelledError:
                # Task is being closed. When we drop Python < 3.8 we can drop
                # this check (since CancelledError is not a subclass of
                # Exception then).
                break
            except Exception:
                err_count += 1
                # Only warn if multiple subsequent attempts fail, and only once
                # per set of subsequent failed attempts. This way we're not
                # excessively noisy during a connection blip, but we also don't
                # silently fail.
                if err_count == warn_at:
                    logger.warning(
                        "Failed to sync cluster info multiple times - perhaps "
                        "there's a connection issue? Error:",
                        exc_info=True,
                    )
            # Sleep, with error backoff
            interval = min(max_interval, self._sync_interval * 1.5**err_count)
            await asyncio.sleep(interval)

        async def _close(self):
            try:
                if self._sync_cloud_info_task:
                    self._sync_cloud_info_task.cancel()
                    with suppress(asyncio.CancelledError):
                        await self._sync_cloud_info_task
             finaly:
                 return await super()._close()

Uh oh!

Add cluster manager info to scheduler #6489

Description

Metadata

Metadata

Assignees

Labels

Type

Fields

Projects

Milestone

Relationships

Development

Issue actions