feat: add postgres telemetry recorder with two-tier degradation
This commit is contained in:
+43
-11
@@ -162,9 +162,13 @@ class GatewayClient:
|
||||
transport_aclose = getattr(self._transport, "aclose", None)
|
||||
if transport_aclose is not None:
|
||||
await transport_aclose()
|
||||
telemetry_close = getattr(self._telemetry, "close", None)
|
||||
if telemetry_close is not None:
|
||||
telemetry_close()
|
||||
telemetry_aclose = getattr(self._telemetry, "aclose", None)
|
||||
if telemetry_aclose is not None:
|
||||
await telemetry_aclose() # Postgres 等异步后端
|
||||
else:
|
||||
telemetry_close = getattr(self._telemetry, "close", None)
|
||||
if telemetry_close is not None:
|
||||
telemetry_close()
|
||||
cache_aclose = getattr(self._cache, "aclose", None)
|
||||
if cache_aclose is not None:
|
||||
await cache_aclose()
|
||||
@@ -196,14 +200,8 @@ class GatewayClient:
|
||||
scope=settings.scope,
|
||||
sources=sources,
|
||||
selector=_build_selector(settings.selector),
|
||||
limiter=limiter
|
||||
or InMemoryLimiter(
|
||||
scope=settings.scope,
|
||||
sources={s.name: s for s in sources},
|
||||
global_limits=settings.global_limits,
|
||||
lease_ttl_s=settings.lease_ttl_s,
|
||||
),
|
||||
breaker=breaker or InMemoryGate(config=settings.breaker),
|
||||
limiter=limiter or _build_limiter(settings, sources),
|
||||
breaker=breaker or _build_breaker(settings),
|
||||
transport=OpenAICompatTransport(registry=registry),
|
||||
retry=settings.retry,
|
||||
backpressure=settings.backpressure,
|
||||
@@ -240,6 +238,35 @@ class GatewayClient:
|
||||
)
|
||||
|
||||
|
||||
def _build_limiter(settings: GatewaySettings, sources: list[SourceConfig]) -> RateLimiter:
|
||||
if settings.limiter_backend == "redis":
|
||||
from polygateway.backends.redis.limiter import RedisLimiter
|
||||
|
||||
assert settings.redis_url is not None # 内部不变量: config 已校验
|
||||
return RedisLimiter.from_url(
|
||||
settings.redis_url,
|
||||
scope=settings.scope,
|
||||
sources={s.name: s for s in sources},
|
||||
global_limits=settings.global_limits,
|
||||
lease_ttl_s=settings.lease_ttl_s,
|
||||
)
|
||||
return InMemoryLimiter(
|
||||
scope=settings.scope,
|
||||
sources={s.name: s for s in sources},
|
||||
global_limits=settings.global_limits,
|
||||
lease_ttl_s=settings.lease_ttl_s,
|
||||
)
|
||||
|
||||
|
||||
def _build_breaker(settings: GatewaySettings) -> ProviderGate:
|
||||
if settings.breaker_backend == "redis":
|
||||
from polygateway.backends.redis.breaker import RedisGate
|
||||
|
||||
assert settings.redis_url is not None # 内部不变量: config 已校验
|
||||
return RedisGate.from_url(settings.redis_url, config=settings.breaker, scope=settings.scope)
|
||||
return InMemoryGate(config=settings.breaker)
|
||||
|
||||
|
||||
def _build_selector(name: str) -> SourceSelector:
|
||||
return RoundRobinSelector() if name == "round_robin" else LeastInflightSelector()
|
||||
|
||||
@@ -258,6 +285,11 @@ def _build_cache(settings: GatewaySettings) -> CacheBackend | None:
|
||||
def _build_telemetry(settings: GatewaySettings) -> TelemetryRecorder | None:
|
||||
if settings.telemetry_backend == "none":
|
||||
return None
|
||||
if settings.telemetry_backend == "postgres":
|
||||
from polygateway.telemetry.postgres import PostgresRecorder
|
||||
|
||||
assert settings.telemetry_pg_dsn is not None # 内部不变量: config 已校验
|
||||
return PostgresRecorder(settings.telemetry_pg_dsn)
|
||||
from polygateway.telemetry.sqlite import SQLiteRecorder
|
||||
|
||||
assert settings.telemetry_sqlite_path is not None # 内部不变量: config 已校验
|
||||
|
||||
@@ -0,0 +1,136 @@
|
||||
"""Postgres 遥测后端(M2 设计 §5): asyncpg lazy 池 + 两级降级。
|
||||
|
||||
参考仓无先例(三项目遥测全 SQLite);asyncpg 工程写法取 GovDoc
|
||||
`taskrun/postgres_store.py`($n 占位、`CREATE TABLE IF NOT EXISTS`、
|
||||
`ON CONFLICT DO NOTHING`),但其"失败冒泡"方向按遥测铁律**有意反转**:
|
||||
① 结构性失败(建池/建表)→ warning 一次后永久降级(池置 None 短路);
|
||||
② 运行时单条写失败 → 逐条 warning 丢弃,不降级不重试(连接抖动由
|
||||
asyncpg 池自恢复;避免浸泡开头一次抖动导致后续全程失遥测)。
|
||||
构造不连库(lazy),18 列 schema 与 SQLite 版同名同序。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from loguru import logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import asyncpg
|
||||
|
||||
_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS llm_calls (
|
||||
call_id TEXT PRIMARY KEY,
|
||||
parent_call_id TEXT,
|
||||
session_id TEXT,
|
||||
model TEXT NOT NULL,
|
||||
provider TEXT NOT NULL,
|
||||
source_name TEXT NOT NULL,
|
||||
messages TEXT NOT NULL,
|
||||
response TEXT NOT NULL,
|
||||
thinking TEXT NOT NULL DEFAULT '',
|
||||
prompt_tokens INTEGER NOT NULL,
|
||||
completion_tokens INTEGER NOT NULL,
|
||||
usage_source TEXT NOT NULL,
|
||||
latency_ms INTEGER NOT NULL,
|
||||
ttft_ms DOUBLE PRECISION,
|
||||
max_inter_token_ms DOUBLE PRECISION,
|
||||
cache_hit BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
error TEXT,
|
||||
cost DOUBLE PRECISION,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
"""
|
||||
|
||||
_COLUMNS = (
|
||||
"call_id",
|
||||
"parent_call_id",
|
||||
"session_id",
|
||||
"model",
|
||||
"provider",
|
||||
"source_name",
|
||||
"messages",
|
||||
"response",
|
||||
"thinking",
|
||||
"prompt_tokens",
|
||||
"completion_tokens",
|
||||
"usage_source",
|
||||
"latency_ms",
|
||||
"ttft_ms",
|
||||
"max_inter_token_ms",
|
||||
"cache_hit",
|
||||
"error",
|
||||
"cost",
|
||||
)
|
||||
|
||||
_INSERT = (
|
||||
f"INSERT INTO llm_calls ({', '.join(_COLUMNS)}) "
|
||||
f"VALUES ({', '.join(f'${i + 1}' for i in range(len(_COLUMNS)))}) "
|
||||
"ON CONFLICT (call_id) DO NOTHING"
|
||||
)
|
||||
|
||||
|
||||
class PostgresRecorder:
|
||||
"""TelemetryRecorder 端口的 Postgres 实现;asyncpg 原生异步,无线程桥接。"""
|
||||
|
||||
def __init__(self, dsn: str, *, pool: asyncpg.Pool | None = None) -> None:
|
||||
try:
|
||||
import asyncpg # noqa: F401 - 仅探测 extra 是否安装
|
||||
except ImportError as exc:
|
||||
raise ImportError(
|
||||
"Postgres 遥测未启用: 安装 pip install 'polygateway[postgres]' 后重试"
|
||||
) from exc
|
||||
self._dsn = dsn
|
||||
self._pool: asyncpg.Pool | None = pool
|
||||
self._external_pool = pool is not None
|
||||
self._schema_ready = False
|
||||
self._failed = False # 结构性降级标志: 置位后所有写入短路
|
||||
self._init_lock = asyncio.Lock()
|
||||
|
||||
async def _ensure_ready(self) -> asyncpg.Pool | None:
|
||||
"""lazy 建池+建表;结构性失败 warning 一次后永久降级(设计 §5 两级之一)。"""
|
||||
if self._failed:
|
||||
return None
|
||||
if self._schema_ready:
|
||||
return self._pool
|
||||
async with self._init_lock:
|
||||
if self._failed or self._schema_ready:
|
||||
return None if self._failed else self._pool
|
||||
try:
|
||||
if self._pool is None:
|
||||
import asyncpg
|
||||
|
||||
self._pool = await asyncpg.create_pool(self._dsn, timeout=10)
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute(_DDL)
|
||||
self._schema_ready = True
|
||||
return self._pool
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
self._failed = True
|
||||
logger.warning("Postgres 遥测初始化失败,后续记录降级为 no-op: {}", exc)
|
||||
return None
|
||||
|
||||
async def record_llm_call(self, **fields: object) -> None:
|
||||
"""写一行遥测;单条失败逐条 warning 丢弃(两级降级之二),绝不冒泡。"""
|
||||
pool = await self._ensure_ready()
|
||||
if pool is None:
|
||||
return
|
||||
row = tuple(fields[col] for col in _COLUMNS)
|
||||
try:
|
||||
async with pool.acquire() as conn:
|
||||
await conn.execute(_INSERT, *row)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
# 遥测铁律: 丢一条 < 拖垮调用;仅记 warning(非 pass),池自恢复
|
||||
logger.warning("Postgres 遥测写入失败(丢弃该行): {}", exc)
|
||||
|
||||
async def aclose(self) -> None:
|
||||
"""幂等关闭自建池;注入的池归注入方管理。"""
|
||||
pool, self._pool = self._pool, None
|
||||
self._schema_ready = False
|
||||
if pool is not None and not self._external_pool:
|
||||
await pool.close()
|
||||
Reference in New Issue
Block a user