iomgaa/PolyGateway
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 11 Wiki Activity

test: cover the ten call observability columns on real Postgres

Browse Source 
Migrate the PG telemetry fixtures to the 36-field recorder and add the
storage compatibility acceptance the plan calls for.

Mechanical migration:
- _EXPECTED_COLUMNS 27 -> 37 physical columns
- _record_minimal gains the ten keys in the same shape as the unit suite
- _PRE_TENANT_COLUMNS now excludes 14 columns, derived from
  _CALL_OBSERVABILITY_COLUMNS instead of a second hand-written list, and
  the two manual-mode warnings assert a notice derived from COLUMNS order
  so a column that silently drops out of the warning turns the test red

New TestCallObservabilityColumnsAcceptance, all on a 27-column 1.3.4
shaped table built by the existing pg_sandbox factory:
- auto appends the ten columns in the same order as a fresh database and
  old rows keep NULL in every one of them (no backfill, no sentinel)
- manual sends no DDL, trims the INSERT, and still round-trips the other
  26 columns value by value
- an old-version writer using insert_sql with the 1.3.4 column set and a
  new-version writer share one table, and event_kind filtering counts
  neither the old rows as failures nor as successes

_minimal_fields is split out of _record_minimal so the simulated old
process reuses the same values rather than copying them.

Verified against the real lab Postgres: 30 passed.
This commit is contained in:
iomgaa
2026-09-09 12:17:34 -04:00
parent 393f2bf617
commit 7b2f6105f3
1 changed files with 363 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_postgres_telemetry.py
+363 -26
View File
@@ -28,7 +28,7 @@ import pytest
from dotenv import dotenv_values
from polygateway.telemetry.postgres import PostgresRecorder
from polygateway.telemetry.schema import COLUMNS, telemetry_schema_sql
from polygateway.telemetry.schema import COLUMNS, insert_sql, telemetry_schema_sql
_EXPECTED_COLUMNS = [
"call_id",
@@ -58,8 +58,31 @@ _EXPECTED_COLUMNS = [
"meta",
"thinking_observation",
"reasoning_effort",
# —— 1.3.5 逻辑调用统计与结构化失败诊断的十列(issue #19/#23)——
"scope",
"operation",
"logical_call_id",
"event_kind",
"http_status_code",
"error_type",
"cause_type",
"error_body",
"attempts",
"total_latency_ms",
]
# 1.3.5 之前那张表的 27 个物理列(26 个 INSERT 字段 + created_at)。写成固定切片
# 而非 `[:-10]`: 后者会随下一次补列静默漂移到另一张表上,而漂移的表现是
# "旧表补列"用例悄悄改测了别的形态。
_PRE_135_COLUMNS = _EXPECTED_COLUMNS[:27]
# 1.3.4 版本的 recorder 实际写入的列(物理列去掉库从不显式写的 created_at)
_PRE_135_WRITTEN_COLUMNS = [c for c in _PRE_135_COLUMNS if c != "created_at"]
# 1.3.5 新增的十列,按 `COLUMNS`(即 INSERT 字段序)排列: manual 档告警逐字比对与
# "旧行新列为 NULL"两处共用同一份,免得两边各抄一份后各自漂移。
_CALL_OBSERVABILITY_COLUMNS = [c for c in COLUMNS if c not in _PRE_135_WRITTEN_COLUMNS]
def _dsn() -> str | None:
"""读 `.env` 的 DSN 并剥掉 SQLAlchemy 风格的 `+driver` 后缀;未配置返回 None。"""
@@ -92,13 +115,13 @@ async def template_admin_dsn() -> str:
return value
async def _record_minimal(
recorder: PostgresRecorder, call_id: str | None = None, **overrides
) -> dict[str, object]:
"""记一行最小遥测,并**返回实际提交的字段**供调用方逐列比对回读结果。
def _minimal_fields(call_id: str | None = None, **overrides) -> dict[str, object]:
"""一行最小遥测的**完整字段字典**(不写库),供 recorder 写入与旧版本进程模拟共用。
返回值不是顺手加的: 逐列断言若在测试里另抄一份期望值,抄错的那一列会以
"库写错列位"的形态误报,而漏抄的列则悄悄不被验证。
独立出来不是顺手拆的: "新旧进程混写"那条用例要以 1.3.4 的列集直接发 INSERT,
若它另抄一份取值,抄错的那一列会以"库写错列位"的形态误报。
调用方随后逐列比对回读结果,故返回的就是实际提交的那一份。
"""
fields: dict[str, object] = {
"call_id": call_id if call_id is not None else "c1",
@@ -130,8 +153,30 @@ async def _record_minimal(
"thinking_observation": "unknown",
# 同理: `Effort` 归一成裸 str,不表态则是 None(与 'low' 必须分得开)
"reasoning_effort": None,
# —— 1.3.5 十列: 默认形态即"一次普通尝试行"(与单测 `_record_minimal` 同款)——
"scope": "LLM",
"operation": "chat",
# 库内现场构造的请求没有上下文 → NULL,不造 ID
"logical_call_id": None,
"event_kind": "attempt",
# 诊断四列只在失败的 attempt 行上非空;成功行不统一填 200
"http_status_code": None,
"error_type": None,
"cause_type": None,
"error_body": None,
# 逻辑快照两列只属终态行
"attempts": None,
"total_latency_ms": None,
}
fields.update(overrides)
return fields
async def _record_minimal(
recorder: PostgresRecorder, call_id: str | None = None, **overrides
) -> dict[str, object]:
"""记一行最小遥测,并**返回实际提交的字段**供调用方逐列比对回读结果。"""
fields = _minimal_fields(call_id, **overrides)
await recorder.record_llm_call(**fields)
return fields
@@ -167,6 +212,17 @@ async def _fetch(dsn: str, sql: str, *args):
await conn.close()
async def _execute_args(dsn: str, sql: str, *args) -> None:
"""带参数执行单条语句(扩展协议);用于模拟旧版本进程按旧列集发出的 INSERT。"""
import asyncpg
conn = await asyncpg.connect(dsn, timeout=10)
try:
await conn.execute(sql, *args)
finally:
await conn.close()
async def _execute_script(dsn: str, sql: str) -> None:
"""整段执行多语句脚本(不带参数,走简单查询协议)——模拟下游把脚本贴进 psql。"""
import asyncpg
@@ -588,14 +644,24 @@ _PRE_TENANT_INSERT = (
)
# `_PRE_TENANT_DDL` 的物理列(23 个): 由 `_EXPECTED_COLUMNS` 去掉此后新增的四列
# `_PRE_TENANT_DDL` 的物理列(23 个): 由 `_EXPECTED_COLUMNS` 去掉此后新增的列
# 派生而非另抄一份——两份常量必然漂移,而漂移的表现是"manual 档没补列"这条断言假绿。
# 去掉后的顺序与 DDL 逐字一致(这四列在 DDL 里本就排在末尾)。
_PRE_TENANT_COLUMNS = [
c
for c in _EXPECTED_COLUMNS
if c not in ("tenant_id", "meta", "thinking_observation", "reasoning_effort")
]
# 去掉后的顺序与 DDL 逐字一致(这些列在 DDL 里本就排在末尾)。
_PRE_TENANT_ABSENT = (
"tenant_id",
"meta",
"thinking_observation",
"reasoning_effort",
*_CALL_OBSERVABILITY_COLUMNS,
)
_PRE_TENANT_COLUMNS = [c for c in _EXPECTED_COLUMNS if c not in _PRE_TENANT_ABSENT]
# 这张表缺的 14 个维度,按 `COLUMNS`(即告警的排列序)列出: manual 档告警逐字比对用。
# 逐字而非前缀断言,是为了让"将来漏进告警的新列"当场红(设计 §4.2 的纪律)。
_PRE_TENANT_MISSING_NOTICE = (
f"以下维度不会被记录: {', '.join(c for c in COLUMNS if c in _PRE_TENANT_ABSENT)}。"
)
# 回读要逐列比对的字段: 物理列去掉库从不显式写的 created_at,恰好 22 个
_PRE_TENANT_WRITTEN_COLUMNS = [c for c in _PRE_TENANT_COLUMNS if c != "created_at"]
@@ -703,7 +769,7 @@ class TestCallerDimensionsAcceptance:
"WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
schema,
)
# 22 → 26 个 recorder 字段(加 created_at 共 27 个物理列),且新列追加在末尾
# 22 → 36 个 recorder 字段(加 created_at 共 37 个物理列),且新列追加在末尾
assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
rows = await _fetch(
schema_dsn,
@@ -908,11 +974,8 @@ class TestManualSchemaModeAcceptance:
assert [m for m in captured_warnings if "补列失败" in m] == []
notices = [m for m in captured_warnings if "auto_migrate=False" in m]
assert len(notices) == 1 # 准备期一次讲清,不逐行刷屏
# 逐字钉住四个维度: 前缀断言会让将来漏进告警的新列照样绿
assert (
"以下维度不会被记录: tenant_id, meta, thinking_observation, reasoning_effort。"
in notices[0]
)
# 逐字钉住缺的每一个维度: 前缀断言会让将来漏进告警的新列照样绿
assert _PRE_TENANT_MISSING_NOTICE in notices[0]
finally:
await recorder.aclose()
@@ -938,11 +1001,8 @@ class TestManualSchemaModeAcceptance:
assert recorder.telemetry_status.degraded is False
notices = [m for m in captured_warnings if "auto_migrate=False" in m]
assert len(notices) == 1 # 准备期一次,第二行不再重复
# 逐字钉住四个维度: 前缀断言会让将来漏进告警的新列照样绿
assert (
"以下维度不会被记录: tenant_id, meta, thinking_observation, reasoning_effort。"
in notices[0]
)
# 逐字钉住缺的每一个维度: 前缀断言会让将来漏进告警的新列照样绿
assert _PRE_TENANT_MISSING_NOTICE in notices[0]
# 提示里的 SQL 必须可直接粘贴执行,而不是只报个列名
assert (
"ALTER TABLE llm_calls ADD COLUMN tenant_id TEXT NOT NULL DEFAULT '';" in notices[0]
@@ -974,6 +1034,283 @@ class TestManualSchemaModeAcceptance:
await recorder.aclose()
# 1.3.5 之前(1.3.4 发布形态)的表: 26 个 recorder 字段 + created_at = 27 个物理列,
# 没有本版新增的任何一列。裸表名: 由 `pg_sandbox` 在沙箱 schema 的 search_path 下执行。
_PRE_135_DDL = """
CREATE TABLE 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(),
cached_prompt_tokens INTEGER,
model_reported TEXT,
sampling TEXT,
reasoning_tokens INTEGER,
tenant_id TEXT NOT NULL DEFAULT '',
meta JSONB NOT NULL DEFAULT '{}'::jsonb,
thinking_observation TEXT,
reasoning_effort TEXT
)
"""
# 一行 1.3.5 之前写下的历史数据(只列 NOT NULL 列,与当年 recorder 的写入等价)。
# 工厂的 `extra` 逐条裸执行、不接受查询参数,故 call_id 内联成字面量。
_PRE_135_INSERT = (
"INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, "
"prompt_tokens, completion_tokens, usage_source, latency_ms) "
"VALUES ('pre135', 'm', 'p', 's1', '[]', 'old body', 1, 2, 'measured', 10)"
)
# 这张表缺的正是本版十列;manual 档告警要逐字比对的那句。
_PRE_135_MISSING_NOTICE = f"以下维度不会被记录: {', '.join(_CALL_OBSERVABILITY_COLUMNS)}。"
@pytest.fixture
async def pre_135_schema(pg_sandbox) -> tuple[str, str]:
"""一次性沙箱里造一张 **1.3.4 形态的 27 列表**,并留一行本版之前的历史数据。
共享表 `llm_calls` 一个字节都不碰: 本机那张表升级一次就再也回不到旧形态,
指望它还是旧形态的测试第二次跑就会空转(与 `pre_tenant_schema` 同款理由)。
"""
sandbox = await pg_sandbox(ddl=_PRE_135_DDL, extra=(_PRE_135_INSERT,))
return sandbox.dsn, sandbox.schema
class TestCallObservabilityColumnsAcceptance:
"""1.3.5(issue #19/#23)的 PG 存储兼容验收: auto 追加 / manual 裁剪 / 旧行 NULL / 混写。
单元层在 SQLite 上断的是同一族语义,这里断的是**真实 PG 上确实如此**——
两端的 DDL、补列语句与列序是两份文本(`SQLITE_BACKFILL` 与 `_PG_BACKFILL_DECLS`),
只有真表能证明它们没有分叉。
"""
async def test_pre_135_table_gains_the_ten_columns_and_old_rows_stay_null(self, pre_135_schema):
"""27 列旧表 auto 补齐到 37 列,新行两类取值读得回,**历史行十列一律 NULL**。
旧行不回填是本版的明示决策(设计 §5): NULL 表达的是"补列之前根本没记过
这件事",与任何哨兵值都不是一回事。若哪天有人给这十列加了 DEFAULT,历史行
会被就地改写成"记过且值为 X",归因查询从此分不清真实缺口——故这条断言的
方向是 NULL,不是空串也不是 0。
"""
schema_dsn, schema = pre_135_schema
recorder = _recorder(schema_dsn, auto_migrate=True)
try:
# attempt 行: 诊断四列非空、逻辑快照两列 NULL
await _record_minimal(
recorder,
call_id="att",
scope="LLM",
operation="chat",
logical_call_id="lcid-1",
event_kind="attempt",
http_status_code=503,
error_type="TransientError",
cause_type="ReadTimeout",
error_body="upstream said 503",
)
# 终态行: 逻辑快照两列非空、诊断三列 NULL(不搬最后一次 attempt 的现场)
await _record_minimal(
recorder,
call_id="term",
scope="LLM",
operation="chat",
logical_call_id="lcid-1",
event_kind="terminal_failure",
error="AllSourcesExhausted: 全部源已耗尽",
error_type="AllSourcesExhausted",
attempts=3,
total_latency_ms=4200,
)
cols = await _fetch(
schema_dsn,
"SELECT column_name FROM information_schema.columns "
"WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
schema,
)
# 26 → 36 个 recorder 字段(加 created_at 共 37 个物理列),新列追加在末尾:
# 列序与新建库一致才不会让两条升级路径分叉
assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
names = ", ".join(_CALL_OBSERVABILITY_COLUMNS)
rows = await _fetch(
schema_dsn,
f"SELECT call_id, {names} FROM llm_calls "
"WHERE call_id = ANY($1::text[]) ORDER BY call_id",
["att", "pre135", "term"],
)
by_id = {r["call_id"]: r for r in rows}
assert dict(by_id["att"]) == {
"call_id": "att",
"scope": "LLM",
"operation": "chat",
"logical_call_id": "lcid-1",
"event_kind": "attempt",
"http_status_code": 503,
"error_type": "TransientError",
"cause_type": "ReadTimeout",
"error_body": "upstream said 503",
"attempts": None,
"total_latency_ms": None,
}
assert dict(by_id["term"]) == {
"call_id": "term",
"scope": "LLM",
"operation": "chat",
"logical_call_id": "lcid-1",
"event_kind": "terminal_failure",
"http_status_code": None,
"error_type": "AllSourcesExhausted",
"cause_type": None,
"error_body": None,
"attempts": 3,
"total_latency_ms": 4200,
}
# 历史行: 十列逐列 NULL(整体比对,漏掉任一列都红)
assert dict(by_id["pre135"]) == {
"call_id": "pre135",
**dict.fromkeys(_CALL_OBSERVABILITY_COLUMNS),
}
finally:
await recorder.aclose()
async def test_manual_trims_the_insert_on_a_pre_135_table(
self, pre_135_schema, captured_warnings
):
"""27 列旧表 + manual: 一条 DDL 都不发,写入按现有列裁剪后照样落库。
与上一条恰成对照: 同一张表、同一份负载,只有 `auto_migrate` 不同,列数就必须是
27 与 37 之别。裁剪是关掉 ALTER 的前提——不裁剪的话每行 INSERT 都撞缺列
(SQLSTATE 42703)而被整行丢弃,那是把自动补列换成遥测静默全失。
"""
schema_dsn, schema = pre_135_schema
recorder = _recorder(schema_dsn, auto_migrate=False)
try:
recorded = await _record_minimal(
recorder, call_id="man135", scope="LLM", logical_call_id="lcid-x"
)
cols = await _fetch(
schema_dsn,
"SELECT column_name FROM information_schema.columns "
"WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
schema,
)
# 表结构逐字不动: 既没多出十列,也没被顺手改了列序
assert [r["column_name"] for r in cols] == _PRE_135_COLUMNS
names = ", ".join(_PRE_135_WRITTEN_COLUMNS)
rows = await _fetch(
schema_dsn, f"SELECT {names} FROM llm_calls WHERE call_id = $1", "man135"
)
assert len(rows) == 1 # 裁剪后的 INSERT 真写进去了,不是被 PG 拒收
# 其余 26 列逐列与提交值相等: 少写十列最容易引发的错是剩下的值整体错位
assert dict(rows[0]) == {c: recorded[c] for c in _PRE_135_WRITTEN_COLUMNS}
assert [m for m in captured_warnings if "写入失败" in m] == []
assert [m for m in captured_warnings if "补列失败" in m] == []
assert recorder.telemetry_status.degraded is False
notices = [m for m in captured_warnings if "auto_migrate=False" in m]
assert len(notices) == 1 # 准备期一次讲清,不逐行刷屏
# 逐字钉住十个维度: 前缀断言会让将来漏进告警的新列照样绿
assert _PRE_135_MISSING_NOTICE in notices[0]
# 提示里的 SQL 必须可直接粘贴执行(首列与末列各验一条,含类型)
assert "ALTER TABLE llm_calls ADD COLUMN scope TEXT;" in notices[0]
assert "ALTER TABLE llm_calls ADD COLUMN total_latency_ms INTEGER;" in notices[0]
finally:
await recorder.aclose()
async def test_old_and_new_writers_share_one_table(self, fresh_schema, captured_warnings):
"""滚动升级期的混写: 已补列的表上,旧版本进程按 26 列写、新版本按 36 列写。
这是升级窗口里必然出现的形态(先升一个 worker,其余仍是 1.3.4),而它的失败
方式是静默的: 若新列带了 NOT NULL 或旧列集的 INSERT 被拒,旧 worker 的遥测
会整段消失而只留逐行 warning。故这里既断三行都在、也断没有写入失败 warning。
旧进程用 `insert_sql("postgres", 旧列集)` 而不是另抄一条 SQL: 1.3.4 的
recorder 发出的正是同一函数按当年列集拼出的语句,另抄一份只会各自漂移。
"""
fresh_dsn, schema = fresh_schema
recorder = _recorder(fresh_dsn, auto_migrate=True)
try:
# 新版本进程: 建表(37 列)并写一条带完整新列的终态行
await _record_minimal(
recorder,
call_id="new-1",
scope="LLM",
operation="chat",
logical_call_id="lcid-mix",
event_kind="terminal_failure",
error_type="AllSourcesExhausted",
attempts=2,
total_latency_ms=1500,
)
# 旧版本进程: 同一张表,按 1.3.4 的 26 列集写入
legacy_fields = _minimal_fields("old-1", response="from a 1.3.4 worker")
await _execute_args(
fresh_dsn,
insert_sql("postgres", _PRE_135_WRITTEN_COLUMNS),
*(legacy_fields[c] for c in _PRE_135_WRITTEN_COLUMNS),
)
# 新版本进程继续写: 旧进程的写入不得污染后续(列集是每进程各自探测的)
await _record_minimal(
recorder, call_id="new-2", scope="LLM", operation="embed", event_kind="attempt"
)
cols = await _fetch(
fresh_dsn,
"SELECT column_name FROM information_schema.columns "
"WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
schema,
)
assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS # 旧进程不改表结构
rows = await _fetch(
fresh_dsn,
"SELECT call_id, response, scope, operation, event_kind, attempts, "
"total_latency_ms FROM llm_calls ORDER BY call_id",
)
assert [r["call_id"] for r in rows] == ["new-1", "new-2", "old-1"]
by_id = {r["call_id"]: r for r in rows}
assert by_id["old-1"]["response"] == "from a 1.3.4 worker"
# 旧进程写下的行,新列一律 NULL——它没写,也不该被谁替它填
assert (by_id["old-1"]["scope"], by_id["old-1"]["event_kind"]) == (None, None)
assert (by_id["old-1"]["attempts"], by_id["old-1"]["total_latency_ms"]) == (None, None)
assert (by_id["new-1"]["attempts"], by_id["new-1"]["total_latency_ms"]) == (2, 1500)
assert by_id["new-2"]["operation"] == "embed"
assert [m for m in captured_warnings if "写入失败" in m] == []
# 下游可见变化的机械化依据: 新口径"计失败调用"按 event_kind 过滤,
# 混写期旧行(event_kind 为 NULL)既不会被误计成失败,也不会被误计成成功
terminal = await _fetch(
fresh_dsn,
"SELECT count(*) AS n FROM llm_calls WHERE event_kind = 'terminal_failure'",
)
assert terminal[0]["n"] == 1
unclassified = await _fetch(
fresh_dsn, "SELECT count(*) AS n FROM llm_calls WHERE event_kind IS NULL"
)
assert unclassified[0]["n"] == 1
finally:
await recorder.aclose()
_PHYSICAL_COLUMNS_SQL = (
"SELECT column_name FROM information_schema.columns "
"WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position"
@@ -1000,7 +1337,7 @@ class TestPublishedSchemaScript:
await _execute_script(fresh_dsn, script)
actual = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)]
# 物理列 = 26 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份
# 物理列 = 36 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份
assert set(actual) == set(COLUMNS) | {"created_at"}
# 列序也不许漂: 新列必须排在 created_at 之后,否则新建库与 ALTER 升级的列序分叉
assert actual == _EXPECTED_COLUMNS