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: