Update scheduler_info across cluster instances (#728)

* Update scheduler_info across cluster instances

* Check cluster name and add test

---------

Co-authored-by: Jacob Tomlinson <[email protected]>

Author: Matt711

dask_kubernetes/operator/kubecluster/kubecluster.py

@@ -791,6 +791,9 @@ async def _scale(self, n, worker_group="default"):
                 name=f"{self.name}-{worker_group}",
                 body={"spec": {"replicas": n}},
             )
+            for instance in self._instances:
+                if instance.name == self.name:
+                    instance.scheduler_info = self.scheduler_info
 
     def adapt(self, minimum=None, maximum=None):
         """Turn on adaptivity

dask_kubernetes/operator/kubecluster/tests/test_kubecluster.py

@@ -104,6 +104,14 @@ def test_cluster_from_name(kopf_runner, docker_image, ns):
                 assert firstcluster == secondcluster
 
 
+def test_cluster_scheduler_info_updated(kopf_runner, docker_image, ns):
+    with kopf_runner:
+        with KubeCluster(name="abc", namespace=ns, image=docker_image) as firstcluster:
+            with KubeCluster.from_name("abc", namespace=ns) as secondcluster:
+                firstcluster.scale(1)
+                assert firstcluster.scheduler_info == secondcluster.scheduler_info
+
+
 def test_additional_worker_groups(kopf_runner, docker_image):
     with kopf_runner:
         with KubeCluster(