Max Ronzhin
c4485d43b1
_refresh_pipeline_sessions_gauge и metrics_endpoint ходили через Depends(get_session) — основной пул, разделяемый с API-запросами. В инциденте 07.08 это дало 16 падений в api/metrics.py ровно тогда, когда метрики были нужнее всего (пул исчерпан). db_up/db_pool_* уже были развязаны в 0.0.32, эта метрика — нет (мешал тестовый харнесс). Добавлен get_metrics_session (core/db.py) — отдельный движок с NullPool, как у check_db_up, но с полноценной ORM-сессией для репозитория. Тестовый харнесс (app-фикстура) подменяет её на ту же savepoint-сессию, что и get_session, — иначе /metrics не видел бы данные теста.
111 lines
6.1 KiB
Python
111 lines
6.1 KiB
Python
"""Настройка асинхронного движка SQLAlchemy и сеанса."""
|
||
|
||
from collections.abc import AsyncGenerator
|
||
from typing import cast
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import (
|
||
AsyncEngine,
|
||
AsyncSession,
|
||
async_sessionmaker,
|
||
create_async_engine,
|
||
)
|
||
from sqlalchemy.pool import NullPool, QueuePool
|
||
|
||
from core.config import get_settings
|
||
|
||
settings = get_settings()
|
||
|
||
engine: AsyncEngine = create_async_engine(
|
||
settings.database_url,
|
||
pool_pre_ping=True,
|
||
# Параметры пула — в настройках (`core/config.py`, там же расчёт бюджета
|
||
# соединений на инстанс). Дефолт SQLAlchemy 5 + 10 под нагрузкой
|
||
# выгребался за секунды.
|
||
pool_size=settings.db_pool_size,
|
||
max_overflow=settings.db_max_overflow,
|
||
pool_timeout=settings.db_pool_timeout,
|
||
)
|
||
|
||
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
|
||
|
||
|
||
async def get_session() -> AsyncGenerator[AsyncSession, None]:
|
||
"""Зависимость FastAPI, возвращающая `AsyncSession`."""
|
||
async with async_session_maker() as session:
|
||
yield session
|
||
|
||
|
||
# --- Проверка доступности БД вне основного пула (сессия 33) ----------------
|
||
#
|
||
# Отдельный движок с `NullPool`: каждый вызов открывает новое соединение и
|
||
# закрывает его сразу после — бюджет соединений не пересекается с
|
||
# `engine.pool` (10 + 10 overflow × число воркеров uvicorn). Это единственный
|
||
# способ отличить «БД лежит» от «основной пул занят под нагрузкой»: проверка
|
||
# через `get_session()` в момент исчерпания пула сама встала бы в очередь на
|
||
# `db_pool_timeout` и не смогла бы ответить, пока не появится случайно
|
||
# освободившееся место — то есть не отличила бы два принципиально разных
|
||
# состояния. Короткий `timeout` на соединение (не путать с `db_pool_timeout`
|
||
# основного пула) — чтобы зависший, а не оборванный TCP (Postgres отвечает,
|
||
# но не может продвинуться) не держал проверку до дефолтных 60 секунд asyncpg.
|
||
_probe_engine: AsyncEngine = create_async_engine(
|
||
settings.database_url,
|
||
poolclass=NullPool,
|
||
connect_args={"timeout": settings.db_probe_timeout_s},
|
||
)
|
||
|
||
|
||
async def check_db_up() -> bool:
|
||
"""`True`, если БД отвечает на `SELECT 1` по отдельному от основного пула соединению."""
|
||
try:
|
||
async with _probe_engine.connect() as connection:
|
||
await connection.execute(text("SELECT 1"))
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
return True
|
||
|
||
|
||
# --- Сессия для метрик, читающих данные (не только «жив/мёртв»), вне основного
|
||
# пула (сессия 37, доделка session 32/33 — см. докстринг `api/metrics.py`) ----
|
||
#
|
||
# `check_db_up` выше обходится голым соединением ("SELECT 1"), но
|
||
# `vidconf_pipeline_sessions` нужна полноценная ORM-сессия (репозиторий,
|
||
# группировка по статусу) — `NullPool`, как и у `_probe_engine`: каждый вызов
|
||
# открывает новое соединение и сразу закрывает его, бюджет основного пула
|
||
# (`engine.pool`) не расходуется. `async_sessionmaker` — тот же паттерн, что
|
||
# `async_session_maker` выше, просто на другом движке.
|
||
_metrics_probe_engine: AsyncEngine = create_async_engine(
|
||
settings.database_url,
|
||
poolclass=NullPool,
|
||
connect_args={"timeout": settings.db_probe_timeout_s},
|
||
)
|
||
_metrics_session_maker = async_sessionmaker(_metrics_probe_engine, expire_on_commit=False)
|
||
|
||
|
||
async def get_metrics_session() -> AsyncGenerator[AsyncSession, None]:
|
||
"""Зависимость FastAPI для метрик, которым нужна БД, но не основной пул.
|
||
|
||
В отличие от `get_session()` (основной пул `engine.pool`, конкурирует за
|
||
те же 10+10×воркеров соединений, что и API-запросы), сессия здесь открыта
|
||
на `_metrics_probe_engine` — исчерпание основного пула эту метрику не
|
||
заденет, как и `vidconf_db_up`/`vidconf_db_pool_*`. Тестовый харнесс
|
||
(`tests/conftest.py`) подменяет и её на savepoint-сессию теста — так же,
|
||
как `get_session` — иначе тест `vidconf_pipeline_sessions` не видел бы
|
||
данные, ещё не закоммиченные за пределы savepoint.
|
||
"""
|
||
async with _metrics_session_maker() as session:
|
||
yield session
|
||
|
||
|
||
def db_pool_checked_out() -> int:
|
||
"""Число соединений основного пула, занятых прямо сейчас — без обращения к БД.
|
||
|
||
SQLAlchemy держит счётчик в памяти самого объекта пула (`engine.pool`),
|
||
поэтому его можно прочитать в любой момент, даже когда все соединения
|
||
заняты или БД недоступна — именно это нужно алерту на исчерпание пула
|
||
(метрика не должна зависеть от того, что измеряет). Размер и лимит
|
||
overflow — конфигурация (`Settings.db_pool_size`/`db_max_overflow`),
|
||
их не нужно снимать с объекта пула отдельно.
|
||
"""
|
||
return cast(QueuePool, engine.pool).checkedout()
|