From 34f2c1343412ec0496de5533256606f868da2813 Mon Sep 17 00:00:00 2001
From: bstadlbauer <11799671+bstadlbauer@users.noreply.github.com>
Date: Thu, 12 Jan 2023 10:15:42 +0100
Subject: [PATCH] Set `lifetime-stagger` default value to `None` (#7445)

---
 distributed/cli/dask_worker.py            |  6 ++---
 distributed/cli/tests/test_dask_worker.py | 33 +++++++++++++++++++++++
 2 files changed, 35 insertions(+), 4 deletions(-)

diff --git a/distributed/cli/dask_worker.py b/distributed/cli/dask_worker.py
index c000ed1d2d..ad2b3c754a 100755
--- a/distributed/cli/dask_worker.py
+++ b/distributed/cli/dask_worker.py
@@ -190,8 +190,7 @@
 @click.option(
     "--lifetime-stagger",
     type=str,
-    default="0 seconds",
-    show_default=True,
+    default=None,
     help="Random amount by which to stagger lifetime values",
 )
 @click.option(
@@ -204,8 +203,7 @@
 @click.option(
     "--lifetime-restart/--no-lifetime-restart",
     "lifetime_restart",
-    default=False,
-    show_default=True,
+    default=None,
     required=False,
     help="Whether or not to restart the worker after the lifetime lapses. "
     "This assumes that you are using the --lifetime and --nanny keywords",
diff --git a/distributed/cli/tests/test_dask_worker.py b/distributed/cli/tests/test_dask_worker.py
index 3750b24e0b..7212ee36d7 100644
--- a/distributed/cli/tests/test_dask_worker.py
+++ b/distributed/cli/tests/test_dask_worker.py
@@ -610,6 +610,39 @@ def dask_setup(worker):
         assert foo == "setup"
 
 
+@pytest.mark.slow
+@gen_cluster(nthreads=[], client=True)
+async def test_set_lifetime_stagger_via_env_var(c, s):
+    # Ensure that lifetime stagger can be set via an environment variable
+    env = os.environ.copy()
+    env["DASK_DISTRIBUTED__WORKER__LIFETIME__DURATION"] = "10 seconds"
+    env["DASK_DISTRIBUTED__WORKER__LIFETIME__STAGGER"] = "2 seconds"
+    with popen(["dask", "worker", s.address], env=env), popen(
+        ["dask", "worker", s.address], env=env
+    ):
+        await c.wait_for_workers(2)
+        [lifetime1, lifetime2] = (
+            await c.run(lambda dask_worker: dask_worker.lifetime)
+        ).values()
+        assert lifetime1 != lifetime2
+        assert 8 <= lifetime1 <= 12
+        assert 8 <= lifetime2 <= 12
+
+
+@pytest.mark.slow
+@gen_cluster(nthreads=[], client=True)
+async def test_set_lifetime_restart_via_env_var(c, s):
+    # Ensure that lifetime restart can be set via an environment variable
+    env = os.environ.copy()
+    env["DASK_DISTRIBUTED__WORKER__LIFETIME__RESTART"] = "True"
+    with popen(["dask", "worker", s.address], env=env):
+        await c.wait_for_workers(1)
+        [lifetime_restart] = (
+            await c.run(lambda dask_worker: dask_worker.lifetime_restart)
+        ).values()
+        assert lifetime_restart
+
+
 @pytest.mark.slow
 @pytest.mark.parametrize("nanny", ["--nanny", "--no-nanny"])
 def test_timeout(nanny):