Work around dask/distributed#7940 (#738)

* Poke the CI

* Bump Kubernetes versions

* Verify things pass with dask/distributed@74a1bcd

* Remove deprecated CI runs to speed up testing

* Pin to commit that I suspect is causing this, expect this run to fail

* Pin to previous commit which I expect to pass

* Implement workaround

* Revert removed CI files

* Copy asynchronous checking from SpecCluster

* Add comment

Author: jacobtomlinson

.github/workflows/operator.yaml

@@ -40,14 +40,14 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.9", "3.10"]
-        kubernetes-version: ["1.27.0"]
+        kubernetes-version: ["1.27.3"]
         include:
         - python-version: "3.10"
-          kubernetes-version: "1.24.13"
+          kubernetes-version: "1.24.15"
         - python-version: "3.10"
-          kubernetes-version: "1.25.9"
+          kubernetes-version: "1.25.11"
         - python-version: "3.10"
-          kubernetes-version: "1.26.4"
+          kubernetes-version: "1.26.6"
 
     env:
       KUBECONFIG: .pytest-kind/pytest-kind/kubeconfig

dask_kubernetes/operator/kubecluster/kubecluster.py

@@ -178,9 +178,10 @@ def __init__(
         scheduler_service_type=None,
         custom_cluster_spec=None,
         scheduler_forward_port=None,
+        loop=None,
+        asynchronous=False,
         **kwargs,
     ):
-
         name = dask.config.get("kubernetes.name", override_with=name)
         self.namespace = (
             dask.config.get("kubernetes.namespace", override_with=namespace)
@@ -261,8 +262,18 @@ def __init__(
         self._rich_spinner = Spinner("dots", speed=0.5)
         self._startup_component_status = {}
 
-        super().__init__(name=name, **kwargs)
-        if not self.asynchronous:
+        super().__init__(name=name, loop=loop, asynchronous=asynchronous, **kwargs)
+
+        # If https://github.com/dask/distributed/pull/7941 is merged we can
+        # simplify the next 8 lines to ``if not self.called_from_running_loop:``
+        try:
+            called_from_running_loop = (
+                getattr(loop, "asyncio_loop", None) is asyncio.get_running_loop()
+            )
+        except RuntimeError:
+            called_from_running_loop = asynchronous
+
+        if not called_from_running_loop:
             self._loop_runner.start()
             self.sync(self._start)
 
@@ -543,7 +554,7 @@ async def generate_rich_output(self):
         table.add_column("Component")
         table.add_column("Status", justify="right")
 
-        for (label, component) in [
+        for label, component in [
             ("DaskCluster", "cluster"),
             ("Scheduler Pod", "schedulerpod"),
             ("Scheduler Service", "schedulerservice"),