diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index 1d81e1c..6143668 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -26,13 +26,30 @@ from polygateway.middleware.cache import digest_messages from polygateway.types import canonical_sampling_json, merge_sampling if TYPE_CHECKING: - from collections.abc import Callable + from collections.abc import Callable, Mapping + from typing import Any from polygateway.ports import CallNext, TelemetryRecorder from polygateway.pricing import PricingTable from polygateway.types import ChatRequest, LLMResponse, SourceConfig +def _canonical_meta_json(meta: Mapping[str, Any]) -> str: + """把调用方自定义维度定型为 JSON 文本(issue #11);空 dict 落字面量 `'{}'`。 + + `sort_keys=True` 让同一份维度在任意两行里字节一致,可直接等值比对与去重; + `ensure_ascii=False` 保留中文原文,避免落库成 `\\uXXXX` 串而无法肉眼审计。 + + `allow_nan=False` 是**第二道闸**(主防线是 `types.validate_caller_dimensions` + 在公共入口的校验): `json.dumps` 默认把 `nan` 写成裸 `NaN` 字面量,那不是合法 + JSON,PG 的 JSONB 会拒收;而写入失败会被 `_record` 的降级 try 吞成 warning, + 等于把调用方的输入错误转化成静默丢遥测。宁可在这里显式抛。 + """ + if not meta: + return "{}" + return json.dumps(dict(meta), sort_keys=True, ensure_ascii=False, allow_nan=False) + + @dataclass(frozen=True) class _AttemptUsage: """一次尝试的用量视图;默认值即"失败尝试"档(无用量可言,记 0 并标 unavailable)。 @@ -72,7 +89,7 @@ class _AttemptUsage: class TelemetryEmitter: - """从请求与结果组装 21 字段并写入 recorder;一切写失败降级 warning。""" + """从请求与结果组装 24 字段并写入 recorder;一切写失败降级 warning。""" def __init__(self, recorder: TelemetryRecorder, *, pricing: PricingTable | None = None) -> None: self._recorder = recorder @@ -111,6 +128,8 @@ class TelemetryEmitter: reasoning_tokens=usage.reasoning_tokens, # 唯一有"生效源"的入口,故是唯一能并上 extra_body 的(设计决策 D) sampling=canonical_sampling_json(merge_sampling(source.extra_body, request.sampling)), + tenant_id=request.tenant_id, + meta=request.meta, ) async def emit_cache_hit(self, *, request: ChatRequest, response: LLMResponse) -> None: @@ -139,6 +158,11 @@ class TelemetryEmitter: # 由最外层 TelemetryMW 调用,手上没有 source。缓存命中行无损: # sampling 已进缓存 key,能命中即意味调用级参数与历史那次逐字相同 sampling=canonical_sampling_json(request.sampling), + # 与上面的 model/prompt_tokens 相反,维度读 request 而非 response: + # 维度回答的是"本次调用由谁发起",不是历史那次。读历史会把本次调用 + # 记到上一个租户头上,两边的账同时错且无任何报错(issue #11 设计 §4.3) + tenant_id=request.tenant_id, + meta=request.meta, ) async def emit_terminal_failure( @@ -166,6 +190,9 @@ class TelemetryEmitter: reasoning_tokens=None, # 无具体源,与 model/provider/source_name 置空同一先例(设计决策 D) sampling=canonical_sampling_json(request.sampling), + # 源不可知,但租户归属是已知的——终态失败行恰是审计最需要的 + tenant_id=request.tenant_id, + meta=request.meta, ) async def _record( @@ -190,6 +217,9 @@ class TelemetryEmitter: model_reported: str | None, sampling: str | None, reasoning_tokens: int | None, + # issue #11: 未归一化的调用方维度,归一化在本方法内收口(recorder 只落库) + tenant_id: str | None, + meta: Mapping[str, Any], ) -> None: try: # 成本换算(M2 §6): 成功行按单价换算;缓存命中 0.0(未产生新调用); @@ -231,6 +261,10 @@ class TelemetryEmitter: model_reported=model_reported, sampling=sampling, reasoning_tokens=reasoning_tokens, + # 空串是哨兵而非 NULL: NULL 的 tenant_id 在 PG 的 RLS policy 下 + # 对所有人永久不可见,空串则可用一条 SQL 审计出未归属的行 + tenant_id=tenant_id or "", + meta=_canonical_meta_json(meta), ) except asyncio.CancelledError: raise diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index c64ccd7..66c80c2 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -245,10 +245,15 @@ class StructuredOutputStrategy(Protocol): @runtime_checkable class TelemetryRecorder(Protocol): - """遥测后端;22 字段冻结(M1 设计 §4.4 + issue #3/#4),唯一调用点是 TelemetryEmitter。 + """遥测后端;24 字段冻结(M1 设计 §4.4 + issue #3/#4/#11),唯一调用点是 TelemetryEmitter。 新增参数不设默认值: 库外无第三方实现者(三项目迁移时删除了各自的同名 Protocol),完整签名的成本为零,而少写一列会被 emitter 的降级吞成 warning。 + + `tenant_id` 与 `meta` 到达 recorder 时**已由 emitter 归一化**——`tenant_id` + 的 `None` 已转空串,`meta` 已序列化为 JSON 字符串(空 dict 为 `'{}'`)。 + recorder 只负责落库,不做任何语义判断,与 `sampling` 列由 + `canonical_sampling_json()` 在 emitter 侧定型是同一先例。 """ async def record_llm_call( @@ -276,4 +281,6 @@ class TelemetryRecorder(Protocol): model_reported: str | None, sampling: str | None, reasoning_tokens: int | None, + tenant_id: str, + meta: str, ) -> None: ... diff --git a/src/polygateway/telemetry/postgres.py b/src/polygateway/telemetry/postgres.py index 24e7986..bd99511 100644 --- a/src/polygateway/telemetry/postgres.py +++ b/src/polygateway/telemetry/postgres.py @@ -48,7 +48,9 @@ CREATE TABLE IF NOT EXISTS llm_calls ( cached_prompt_tokens INTEGER, model_reported TEXT, sampling TEXT, - reasoning_tokens INTEGER + reasoning_tokens INTEGER, + tenant_id TEXT NOT NULL DEFAULT '', + meta JSONB NOT NULL DEFAULT '{}'::jsonb ); """ @@ -58,6 +60,15 @@ _BACKFILL = ( ("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 解析 @@ -92,6 +103,8 @@ _COLUMNS = ( "model_reported", "sampling", "reasoning_tokens", + "tenant_id", + "meta", ) _INSERT = ( diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index 8483c68..4cb5a64 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -46,7 +46,9 @@ CREATE TABLE IF NOT EXISTS llm_calls ( cached_prompt_tokens INTEGER, model_reported TEXT, sampling TEXT, - reasoning_tokens INTEGER + reasoning_tokens INTEGER, + tenant_id TEXT NOT NULL DEFAULT '', + meta TEXT NOT NULL DEFAULT '{}' ); """ @@ -57,6 +59,10 @@ _BACKFILL_COLUMNS = ( ("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 = ( @@ -82,6 +88,8 @@ _COLUMNS = ( "model_reported", "sampling", "reasoning_tokens", + "tenant_id", + "meta", ) _INSERT = ( @@ -137,7 +145,7 @@ class SQLiteRecorder: logger.warning("SQLite 遥测补列失败(写入将逐行降级): {}", exc) async def record_llm_call(self, **fields: object) -> None: - """写一行遥测;字段集合即 21 字段冻结签名(ports.TelemetryRecorder)。""" + """写一行遥测;字段集合即 24 字段冻结签名(ports.TelemetryRecorder)。""" if self._conn is None: return row = tuple(fields[col] for col in _COLUMNS) diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index d1af681..2db3f58 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -45,6 +45,8 @@ _EXPECTED_COLUMNS = [ "model_reported", "sampling", "reasoning_tokens", + "tenant_id", + "meta", ] # run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见 @@ -110,6 +112,9 @@ async def _record_minimal( "model_reported": None, "sampling": None, "reasoning_tokens": None, + # 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}' + "tenant_id": "", + "meta": "{}", } fields.update(overrides) await recorder.record_llm_call(**fields) diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index fe722eb..c5009ee 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -118,6 +118,8 @@ class _DummyRecorder: model_reported, sampling, reasoning_tokens, + tenant_id, + meta, ) -> None: ... @@ -205,6 +207,29 @@ class TestGateUpdate: ) +class TestTelemetryRecorderSignature: + """`record_llm_call` 的冻结签名以 `inspect.signature` 实测,不凭记忆断言。 + + 该 Protocol 的纪律是新增参数**不设默认值**(ports.py docstring):库外无第三方 + 实现者,而带默认值的参数会让 emitter 漏传时静默落默认值——遥测里的租户归属 + 一旦静默错位,事后无从分辨是"没传"还是"就是空的"。 + """ + + def test_caller_dimensions_are_declared(self): + import inspect + + params = inspect.signature(TelemetryRecorder.record_llm_call).parameters + assert {"tenant_id", "meta"} <= set(params) + + @pytest.mark.parametrize("name", ["tenant_id", "meta"]) + def test_caller_dimensions_have_no_default(self, name): + import inspect + + param = inspect.signature(TelemetryRecorder.record_llm_call).parameters[name] + assert param.default is inspect.Parameter.empty + assert param.kind is inspect.Parameter.KEYWORD_ONLY + + class TestOcrPorts: """M3 三个 OCR Protocol(设计 §3.2): runtime_checkable 结构判定。""" diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index c332777..5c3efcc 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -40,6 +40,8 @@ _EXPECTED_COLUMNS = [ "model_reported", "sampling", "reasoning_tokens", + "tenant_id", + "meta", ] @@ -104,11 +106,38 @@ async def _record_minimal(recorder, call_id="c1", **overrides): "model_reported": None, "sampling": None, "reasoning_tokens": None, + # 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}' + "tenant_id": "", + "meta": "{}", } fields.update(overrides) await recorder.record_llm_call(**fields) +class TestBackendColumnParity: + """两个后端的 `_COLUMNS` 必须逐字同名同序(issue #11)。 + + emitter 只组装一份 `fields`,两个后端各自按自己的 `_COLUMNS` 取值;两份清单 + 一旦分叉,同一次调用在 SQLite 上写得进、在 PG 上抛 KeyError 被降级吞掉, + 差异只在换后端时才暴露。列**序**同样断言: 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 + + assert SQLITE_COLUMNS == PG_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 + + assert SQLITE_COLUMNS[-2:] == ("tenant_id", "meta") + assert PG_COLUMNS[-2:] == ("tenant_id", "meta") + + class TestSQLiteRecorder: async def test_schema_has_frozen_columns(self, tmp_path): recorder = SQLiteRecorder(tmp_path / "t.db") @@ -327,6 +356,8 @@ class TestPostgresBackfillDiscipline: "model_reported", "sampling", "reasoning_tokens", + "tenant_id", + "meta", ] def _recorder(self, conn): @@ -379,6 +410,8 @@ class TestPostgresTableProbe: "model_reported", "sampling", "reasoning_tokens", + "tenant_id", + "meta", ] def _recorder(self, conn): @@ -613,6 +646,110 @@ class TestEmitterSamplingColumn: assert rec.rows[0]["sampling"] is None +class TestEmitterCallerDimensions: + """issue #11: 三个 emit 入口统一从 `request` 读维度,`_record` 落库前归一化。 + + 维度只有一个读取点(`request`),否则同一列在三种行里口径分叉——那正是 + "遥测调用点收敛为单一 helper"这条铁律要防的形态。 + """ + + _META = {"z_last": "z", "a_first": 1, "m_mid": True} + _REQ_A = ChatRequest( + messages=[{"role": "user", "content": "hi"}], + session_id="sess-1", + tenant_id="tenant-a", + meta=_META, + ) + + @pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"]) + async def test_every_entry_point_carries_the_dimensions(self, emit): + """三条路径写出的行都必须带维度: 漏掉任一条,该租户的账就永远对不上。""" + rec = _MemoryRecorder() + emitter = TelemetryEmitter(rec) + if emit == "attempt": + await emitter.emit_attempt( + request=self._REQ_A, + source=_source(), + call_id="c", + latency_ms=1, + response=_resp(), + error=None, + ) + elif emit == "cache_hit": + await emitter.emit_cache_hit(request=self._REQ_A, response=_resp(cache_hit=True)) + else: + await emitter.emit_terminal_failure( + request=self._REQ_A, call_id="c", latency_ms=1, error="dead" + ) + row = rec.rows[0] + assert row["tenant_id"] == "tenant-a" + assert json.loads(row["meta"]) == self._META + + async def test_cache_hit_records_the_current_caller_not_the_cached_one(self): + """缓存命中行的维度是"本次由谁发起",不是历史那次——最容易实现反的一处。 + + 历史那次由租户 B 发起并把响应留在了缓存里;本次由租户 A 发起并命中。 + 若读了历史那次的归属,租户 A 的调用会记到 B 头上,而 A 的账面凭空少一行 + ——两个租户的账同时错,且错得没有任何报错。 + """ + historical = ChatRequest( + messages=[{"role": "user", "content": "hi"}], + tenant_id="tenant-b", + meta={"batch": "old-batch"}, + ) + rec = _MemoryRecorder() + mw = TelemetryMW(TelemetryEmitter(rec)) + + async def terminal(request): + # 缓存层回放的是历史那次的响应对象(其 call_id 属于 historical 那次) + return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid") + + assert historical.tenant_id == "tenant-b" # 历史归属确实不同,否则本用例是空转 + await mw(self._REQ_A, terminal) + + row = rec.rows[0] + assert row["cache_hit"] is True + assert row["tenant_id"] == "tenant-a" + assert "old-batch" not in row["meta"] + + async def test_absent_dimensions_land_as_sentinels(self): + """未传维度落哨兵值: `tenant_id` 空串、`meta` 字面量 `'{}'`,都不是 NULL。 + + NULL 的 `tenant_id` 在 PG 的 RLS policy 下对所有人永久不可见(设计 §4.4), + 空串则可用一条 SQL 审计出还有多少行未归属;`meta` 同理,`'{}'` 可被 + JSON 函数直接查询,NULL 则要每条查询都额外判空。 + """ + rec = _MemoryRecorder() + await TelemetryEmitter(rec).emit_attempt( + request=_REQ, # tenant_id=None, meta={} + source=_source(), + call_id="c", + latency_ms=1, + response=_resp(), + error=None, + ) + row = rec.rows[0] + assert row["tenant_id"] == "" + assert row["meta"] == "{}" + + async def test_meta_is_serialized_with_sorted_keys(self): + """键序固定,同一份维度在任意两行里字节一致,可直接做等值比对与去重。""" + rec = _MemoryRecorder() + await TelemetryEmitter(rec).emit_terminal_failure( + request=self._REQ_A, call_id="c", latency_ms=1, error="dead" + ) + assert list(json.loads(rec.rows[0]["meta"])) == ["a_first", "m_mid", "z_last"] + + async def test_non_ascii_meta_stays_readable(self): + """`ensure_ascii=False`: 中文维度按原文落库,而非 `\\uXXXX` 转义串。""" + rec = _MemoryRecorder() + req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"dept": "研发"}) + await TelemetryEmitter(rec).emit_terminal_failure( + request=req, call_id="c", latency_ms=1, error="dead" + ) + assert "研发" in rec.rows[0]["meta"] + + class TestCostWithCachedTier: """issue #3: 命中部分按缓存单价计费,避免 cost 系统性高估。"""