Delay lookup of allowed failures. (dask#2761)

This allows for setting the config after importing distributed

xref dask/dask-examples#75 (comment)

distributed/scheduler.py

@@ -73,8 +73,6 @@
 logger = logging.getLogger(__name__)
 
 
-ALLOWED_FAILURES = dask.config.get("distributed.scheduler.allowed-failures")
-
 LOG_PDB = dask.config.get("distributed.admin.pdb-on-err")
 DEFAULT_DATA_SIZE = dask.config.get("distributed.scheduler.default-data-size")
 
@@ -829,7 +827,7 @@ def __init__(
         synchronize_worker_interval="60s",
         services=None,
         service_kwargs=None,
-        allowed_failures=ALLOWED_FAILURES,
+        allowed_failures=None,
         extensions=None,
         validate=False,
         scheduler_file=None,
@@ -846,6 +844,8 @@ def __init__(
         self._setup_logging(logger)
 
         # Attributes
+        if allowed_failures is None:
+            allowed_failures = dask.config.get("distributed.scheduler.allowed-failures")
         self.allowed_failures = allowed_failures
         self.validate = validate
         self.status = None

distributed/tests/test_scheduler.py

@@ -1607,3 +1607,17 @@ async def test_async_context_manager():
             assert w.status == "running"
             assert s.workers
         assert not s.workers
+
+
+@pytest.mark.asyncio
+async def test_allowed_failures_config():
+    async with Scheduler(port=0, allowed_failures=10) as s:
+        assert s.allowed_failures == 10
+
+    with dask.config.set({"distributed.scheduler.allowed_failures": 100}):
+        async with Scheduler(port=0) as s:
+            assert s.allowed_failures == 100
+
+    with dask.config.set({"distributed.scheduler.allowed_failures": 0}):
+        async with Scheduler(port=0) as s:
+            assert s.allowed_failures == 0