refactor: make the telemetry schema a single source of truth
DDL, column order and backfill statements lived twice, once in each recorder. A public telemetry_schema_sql() would have made three copies, and the drift shows up downstream as "I ran the printed SQL and the library still reports a missing column". Move both DDLs, both backfill lists and the 24 INSERT fields into telemetry/schema.py verbatim; the recorders now import them and build _INSERT through insert_sql(backend, COLUMNS) at import time. The generated statements are byte-identical to the previous constants, so runtime behaviour is unchanged (the postgres conflict target stays bound to call_id for now). insert_sql() validates its columns against COLUMNS: from the next task on those names come from database probing, not from a constant, so the subset check is the gate on the only injection surface. The new telemetry_schema_sql() prints a paste-ready migration script; its postgres backfill deliberately uses ADD COLUMN IF NOT EXISTS while the library's own statements do not, because that form takes an ACCESS EXCLUSIVE lock even when the column exists. Both variants are derived from one declaration list so their column sets cannot drift.
This commit is contained in:
+126
-22
@@ -115,28 +115,133 @@ async def _record_minimal(recorder, call_id="c1", **overrides):
|
||||
await recorder.record_llm_call(**fields)
|
||||
|
||||
|
||||
class TestBackendColumnParity:
|
||||
"""两个后端的 `_COLUMNS` 必须逐字同名同序(issue #11)。
|
||||
# 搬迁前(1.2.1)两个 recorder 各自持有的 INSERT 常量原文,逐字冻结在此。
|
||||
# 这两条字符串是"纯搬迁不改行为"的机械证据: 构造逻辑换了地方,产物必须一字不差。
|
||||
_FROZEN_SQLITE_INSERT = (
|
||||
"INSERT OR IGNORE INTO llm_calls (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, cached_prompt_tokens, "
|
||||
"model_reported, sampling, reasoning_tokens, tenant_id, meta) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
|
||||
)
|
||||
_FROZEN_PG_INSERT = (
|
||||
"INSERT INTO llm_calls (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, cached_prompt_tokens, model_reported, "
|
||||
"sampling, reasoning_tokens, tenant_id, meta) "
|
||||
"VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, "
|
||||
"$19, $20, $21, $22, $23, $24) "
|
||||
"ON CONFLICT (call_id) DO NOTHING"
|
||||
)
|
||||
|
||||
emitter 只组装一份 `fields`,两个后端各自按自己的 `_COLUMNS` 取值;两份清单
|
||||
一旦分叉,同一次调用在 SQLite 上写得进、在 PG 上抛 KeyError 被降级吞掉,
|
||||
差异只在换后端时才暴露。列**序**同样断言: INSERT 用位置占位符,顺序错位
|
||||
会把值写进错误的列而不报错。
|
||||
|
||||
def _first_occurrence_order(text: str, names: list[str]) -> list[str]:
|
||||
"""按各列名在 text 中首次出现的位置排序,用于比对"列名出现顺序"。"""
|
||||
found = [(text.index(name), name) for name in names if name in text]
|
||||
return [name for _, name in sorted(found)]
|
||||
|
||||
|
||||
class TestSchemaModule:
|
||||
"""`telemetry/schema.py` 是 schema 单一事实源(issue #13 Task 1)。
|
||||
|
||||
库执行的 DDL 与打印给下游的 SQL 必须同源: 常量分散在两个 recorder 里各存一份时,
|
||||
公共函数再写一份就是三份,漂移的表现是"下游照打印的 SQL 建完表,库仍报缺列"。
|
||||
"""
|
||||
|
||||
def test_columns_and_ddl_are_frozen(self):
|
||||
"""列序与两端 DDL 逐字未变(搬迁不得改动任何一个字符)。"""
|
||||
from polygateway.telemetry.schema import COLUMNS, PG_BACKFILL, PG_DDL, SQLITE_DDL
|
||||
|
||||
# COLUMNS 是 INSERT 字段序,不含数据库自填的 created_at
|
||||
assert list(COLUMNS) == [c for c in _EXPECTED_COLUMNS if c != "created_at"]
|
||||
assert len(COLUMNS) == 24
|
||||
# 两端 DDL 的列出现顺序 == 物理列序(created_at 在第 19 位)
|
||||
for ddl in (SQLITE_DDL, PG_DDL):
|
||||
assert _first_occurrence_order(ddl, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
|
||||
assert "CREATE TABLE IF NOT EXISTS llm_calls" in SQLITE_DDL
|
||||
assert "created_at TEXT NOT NULL DEFAULT (datetime('now'))" in SQLITE_DDL
|
||||
assert "created_at TIMESTAMPTZ NOT NULL DEFAULT now()" in PG_DDL
|
||||
assert "meta JSONB NOT NULL DEFAULT '{}'::jsonb" in PG_DDL
|
||||
# 库内执行的补列语句不带 IF NOT EXISTS(它即便列已存在也先取 ACCESS EXCLUSIVE 锁)
|
||||
assert PG_BACKFILL[0] == (
|
||||
"cached_prompt_tokens",
|
||||
"ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER",
|
||||
)
|
||||
assert PG_BACKFILL[-1] == (
|
||||
"meta",
|
||||
"ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb",
|
||||
)
|
||||
assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL)
|
||||
|
||||
def test_insert_sql_reproduces_the_frozen_statements(self):
|
||||
"""`insert_sql(backend, COLUMNS)` 必须与搬迁前的 `_INSERT` 逐字节相同。"""
|
||||
from polygateway.telemetry.schema import COLUMNS, insert_sql
|
||||
|
||||
assert insert_sql("sqlite", COLUMNS) == _FROZEN_SQLITE_INSERT
|
||||
assert insert_sql("postgres", COLUMNS) == _FROZEN_PG_INSERT
|
||||
# 裁剪列表按位置占位符重新编号,不留空洞
|
||||
assert insert_sql("postgres", ["call_id", "model"]) == (
|
||||
"INSERT INTO llm_calls (call_id, model) VALUES ($1, $2) "
|
||||
"ON CONFLICT (call_id) DO NOTHING"
|
||||
)
|
||||
|
||||
def test_insert_sql_rejects_foreign_columns_and_backends(self):
|
||||
"""列名来自数据库探测结果而非常量,子集校验是唯一的注入面闸门。"""
|
||||
from polygateway.telemetry.schema import COLUMNS, insert_sql
|
||||
|
||||
with pytest.raises(ValueError, match="call_id_x"):
|
||||
insert_sql("sqlite", ["call_id_x"])
|
||||
with pytest.raises(ValueError):
|
||||
insert_sql("sqlite", ["call_id", "meta); DROP TABLE llm_calls; --"])
|
||||
with pytest.raises(ValueError, match="mysql"):
|
||||
insert_sql("mysql", COLUMNS)
|
||||
|
||||
def test_schema_sql_is_paste_ready_and_same_source(self):
|
||||
"""打印给下游的脚本与库执行的 DDL 同源,且对人可重复执行。"""
|
||||
from polygateway.telemetry.schema import PG_BACKFILL, SQLITE_BACKFILL, telemetry_schema_sql
|
||||
|
||||
pg = telemetry_schema_sql("postgres")
|
||||
lite = telemetry_schema_sql("sqlite")
|
||||
for script in (pg, lite):
|
||||
# 24 个 INSERT 字段 + created_at 全在,且首次出现顺序与建表 DDL 一致
|
||||
assert _first_occurrence_order(script, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
|
||||
assert "CREATE TABLE IF NOT EXISTS llm_calls" in script
|
||||
# 人执行的那份必须幂等: PG 用 ADD COLUMN IF NOT EXISTS(与库内那份有意不同)
|
||||
for column, _ in PG_BACKFILL:
|
||||
assert f"ALTER TABLE llm_calls ADD COLUMN IF NOT EXISTS {column} " in pg
|
||||
# SQLite 无该语法(写上去直接语法错误),只能以注释交代执行前提
|
||||
lite_alters = [line for line in lite.splitlines() if line.startswith("ALTER TABLE")]
|
||||
assert len(lite_alters) == len(SQLITE_BACKFILL)
|
||||
assert all("IF NOT EXISTS" not in line for line in lite_alters)
|
||||
for column, _ in SQLITE_BACKFILL:
|
||||
assert f"ALTER TABLE llm_calls ADD COLUMN {column} " in lite
|
||||
assert "不存在" in lite
|
||||
with pytest.raises(ValueError, match="mysql"):
|
||||
telemetry_schema_sql("mysql")
|
||||
|
||||
|
||||
class TestBackendColumnParity:
|
||||
"""两个后端的列清单必须逐字同名同序(issue #11)。
|
||||
|
||||
emitter 只组装一份 `fields`,两个后端各按自己的清单取值;两份清单一旦分叉,
|
||||
同一次调用在 SQLite 上写得进、在 PG 上抛 KeyError 被降级吞掉,差异只在换后端时
|
||||
才暴露。issue #13 起两端共用 `schema.COLUMNS`,故这里断言的是"共用"本身
|
||||
(同一个对象则永远无从分叉),列**序**仍单独断言: INSERT 用位置占位符,
|
||||
顺序错位会把值写进错误的列而不报错。
|
||||
"""
|
||||
|
||||
def test_two_backends_agree_on_columns(self):
|
||||
from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
|
||||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||||
from polygateway.telemetry import postgres, sqlite
|
||||
from polygateway.telemetry.schema import COLUMNS
|
||||
|
||||
assert SQLITE_COLUMNS == PG_COLUMNS
|
||||
assert sqlite.COLUMNS is COLUMNS
|
||||
assert postgres.COLUMNS is COLUMNS
|
||||
|
||||
def test_caller_dimensions_are_appended_last(self):
|
||||
"""新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。"""
|
||||
from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
|
||||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||||
from polygateway.telemetry.schema import COLUMNS
|
||||
|
||||
assert SQLITE_COLUMNS[-2:] == ("tenant_id", "meta")
|
||||
assert PG_COLUMNS[-2:] == ("tenant_id", "meta")
|
||||
assert COLUMNS[-2:] == ("tenant_id", "meta")
|
||||
|
||||
|
||||
class TestSQLiteRecorder:
|
||||
@@ -516,10 +621,10 @@ class TestPostgresBackfillDiscipline:
|
||||
async def test_missing_columns_are_added_once(self):
|
||||
conn = _FakePgConn(self._LEGACY)
|
||||
await _record_minimal(self._recorder(conn))
|
||||
from polygateway.telemetry.postgres import _BACKFILL
|
||||
from polygateway.telemetry.schema import PG_BACKFILL
|
||||
|
||||
altered = [s for s in conn.statements if s.startswith("ALTER TABLE")]
|
||||
assert len(altered) == len(_BACKFILL) # 旧表缺全部补列,故一列一条 ALTER
|
||||
assert len(altered) == len(PG_BACKFILL) # 旧表缺全部补列,故一列一条 ALTER
|
||||
assert all("IF NOT EXISTS" not in s for s in altered) # 探测已确认缺列,无需再判
|
||||
|
||||
|
||||
@@ -604,16 +709,15 @@ class _MemoryRecorder:
|
||||
|
||||
|
||||
class TestEmitterRecorderContract:
|
||||
"""emitter 的实参键集合必须与两个后端的 _COLUMNS 完全一致(issue #3)。
|
||||
"""emitter 的实参键集合必须与后端的 `schema.COLUMNS` 完全一致(issue #3)。
|
||||
|
||||
两个后端的 `row = tuple(fields[col] for col in _COLUMNS)` 都在 try **之外**,
|
||||
两个后端的 `row = tuple(fields[col] for col in COLUMNS)` 都在 try **之外**,
|
||||
emitter 漏传一个键就抛 KeyError,被 `_record` 的 except Exception 吞成 warning
|
||||
→ 遥测静默全丢。而 8 个 `**fields` 形态的 fake 一个都拦不住,故显式断言。
|
||||
"""
|
||||
|
||||
async def test_emitter_supplies_exactly_the_backend_columns(self):
|
||||
from polygateway.telemetry.postgres import _COLUMNS as PG_COLUMNS
|
||||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||||
from polygateway.telemetry.schema import COLUMNS
|
||||
|
||||
rec = _MemoryRecorder()
|
||||
await TelemetryEmitter(rec).emit_attempt(
|
||||
@@ -624,11 +728,11 @@ class TestEmitterRecorderContract:
|
||||
response=_resp(),
|
||||
error=None,
|
||||
)
|
||||
assert set(rec.rows[0]) == set(SQLITE_COLUMNS) == set(PG_COLUMNS)
|
||||
assert set(rec.rows[0]) == set(COLUMNS)
|
||||
|
||||
@pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"])
|
||||
async def test_every_entry_point_supplies_the_same_keys(self, emit):
|
||||
from polygateway.telemetry.sqlite import _COLUMNS as SQLITE_COLUMNS
|
||||
from polygateway.telemetry.schema import COLUMNS
|
||||
|
||||
rec = _MemoryRecorder()
|
||||
emitter = TelemetryEmitter(rec)
|
||||
@@ -647,7 +751,7 @@ class TestEmitterRecorderContract:
|
||||
await emitter.emit_terminal_failure(
|
||||
request=_REQ, call_id="c", latency_ms=1, error="dead"
|
||||
)
|
||||
assert set(rec.rows[0]) == set(SQLITE_COLUMNS)
|
||||
assert set(rec.rows[0]) == set(COLUMNS)
|
||||
|
||||
|
||||
class TestEmitterObservabilityFields:
|
||||
|
||||
Reference in New Issue
Block a user