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

feat: add stall_ms and stall_count telemetry columns

Browse Source 
Third approved port change: TelemetryRecorder.record_llm_call gains
stall_ms/stall_count (assembly gate derives from this signature).
Schema five edit points (both DDLs, both backfill decls, COLUMNS)
with INTEGER nullable columns appended last; telemetry_schema_sql
regenerates from the same constants untouched. Emitter forwards the
pair from retry._emit on the success branch only (cancelled hedge
loser rows stay NULL by construction); cache-hit and terminal rows
pass explicit None. Both **fields recorders pick the columns up with
zero signature change.
This commit is contained in:
iomgaa
2026-09-14 01:16:33 -04:00
parent 53bcf43172
commit e0b0be09a7
8 changed files with 220 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_postgres_telemetry.py
+6 -3
View File
@@ -69,6 +69,9 @@ _EXPECTED_COLUMNS = [
"error_body",
"attempts",
"total_latency_ms",
# —— 1.3.8 沉默归因两列(issue #27)——
"stall_ms",
"stall_count",
]
# 1.3.5 之前那张表的 27 个物理列(26 个 INSERT 字段 + created_at)。写成固定切片
@@ -769,7 +772,7 @@ class TestCallerDimensionsAcceptance:
"WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
schema,
)
# 22 → 36 个 recorder 字段(加 created_at 共 37 个物理列),且新列追加在末尾
# 22 → 38 个 recorder 字段(加 created_at 共 39 个物理列),且新列追加在末尾
assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
rows = await _fetch(
schema_dsn,
@@ -1142,7 +1145,7 @@ class TestCallObservabilityColumnsAcceptance:
"WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
schema,
)
# 26 → 36 个 recorder 字段(加 created_at 共 37 个物理列),新列追加在末尾:
# 26 → 38 个 recorder 字段(加 created_at 共 39 个物理列),新列追加在末尾:
# 列序与新建库一致才不会让两条升级路径分叉
assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
@@ -1337,7 +1340,7 @@ class TestPublishedSchemaScript:
await _execute_script(fresh_dsn, script)
actual = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)]
# 物理列 = 36 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份
# 物理列 = 38 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份
assert set(actual) == set(COLUMNS) | {"created_at"}
# 列序也不许漂: 新列必须排在 created_at 之后,否则新建库与 ALTER 升级的列序分叉
assert actual == _EXPECTED_COLUMNS
tests/unit/test_hedge.py
+17
View File
@@ -142,6 +142,10 @@ class RecordingEmitter:
error,
reasoning_applies,
operation,
# 1.3.8: `_emit` 显式转发 stall 两值(审查 B1)——显式全签名 fake 不加则 TypeError;
# 默认 None 只护调用方,护不住被调的显式签名
stall_ms=None,
stall_count=None,
):
self.rows.append(
{
@@ -151,6 +155,9 @@ class RecordingEmitter:
if request.call_context is not None
else None,
"error": error,
# 1.3.8: 逐行捕获 stall 归因(取消输家行须恒 None,设计 §11-B2)
"stall_ms": stall_ms,
"stall_count": stall_count,
}
)
@@ -455,6 +462,16 @@ class TestHedgeSettlementAndSignals:
assert winner["error"] is None
assert loser["logical_call_id"] == winner["logical_call_id"] == ctx.logical_call_id
async def test_hedge_cancelled_row_has_null_stall(self):
"""被取消输家无 TransportResult: `hedge_cancelled` 行 stall 两值恒 None(1.3.8 B2)。"""
emitter = RecordingEmitter()
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, emitter=emitter)
await asyncio.wait_for(mw(_req()), timeout=5)
loser = next(r for r in emitter.rows if r["error"] == "hedge_cancelled")
assert loser["stall_ms"] is None
assert loser["stall_count"] is None
async def test_loser_does_not_feed_breaker(self):
"""输家取消不喂熔断失败计数、不喂健康分;赢家照常 record_success(④)。"""
selector = RecordingSelector()
tests/unit/test_ports.py
+3 -1
View File
@@ -147,6 +147,8 @@ class _DummyRecorder:
error_body,
attempts,
total_latency_ms,
stall_ms,
stall_count,
) -> None: ...
@@ -319,7 +321,7 @@ class TestTelemetryRecorderSignature:
"attempts",
"total_latency_ms",
} <= set(params)
assert len(params) - 1 == 36 # 减掉 self
assert len(params) - 1 == 38 # 减掉 self
@pytest.mark.parametrize(
"name",
tests/unit/test_telemetry.py
+146 -14
View File
@@ -89,6 +89,9 @@ _EXPECTED_COLUMNS = [
"error_body",
"attempts",
"total_latency_ms",
# —— 1.3.8 沉默归因两列(issue #27)——
"stall_ms",
"stall_count",
]
# 1.3.5 之前的 26 个 INSERT 字段(旧表形态),供冻结 INSERT 语句与旧表补列用例复用
@@ -181,6 +184,9 @@ async def _record_minimal(recorder, call_id="c1", **overrides):
# 逻辑快照两列只属终态行
"attempts": None,
"total_latency_ms": None,
# —— 1.3.8 沉默归因两列: 默认 None = 非流式/无沉默观测(合法事实)——
"stall_ms": None,
"stall_count": None,
}
fields.update(overrides)
await recorder.record_llm_call(**fields)
@@ -262,7 +268,7 @@ class TestSchemaModule:
# COLUMNS 是 INSERT 字段序,不含数据库自填的 created_at
assert list(COLUMNS) == [c for c in _EXPECTED_COLUMNS if c != "created_at"]
assert len(COLUMNS) == 36
assert len(COLUMNS) == 38
# 两端 DDL 的列出现顺序 == 物理列序(created_at 在第 19 位)
for ddl in (SQLITE_DDL, PG_DDL):
assert _first_occurrence_order(ddl, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
@@ -276,8 +282,8 @@ class TestSchemaModule:
"ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER",
)
assert PG_BACKFILL[-1] == (
"total_latency_ms",
"ALTER TABLE llm_calls ADD COLUMN total_latency_ms INTEGER",
"stall_count",
"ALTER TABLE llm_calls ADD COLUMN stall_count INTEGER",
)
assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL)
@@ -291,10 +297,10 @@ class TestSchemaModule:
assert insert_sql("sqlite", _PRE_135_COLUMNS) == _FROZEN_SQLITE_INSERT
assert insert_sql("postgres", _PRE_135_COLUMNS) == _FROZEN_PG_INSERT
# 全量 36 列: 占位符随列数增长,且不留空洞
assert insert_sql("postgres", COLUMNS).count("$") == 36
assert "$36)" in insert_sql("postgres", COLUMNS)
assert insert_sql("sqlite", COLUMNS).count("?") == 36
# 全量 38 列: 占位符随列数增长,且不留空洞
assert insert_sql("postgres", COLUMNS).count("$") == 38
assert "$38)" in insert_sql("postgres", COLUMNS)
assert insert_sql("sqlite", COLUMNS).count("?") == 38
# 裁剪列表按位置占位符重新编号,不留空洞
assert insert_sql("postgres", ["call_id", "model"]) == (
"INSERT INTO llm_calls (call_id, model) VALUES ($1, $2) ON CONFLICT DO NOTHING"
@@ -371,7 +377,130 @@ class TestBackendColumnParity:
"""新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。"""
from polygateway.telemetry.schema import COLUMNS
assert COLUMNS[-4:] == ("cause_type", "error_body", "attempts", "total_latency_ms")
assert COLUMNS[-6:] == (
"cause_type",
"error_body",
"attempts",
"total_latency_ms",
"stall_ms",
"stall_count",
)
# 1.3.7(1.3.8 补列前)的表形态: 36 个 INSERT 字段 + created_at = 37 物理列。
# 冻结字面量而非由常量切片派生: 后者会随 schema 改动静默漂移到另一张表上
# (与 _PRE_TENANT_DDL 同款纪律)。
_PRE_138_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 '{}',
thinking_observation TEXT,
reasoning_effort TEXT,
scope TEXT,
operation TEXT,
logical_call_id TEXT,
event_kind TEXT,
http_status_code INTEGER,
error_type TEXT,
cause_type TEXT,
error_body TEXT,
attempts INTEGER,
total_latency_ms INTEGER
);
"""
class TestStallColumns:
"""1.3.8 沉默归因两列(issue #27): 落行、缺省 NULL、迁移幂等与 manual 告警。
新列可空、无默认值、追加末尾——旧行的 NULL 表达"补列之前根本没记过这件事"。
"""
async def test_stall_values_round_trip_and_default_is_null(self, tmp_path):
"""stall 值原样落行读回;未观测行(默认 None)两列是 NULL 而非 0。"""
db = tmp_path / "stall.db"
recorder = SQLiteRecorder(db, auto_migrate=True)
await _record_minimal(recorder, call_id="c-stall", stall_ms=3200, stall_count=2)
await _record_minimal(recorder, call_id="c-plain")
recorder.close()
conn = sqlite3.connect(db)
assert conn.execute(
"SELECT stall_ms, stall_count FROM llm_calls WHERE call_id = 'c-stall'"
).fetchone() == (3200, 2)
# "观测过且零沉默"(0)与"无从观测"(NULL)是两回事: 未传值的行必须是 NULL
assert conn.execute(
"SELECT stall_ms, stall_count FROM llm_calls WHERE call_id = 'c-plain'"
).fetchone() == (None, None)
conn.close()
async def test_auto_migrate_backfills_the_two_columns_idempotently(self, tmp_path):
"""1.3.7 形态旧表(37 物理列)经 auto 补两列;二次初始化幂等不炸。"""
db = tmp_path / "pre138.db"
conn = sqlite3.connect(db)
conn.execute(_PRE_138_DDL)
conn.commit()
conn.close()
recorder = SQLiteRecorder(db, auto_migrate=True)
await _record_minimal(recorder, call_id="c-new", stall_ms=7, stall_count=1)
recorder.close()
# 二次初始化: 探测到列已齐,一条 ALTER 都不发(幂等)
recorder2 = SQLiteRecorder(db, auto_migrate=True)
await _record_minimal(recorder2, call_id="c-new2")
recorder2.close()
conn = sqlite3.connect(db)
assert [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] == _EXPECTED_COLUMNS
assert conn.execute(
"SELECT stall_ms, stall_count FROM llm_calls WHERE call_id = 'c-new'"
).fetchone() == (7, 1)
conn.close()
async def test_manual_mode_warning_names_both_new_alters(self, tmp_path, captured_warnings):
"""manual 档缺列告警逐字含两条新 ALTER,物理列原封不动。"""
db = tmp_path / "manual138.db"
conn = sqlite3.connect(db)
conn.execute(_PRE_138_DDL)
conn.commit()
conn.close()
recorder = SQLiteRecorder(db, auto_migrate=False)
await _record_minimal(recorder, call_id="c-manual", stall_ms=9, stall_count=3)
recorder.close()
conn = sqlite3.connect(db)
assert len([r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]) == 37
conn.close()
assert len(captured_warnings) == 1
message = captured_warnings[0]
assert "stall_ms" in message and "stall_count" in message
assert "ALTER TABLE llm_calls ADD COLUMN stall_ms INTEGER" in message
assert "ALTER TABLE llm_calls ADD COLUMN stall_count INTEGER" in message
class TestSQLiteRecorder:
@@ -647,7 +776,7 @@ class TestSQLiteCallerDimensionsAcceptance:
conn = sqlite3.connect(db)
cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]
assert cols == _EXPECTED_COLUMNS # 22 → 36 个 recorder 字段(+ created_at 共 37 物理列)
assert cols == _EXPECTED_COLUMNS # 22 → 38 个 recorder 字段(+ created_at 共 39 物理列)
rows = dict(conn.execute("SELECT call_id, tenant_id FROM llm_calls").fetchall())
assert rows["new-row"] == "tenant-a"
assert rows["old-row"] == "" # 不是 None: NULL 会被 RLS 静默吞掉
@@ -697,7 +826,7 @@ class TestSQLiteSchemaMode:
"""issue #13: `auto_migrate` 两档——auto 保持自动补列,manual 只裁剪写入不发 DDL。
列数断言一律按**物理列数**写: 旧表 22 个 INSERT 字段 + `created_at` = 23,
补齐后 36 + `created_at` = 37。混用 INSERT 字段数与物理列数是本处最易错的地方。
补齐后 38 + `created_at` = 39。混用 INSERT 字段数与物理列数是本处最易错的地方。
"""
def _physical_columns(self, db: Path) -> list[str]:
@@ -736,7 +865,7 @@ class TestSQLiteSchemaMode:
assert "ALTER TABLE" in message # 给出可直接执行的补列 SQL
async def test_auto_mode_still_upgrades_the_legacy_table(self, tmp_path):
"""auto + 同款旧表: 现状回归,补列后物理列数 23 → 37。"""
"""auto + 同款旧表: 现状回归,补列后物理列数 23 → 39。"""
db = tmp_path / "auto_legacy.db"
_make_pre_tenant_db(db)
@@ -745,10 +874,10 @@ class TestSQLiteSchemaMode:
recorder.close()
assert self._physical_columns(db) == _EXPECTED_COLUMNS
assert len(self._physical_columns(db)) == 37
assert len(self._physical_columns(db)) == 39
async def test_manual_mode_still_creates_a_fresh_table(self, tmp_path):
"""manual 只管 ALTER,不管 CREATE: 全新库照建,37 个物理列齐全(设计 §4.2)。"""
"""manual 只管 ALTER,不管 CREATE: 全新库照建,39 个物理列齐全(设计 §4.2)。"""
db = tmp_path / "manual_fresh.db"
recorder = SQLiteRecorder(db, auto_migrate=False)
await _record_minimal(recorder, call_id="c-fresh", tenant_id="tenant-a")
@@ -985,6 +1114,9 @@ class TestPostgresBackfillDiscipline:
"error_body",
"attempts",
"total_latency_ms",
# 1.3.8 两列: 稳态定义随补列前移,否则每进程首写都去抢锁
"stall_ms",
"stall_count",
]
def _recorder(self, conn):
@@ -3353,7 +3485,7 @@ class TestRecorderShapeGate:
class NarrowRecorder:
async def record_llm_call(self, *, call_id, model) -> None: ...
# 若闸内硬编码了 36 个名字,这里必然拒绝
# 若闸内硬编码了 38 个名字,这里必然拒绝
tele.TelemetryEmitter(NarrowRecorder(), scope="LLM", text_cap=None)