Issue/7694 executor cleanup

changelogs/unreleased/7694-executor-cleanup.yml

@@ -0,0 +1,4 @@
+description: Add policy to cleanup old executors
+issue-nr: 7694
+change-type: patch
+destination-branches: [master]

src/inmanta/agent/agent.py

@@ -639,15 +639,26 @@ def periodic_schedule(
             action: Callable[[], Coroutine[object, None, object]],
             interval: Union[int, str],
             splay_value: int,
-            initial_time: datetime.datetime,
         ) -> bool:
+            """
+            Schedule a periodic task
+
+            :param kind: Name of the task (value to display in logs)
+            :param action: The action to schedule periodically
+            :param interval: The interval at which to schedule the task. Can be specified as either a number of
+                seconds, or a cron string.
+            :param splay_value: When specifying the interval as a number of seconds, this parameter specifies
+                the number of seconds by which to delay the initial execution of this action.
+            """
+            now = datetime.datetime.now().astimezone()
+
             if isinstance(interval, int) and interval > 0:
                 self.logger.info(
                     "Scheduling periodic %s with interval %d and splay %d (first run at %s)",
                     kind,
                     interval,
                     splay_value,
-                    (initial_time + datetime.timedelta(seconds=splay_value)).strftime(const.TIME_LOGFMT),
+                    (now + datetime.timedelta(seconds=splay_value)).strftime(const.TIME_LOGFMT),
                 )
                 interval_schedule: IntervalSchedule = IntervalSchedule(
                     interval=float(interval), initial_delay=float(splay_value)
@@ -674,8 +685,9 @@ def periodic_schedule(
                 )
             )
             self.ensure_deploy_on_start = False
-        periodic_schedule("deploy", deploy_action, self._deploy_interval, self._deploy_splay_value, now)
-        periodic_schedule("repair", repair_action, self._repair_interval, self._repair_splay_value, now)
+
+        periodic_schedule("deploy", deploy_action, self._deploy_interval, self._deploy_splay_value)
+        periodic_schedule("repair", repair_action, self._repair_interval, self._repair_splay_value)
 
     def _enable_time_trigger(self, action: TaskMethod, schedule: TaskSchedule) -> None:
         self.process._sched.add_action(action, schedule)

src/inmanta/agent/config.py

@@ -149,6 +149,23 @@
     is_time,
 )
 
+executor_cap_per_agent = Option[int](
+    "config",
+    "executor-cap-per-agent",
+    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,
+)
+
+executor_retention = Option[int](
+    "config",
+    "executor-retention",
+    60,
+    "Amount of time (in seconds) to wait before cleaning up inactive executors.",
+    is_time,
+)
+
 agent_get_resource_backoff: Option[float] = Option(
     "config",
     "agent-get-resource-backoff",

src/inmanta/agent/executor.py

@@ -471,11 +471,14 @@ class ExecutorManager(abc.ABC, typing.Generic[E]):
     @abc.abstractmethod
     async def get_executor(self, agent_name: str, agent_uri: str, code: typing.Collection[ResourceInstallSpec]) -> E:
         """
-        Retrieves an Executor based on the agent name and blueprint.
+        Retrieves an Executor for a given agent with the relevant handler code loaded in its venv.
         If an Executor does not exist for the given configuration, a new one is created.
 
         :param agent_name: The name of the agent for which an Executor is being retrieved or created.
-        :param blueprint: The ExecutorBlueprint defining the configuration for the Executor.
+        :param agent_uri: The name of the host on which the agent is running.
+        :param code: Collection of ResourceInstallSpec defining the configuration for the Executor i.e.
+            which resource types it can act on and all necessary information to install the relevant
+            handler code in its venv.
         :return: An Executor instance
         """
         pass

src/inmanta/agent/forking_executor.py

@@ -20,6 +20,7 @@
 import collections
 import concurrent.futures
 import concurrent.futures.thread
+import datetime
 import functools
 import logging
 import logging.config
@@ -29,6 +30,7 @@
 import typing
 import uuid
 from asyncio import transports
+from datetime import timedelta
 
 import inmanta.agent.cache
 import inmanta.agent.executor
@@ -42,9 +44,11 @@
 import inmanta.protocol.ipc_light
 import inmanta.signals
 import inmanta.util
+from inmanta import util
 from inmanta.agent import executor
 from inmanta.data.model import ResourceType
 from inmanta.protocol.ipc_light import FinalizingIPCClient, IPCServer, LogReceiver, LogShipper
+from inmanta.util import IntervalSchedule
 from setproctitle import setproctitle
 
 LOGGER = logging.getLogger(__name__)
@@ -516,6 +520,13 @@ def __init__(
 
         self._locks: inmanta.util.NamedLock = inmanta.util.NamedLock()
 
+        self.executor_retention_time = inmanta.agent.config.executor_retention.get()
+        self.max_executors_per_agent = inmanta.agent.config.executor_cap_per_agent.get()
+
+        self._sched = util.Scheduler("MP manager")
+
+        self._sched.add_action(self.cleanup_inactive_executors, IntervalSchedule(2))
+
     def __add_executor(self, theid: executor.ExecutorId, the_executor: MPExecutor) -> None:
         self.executor_map[theid] = the_executor
         self.agent_map[theid.agent_name].add(theid)
@@ -548,11 +559,14 @@ async def get_executor(
         self, agent_name: str, agent_uri: str, code: typing.Collection[executor.ResourceInstallSpec]
     ) -> MPExecutor:
         """
-        Retrieves an Executor based on the agent name and ResourceInstallSpec.
+        Retrieves an Executor for a given agent with the relevant handler code loaded in its venv.
         If an Executor does not exist for the given configuration, a new one is created.
 
         :param agent_name: The name of the agent for which an Executor is being retrieved or created.
-        :param code: The set of sources to be installed on the executor.
+        :param agent_uri: The name of the host on which the agent is running.
+        :param code: Collection of ResourceInstallSpec defining the configuration for the Executor i.e.
+            which resource types it can act on and all necessary information to install the relevant
+            handler code in its venv.
         :return: An Executor instance
         """
         blueprint = executor.ExecutorBlueprint.from_specs(code)
@@ -562,8 +576,7 @@ async def get_executor(
             if not it.closing:
                 LOGGER.debug("Found existing executor for agent %s with id %s", agent_name, executor_id.identity())
                 return it
-        # Acquire a lock based on the blueprint's hash
-        # We don't care about URI here
+        # Acquire a lock based on the executor's identity (agent name, agent uri and blueprint hash)
         async with self._locks.get(executor_id.identity()):
             if executor_id in self.executor_map:
                 it = self.executor_map[executor_id]
@@ -575,6 +588,14 @@ async def get_executor(
                         "Found stale executor for agent %s with id %s, waiting for close", agent_name, executor_id.identity()
                     )
                     await it.join(2.0)
+            n_executors_for_agent = len(self.agent_map[executor_id.agent_name])
+            if n_executors_for_agent >= self.max_executors_per_agent:
+                # Close oldest executor:
+                executor_ids = self.agent_map[executor_id.agent_name]
+                oldest_executor = min([self.executor_map[id] for id in executor_ids], key=lambda e: e.connection.last_used_at)
+
+                await oldest_executor.stop()
+
             my_executor = await self.create_executor(executor_id)
             self.__add_executor(executor_id, my_executor)
             if my_executor.failed_resource_sources:
@@ -628,7 +649,7 @@ async def create_executor(self, executor_id: executor.ExecutorId) -> MPExecutor:
     async def make_child_and_connect(
         self, executor_id: executor.ExecutorId, venv: executor.ExecutorVirtualEnvironment
     ) -> MPExecutor:
-        """Async code to make a child process as share a socker with it"""
+        """Async code to make a child process and share a socket with it"""
         loop = asyncio.get_running_loop()
         name = executor_id.agent_name
 
@@ -657,13 +678,15 @@ def _child_closed(self, child_handle: MPExecutor) -> None:
     def _make_child(self, name: str, log_level: int, cli_log: bool) -> tuple[multiprocessing.Process, socket.socket]:
         """Sync code to make a child process and share a socket with it"""
         parent_conn, child_conn = socket.socketpair()
+        # Fork an ExecutorServer
         p = multiprocessing.Process(
             target=mp_worker_entrypoint,
             args=(child_conn, name, log_level, cli_log, inmanta.config.Config.config_as_dict()),
             name=f"agent.executor.{name}",
         )
         p.start()
         child_conn.close()
+
         return p, parent_conn
 
     async def stop(self) -> None:
@@ -681,3 +704,10 @@ async def stop_for_agent(self, agent_name: str) -> list[MPExecutor]:
         children = [self.executor_map[child_id] for child_id in children_ids]
         await asyncio.gather(*(child.stop() for child in children))
         return children
+
+    async def cleanup_inactive_executors(self) -> None:
+        now = datetime.datetime.now().astimezone()
+
+        for _executor in self.executor_map.values():
+            if now - _executor.connection.last_used_at > timedelta(seconds=self.executor_retention_time):
+                await _executor.stop()

src/inmanta/agent/in_process_executor.py

@@ -474,10 +474,14 @@ async def get_executor(
         self, agent_name: str, agent_uri: str, code: typing.Collection[executor.ResourceInstallSpec]
     ) -> InProcessExecutor:
         """
-        Creates an Executor based with the specified agent name and blueprint.
-        It ensures the required virtual environment is prepared and source code is loaded.
-
-        :param executor_id: executor identifier containing an agent name and a blueprint configuration.
+        Retrieves an Executor for a given agent with the relevant handler code loaded in its venv.
+        If an Executor does not exist for the given configuration, a new one is created.
+
+        :param agent_name: The name of the agent for which an Executor is being retrieved or created.
+        :param agent_uri: The name of the host on which the agent is running.
+        :param code: Collection of ResourceInstallSpec defining the configuration for the Executor i.e.
+            which resource types it can act on and all necessary information to install the relevant
+            handler code in its venv.
         :return: An Executor instance
         """
         if agent_name in self.executors:

src/inmanta/data/__init__.py

@@ -2363,6 +2363,8 @@ def validate_cron(value: str, allow_empty: bool = True) -> str:
 AVAILABLE_VERSIONS_TO_KEEP = "available_versions_to_keep"
 RECOMPILE_BACKOFF = "recompile_backoff"
 ENVIRONMENT_METRICS_RETENTION = "environment_metrics_retention"
+EXECUTOR_CAP_PER_AGENT = "executor_cap_per_agent"
+EXECUTOR_RETENTION = "executor_retention"
 
 
 class Setting:
@@ -2646,6 +2648,26 @@ def to_dto(self) -> m.Environment:
             "Default=336 hours (2 weeks). Set to 0 to disable automatic cleanups.",
             validator=convert_int,
         ),
+        EXECUTOR_CAP_PER_AGENT: Setting(
+            name=EXECUTOR_CAP_PER_AGENT,
+            typ="int",
+            default=3,
+            doc="Set the upper bound on the number of concurrent executors per agent for this environment. "
+            "If this limit is already reached for this agent when attempting to create a new executor, the "
+            "oldest executor of this agent will be stopped. This setting can be used in combination with "
+            "the :inmanta.environment-settings:setting:`executor_retention` setting to define a policy to "
+            "manage executors.",
+            validator=convert_int,
+        ),
+        EXECUTOR_RETENTION: Setting(
+            name=EXECUTOR_RETENTION,
+            typ="int",
+            default=60,
+            doc="The number of seconds to wait before cleaning up inactive executors. This setting "
+            "can be used in combination with the :inmanta.environment-settings:setting:`executor_cap_per_agent` "
+            "setting to define a policy to manage executors.",
+            validator=convert_int,
+        ),
     }
 
     @classmethod

src/inmanta/protocol/ipc_light.py

@@ -18,6 +18,7 @@
 
 import abc
 import asyncio
+import datetime
 import functools
 import logging
 import pickle
@@ -226,8 +227,12 @@ class IPCClient(IPCFrameProtocol, typing.Generic[ServerContext]):
     def __init__(self, name: str):
         super().__init__(name)
         # TODO timeouts
-        self.requests: dict[uuid.UUID, Future[object]] = {}
+
         # All outstanding calls
+        self.requests: dict[uuid.UUID, Future[object]] = {}
+
+        # Keeps track of when this client was active last
+        self.last_used_at = datetime.datetime.now().astimezone()
 
     @typing.overload
     def call(
@@ -239,6 +244,8 @@ def call(self, method: IPCMethod[ServerContext, ReturnType], has_reply: typing.L
 
     def call(self, method: IPCMethod[ServerContext, ReturnType], has_reply: bool = True) -> Future[ReturnType] | None:
         """Call a method with given arguments"""
+        self.last_used_at = datetime.datetime.now().astimezone()
+
         request = IPCRequestFrame(
             id=uuid.uuid4() if has_reply else None,
             method=method,

src/inmanta/server/agentmanager.py

@@ -615,6 +615,7 @@ def _get_session_to_failover_agent(self, tid: uuid.UUID, endpoint: str) -> Optio
         current_active_session = self.tid_endpoint_to_session[(tid, endpoint)]
         for session in self.sessions.values():
             if endpoint in session.endpoint_names and session.tid == tid:
+                # not current_active_session below redundant ??
                 if not current_active_session or session.id != current_active_session.id:
                     return session
         return None
@@ -1213,6 +1214,9 @@ async def _make_agent_config(
         agent_repair_splay: int = cast(int, await env.get(data.AUTOSTART_AGENT_REPAIR_SPLAY_TIME, connection=connection))
         agent_repair_interval: str = cast(str, await env.get(data.AUTOSTART_AGENT_REPAIR_INTERVAL, connection=connection))
 
+        executor_cap_per_agent: int = cast(int, await env.get(data.EXECUTOR_CAP_PER_AGENT, connection=connection))
+        executor_retention: int = cast(int, await env.get(data.EXECUTOR_RETENTION, connection=connection))
+
         # generate config file
         config = f"""[config]
 state-dir=%(statedir)s
@@ -1225,6 +1229,8 @@ async def _make_agent_config(
 agent-deploy-interval=%(agent_deploy_interval)s
 agent-repair-splay-time=%(agent_repair_splay)d
 agent-repair-interval=%(agent_repair_interval)s
+executor-cap-per-agent=%(executor_cap_per_agent)d
+executor-retention=%(executor_retention)d
 
 agent-get-resource-backoff=%(agent_get_resource_backoff)f
 
@@ -1244,6 +1250,8 @@ async def _make_agent_config(
             "agent_repair_interval": agent_repair_interval,
             "serveradress": server_config.server_address.get(),
             "agent_get_resource_backoff": agent_cfg.agent_get_resource_backoff.get(),
+            "executor_cap_per_agent": executor_cap_per_agent,
+            "executor_retention": executor_retention,
         }
 
         if server_config.server_enable_auth.get():