Replace kubernetes_asyncio with kr8s in cluster discovery (#784)

* Replace kubernetes_asyncio with kr8s in cluster discovery

* Ensure DaskCluster kr8s resource is registered

* Allow more reruns

* Nudge GitHub Actions

dask_kubernetes/operator/kubecluster/discovery.py

@@ -1,26 +1,15 @@
 from typing import Callable, AsyncIterator, Tuple
 
-import kubernetes_asyncio as kubernetes
+import kr8s
 
 from dask_kubernetes.operator.kubecluster import KubeCluster
-from dask_kubernetes.common.auth import ClusterAuth
-from dask_kubernetes.common.utils import get_current_namespace
+from dask_kubernetes.operator._objects import DaskCluster  # noqa
 
 
 async def discover() -> AsyncIterator[Tuple[str, Callable]]:
-
-    await ClusterAuth.load_first()
-
     try:
-        async with kubernetes.client.api_client.ApiClient() as api_client:
-            custom_objects_api = kubernetes.client.CustomObjectsApi(api_client)
-            clusters = await custom_objects_api.list_namespaced_custom_object(
-                group="kubernetes.dask.org",
-                version="v1",
-                plural="daskclusters",
-                namespace=get_current_namespace(),
-            )
-        for cluster in clusters["items"]:
-            yield (cluster["metadata"]["name"], KubeCluster)
+        clusters = await kr8s.asyncio.get("daskclusters", namespace=kr8s.ALL)
+        for cluster in clusters:
+            yield (cluster.name, KubeCluster)
     except Exception:
         return

setup.cfg

@@ -39,4 +39,4 @@ parentdir_prefix = dask-kubernetes-
 addopts = -v -x --keep-cluster --durations=10
 timeout = 60
 timeout_func_only = true
-reruns = 3
+reruns = 5