feat: wire host database and extension metrics

This commit is contained in:
jxxghp
2026-08-22 13:22:33 +08:00
parent 996a1a0705
commit 1a71e4809f
4 changed files with 117 additions and 12 deletions
+31 -1
View File
@@ -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
+26 -9
View File
@@ -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