Replace kubernetes_asyncio with kr8s in daskworkergroup_create (#758)

* Replace kubernetes_asyncio with kr8s in daskworkergroup_create

* Increase timeout

* Be explicit with owner kwarg

* Bump kr8s to workaround bug

* Bump kr8s again to workaround bug in Python 3.9 too

* Revert timeout increase

* Reinstate timeout increase

* Switch to kr8s adoption method

* Simplify

* Fix problem with body not being a dict

* Revert workaround due to upstream fix in kr8s 0.8.6

* Fail tests early so we can see tracebacks for those that time out

* Remove maxfail

dask_kubernetes/operator/controller/controller.py

@@ -383,38 +383,17 @@ async def handle_scheduler_service_status(
 
 
 @kopf.on.create("daskworkergroup.kubernetes.dask.org")
-async def daskworkergroup_create(spec, name, namespace, logger, **kwargs):
-    async with kubernetes.client.api_client.ApiClient() as api_client:
-        api = kubernetes.client.CustomObjectsApi(api_client)
-        cluster = await api.get_namespaced_custom_object(
-            group="kubernetes.dask.org",
-            version="v1",
-            plural="daskclusters",
-            namespace=namespace,
-            name=spec["cluster"],
-        )
-        new_spec = dict(spec)
-        kopf.adopt(new_spec, owner=cluster)
-        api.api_client.set_default_header(
-            "content-type", "application/merge-patch+json"
-        )
-        await api.patch_namespaced_custom_object(
-            group="kubernetes.dask.org",
-            version="v1",
-            plural="daskworkergroups",
-            namespace=namespace,
-            name=name,
-            body=new_spec,
-        )
-        logger.info(f"Successfully adopted by {spec['cluster']}")
+async def daskworkergroup_create(body, logger, **kwargs):
+    wg = await DaskWorkerGroup(body)
+    cluster = await wg.cluster()
+    await cluster.adopt(wg)
+    logger.info(f"Successfully adopted by {cluster.name}")
 
     del kwargs["new"]
     await daskworkergroup_replica_update(
-        spec=spec,
-        name=name,
-        namespace=namespace,
+        body=body,
         logger=logger,
-        new=spec["worker"]["replicas"],
+        new=wg.replicas,
         **kwargs,
     )
 

requirements.txt

@@ -5,4 +5,4 @@ kubernetes-asyncio>=12.0.1
 kopf>=1.35.3
 pykube-ng>=22.9.0
 rich>=12.5.1
-kr8s==0.8.5
+kr8s==0.8.6