Change autoscaler adoption to use kr8s (#765)

* Release 2023.7.0

* Make autoscaler adoption more explicit with kr8s

* More minor tweaks to improve readability

* Update objects to use dot notation

* Even more simplification

dask_kubernetes/operator/_objects.py

@@ -85,7 +85,7 @@ async def pods(self) -> List[Pod]:
             Pod.endpoint,
             label_selector=",".join(
                 [
-                    f"dask.org/cluster-name={self.spec['cluster']}",
+                    f"dask.org/cluster-name={self.spec.cluster}",
                     "dask.org/component=worker",
                     f"dask.org/workergroup-name={self.name}",
                 ]
@@ -98,7 +98,7 @@ async def deployments(self) -> List[Deployment]:
             Deployment.endpoint,
             label_selector=",".join(
                 [
-                    f"dask.org/cluster-name={self.spec['cluster']}",
+                    f"dask.org/cluster-name={self.spec.cluster}",
                     "dask.org/component=worker",
                     f"dask.org/workergroup-name={self.name}",
                 ]
@@ -107,7 +107,7 @@ async def deployments(self) -> List[Deployment]:
         )
 
     async def cluster(self) -> DaskCluster:
-        return await DaskCluster.get(self.spec["cluster"], namespace=self.namespace)
+        return await DaskCluster.get(self.spec.cluster, namespace=self.namespace)
 
 
 class DaskAutoscaler(APIObject):
@@ -119,7 +119,7 @@ class DaskAutoscaler(APIObject):
     namespaced = True
 
     async def cluster(self) -> DaskCluster:
-        return await DaskCluster.get(self.spec["cluster"], namespace=self.namespace)
+        return await DaskCluster.get(self.spec.cluster, namespace=self.namespace)
 
 
 class DaskJob(APIObject):

dask_kubernetes/operator/controller/controller.py

@@ -831,11 +831,12 @@ async def handle_runner_status_change_succeeded(meta, namespace, logger, **kwarg
 
 
 @kopf.on.create("daskautoscaler.kubernetes.dask.org")
-async def daskautoscaler_create(name, spec, namespace, logger, patch, **kwargs):
+async def daskautoscaler_create(body, logger, **_):
     """When an autoscaler is created make it a child of the associated cluster for cascade deletion."""
-    cluster = await DaskCluster.get(spec["cluster"], namespace=namespace)
-    kopf.adopt(patch, owner=cluster.raw)
-    logger.info(f"Autoscaler {name} successfully adopted by cluster {spec['cluster']}")
+    autoscaler = await DaskAutoscaler(body)
+    cluster = await autoscaler.cluster()
+    await cluster.adopt(autoscaler)
+    logger.info(f"Autoscaler {autoscaler.name} adopted by cluster {cluster.name}")
 
 
 @kopf.timer("daskautoscaler.kubernetes.dask.org", interval=5.0)

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.4
+kr8s==0.8.5