Merge pull request nolar#853 from nolar/daemon-stopper-not-set

Stop daemons on operator exit

kopf/_cogs/aiokits/aiotasks.py

@@ -321,8 +321,7 @@ async def close(self) -> None:
  Stop accepting new tasks and cancel all running/pending ones.
  """
 
- # Ensure that all pending coros are awaited -- to prevent RuntimeWarnings/ResourceWarnings.
- # But do it via the normal flow, i.e. without exceeding the limit of the scheduler (if any).
+ # Running tasks are cancelled here. Pending tasks are cancelled at actual spawning.
  self._closed = True
  for task in self._running_tasks:
  task.cancel()

kopf/_core/engines/daemons.py

@@ -305,6 +305,7 @@ async def daemon_killer(
  settings=settings,
  daemon=daemon,
  reason=stoppers.DaemonStoppingReason.OPERATOR_EXITING))
+ await scheduler.wait() # prevent insta-cancelling our own coros (daemon stoppers).
  await scheduler.close()
 
 

tests/handling/daemons/conftest.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextlib
 import time
 import unittest.mock
 
@@ -90,12 +91,11 @@ async def background_daemon_killer(settings, memories, operator_paused):
  """
  task = asyncio.create_task(daemon_killer(
  settings=settings, memories=memories, operator_paused=operator_paused))
- yield
- task.cancel()
- try:
+ yield task
+
+ with contextlib.suppress(asyncio.CancelledError):
+ task.cancel()
  await task
- except asyncio.CancelledError:
- pass
 
 
 @pytest.fixture()

tests/handling/daemons/test_daemon_termination.py

@@ -1,12 +1,13 @@
 import asyncio
+import contextlib
 import logging
 
 import pytest
 
 import kopf
 
 
-async def test_daemon_exits_gracefully_and_instantly_on_termination_request(
+async def test_daemon_exits_gracefully_and_instantly_on_resource_deletion(
  settings, resource, dummy, simulate_cycle,
  caplog, assert_logs, k8s_mocked, frozen_time, mocker, timer):
  caplog.set_level(logging.DEBUG)
@@ -39,6 +40,41 @@ async def fn(**kwargs):
  assert k8s_mocked.patch.call_args_list[0][1]['payload']['metadata']['finalizers'] == []
 
 
+async def test_daemon_exits_gracefully_and_instantly_on_operator_exiting(
+ settings, resource, dummy, simulate_cycle, background_daemon_killer,
+ caplog, assert_logs, k8s_mocked, frozen_time, mocker, timer):
+ caplog.set_level(logging.DEBUG)
+
+ # A daemon-under-test.
+ @kopf.daemon(*resource, id='fn')
+ async def fn(**kwargs):
+ dummy.kwargs = kwargs
+ dummy.steps['called'].set()
+ await kwargs['stopped'].wait()
+
+ # 0th cycle: trigger spawning and wait until ready. Assume the finalizers are already added.
+ finalizer = settings.persistence.finalizer
+ event_object = {'metadata': {'finalizers': [finalizer]}}
+ await simulate_cycle(event_object)
+ await dummy.steps['called'].wait()
+
+ # 1st stage: trigger termination due to operator exiting.
+ mocker.resetall()
+ background_daemon_killer.cancel()
+
+ # Check that the daemon has exited near-instantly, with no delays.
+ with timer:
+ await dummy.wait_for_daemon_done()
+
+ assert timer.seconds < 0.01 # near-instantly
+ assert k8s_mocked.sleep.call_count == 0
+ assert k8s_mocked.patch.call_count == 0
+
+ # To prevent double-cancelling of the scheduler's system tasks in the fixture, let them finish:
+ with contextlib.suppress(asyncio.CancelledError):
+ await background_daemon_killer
+
+
 @pytest.mark.usefixtures('background_daemon_killer')
 async def test_daemon_exits_gracefully_and_instantly_on_operator_pausing(
  settings, memories, resource, dummy, simulate_cycle, conflicts_found,