Improve CI stability (#5022)

* Drastically reduced CI failure rates
* Extensive review of all timeouts
* Statistical review of flaky tests
* Generate and publish junit reports for later ingestion by your choice of aggregator
* On Linux and MacOS, pytest-timeout will now kill the individual offending test instead of the whole suite
* pytest-timeout should kick in a lot less frequently now
* gen_test and gen_cluster no longer accept timeout=None (which meant letting pytest-timeout kill everything off)
* Hopefully increased mamba resiliency to transitory network issues

Author: crusaderky

.github/workflows/tests.yaml

@@ -11,17 +11,26 @@ concurrency:
 jobs:
   test:
     runs-on: ${{ matrix.os }}
+    timeout-minutes: 180
 
     strategy:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
         python-version: ["3.7", "3.8", "3.9"]
 
-        # Uncomment to stress-test the test suite for random failures
+        # Uncomment to stress-test the test suite for random failures.
+        # Must also change env.TEST_ID below.
         # This will take a LONG time and delay all PRs across the whole github.com/dask!
+        # To avoid hamstringing other people, change 'on: [push, pull_request]' above
+        # to just 'on: [push]'; this way the stress test will run exclusively in your
+        # branch (https://github.com/<your name>/distributed/actions).
         # run: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]
 
+    env:
+      TEST_ID: ${{ matrix.os }}-${{ matrix.python-version }}
+      # TEST_ID: ${{ matrix.os }}-${{ matrix.python-version }}-${{ matrix.run }}
+
     steps:
       - name: Checkout source
         uses: actions/checkout@v2
@@ -33,13 +42,15 @@ jobs:
         with:
           miniforge-variant: Mambaforge
           miniforge-version: latest
+          condarc-file: continuous_integration/condarc
           use-mamba: true
-          channels: conda-forge,defaults
-          channel-priority: true
           python-version: ${{ matrix.python-version }}
           environment-file: continuous_integration/environment-${{ matrix.python-version }}.yaml
           activate-environment: dask-distributed
-          auto-activate-base: false
+
+      - name: Show conda options
+        shell: bash -l {0}
+        run: conda config --show
 
       - name: Install stacktrace
         shell: bash -l {0}
@@ -83,7 +94,14 @@ jobs:
         if: ${{ matrix.os != 'windows-latest' }}
         run: bash continuous_integration/scripts/setup_ssh.sh
 
+      - name: Reconfigure pytest-timeout
+        shell: bash -l {0}
+        # No SIGALRM available on Windows
+        if: ${{ matrix.os != 'windows-latest' }}
+        run: sed -i.bak 's/timeout_method = thread/timeout_method = signal/' setup.cfg
+
       - name: Test
+        id: run_tests
         shell: bash -l {0}
         env:
           PYTHONFAULTHANDLER: 1
@@ -93,10 +111,21 @@ jobs:
               # https://github.com/dask/distributed/issues/4514
               export DISABLE_IPV6=1
           fi
-
           source continuous_integration/scripts/set_ulimit.sh
-          pytest distributed -m "not avoid_ci" --runslow
+
+          pytest distributed -m "not avoid_ci" --runslow \
+            --junitxml reports/pytest.xml -o junit_suite_name=$TEST_ID
 
       # - name: Debug with tmate on failure
       #   if: ${{ failure() }}
       #   uses: mxschmitt/action-tmate@v3
+
+      - name: Upload test artifacts
+        # ensure this runs even if pytest fails
+        if: >
+          always() &&
+          (steps.run_tests.outcome == 'success' || steps.run_tests.outcome == 'failure')
+        uses: actions/upload-artifact@v2
+        with:
+          name: ${{ env.TEST_ID }}
+          path: reports

continuous_integration/condarc

@@ -0,0 +1,9 @@
+channels:
+  - conda-forge
+  - defaults
+channel_priority: true
+auto_activate_base: false
+remote_backoff_factor: 20
+remote_connect_timeout_secs: 20.0
+remote_max_retries: 10
+remote_read_timeout_secs: 60.0

distributed/cli/tests/test_dask_scheduler.py

@@ -1,3 +1,4 @@
+import psutil
 import pytest
 
 pytest.importorskip("requests")
@@ -14,6 +15,7 @@
 import distributed
 import distributed.cli.dask_scheduler
 from distributed import Client, Scheduler
+from distributed.compatibility import LINUX
 from distributed.metrics import time
 from distributed.utils import get_ip, get_ip_interface, tmpfile
 from distributed.utils_test import (
@@ -118,9 +120,7 @@ def test_dashboard_non_standard_ports(loop):
         requests.get("http://localhost:4832/status/")
 
 
-@pytest.mark.skipif(
-    not sys.platform.startswith("linux"), reason="Need 127.0.0.2 to mean localhost"
-)
+@pytest.mark.skipif(not LINUX, reason="Need 127.0.0.2 to mean localhost")
 def test_dashboard_whitelist(loop):
     pytest.importorskip("bokeh")
     with pytest.raises(Exception):
@@ -144,7 +144,6 @@ def test_dashboard_whitelist(loop):
 
 
 def test_interface(loop):
-    psutil = pytest.importorskip("psutil")
     if_names = sorted(psutil.net_if_addrs())
     for if_name in if_names:
         try:
@@ -168,25 +167,24 @@ def test_interface(loop):
                 start = time()
                 while not len(c.nthreads()):
                     sleep(0.1)
-                    assert time() - start < 5
+                    assert time() - start < 30
                 info = c.scheduler_info()
                 assert "tcp://127.0.0.1" in info["address"]
                 assert all("127.0.0.1" == d["host"] for d in info["workers"].values())
 
 
-@pytest.mark.flaky(reruns=10, reruns_delay=5)
 def test_pid_file(loop):
     def check_pidfile(proc, pidfile):
         start = time()
         while not os.path.exists(pidfile):
             sleep(0.01)
-            assert time() < start + 5
+            assert time() < start + 30
 
         text = False
         start = time()
         while not text:
             sleep(0.01)
-            assert time() < start + 5
+            assert time() < start + 30
             with open(pidfile) as f:
                 text = f.read()
         pid = int(text)

distributed/cli/tests/test_dask_worker.py

@@ -6,14 +6,14 @@
 pytest.importorskip("requests")
 
 import os
-import sys
 from multiprocessing import cpu_count
 from time import sleep
 
 import requests
 
 import distributed.cli.dask_worker
 from distributed import Client, Scheduler
+from distributed.compatibility import LINUX
 from distributed.deploy.utils import nprocesses_nthreads
 from distributed.metrics import time
 from distributed.utils import parse_ports, sync, tmpfile
@@ -275,9 +275,7 @@ def test_nprocs_expands_name(loop):
                     assert len(set(names)) == 4
 
 
-@pytest.mark.skipif(
-    not sys.platform.startswith("linux"), reason="Need 127.0.0.2 to mean localhost"
-)
+@pytest.mark.skipif(not LINUX, reason="Need 127.0.0.2 to mean localhost")
 @pytest.mark.parametrize("nanny", ["--nanny", "--no-nanny"])
 @pytest.mark.parametrize(
     "listen_address", ["tcp://0.0.0.0:39837", "tcp://127.0.0.2:39837"]
@@ -311,9 +309,7 @@ def func(dask_worker):
                 assert client.run(func) == {"tcp://127.0.0.2:39837": listen_address}
 
 
-@pytest.mark.skipif(
-    not sys.platform.startswith("linux"), reason="Need 127.0.0.2 to mean localhost"
-)
+@pytest.mark.skipif(not LINUX, reason="Need 127.0.0.2 to mean localhost")
 @pytest.mark.parametrize("nanny", ["--nanny", "--no-nanny"])
 @pytest.mark.parametrize("host", ["127.0.0.2", "0.0.0.0"])
 def test_respect_host_listen_address(loop, nanny, host):

distributed/comm/tests/test_comms.py

@@ -29,7 +29,6 @@
 )
 from distributed.comm.registry import backends, get_backend
 from distributed.comm.tcp import TCP, TCPBackend, TCPConnector
-from distributed.compatibility import WINDOWS
 from distributed.metrics import time
 from distributed.protocol import Serialized, deserialize, serialize, to_serialize
 from distributed.utils import get_ip, get_ipv6
@@ -110,7 +109,7 @@ async def debug_loop():
     while True:
         loop = ioloop.IOLoop.current()
         print(".", loop, loop._handlers)
-        await asyncio.sleep(0.50)
+        await asyncio.sleep(0.5)
 
 
 #
@@ -1107,7 +1106,6 @@ def check_out(deserialize_flag, out_value):
     await check_connector_deserialize(addr, True, msg, partial(check_out, True))
 
 
-@pytest.mark.flaky(reruns=10, reruns_delay=5, condition=WINDOWS)
 @pytest.mark.asyncio
 async def test_tcp_deserialize():
     await check_deserialize("tcp://")

distributed/comm/tests/test_ucx.py

@@ -276,7 +276,7 @@ async def test_ucx_localcluster(processes, cleanup):
     ) as cluster:
         async with Client(cluster, asynchronous=True) as client:
             x = client.submit(inc, 1)
-            await x.result()
+            await x
             assert x.key in cluster.scheduler.tasks
             if not processes:
                 assert any(w.data == {x.key: 2} for w in cluster.workers.values())

distributed/comm/tests/test_ws.py

@@ -122,9 +122,9 @@ async def test_collections(cleanup):
 async def test_large_transfer(cleanup):
     np = pytest.importorskip("numpy")
     async with Scheduler(protocol="ws://") as s:
-        async with Worker(s.address, protocol="ws://") as w:
+        async with Worker(s.address, protocol="ws://"):
             async with Client(s.address, asynchronous=True) as c:
-                future = await c.scatter(np.random.random(1000000))
+                await c.scatter(np.random.random(1_000_000))
 
 
 @pytest.mark.asyncio

distributed/compatibility.py

@@ -8,6 +8,7 @@
 logging_names.update(logging._nameToLevel)
 
 PYPY = platform.python_implementation().lower() == "pypy"
+LINUX = sys.platform == "linux"
 MACOS = sys.platform == "darwin"
 WINDOWS = sys.platform.startswith("win")
 TORNADO6 = tornado.version_info[0] >= 6

distributed/dashboard/tests/test_scheduler_bokeh.py

@@ -17,7 +17,6 @@
 from dask.utils import stringify
 
 from distributed.client import wait
-from distributed.compatibility import MACOS
 from distributed.dashboard import scheduler
 from distributed.dashboard.components.scheduler import (
     AggregateAction,
@@ -93,10 +92,8 @@ async def test_counters(c, s, a, b):
     await asyncio.sleep(0.1)
     ss.update()
 
-    start = time()
     while not len(ss.digest_sources["tick-duration"][0].data["x"]):
-        await asyncio.sleep(1)
-        assert time() < start + 5
+        await asyncio.sleep(0.01)
 
 
 @gen_cluster(client=True)
@@ -184,15 +181,15 @@ async def test_task_stream_clear_interval(c, s, a, b):
 
     await wait(c.map(inc, range(10)))
     ts.update()
-    await asyncio.sleep(0.010)
+    await asyncio.sleep(0.01)
     await wait(c.map(dec, range(10)))
     ts.update()
 
     assert len(set(map(len, ts.source.data.values()))) == 1
     assert ts.source.data["name"].count("inc") == 10
     assert ts.source.data["name"].count("dec") == 10
 
-    await asyncio.sleep(0.300)
+    await asyncio.sleep(0.3)
     await wait(c.map(inc, range(10, 20)))
     ts.update()
 
@@ -848,7 +845,6 @@ async def test_aggregate_action(c, s, a, b):
     assert ("compute") in mbk.action_source.data["names"]
 
 
-@pytest.mark.flaky(reruns=10, reruns_delay=5, condition=MACOS)
 @gen_cluster(client=True, scheduler_kwargs={"dashboard": True})
 async def test_compute_per_key(c, s, a, b):
     mbk = ComputePerKey(s)

distributed/deploy/local.py

@@ -136,8 +136,8 @@ def __init__(
 
         if threads_per_worker == 0:
             warnings.warn(
-                "Setting `threads_per_worker` to 0 is discouraged. "
-                "Please set to None or to a specific int to get best behavior."
+                "Setting `threads_per_worker` to 0 has been deprecated. "
+                "Please set to None or to a specific int."
             )
             threads_per_worker = None