[BUG] - Failing to create a new dask cluster

[marcelovilla]

Describe the bug

When trying to create a dask cluster, I am seeing the following error:

ValueError                                Traceback (most recent call last)
Cell In[2], line 2
      1 # create a new cluster with our options
----> 2 cluster = gateway.new_cluster(options)
      3 # view the cluster UI
      4 cluster

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:641, in Gateway.new_cluster(self, cluster_options, shutdown_on_close, **kwargs)
    618 def new_cluster(self, cluster_options=None, shutdown_on_close=True, **kwargs):
    619     """Submit a new cluster to the gateway, and wait for it to be started.
    620 
    621     Same as calling ``submit`` and ``connect`` in one go.
   (...)
    639     cluster : GatewayCluster
    640     """
--> 641     return GatewayCluster(
    642         address=self.address,
    643         proxy_address=self.proxy_address,
    644         public_address=self._public_address,
    645         auth=self.auth,
    646         asynchronous=self.asynchronous,
    647         loop=self.loop,
    648         cluster_options=cluster_options,
    649         shutdown_on_close=shutdown_on_close,
    650         **kwargs,
    651     )

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:816, in GatewayCluster.__init__(self, address, proxy_address, public_address, auth, cluster_options, shutdown_on_close, asynchronous, loop, **kwargs)
    804 def __init__(
    805     self,
    806     address=None,
   (...)
    814     **kwargs,
    815 ):
--> 816     self._init_internal(
    817         address=address,
    818         proxy_address=proxy_address,
    819         public_address=public_address,
    820         auth=auth,
    821         cluster_options=cluster_options,
    822         cluster_kwargs=kwargs,
    823         shutdown_on_close=shutdown_on_close,
    824         asynchronous=asynchronous,
    825         loop=loop,
    826     )

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:921, in GatewayCluster._init_internal(self, address, proxy_address, public_address, auth, cluster_options, cluster_kwargs, shutdown_on_close, asynchronous, loop, name)
    919     self.status = "starting"
    920 if not self.asynchronous:
--> 921     self.gateway.sync(self._start_internal)

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:344, in Gateway.sync(self, func, *args, **kwargs)
    340 future = asyncio.run_coroutine_threadsafe(
    341     func(*args, **kwargs), self.loop.asyncio_loop
    342 )
    343 try:
--> 344     return future.result()
    345 except BaseException:
    346     future.cancel()

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/concurrent/futures/_base.py:456, in Future.result(self, timeout)
    454     raise CancelledError()
    455 elif self._state == FINISHED:
--> 456     return self.__get_result()
    457 else:
    458     raise TimeoutError()

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/concurrent/futures/_base.py:401, in Future.__get_result(self)
    399 if self._exception:
    400     try:
--> 401         raise self._exception
    402     finally:
    403         # Break a reference cycle with the exception in self._exception
    404         self = None

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:935, in GatewayCluster._start_internal(self)
    933     self._start_task = asyncio.ensure_future(self._start_async())
    934 try:
--> 935     await self._start_task
    936 except BaseException:
    937     # On exception, cleanup
    938     await self._stop_internal()

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:948, in GatewayCluster._start_async(self)
    946 if self.status == "created":
    947     self.status = "starting"
--> 948     self.name = await self.gateway._submit(
    949         cluster_options=self._cluster_options, **self._cluster_kwargs
    950     )
    951 # Connect to cluster
    952 try:

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:536, in Gateway._submit(self, cluster_options, **kwargs)
    534     options = self._config_cluster_options()
    535     options.update(kwargs)
--> 536 resp = await self._request("POST", url, json={"cluster_options": options})
    537 data = await resp.json()
    538 return data["name"]

File /home/conda/nebari-git/5475fd44-1708374389-2-dask/lib/python3.11/site-packages/dask_gateway/client.py:414, in Gateway._request(self, method, url, json)
    411     msg = await resp.text()
    413 if resp.status in {404, 422}:
--> 414     raise ValueError(msg)
    415 elif resp.status == 409:
    416     raise GatewayClusterError(msg)

ValueError: 'dict' object has no attribute 'provider'

Expected behavior

I would expect the cluster to start normally.

OS and architecture in which you are running Nebari

AWS

How to Reproduce the problem?

Run the following code using the nebari-git-dask environment or any other environment with dask on it:

from dask_gateway import Gateway
# instantiate dask gateway
gateway = Gateway()

# view the cluster options UI
options = gateway.cluster_options()
options

# create a new cluster with our options
cluster = gateway.new_cluster(options)
# view the cluster UI
cluster

Command output

No response

Versions and dependencies used.

This happened testing Nebari against 195738e

Compute environment

AWS

Integrations

Dask

Anything else?

No response