From 1471e0a2c63ea69e7ca46ea7cd393d244c911c23 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 19 Aug 2026 11:18:06 -0400 Subject: [PATCH] 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. --- src/polygateway/telemetry/postgres.py | 88 +------ src/polygateway/telemetry/schema.py | 228 +++++++++++++++++++ src/polygateway/telemetry/sqlite.py | 81 +------ tests/integration/test_postgres_telemetry.py | 4 +- tests/unit/test_telemetry.py | 148 ++++++++++-- 5 files changed, 367 insertions(+), 182 deletions(-) create mode 100644 src/polygateway/telemetry/schema.py diff --git a/src/polygateway/telemetry/postgres.py b/src/polygateway/telemetry/postgres.py index 3f35d96..b0e819c 100644 --- a/src/polygateway/telemetry/postgres.py +++ b/src/polygateway/telemetry/postgres.py @@ -21,56 +21,11 @@ from typing import TYPE_CHECKING from loguru import logger +from polygateway.telemetry.schema import COLUMNS, PG_BACKFILL, PG_DDL, insert_sql + 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(), - cached_prompt_tokens INTEGER, - model_reported TEXT, - sampling TEXT, - reasoning_tokens INTEGER, - tenant_id TEXT NOT NULL DEFAULT '', - meta JSONB NOT NULL DEFAULT '{}'::jsonb -); -""" - -# 新列排在 created_at 之后: 与旧表 ALTER 追加的位置一致(见 sqlite.py 同款注释) -_BACKFILL = ( - ("cached_prompt_tokens", "ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER"), - ("model_reported", "ALTER TABLE llm_calls ADD COLUMN model_reported TEXT"), - ("sampling", "ALTER TABLE llm_calls ADD COLUMN sampling TEXT"), - ("reasoning_tokens", "ALTER TABLE llm_calls ADD COLUMN reasoning_tokens INTEGER"), - # 两个默认值都是非易失常量,PG 11+ 只改 catalog 不重写全表,故大表补列亦是秒级 - ( - "tenant_id", - "ALTER TABLE llm_calls ADD COLUMN tenant_id TEXT NOT NULL DEFAULT ''", - ), - ( - "meta", - "ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb", - ), -) - # 探测表是否存在;不需要任何权限,且与 INSERT 走同一套 search_path 解析 _TABLE_EXISTS = "SELECT to_regclass('llm_calls')" @@ -80,38 +35,7 @@ _EXISTING_COLUMNS = ( "WHERE attrelid = to_regclass('llm_calls') AND attnum > 0 AND NOT attisdropped" ) -_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", - "cached_prompt_tokens", - "model_reported", - "sampling", - "reasoning_tokens", - "tenant_id", - "meta", -) - -_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" -) +_INSERT = insert_sql("postgres", COLUMNS) class PostgresRecorder: @@ -200,7 +124,7 @@ class PostgresRecorder: await self._backfill_columns(conn) # 旧表可能缺列;失败只逐行降级 return True try: - await conn.execute(_DDL) # type: ignore[attr-defined] + await conn.execute(PG_DDL) # type: ignore[attr-defined] except asyncio.CancelledError: raise except Exception as exc: @@ -223,7 +147,7 @@ class PostgresRecorder: """ try: existing = {row["attname"] for row in await conn.fetch(_EXISTING_COLUMNS)} # type: ignore[attr-defined] - for column, statement in _BACKFILL: + for column, statement in PG_BACKFILL: if column not in existing: await conn.execute(statement) # type: ignore[attr-defined] except asyncio.CancelledError: @@ -236,7 +160,7 @@ class PostgresRecorder: pool = await self._ensure_ready() if pool is None: return - row = tuple(fields[col] for col in _COLUMNS) + row = tuple(fields[col] for col in COLUMNS) try: async with pool.acquire() as conn: await conn.execute(_INSERT, *row) diff --git a/src/polygateway/telemetry/schema.py b/src/polygateway/telemetry/schema.py new file mode 100644 index 0000000..99ab544 --- /dev/null +++ b/src/polygateway/telemetry/schema.py @@ -0,0 +1,228 @@ +"""遥测表 `llm_calls` 的 schema 单一事实源: 列序、两端 DDL、补列语句与 INSERT 构造。 + +两个 recorder(`sqlite.py` / `postgres.py`)与公共函数 `telemetry_schema_sql` 共用本模块。 +收敛的理由是**正确性**而非整洁: 打印给下游的 SQL 必须与库真正执行的 DDL 同源——常量在 +多处各存一份必然漂移,而漂移的表现是"下游照打印的 SQL 建完表,库仍报缺列"。 + +**`COLUMNS` 是 INSERT 字段序,不是物理列序**: 数据库自填的 `created_at` 不在其中(它带 +`DEFAULT now()` / `datetime('now')`,库从不显式写它)。物理表列 = 24 个 INSERT 字段 + +`created_at` = 25;列数断言一律按物理列数写,两套口径混用是最易错处。 + +本模块只依赖标准库: `telemetry/` 与 `backends/`、`transports/`、`structured/` 同层且 +互不依赖(import-linter 契约执法)。 +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Sequence + +TABLE = "llm_calls" + +# 支持的后端;`insert_sql` / `telemetry_schema_sql` 的取值域 +_BACKENDS = ("sqlite", "postgres") + +SQLITE_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 REAL, + max_inter_token_ms REAL, + cache_hit INTEGER NOT NULL DEFAULT 0, + error TEXT, + cost REAL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + cached_prompt_tokens INTEGER, + model_reported TEXT, + sampling TEXT, + reasoning_tokens INTEGER, + tenant_id TEXT NOT NULL DEFAULT '', + meta TEXT NOT NULL DEFAULT '{}' +); +""" + +PG_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(), + cached_prompt_tokens INTEGER, + model_reported TEXT, + sampling TEXT, + reasoning_tokens INTEGER, + tenant_id TEXT NOT NULL DEFAULT '', + meta JSONB NOT NULL DEFAULT '{}'::jsonb +); +""" + +# 新列必须排在 created_at 之后: 旧表只能经 ALTER 追加到末尾,新建库若把它们 +# 插在前面,两条路径的物理列序会分叉(列序断言测试无合规修法)。 +SQLITE_BACKFILL = ( + ("cached_prompt_tokens", "INTEGER"), + ("model_reported", "TEXT"), + ("sampling", "TEXT"), + ("reasoning_tokens", "INTEGER"), + # NOT NULL 补列必须带非 NULL 常量默认值,否则 SQLite 直接拒绝该 ALTER + # ("Cannot add a NOT NULL column with default value NULL"),补列全盘失败。 + ("tenant_id", "TEXT NOT NULL DEFAULT ''"), + ("meta", "TEXT NOT NULL DEFAULT '{}'"), +) + +# PG 补列的列定义。语句由此派生成两份文本(见下),使"库内执行的那份"与"打印给 +# 下游的那份"的列集合与列定义**无法分叉**——本模块存在的全部理由就是不许漂移。 +_PG_BACKFILL_DECLS = ( + ("cached_prompt_tokens", "INTEGER"), + ("model_reported", "TEXT"), + ("sampling", "TEXT"), + ("reasoning_tokens", "INTEGER"), + # 两个默认值都是非易失常量,PG 11+ 只改 catalog 不重写全表,故大表补列亦是秒级 + ("tenant_id", "TEXT NOT NULL DEFAULT ''"), + ("meta", "JSONB NOT NULL DEFAULT '{}'::jsonb"), +) + +# 新列排在 created_at 之后: 与旧表 ALTER 追加的位置一致(见 SQLITE_BACKFILL 同款注释)。 +# **库内执行的这份有意不带 `IF NOT EXISTS`**: PG 对它即便列已存在也会先取 ACCESS +# EXCLUSIVE 锁,而遥测是业务路径上的内联 await,故库侧一律"先探测后 ALTER" +# (postgres.py `_backfill_columns` 记有实测)。给人执行的那份见 `telemetry_schema_sql`。 +PG_BACKFILL = tuple( + (column, f"ALTER TABLE {TABLE} ADD COLUMN {column} {decl}") + for column, decl in _PG_BACKFILL_DECLS +) + +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", + "cached_prompt_tokens", + "model_reported", + "sampling", + "reasoning_tokens", + "tenant_id", + "meta", +) + +_COLUMN_SET = frozenset(COLUMNS) + + +def insert_sql(backend: str, columns: Sequence[str]) -> str: + """按给定列构造 INSERT;列必须是 `COLUMNS` 的子集,否则 ValueError。 + + 子集校验是**注入面的闸**: 列名来自数据库探测结果,不是常量,不校验就等于把外部 + 字符串拼进 SQL(占位符只保护值,保护不了列名)。sqlite 用 `?`、postgres 用 `$n`, + 两端的重复键处理都不绑定具体约束名(`INSERT OR IGNORE` / `ON CONFLICT`)。 + + Args: + backend: `"sqlite"` 或 `"postgres"`。 + columns: 要写入的列,顺序即占位符顺序(调用方须按同序取值)。 + + Returns: + 完整的 INSERT 语句。 + + Raises: + ValueError: backend 不在取值域内,或 columns 含 `COLUMNS` 之外的列名。 + """ + if backend not in _BACKENDS: + raise ValueError(f"未知遥测后端 {backend!r}: 只支持 {list(_BACKENDS)}") + selected = tuple(columns) + unknown = [column for column in selected if column not in _COLUMN_SET] + if unknown: + raise ValueError(f"列名不在遥测 schema 内(拒绝拼进 SQL): {unknown}") + names = ", ".join(selected) + if backend == "sqlite": + placeholders = ", ".join("?" for _ in selected) + return f"INSERT OR IGNORE INTO {TABLE} ({names}) VALUES ({placeholders})" + placeholders = ", ".join(f"${i + 1}" for i in range(len(selected))) + return f"INSERT INTO {TABLE} ({names}) VALUES ({placeholders}) ON CONFLICT (call_id) DO NOTHING" + + +def telemetry_schema_sql(backend: str) -> str: + """返回可直接粘进迁移文件的完整脚本(建表 + 各补列语句 + 注释)。 + + 给不愿意让库在自己的生产表上发 DDL 的下游用: 输出与库运行时执行的 DDL 同源, + 照它建完表,库探测到的列就是齐的。 + + **补列语句与库内执行的那份是两套文本,不是一份**: 这份给人执行,必须可重复执行, + 故 PG 变体带 `ADD COLUMN IF NOT EXISTS`(它会先取 ACCESS EXCLUSIVE 锁,但执行时机 + 由 DBA 自己挑,锁风险可控);库内那份不带,靠先探测后 ALTER 规避锁。SQLite 没有 + `ADD COLUMN IF NOT EXISTS` 语法,只能以注释交代"仅当该列不存在时执行"。 + + Args: + backend: `"sqlite"` 或 `"postgres"`。 + + Returns: + 含注释的完整 SQL 脚本。 + + Raises: + ValueError: backend 不在取值域内。 + """ + if backend not in _BACKENDS: + raise ValueError(f"未知遥测后端 {backend!r}: 只支持 {list(_BACKENDS)}") + if backend == "sqlite": + ddl = SQLITE_DDL + notes = ( + f"-- 旧表补列(库升级后新增的列)。SQLite 无 ADD COLUMN IF NOT EXISTS 语法,\n" + f"-- 以下每条**仅当该列不存在时执行**(先 PRAGMA table_info({TABLE}) 对照)。" + ) + alters = [ + f"ALTER TABLE {TABLE} ADD COLUMN {column} {decl};" for column, decl in SQLITE_BACKFILL + ] + else: + ddl = PG_DDL + notes = ( + "-- 旧表补列(库升级后新增的列)。带 IF NOT EXISTS,整段可重复执行;\n" + "-- 注意它即便列已存在也会先取 ACCESS EXCLUSIVE 锁,请挑低峰执行。" + ) + alters = [ + f"ALTER TABLE {TABLE} ADD COLUMN IF NOT EXISTS {column} {decl};" + for column, decl in _PG_BACKFILL_DECLS + ] + header = ( + f"-- PolyGateway 遥测表 {TABLE}({backend})\n" + f'-- 由 polygateway.telemetry_schema_sql("{backend}") 生成,与库运行时执行的 DDL 同源。\n' + "-- 新建库执行整段;已有旧表则建表语句自动跳过,只需关注下方补列语句。" + ) + return "\n".join([header, "", ddl.strip(), "", notes, *alters, ""]) diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index 4cb5a64..abf3d1b 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -22,80 +22,9 @@ from pathlib import Path from loguru import logger -_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 REAL, - max_inter_token_ms REAL, - cache_hit INTEGER NOT NULL DEFAULT 0, - error TEXT, - cost REAL, - created_at TEXT NOT NULL DEFAULT (datetime('now')), - cached_prompt_tokens INTEGER, - model_reported TEXT, - sampling TEXT, - reasoning_tokens INTEGER, - tenant_id TEXT NOT NULL DEFAULT '', - meta TEXT NOT NULL DEFAULT '{}' -); -""" +from polygateway.telemetry.schema import COLUMNS, SQLITE_BACKFILL, SQLITE_DDL, insert_sql -# 新列必须排在 created_at 之后: 旧表只能经 ALTER 追加到末尾,新建库若把它们 -# 插在前面,两条路径的物理列序会分叉(列序断言测试无合规修法)。 -_BACKFILL_COLUMNS = ( - ("cached_prompt_tokens", "INTEGER"), - ("model_reported", "TEXT"), - ("sampling", "TEXT"), - ("reasoning_tokens", "INTEGER"), - # NOT NULL 补列必须带非 NULL 常量默认值,否则 SQLite 直接拒绝该 ALTER - # ("Cannot add a NOT NULL column with default value NULL"),补列全盘失败。 - ("tenant_id", "TEXT NOT NULL DEFAULT ''"), - ("meta", "TEXT NOT NULL DEFAULT '{}'"), -) - -_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", - "cached_prompt_tokens", - "model_reported", - "sampling", - "reasoning_tokens", - "tenant_id", - "meta", -) - -_INSERT = ( - f"INSERT OR IGNORE INTO llm_calls ({', '.join(_COLUMNS)}) " - f"VALUES ({', '.join('?' for _ in _COLUMNS)})" -) +_INSERT = insert_sql("sqlite", COLUMNS) class SQLiteRecorder: @@ -110,7 +39,7 @@ class SQLiteRecorder: conn = sqlite3.connect(path, check_same_thread=False, timeout=10.0) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=5000") - conn.execute(_DDL) + conn.execute(SQLITE_DDL) conn.commit() self._conn = conn except (OSError, sqlite3.Error) as exc: @@ -132,7 +61,7 @@ class SQLiteRecorder: except sqlite3.Error as exc: logger.warning("SQLite 遥测列探测失败(写入将逐行降级): {}", exc) return - for column, decl in _BACKFILL_COLUMNS: + for column, decl in SQLITE_BACKFILL: if column in existing: continue # 逐列独立 try: 一列撞上 duplicate 不得让后面的列漏补 @@ -148,7 +77,7 @@ class SQLiteRecorder: """写一行遥测;字段集合即 24 字段冻结签名(ports.TelemetryRecorder)。""" if self._conn is None: return - row = tuple(fields[col] for col in _COLUMNS) + row = tuple(fields[col] for col in COLUMNS) try: await asyncio.to_thread(self._write, row) except (OSError, sqlite3.Error) as exc: diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 5850c2f..a9b3f9e 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -320,7 +320,7 @@ async def least_privilege_dsn(dsn): """ import asyncpg - from polygateway.telemetry.postgres import _DDL + from polygateway.telemetry.schema import PG_DDL name = f"pgwtest_lp_{uuid4().hex[:8]}" admin = await asyncpg.connect(dsn, timeout=10) @@ -332,7 +332,7 @@ async def least_privilege_dsn(dsn): await admin.execute(f"CREATE ROLE {name} LOGIN PASSWORD '{_PROBE_PASSWORD}'") await admin.execute(f"CREATE SCHEMA {name}") await admin.execute(f"SET search_path = {name}") - await admin.execute(_DDL) # 表由**别的账号**建好,与现场一致 + await admin.execute(PG_DDL) # 表由**别的账号**建好,与现场一致 await admin.execute(f"GRANT USAGE ON SCHEMA {name} TO {name}") await admin.execute(f"GRANT SELECT, INSERT ON {name}.llm_calls TO {name}") # 关键: 绝不 GRANT CREATE ON SCHEMA —— 缺的正是这一项 diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index afec6b2..f9a5083 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -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: