Issue/pool reporting

changelogs/unreleased/8228-database_pool_report.yml

@@ -0,0 +1,6 @@
+description: Update database pool status reporting for the new scheduler
+issue-nr: 8228
+change-type: minor
+destination-branches: [master]
+sections:
+  minor-improvement: "{{description}}"

mypy-baseline.txt

@@ -101,7 +101,7 @@ src/inmanta/ast/type.py:0: error: No overload variant of "reversed" matches argu
 src/inmanta/ast/type.py:0: note: Possible overload variants:
 src/inmanta/ast/type.py:0: note: def [_T] __new__(cls, Reversible[_T], /) -> reversed[_T]
 src/inmanta/ast/type.py:0: note: def [_T] __new__(cls, SupportsLenAndGetItem[_T], /) -> reversed[_T]
-src/inmanta/server/services/databaseservice.py:0: error: Incompatible return value type (got "dict[str, object]", expected "dict[str, BaseModel | UUID | bool | int | float | datetime | str | Sequence[BaseModel | UUID | bool | int | float | datetime | str] | Mapping[str, BaseModel | UUID | bool | int | float | datetime | str]]")  [return-value]
+src/inmanta/server/services/metricservice.py:0: error: Incompatible types in assignment (expression has type "float", variable has type "int")  [assignment]
 src/inmanta/protocol/decorators.py:0: error: "FuncT" has no attribute "__protocol_method__"  [attr-defined]
 src/inmanta/protocol/decorators.py:0: error: "FuncT" has no attribute "__protocol_mapping__"  [attr-defined]
 src/inmanta/protocol/decorators.py:0: error: "FuncT" has no attribute "__api_version__"  [attr-defined]
@@ -146,10 +146,6 @@ src/inmanta/execute/runtime.py:0: error: Argument 2 to "assign" of "AssignableNo
 src/inmanta/execute/runtime.py:0: error: Argument 1 to "AttributeException" has incompatible type "Instance"; expected "Locatable"  [arg-type]
 src/inmanta/execute/runtime.py:0: error: Missing type parameters for generic type "ResultVariable"  [type-arg]
 src/inmanta/execute/runtime.py:0: error: Argument 1 to "RuntimeException" has incompatible type "Instance"; expected "Locatable | None"  [arg-type]
-src/inmanta/data/model.py:0: error: Return type "str" of "__getstate__" incompatible with return type "dict[Any, Any]" in supertype "BaseModel"  [override]
-src/inmanta/data/model.py:0: error: Argument 1 of "__setstate__" is incompatible with supertype "BaseModel"; supertype defines the argument type as "dict[Any, Any]"  [override]
-src/inmanta/data/model.py:0: note: This violates the Liskov substitution principle
-src/inmanta/data/model.py:0: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides
 src/inmanta/data/model.py:0: error: Function is missing a type annotation  [no-untyped-def]
 src/inmanta/data/model.py:0: error: Unused "type: ignore" comment, use narrower [prop-decorator] instead of [misc] code  [unused-ignore]
 src/inmanta/ast/attribute.py:0: error: Incompatible types in assignment (expression has type "property", variable has type "Type")  [assignment]
@@ -555,8 +551,11 @@ src/inmanta/data/__init__.py:0: error: Incompatible return value type (got "Reso
 src/inmanta/data/__init__.py:0: error: Argument 1 to "add_logs" of "ResourceAction" has incompatible type "list[dict[str, object]]"; expected "str | None"  [arg-type]
 src/inmanta/data/__init__.py:0: error: "object" has no attribute "__iter__"; maybe "__dir__" or "__str__"? (not iterable)  [attr-defined]
 src/inmanta/data/__init__.py:0: error: Argument "changes" to "ResourceAction" has incompatible type "dict[ResourceVersionIdStr, dict[str, object]] | None"; expected "dict[str, Any] | None"  [arg-type]
+src/inmanta/data/__init__.py:0: error: Function is missing a return type annotation  [no-untyped-def]
 src/inmanta/data/__init__.py:0: error: Missing type parameters for generic type "dict"  [type-arg]
 src/inmanta/data/__init__.py:0: error: Unsupported left operand type for + ("object")  [operator]
+src/inmanta/data/__init__.py:0: error: Function is missing a type annotation  [no-untyped-def]
+src/inmanta/data/__init__.py:0: error: Call to untyped function "convert_or_ignore" in typed context  [no-untyped-call]
 src/inmanta/data/__init__.py:0: error: Incompatible return value type (got "Sequence[Resource]", expected "list[Resource]")  [return-value]
 src/inmanta/data/__init__.py:0: error: Argument 1 to "ResourceIdStr" has incompatible type "object"; expected "str"  [arg-type]
 src/inmanta/data/__init__.py:0: error: Enum index should be a string (actual index type "object")  [misc]
@@ -1102,6 +1101,7 @@ src/inmanta/server/services/orchestrationservice.py:0: error: Incompatible types
 src/inmanta/server/services/orchestrationservice.py:0: error: Incompatible types in assignment (expression has type "set[str | None]", variable has type "set[str]")  [assignment]
 src/inmanta/server/services/orchestrationservice.py:0: error: Argument "agents" to "_trigger_auto_deploy" of "OrchestrationService" has incompatible type "Collection[str]"; expected "Sequence[str] | None"  [arg-type]
 src/inmanta/server/services/orchestrationservice.py:0: error: Incompatible return value type (got "ReturnValueWithMeta[Sequence[DesiredStateVersion]]", expected "ReturnValue[list[DesiredStateVersion]]")  [return-value]
+src/inmanta/server/agentmanager.py:0: error: Argument 1 to "ensure_agent_registered" of "AgentManager" has incompatible type "Environment | None"; expected "Environment"  [arg-type]
 src/inmanta/agent/forking_executor.py:0: error: Argument 1 of "connection_made" is incompatible with supertype "BaseProtocol"; supertype defines the argument type as "BaseTransport"  [override]
 src/inmanta/agent/forking_executor.py:0: note: This violates the Liskov substitution principle
 src/inmanta/agent/forking_executor.py:0: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides

src/inmanta/agent/agent_new.py

@@ -24,15 +24,17 @@
 from concurrent.futures.thread import ThreadPoolExecutor
 from typing import Any, Callable, Coroutine, Optional, Union
 
-from inmanta import config, const, protocol
+import inmanta.server.config as opt
+from inmanta import config, const, data, protocol
 from inmanta.agent import config as cfg
 from inmanta.agent import executor, forking_executor
 from inmanta.agent.reporting import collect_report
 from inmanta.const import AGENT_SCHEDULER_ID
-from inmanta.data.model import AttributeStateChange, ResourceVersionIdStr
+from inmanta.data.model import AttributeStateChange, DataBaseReport, ResourceVersionIdStr
 from inmanta.deploy import scheduler
 from inmanta.deploy.work import TaskPriority
 from inmanta.protocol import SessionEndpoint, methods, methods_v2
+from inmanta.server.services.databaseservice import DatabaseMonitor
 from inmanta.types import Apireturn
 from inmanta.util import (
     CronSchedule,
@@ -84,6 +86,20 @@ def __init__(
 
         self._set_deploy_and_repair_intervals()
 
+    async def start(self) -> None:
+        # Make mypy happy
+        assert data.Resource._connection_pool is not None
+        self._db_monitor = DatabaseMonitor(
+            data.Resource._connection_pool,
+            opt.db_name.get(),
+            opt.db_host.get(),
+            "scheduler",
+            str(self.environment),
+        )
+        self._db_monitor.start()
+
+        await super().start()
+
     def _set_deploy_and_repair_intervals(self) -> None:
         """
         Fetch the settings related to automatic deploys and repairs from the config
@@ -341,6 +357,20 @@ async def get_facts(self, env: uuid.UUID, agent: str, resource: dict[str, Any])
     async def get_status(self) -> Apireturn:
         return 200, collect_report(self)
 
+    @protocol.handle(methods_v2.get_db_status)
+    async def get_db_status(self) -> DataBaseReport:
+        if self._db_monitor is None:
+            return DataBaseReport(
+                connected=False,
+                database="",
+                host="",
+                max_pool=0,
+                open_connections=0,
+                free_connections=0,
+                pool_exhaustion_count=0,
+            )
+        return await self._db_monitor.get_status()
+
     def check_storage(self) -> dict[str, str]:
         """
         Check if the server storage is configured and ready to use. Ultimately, this is

src/inmanta/app.py

@@ -67,6 +67,7 @@
 from inmanta.server import config as opt
 from inmanta.server.bootloader import InmantaBootloader
 from inmanta.server.services.databaseservice import initialize_database_connection_pool
+from inmanta.server.services.metricservice import MetricsService
 from inmanta.signals import safe_shutdown, setup_signal_handlers
 from inmanta.util import get_compiler_version
 from inmanta.warnings import WarningsManager
@@ -156,6 +157,9 @@ async def start() -> None:
             connection_pool_max_size=agent_config.scheduler_db_connection_pool_max_size.get(),
             connection_timeout=agent_config.scheduler_db_connection_timeout.get(),
         )
+        # also report metrics if this is relevant
+        metrics_reporter = MetricsService()
+        metrics_reporter.start_metric_reporters()
         await a.start()
 
     LOGGER.info("Agent with Resource scheduler starting now")

src/inmanta/data/model.py

@@ -24,7 +24,7 @@
 from collections import abc
 from collections.abc import Sequence
 from enum import Enum, StrEnum
-from typing import ClassVar, NewType, Optional, Self, Union
+from typing import ClassVar, Mapping, NewType, Optional, Self, Union
 
 import pydantic
 import pydantic.schema
@@ -90,7 +90,7 @@ class SliceStatus(BaseModel):
     """
 
     name: str
-    status: dict[str, ArgumentTypes]
+    status: Mapping[str, ArgumentTypes | Mapping[str, ArgumentTypes]]
 
 
 class FeatureStatus(BaseModel):
@@ -870,3 +870,30 @@ def has_source(self) -> bool:
 
 
 LEGACY_PIP_DEFAULT = PipConfig(use_system_config=True)
+
+
+class DataBaseReport(BaseModel):
+    connected: bool
+    database: str
+    host: str
+    max_pool: int
+    open_connections: int
+    free_connections: int
+    pool_exhaustion_count: int
+
+    def __add__(self, other: "DataBaseReport") -> "DataBaseReport":
+        if not isinstance(other, DataBaseReport):
+            return NotImplemented
+        if other.database != self.database:
+            return NotImplemented
+        if other.host != self.host:
+            return NotImplemented
+        return DataBaseReport(
+            connected=self.connected and other.connected,
+            database=self.database,
+            host=self.host,
+            max_pool=self.max_pool + other.max_pool,
+            open_connections=self.open_connections + other.open_connections,
+            free_connections=self.free_connections + other.free_connections,
+            pool_exhaustion_count=self.pool_exhaustion_count + other.pool_exhaustion_count,
+        )

src/inmanta/protocol/methods_v2.py

@@ -24,7 +24,7 @@
 
 from inmanta.const import AgentAction, ApiDocsFormat, Change, ClientType, ParameterSource, ResourceState
 from inmanta.data import model
-from inmanta.data.model import LinkedDiscoveredResource, PipConfig, ResourceIdStr
+from inmanta.data.model import DataBaseReport, LinkedDiscoveredResource, PipConfig, ResourceIdStr
 from inmanta.protocol import methods
 from inmanta.protocol.common import ReturnValue
 from inmanta.protocol.decorators import typedmethod
@@ -496,6 +496,15 @@ def update_agent_map(agent_map: dict[str, str]) -> None:
     """
 
 
+@typedmethod(
+    path="/db_status", api=False, server_agent=True, enforce_auth=False, operation="POST", client_types=[], api_version=2
+)
+def get_db_status() -> DataBaseReport:
+    """
+    Get a report of the DB connection pool status
+    """
+
+
 @typedmethod(
     path="/compiledata/<id>",
     operation="GET",

src/inmanta/reporter.py

@@ -80,6 +80,8 @@ class InfluxReporter(AsyncReporter):
     """
     InfluxDB reporter using native http api
     (based on https://influxdb.com/docs/v1.1/guides/writing_data.html)
+
+    If metric name end with `,a=b` these will append as tags `a=b`
     """
 
     def __init__(

src/inmanta/server/__init__.py

@@ -19,7 +19,7 @@
 # flake8: noqa: F401
 
 SLICE_SERVER = "core.server"
-SLICE_AGENT_MANAGER = "core.agentmanager"
+SLICE_AGENT_MANAGER = "core.scheduler_manager"
 SLICE_AUTOSTARTED_AGENT_MANAGER = "core.autostarted_agent_manager"
 SLICE_SESSION_MANAGER = "core.session"
 SLICE_DATABASE = "core.database"

src/inmanta/server/agentmanager.py

@@ -27,6 +27,7 @@
 from collections.abc import Iterable, Iterator, Mapping, Sequence, Set
 from datetime import datetime
 from enum import Enum
+from functools import reduce
 from typing import Any, Optional, Union, cast
 from uuid import UUID
 
@@ -38,9 +39,9 @@
 from inmanta import const, data, tracing
 from inmanta.agent import config as agent_cfg
 from inmanta.config import Config
-from inmanta.const import UNDEPLOYABLE_NAMES, AgentAction, AgentStatus
-from inmanta.data import APILIMIT, InvalidSort, model
-from inmanta.data.model import ResourceIdStr
+from inmanta.const import AGENT_SCHEDULER_ID, UNDEPLOYABLE_NAMES, AgentAction, AgentStatus
+from inmanta.data import APILIMIT, Environment, InvalidSort, model
+from inmanta.data.model import DataBaseReport, ResourceIdStr
 from inmanta.protocol import encode_token, handle, methods, methods_v2
 from inmanta.protocol.common import ReturnValue
 from inmanta.protocol.exceptions import BadRequest, Forbidden, NotFound, ShutdownInProgress
@@ -168,12 +169,67 @@ def __init__(self, closesessionsonstart: bool = True, fact_back_off: Optional[in
 
         self.closesessionsonstart: bool = closesessionsonstart
 
-    async def get_status(self) -> dict[str, ArgumentTypes]:
-        return {
+    async def get_status(self) -> Mapping[str, ArgumentTypes | Mapping[str, ArgumentTypes]]:
+        # The basic report
+
+        out: dict[str, ArgumentTypes | Mapping[str, ArgumentTypes]] = {
             "resource_facts": len(self._fact_resource_block_set),
             "sessions": len(self.sessions),
         }
 
+        # Try to get more info from scheduler, but make sure not to timeout
+        schedulers = self.get_all_schedulers()
+        deadline = 0.9 * Server.GET_SERVER_STATUS_TIMEOUT
+
+        async def get_report(env: uuid.UUID, session: protocol.Session) -> tuple[uuid.UUID, DataBaseReport]:
+            result = await asyncio.wait_for(session.client.get_db_status(), deadline)
+            assert result.code == 200
+            # Mypy can't help here, ....
+            return (env, DataBaseReport(**result.result["data"]))
+
+        # Get env name mapping in parallel with next call
+        env_mapping = asyncio.create_task(
+            asyncio.wait_for(Environment.get_list(details=False, is_marked_for_deletion=False), deadline)
+        )
+        # Get the reports of all schedulers
+        results = await asyncio.gather(*[get_report(env, scheduler) for env, scheduler in schedulers], return_exceptions=True)
+        try:
+            # This can timeout, but not likely
+            env_mapping_result = await env_mapping
+            uuid_to_name = {env.id: env.name for env in env_mapping_result}
+        except TimeoutError:
+            # default to uuid's
+            uuid_to_name = {}
+
+        # Filter out timeouts and other errors
+        good_reports = [x[1] for x in results if not isinstance(x, BaseException)]
+
+        def short_report(report: DataBaseReport) -> Mapping[str, ArgumentTypes]:
+            return {
+                "connected": report.connected,
+                "max_pool": report.max_pool,
+                "open_connections": report.open_connections,
+                "free_connections": report.free_connections,
+                "pool_exhaustion_count": report.pool_exhaustion_count,
+            }
+
+        if good_reports:
+            total: DataBaseReport = reduce(lambda x, y: x + y, good_reports)
+            out["database"] = total.database
+            out["host"] = total.host
+            out["total"] = short_report(total)
+
+        for result in results:
+            if isinstance(result, BaseException):
+                logging.debug("Failed to collect database status for scheduler", exc_info=True)
+            else:
+                the_uuid = result[0]
+                # Default name to uuid
+                name = uuid_to_name.get(the_uuid, str(the_uuid))
+                out[name] = short_report(result[1])
+
+        return out
+
     def get_dependencies(self) -> list[str]:
         return [SLICE_DATABASE, SLICE_SESSION_MANAGER]
 
@@ -185,7 +241,6 @@ async def prestart(self, server: protocol.Server) -> None:
         autostarted_agent_manager = server.get_slice(SLICE_AUTOSTARTED_AGENT_MANAGER)
         assert isinstance(autostarted_agent_manager, AutostartedAgentManager)
         self._autostarted_agent_manager = autostarted_agent_manager
-
         presession = server.get_slice(SLICE_SESSION_MANAGER)
         assert isinstance(presession, SessionManager)
         presession.add_listener(self)
@@ -197,13 +252,27 @@ async def start(self) -> None:
             await self._expire_all_sessions_in_db()
 
         self.add_background_task(self._process_session_listener_actions())
+        # Schedule cleanup agentprocess and agentinstance tables
+        agent_process_purge_interval = opt.agent_process_purge_interval.get()
+        if agent_process_purge_interval > 0:
+            self.schedule(
+                self._purge_agent_processes, interval=agent_process_purge_interval, initial_delay=0, cancel_on_stop=False
+            )
 
     async def prestop(self) -> None:
         await super().prestop()
 
     async def stop(self) -> None:
         await super().stop()
 
+    def get_all_schedulers(self) -> list[tuple[uuid.UUID, protocol.Session]]:
+        # Linear scan, but every item should be a hit
+        return [
+            (env_id, session)
+            for (env_id, agent_id), session in self.tid_endpoint_to_session.items()
+            if agent_id == AGENT_SCHEDULER_ID
+        ]
+
     async def halt_agents(self, env: data.Environment, connection: Optional[asyncpg.connection.Connection] = None) -> None:
         """
         Halts all agents for an environment. Persists prior paused state.
@@ -513,6 +582,10 @@ async def _expire_all_sessions_in_db(self) -> None:
                     await data.AgentInstance.expire_all(now=datetime.now().astimezone(), connection=connection)
                     await data.Agent.mark_all_as_non_primary(connection=connection)
 
+    async def _purge_agent_processes(self) -> None:
+        agent_processes_to_keep = opt.agent_processes_to_keep.get()
+        await data.AgentProcess.cleanup(nr_expired_records_to_keep=agent_processes_to_keep)
+
     # Util
     async def _use_new_active_session_for_agent(self, tid: uuid.UUID, endpoint_name: str) -> Optional[protocol.Session]:
         """
@@ -940,7 +1013,7 @@ def __init__(self) -> None:
         self._agent_procs: dict[UUID, subprocess.Process] = {}  # env uuid -> subprocess.Process
         self.agent_lock = asyncio.Lock()  # Prevent concurrent updates on _agent_procs
 
-    async def get_status(self) -> dict[str, ArgumentTypes]:
+    async def get_status(self) -> Mapping[str, ArgumentTypes]:
         return {"processes": len(self._agent_procs)}
 
     async def prestart(self, server: protocol.Server) -> None:
@@ -1282,12 +1355,21 @@ async def _make_agent_config(
 name={opt.db_name.get()}
 username={opt.db_username.get()}
 password={opt.db_password.get()}
+
 [scheduler]
 db-connection-pool-min-size={agent_cfg.scheduler_db_connection_pool_min_size.get()}
 db-connection-pool-max-size={agent_cfg.scheduler_db_connection_pool_max_size.get()}
 db-connection-timeout={agent_cfg.scheduler_db_connection_timeout.get()}
 
-            """
+[influxdb]
+host = {opt.influxdb_host.get()}
+port = {opt.influxdb_port.get()}
+name = {opt.influxdb_name.get()}
+username = {opt.influxdb_username.get()}
+password = {opt.influxdb_password.get()}
+interval = {opt.influxdb_interval.get()}
+tags = {opt.influxdb_tags.get()}
+"""
         return config
 
     async def _fork_inmanta(