Issue/7694 executor cleanup

src/inmanta/agent/config.py

@@ -17,6 +17,7 @@
 """
 
 import enum
+import functools
 import logging
 import typing
 import uuid
@@ -188,7 +189,7 @@ def is_executor_mode(value: str | AgentExecutorMode) -> AgentExecutorMode:
     3,
     "Maximum number of concurrent executors to keep per environment, per agent. If this limit is already reached "
     "when creating a new executor, the oldest one will be stopped first.",
-    is_int,
+    functools.partial(is_lower_bounded_int, 1),
 )
 
 agent_executor_retention_time = Option[int](

src/inmanta/agent/forking_executor.py

@@ -32,7 +32,7 @@
 import uuid
 from abc import abstractmethod
 from asyncio import Future, transports
-from typing import Any, Coroutine, Optional, Sequence
+from typing import Optional, Sequence
 
 import inmanta.agent.cache
 import inmanta.agent.executor
@@ -606,13 +606,9 @@ def __init__(
         self.max_executors_per_agent = inmanta.agent.config.agent_executor_cap.get()
 
         self.stopping: bool = False
-        # Keep a reference to all active backgrounds tasks to make sure they don't disappear
-        # mid-execution and to be able to cancel them when shutting down.
-        self.background_tasks: set[asyncio.Task[None]] = set()
-
-        # Keep a reference to the last scheduled cleanup task for easy cancelation
-        # and rescheduling when the retention time is updated
-        self.next_executor_cleanup_task: Optional[asyncio.Task[None]] = None
+        # We keep a reference to the periodic cleanup task to prevent it
+        # from disappearing mid-execution https://docs.python.org/3.11/library/asyncio-task.html#creating-tasks
+        self.cleanup_job: Optional[asyncio.Task[None]] = None
 
     def __add_executor(self, theid: executor.ExecutorId, the_executor: MPExecutor) -> None:
         self.executor_map[theid] = the_executor
@@ -784,13 +780,10 @@ def _make_child(self, name: str, log_level: int, cli_log: bool) -> tuple[multipr
         return p, parent_conn
 
     async def start(self) -> None:
-        self.start_executor_retention_time_watcher()
-        self.start_periodic_executor_cleanup()
+        self.cleanup_job = asyncio.create_task(self.cleanup_inactive_executors())
 
     async def stop(self) -> None:
         self.stopping = True
-        for task in self.background_tasks:
-            task.cancel()
         await asyncio.gather(*(child.stop() for child in self.children))
 
     async def force_stop(self, grace_time: float) -> None:
@@ -806,49 +799,11 @@ async def stop_for_agent(self, agent_name: str) -> list[MPExecutor]:
         await asyncio.gather(*(child.stop() for child in children))
         return children
 
-    def start_executor_retention_time_watcher(self) -> None:
-        """
-        Schedule a task that periodically checks for updates to the executor
-        retention time. If such an update is detected, the previous recurring cleanup
-        task is canceled and a new recurring cleanup task with the updated retention time
-        is scheduled.
-        """
-
-        async def watch_retention_time() -> None:
-            if self.stopping:
-                # Stop periodic re-scheduling when the manager is shutting down
-                return
-
-            retention_time_from_config = inmanta.agent.config.agent_executor_retention_time.get()
-
-            if self.executor_retention_time != retention_time_from_config:
-                self.executor_retention_time = retention_time_from_config
-                self.start_periodic_executor_cleanup(restart=True)
-
-            await asyncio.sleep(2)
-            self.add_background_task(watch_retention_time())
-
-        self.add_background_task(watch_retention_time())
-
-    def start_periodic_executor_cleanup(self, restart: Optional[bool] = False) -> None:
+    async def cleanup_inactive_executors(self) -> None:
         """
-        Schedule a task that removes inactive executors and re-schedules
-        itself based on the retention time.
-
-        :param restart: Cancel the previously scheduled cleanup task and start a new one.
+        This task periodically cleans up idle executors
         """
-
-        async def cleanup_inactive_executors() -> None:
-            """
-            This task cleans up idle executors and reschedules itself
-            """
-            if self.stopping:
-                # Stop periodic re-scheduling when the manager is shutting down
-                return
-
-            if restart and self.next_executor_cleanup_task:
-                self.next_executor_cleanup_task.cancel()
-
+        while not self.stopping:
             cleanup_start = datetime.datetime.now().astimezone()
 
             reschedule_interval: float = self.executor_retention_time
@@ -884,18 +839,3 @@ async def cleanup_inactive_executors() -> None:
             cleanup_end = datetime.datetime.now().astimezone()
 
             await asyncio.sleep(max(0.0, reschedule_interval - (cleanup_end - cleanup_start).total_seconds()))
-            self.next_executor_cleanup_task = self.add_background_task(cleanup_inactive_executors())
-
-        self.add_background_task(cleanup_inactive_executors())
-
-    def add_background_task(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]:
-        """
-        Wrap a coroutine in an asyncio Task and register it.
-        """
-        task = asyncio.create_task(coro)
-
-        self.background_tasks.add(task)
-        # Make sure the task removes itself from the set when it's done
-        task.add_done_callback(self.background_tasks.discard)
-
-        return task

src/inmanta/config.py

@@ -262,6 +262,14 @@ def is_list(value: str | list[str]) -> list[str]:
     return [] if value == "" else [x.strip() for x in value.split(",")]
 
 
+def is_lower_bounded_int(lower_bound: int, value: str | int) -> int:
+    """lower-bounded int"""
+    to_int = int(value)
+    if to_int < lower_bound:
+        raise ValueError(f"Value can not be lower than {lower_bound}")
+    return to_int
+
+
 def is_map(map_in: str | typing.Mapping[str, str]) -> typing.Mapping[str, str]:
     """List of comma-separated key=value pairs"""
     if isinstance(map_in, typing.Mapping):

tests/forking_agent/test_executor.py

@@ -86,8 +86,8 @@ def set_custom_executor_policy():
     inmanta.agent.config.agent_executor_cap.set("2")
 
     old_retention_value = inmanta.agent.config.agent_executor_retention_time.get()
-    # Clean up executors after 20s of inactivity
-    inmanta.agent.config.agent_executor_retention_time.set("20")
+    # Clean up executors after 2s of inactivity
+    inmanta.agent.config.agent_executor_retention_time.set("2")
 
     yield
 
@@ -108,7 +108,6 @@ async def test_executor_server(set_custom_executor_policy, mpmanager: MPManager,
     Also test that an executor policy can be set:
         - the agent_executor_cap option correctly stops the oldest executor.
         - the agent_executor_retention_time option is used to clean up old executors.
-        - updating the agent_executor_retention_time option correctly re-schedules the cleanup
     """
 
     with pytest.raises(ImportError):
@@ -215,10 +214,6 @@ def test():
         import lorem  # noqa: F401, F811
 
     async def check_automatic_clean_up() -> bool:
-        # Retention time is set to 20 prior to this
-        # Setting it to 2 tests that a new periodic cleanup is schedule when
-        # the setting is updated
-        inmanta.agent.config.agent_executor_retention_time.set("2")
         return len(manager.agent_map["agent2"]) == 0
 
     with caplog.at_level(logging.DEBUG):

tests/test_config.py

@@ -16,12 +16,14 @@
     Contact: [email protected]
 """
 
+import functools
 import logging
 import os
 import random
 import socket
 
 import netifaces
+import pytest
 from tornado import netutil
 
 import inmanta.agent.config as cfg
@@ -347,3 +349,18 @@ def test_option_is_list_empty():
     option: Option = Option("test", "list", "default,values", "documentation", cfg.is_list)
     option.set("")
     assert option.get() == []
+
+
+def test_option_is_lower_bounded_int():
+    lower_bound = 1
+    option: Option = Option(
+        "test", "lb_int", lower_bound, "documentation", functools.partial(cfg.is_lower_bounded_int, lower_bound)
+    )
+    option.set("2")
+    assert option.get() == 2
+
+    option.set("0")
+    with pytest.raises(ValueError) as exc_info:
+        option.get()
+
+    assert f"Value can not be lower than {lower_bound}" in str(exc_info.value)