diff --git a/app/db/engine.py b/app/db/engine.py index 76dd999e9..c7f2cc373 100644 --- a/app/db/engine.py +++ b/app/db/engine.py @@ -7,13 +7,39 @@ import threading from typing import Dict, Optional, cast -from sqlalchemy import NullPool, QueuePool, create_engine, text +from sqlalchemy import NullPool, QueuePool, create_engine, event, text from sqlalchemy.engine import Engine as SyncEngine from sqlalchemy.ext.asyncio import AsyncEngine as SaAsyncEngine, create_async_engine +from sqlalchemy.pool import Pool from app.runtime.config import settings from app.db.diagnostics import _register_database_error_logging from app.runtime.log import logger +from app.runtime.observability import record_metric + + +def _database_backend_label() -> str: + """把数据库类型收敛为有限的观测标签。""" + return "postgresql" if settings.DB_TYPE.lower() == "postgresql" else "sqlite" + + +def _register_database_pool_metrics(engine: SyncEngine) -> None: + """在 SQLAlchemy 池 checkout/checkin 边界维护当前借出连接数。""" + if not isinstance(engine.pool, Pool): + # 引擎构建单测允许注入不具备 PoolEvents 的轻量替身。 + return + backend = _database_backend_label() + + def record_checkout(*_args: object) -> None: + """连接借出后增加当前使用量。""" + record_metric("db.pool.checked_out", 1, backend=backend) + + def record_checkin(*_args: object) -> None: + """连接归还后减少当前使用量。""" + record_metric("db.pool.checked_out", -1, backend=backend) + + event.listen(engine.pool, "checkout", record_checkout) + event.listen(engine.pool, "checkin", record_checkin) def _async_pool_kwargs(pooled: bool) -> dict: @@ -88,6 +114,7 @@ def _get_sqlite_engine(is_async: bool = False, pooled: bool = False): # 创建数据库引擎 engine = create_engine(**_db_kwargs) _register_database_error_logging(engine) + _register_database_pool_metrics(engine) # 设置WAL模式。 # 这是引擎构建里唯一的阻塞 I/O,且发生在 get_engine() 的创建锁内——异步侧因此 @@ -114,6 +141,7 @@ def _get_sqlite_engine(is_async: bool = False, pooled: bool = False): # 创建异步数据库引擎 async_engine = create_async_engine(**_db_kwargs) _register_database_error_logging(async_engine.sync_engine) + _register_database_pool_metrics(async_engine.sync_engine) # 异步侧不再设置 WAL。journal_mode 是数据库文件级的持久属性,同步引擎已经设置过, # 这里重复设置本就是冗余的;而它原本用 asyncio.run() 完成,是异步引擎构建里唯一的 @@ -159,6 +187,7 @@ def _get_postgresql_engine(is_async: bool = False, pooled: bool = False): # 创建数据库引擎 engine = create_engine(**_db_kwargs) _register_database_error_logging(engine) + _register_database_pool_metrics(engine) print(f"PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}") return engine @@ -177,6 +206,7 @@ def _get_postgresql_engine(is_async: bool = False, pooled: bool = False): # 创建异步数据库引擎 async_engine = create_async_engine(**_db_kwargs) _register_database_error_logging(async_engine.sync_engine) + _register_database_pool_metrics(async_engine.sync_engine) print(f"Async PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}") return async_engine diff --git a/app/db/session.py b/app/db/session.py index 85a82d453..f2791750b 100644 --- a/app/db/session.py +++ b/app/db/session.py @@ -21,9 +21,11 @@ from sqlalchemy.orm import Session, scoped_session, sessionmaker import app.db.engine as engine_module from app.db.engine import (_async_pool_enabled, _get_database_engine, - get_engine, get_global_async_engine) + _database_backend_label, get_engine, + get_global_async_engine) from app.runtime.config import global_vars, settings from app.runtime.log import logger +from app.runtime.observability import record_metric # 会话工厂同样惰性:sessionmaker 在构造时就要绑定引擎,模块级构造等于把引擎的 # 创建时机重新拉回 import 期,惰性化就白做了。 @@ -198,14 +200,29 @@ async def _acquire_fallback_slot(): 变得无界。信号量是线程安全且与事件循环无关的,但不能在协程里阻塞获取, 因此用非阻塞获取 + 异步让出。 """ - deadline = time.monotonic() + settings.DB_POOL_TIMEOUT - while not _fallback_slots.acquire(blocking=False): - if time.monotonic() >= deadline: - raise TimeoutError( - f"异步数据库连接配额已耗尽(上限 {settings.DB_ASYNC_FALLBACK_LIMIT})," - f"等待超过 {settings.DB_POOL_TIMEOUT} 秒" - ) - await asyncio.sleep(0.01) + started_at = time.monotonic() + deadline = started_at + settings.DB_POOL_TIMEOUT + outcome = "success" + try: + while not _fallback_slots.acquire(blocking=False): + if time.monotonic() >= deadline: + outcome = "timeout" + record_metric( + "db.pool.timeout", + backend=_database_backend_label(), + ) + raise TimeoutError( + f"异步数据库连接配额已耗尽(上限 {settings.DB_ASYNC_FALLBACK_LIMIT})," + f"等待超过 {settings.DB_POOL_TIMEOUT} 秒" + ) + await asyncio.sleep(0.01) + finally: + record_metric( + "db.pool.wait", + time.monotonic() - started_at, + backend=_database_backend_label(), + outcome=outcome, + ) @asynccontextmanager diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py index dcf45de14..12a0d0c70 100644 --- a/app/runtime/extensions/module/dispatcher.py +++ b/app/runtime/extensions/module/dispatcher.py @@ -9,7 +9,7 @@ from typing import Any, Protocol, cast from app.foundation.reflection import ObjectUtils from app.runtime.execution import run_in_threadpool from app.runtime.log import logger -from app.runtime.observability import observe_duration +from app.runtime.observability import observe_duration, record_metric from app.runtime.extensions.module.contracts import ( diagnose_module_callable, get_module_method_contract, @@ -143,6 +143,7 @@ class ModuleInvocationDispatcher: **kwargs, ) except Exception as err: + self._record_timeout(method, "plugin", err) self._plugin_error_handler( err, plugin_id, @@ -190,6 +191,7 @@ class ModuleInvocationDispatcher: **kwargs, ) except Exception as err: + self._record_timeout(method, "plugin", err) self._plugin_error_handler( err, plugin_id, @@ -237,6 +239,7 @@ class ModuleInvocationDispatcher: **kwargs, ) except Exception as err: + self._record_timeout(method, "system", err) self._system_error_handler( err, module_id, @@ -284,6 +287,7 @@ class ModuleInvocationDispatcher: **kwargs, ) except Exception as err: + self._record_timeout(method, "system", err) self._system_error_handler( err, module_id, @@ -293,6 +297,16 @@ class ModuleInvocationDispatcher: ) return result + @staticmethod + def _record_timeout(method: str, provider_type: str, error: Exception) -> None: + """仅把真实超时归入低基数模块超时指标。""" + if isinstance(error, TimeoutError): + record_metric( + "module.provider.timeout", + method=method, + provider_type=provider_type, + ) + @staticmethod def _diagnose_callable( method: str, diff --git a/app/runtime/extensions/plugin/lifecycle.py b/app/runtime/extensions/plugin/lifecycle.py index 99bdee618..d8d3025f4 100644 --- a/app/runtime/extensions/plugin/lifecycle.py +++ b/app/runtime/extensions/plugin/lifecycle.py @@ -4,11 +4,51 @@ from __future__ import annotations import traceback from collections.abc import Callable -from typing import Any, Optional +from functools import wraps +import time +from typing import Any, Optional, ParamSpec, TypeVar, cast +from app.runtime.observability import record_metric from app.schemas.plugin import PluginRuntimeStatus +P = ParamSpec("P") +R = TypeVar("R") + + +def observe_plugin_lifecycle(operation: str) -> Callable[[Callable[P, R]], Callable[P, R]]: + """为插件生命周期入口记录不含插件标识的低基数耗时。""" + + def decorator(func: Callable[P, R]) -> Callable[P, R]: + """包装单个同步生命周期方法,并保留原始调用签名。""" + + @wraps(func) + def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: + """执行生命周期方法并把失败状态归一为 error。""" + started_at = time.perf_counter() + outcome = "success" + try: + result = func(*args, **kwargs) + statuses = result.values() if isinstance(result, dict) else (result,) + if PluginRuntimeStatus.LOAD_FAILED in statuses: + outcome = "error" + return result + except BaseException: + outcome = "error" + raise + finally: + record_metric( + "plugin.lifecycle.duration", + time.perf_counter() - started_at, + operation=operation, + outcome=outcome, + ) + + return cast(Callable[P, R], wrapper) + + return decorator + + class PluginLifecycle: """管理插件发现、初始化、启停和热重载,不持有市场或 HTTP 路由职责。""" @@ -44,6 +84,7 @@ class PluginLifecycle: self._logger = log self._event_sender = event_sender + @observe_plugin_lifecycle("start") def start( self, plugin_id: Optional[str] = None, @@ -100,6 +141,7 @@ class PluginLifecycle: self._clear_tools() return results + @observe_plugin_lifecycle("initialize") def initialize(self, plugin_id: str, config: dict) -> None: """重新应用指定插件配置并刷新事件注册状态。""" plugin = self._running.get(plugin_id) @@ -112,6 +154,7 @@ class PluginLifecycle: self._disable_events(type(plugin)) self._clear_tools() + @observe_plugin_lifecycle("stop") def stop(self, plugin_id: Optional[str] = None) -> None: """停止指定插件或全部插件,并清理模块缓存。""" if plugin_id: @@ -139,6 +182,7 @@ class PluginLifecycle: self._clear_tools() self._logger.info("插件停止完成") + @observe_plugin_lifecycle("reload") def reload( self, plugin_id: str,