BiSheng 指标日志契约（BS_METRIC）全解：结构化埋点、监控层聚合与可观测性落地指南

BiSheng 指标日志契约（BS_METRIC）全解：结构化埋点、监控层聚合与可观测性落地指南

【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng

导读：本文围绕 BiSheng 开源 LLM DevOps 平台的指标日志契约（BS_METRIC）展开，系统讲解后端各进程（web / Celery / Linsight worker）如何以统一 marker + logfmt 格式输出结构化测量行，以及监控团队如何基于 ELK / Loki / ES 日志管线解析并聚合成 DB QPS、P95、存储成功率、模型 TTFT 等关键指标。读完本文，你将掌握BS_METRIC的完整行格式与字段字典、各指标算法的精确口径、Loki / ES 查询写法、后端开关配置，以及埋点背后的源码级实现原理与踩坑约束。

一、契约的定位：后端只打原始测量，聚合全在监控层

BS_METRIC是 BiSheng 给监控团队的解析依据，正式契约文档见 docs/observability/metric-log-contract.md，设计稿见 features/v2.6.0/042-metric-log-observability/design.md（F042）。

该方案有几个核心定位，理解它们才能正确使用这份契约：

• 只打原始测量：后端各进程向标准日志输出结构化行，P95 / QPS / 成功率等聚合计算全部在监控层（日志管线）完成，后端不引入prometheus_client、不建/metrics端点、不做进程内百分位计算。
• 多进程透明：FastAPI web 进程 + Celery workers + Linsight worker 各自独立运行，大量 DB 查询、模型调用、文件存储发生在 worker 内。日志方案天然对各进程透明——各写各的，采集端集中汇合。
• 数据流闭环：业务代码在四类"最细统一边界"（DB engine cursor 事件、MinIO 存储切面、模型调用 wrapper finally、E+ 站内信客户端）埋点 →emit_metric格式化输出 → loguru 打到 stdout / 日志文件 → 监控层采集、按domain解析、聚合出指标并上屏告警。

二、行格式与通用格式化规则

统一的行格式是marker + domain + logfmt 风格 key=value：

BS_METRIC domain=<域> key=value key=value ...

通用规则（源码实现在 src/backend/bisheng/common/services/metric_log.py 的_fmt_value，见 第 65-76 行）：

规则	说明
数值字段*_ms	单位恒为毫秒；float 保留 3 位小数（round(v, 3)），避免科学计数法
bool字段	渲染为1/0，而非True/False
含空格 / 引号 /=/ 换行 / 制表符 / 竖线的字符串	加双引号包裹，并转义反斜杠、双引号；换行与制表符替换为空格
None字段	省略不打，监控层解析时按缺省处理

采集端正则建议：

• 锚定BS_METRIC domain=；
• 用domain=的精确 token（后接空格或行尾）做分流，避免db_query误匹配db_query_agg。

对应的单元测试见 src/backend/test/metric_log/test_metric_log.py，其中_line_for正是用"前缀 + 空格或行尾"的边界匹配来区分db_query与db_query_agg（第 30-36 行），验证了契约中关于精确 token 分流的约定；test_emit_metric_bool_rendered_as_int、test_emit_metric_float_rounded_not_scientific、test_emit_metric_escapes_values_with_spaces_or_quotes分别覆盖布尔、浮点、转义规则。

三、字段字典：六大 domain 及触发时机

domain	触发时机	字段
db_query	单条 SQL 且elapsed_ms >= db_slow_query_ms（慢查询明细）；失败查询也打	op(SELECT/INSERT/UPDATE/DELETE/OTHER)elapsed_msstatus(ok/error)
db_query_agg	每进程每db_agg_window_s秒	window_scountsum_msle_5 le_10 le_25 le_50 le_100 le_250 le_500 le_1000 le_inf（累积桶计数，单位 ms）
db_pool	周期采样（由查询驱动）；池等待超时时	engine(sync/async)checked_outidlesizecapacityat_capacity(0/1)result(可选=wait_timeout)
obj_storage	每次上传/下载完成	op(put/get)result(ok/error/excluded)http_statuserr_codeelapsed_ms
model_invoke	每次模型调用结束	model_idstatus(success/failed)is_stream(0/1)ttft_mstotal_ms
eplus_notify	E+ 每次真实调用（ok/error）；forwarder 过白名单后的收件人 skip（skipped）	result(ok/error/skipped)http_statusbiz_codeerr_codeelapsed_msactionreason(skipped 时)

各 domain 的开关映射在源码中由_DOMAIN_SWITCH定义（metric_log.py 第 31-38 行）：db_query/db_query_agg/db_pool同受db开关控制，obj_storage、model_invoke、eplus_notify各自独立，未映射的新 domain 默认放行。

3.1 db_query / db_query_agg：慢查询明细 + 周期汇总双行制

DB 是超高频埋点域，逐条 SQL 打一行在高 QPS 下日志量不可接受，而"仅慢查询"又算不出 QPS 与完整 P95。因此采用双行制（设计文档决策 3）：

• db_query：只打超过db_slow_query_ms阈值的慢查询明细，同时失败查询无条件打一行（status=error）；
• db_query_agg：每进程每db_agg_window_s秒输出一行汇总，携带count（供 QPS）与延迟直方图桶计数（供 P95）。

调用链在record_db_query（metric_log.py 第 255-287 行）中实现：status="error"时只发错误明细行且不污染延迟直方图；成功时按阈值决定是否发明细行，随后无条件record进直方图、窗口期满 flush 出db_query_agg，并顺带采样连接池。

埋点接入点位于 src/backend/bisheng/core/database/connection.py 的_install_db_metric_events（第 22-61 行），通过 SQLAlchemy 通用事件监听（不依赖 MySQL / 达梦方言）：

• before_cursor_execute记录time.monotonic()起点；
• after_cursor_execute计算elapsed_ms并调用record_db_query(..., status="ok")；
• handle_error兜底打status="error"行；
• 同步 engine 与异步 engine（取其sync_engine）各自安装一次。

3.2 db_pool：饱和度 Gauge + 等待超时计数

SQLAlchemy不提供"当前排队等连接数"的公开 API，PoolEvents.checkout也在拿到连接之后才触发，因此用饱和度反推（设计决策 6）：周期采样pool.checkedout()/checkedin()/size()，capacity = size + max_overflow，checked_out >= capacity即at_capacity=1；session 上下文捕获 SQLAlchemyTimeoutError时额外打一行result=wait_timeout（connection.py 第 64-70 行、第 241 行）。

采样实现为maybe_emit_pool_gauge（metric_log.py 第 198-230 行）：由查询事件驱动、_PoolSampler保证每窗口最多一次；只读稳定的公开方法checkedout/checkedin/size（坑 10），max_overflow从DatabasePoolConf读取而非私有字段；异步池经sync_engine.pool采样（坑 9）；SQLite 的StaticPool无饱和度语义，检测到缺少checkedout属性时静默跳过。

3.3 obj_storage：成功率口径排除签证过期

对象存储的成功率口径特殊：401/403 签证/权限过期是预期内的客户端刷新场景，不算服务故障，单列result=excluded且不进失败分母；只有超时 / 5xx / 连接错误算result=error；NoSuchKey（对象不存在）视为存储正确响应，算ok。

分类逻辑在 src/backend/bisheng/core/storage/minio/minio_storage.py 的_STORAGE_EXCLUDED_CODES（第 95-97 行）与_classify_storage_exc（第 100-114 行）：

• NoSuchKey→ok；
• HTTP 状态 401 / 403 或 code 命中{AccessDenied, SignatureDoesNotMatch, ExpiredToken, InvalidAccessKeyId, TokenRefreshRequired}→excluded；
• 其余 S3 错误 →error（携带http_status与err_code）；
• 跨租户共享回退异常StorageSharingFallbackError→ok。

埋点以_metered("put" / "get")装饰器 +_storage_metric上下文管理器实现（第 117-170 行），同时支持同步与 async 方法，只包住真正执行 S3 调用的叶子方法，保证每次操作只计一次；http_status/err_code在存储边界内读取——因为 minio 的S3Error是 frozen dataclass，逃逸后重抛会掩盖真实异常（坑 6）。

3.4 model_invoke：复用已有 TTFT 采集

模型调用的 TTFT / status / is_stream在 F042 之前已采集并写 ES telemetry（MODEL_INVOKE），因此不重复造轮子，而是在 src/backend/bisheng/llm/domain/utils.py 的模型调用 wrapper finally 中并行补打一行日志（第 242-251 行）：

emit_metric( "model_invoke", model_id=self.model_id, status="success" if status == StatusEnum.SUCCESS else "failed", is_stream=is_stream, ttft_ms=first_token_cost_time, total_ms=(end_time - start_time) * 1000.0, )

其中ttft_ms是首 Token 延迟（毫秒），total_ms是完整调用耗时。

3.5 eplus_notify：只 meter 真实调用与低频 skip

E+ 站内信埋点有明确的高频短路排除约束（坑 12）：forwarder.maybe_forward_external对每条站内信都会调用，feature_disabled/not_in_whitelist分支几乎每条都命中，在这两个分支打指标会刷屏淹没真实事件——因此不打；只有两类打：

• 客户端每次真实调用（成功result=ok/ 失败result=error），见 src/backend/bisheng/notification/external/cofco_eplus_client.py（第 108-115 行 非 JSON 响应、第 128-135 行code=0成功、第 145-152 行 业务失败、第 167-174 行 异常兜底）；
• forwarder过白名单之后的收件人解析 skip（低频，仅 forwardable 消息才走到），见 src/backend/bisheng/notification/forwarder.py（第 145 行），带reason字段说明跳过原因。

四、示例行与le_*累积桶语义

BS_METRIC domain=db_query op=SELECT elapsed_ms=253.1 status=ok BS_METRIC domain=db_query_agg window_s=10 count=8421 sum_ms=41230 le_5=6100 le_10=7300 le_25=8000 le_50=8250 le_100=8360 le_250=8408 le_500=8418 le_1000=8420 le_inf=8421 BS_METRIC domain=db_pool engine=async checked_out=37 idle=63 size=100 capacity=120 at_capacity=0 BS_METRIC domain=obj_storage op=put result=ok http_status=200 elapsed_ms=45.2 BS_METRIC domain=obj_storage op=get result=error http_status=500 err_code=InternalError elapsed_ms=1203.4 BS_METRIC domain=model_invoke model_id=123 status=success is_stream=1 ttft_ms=340.0 total_ms=5200.0 BS_METRIC domain=eplus_notify result=ok http_status=200 biz_code=0 elapsed_ms=88 action=request_channel

le_*是累积桶（Prometheus histogram 语义）：le_10包含le_5的全部计数，le_inf == count。累积桶跨进程、跨窗口可直接相加，聚合后经histogram_quantile即可得到真实全局 P95。这正是"预算好的百分位跨进程不可合并、原始桶计数可合并"这一核心设计（决策 2）的落点。

桶边界定义于源码_BUCKETS_MS = (5, 10, 25, 50, 100, 250, 500, 1000)（metric_log.py 第 108 行），DbQueryHistogram.record用bisect_left实现"elapsed_ms <= bound落入该桶"的左闭语义（第 127-135 行），maybe_flush在窗口期满后按累积方式输出（第 137-159 行）。直方图与池采样器都是模块级单例（db_histogram、_pool_sampler），按进程聚合——因为sessionmaker每次调用都新建，计数器不能挂在 session 上（坑 7）。

五、指标算法口径（监控层计算指南）

设采集周期内某窗口跨所有进程汇总，各指标口径如下：

指标	口径
DB QPS	sum(db_query_agg.count) / 时间跨度秒
DB 查询耗时 P95 / P50 / P99	对db_query_agg各le_*桶按进程/窗口求和后histogram_quantile(0.95, buckets)
DB 慢查询 TopN / 错误率	db_query明细行按op分组；错误率 =count(status=error) / count(*)
连接池饱和度	db_pool.checked_out / capacity；排队告警 =at_capacity==1持续 N 个采样，或出现db_pool result=wait_timeout（池耗尽、等待超时）
存储成功率	count(result=ok) / (count(result=ok) + count(result=error))，按op(put/get) 分；result=excluded不进分母（401/403 签证过期）
存储耗时	对obj_storage.elapsed_ms原始样本按op分组算分位
模型 TTFT P95	对model_invoke.ttft_ms原始样本算分位；调用成功率 =count(status=success) / count(*)
E+ 调用事件	eplus_notify按result(ok/error/skipped) 分组计数；elapsed_ms分位；错误细分看biz_code/err_code

需要强调：模型、存储、E+ 是低中频域，打原始样本足够；DB 是超高频域，打直方图桶计数控制日志量。若未来存储/模型 QPS 上升导致日志量成为问题，设计文档也明确了可照 DB 的*_agg模式扩展。

六、查询示例

6.1 Loki（LogQL）

DB QPS（5 分钟速率）：

sum(rate({app="bisheng"} |= "BS_METRIC domain=db_query_agg" | logfmt | unwrap count [5m]))

存储成功率（put，排除 excluded）：

sum(count_over_time({app="bisheng"} |= "BS_METRIC domain=obj_storage" | logfmt | op="put" | result="ok" [5m])) / sum(count_over_time({app="bisheng"} |= "BS_METRIC domain=obj_storage" | logfmt | op="put" | result=~"ok|error" [5m]))

模型 TTFT P95：

quantile_over_time(0.95, {app="bisheng"} |= "BS_METRIC domain=model_invoke" | logfmt | unwrap ttft_ms [5m])

DB 查询 P95 需对le_*桶跨进程求和再做histogram_quantile；若管线支持，把db_query_agg的桶转成 Prometheus histogram 系列（每个le_x一条）后用histogram_quantile(0.95, sum by (le)(...))。

6.2 ES（聚合思路）

• 用 ingest / grok 把BS_METRIC domain=... k=v解析成结构化字段（domain、op、result、elapsed_ms…）；
• 成功率：filter result:ok / (result:ok OR result:error)；耗时分位：percentilesagg onelapsed_ms；
• DB P95：对db_query_agg的le_*求sumagg 后在可视化层做histogram_quantile（或转 Prometheus remote-write）。

七、后端开关与运维配置

config.yaml的metric_log段（默认全开），对应的配置模型MetricLogConf定义于 src/backend/bisheng/core/config/settings.py（第 665-689 行），并通过Settings.metric_log挂载到全局配置（第 756 行）：

键	默认	说明
enabled	true	总开关，关闭后不打任何BS_METRIC
db/obj_storage/model_invoke/eplus	true	各域独立开关
db_slow_query_ms	200	慢查询明细阈值（ms）；调小可看到更多db_query明细
db_agg_window_s	10	db_query_agg汇总 +db_pool采样窗口（秒）

开关判断逻辑在_domain_enabled（metric_log.py 第 54-62 行）：配置未加载前fail-open（默认放行避免启动早期丢指标），总开关关闭直接短路，再按 domain 映射查各域开关；record_db_query入口同样先做开关闸门（第 267-269 行），关闭的 domain 是零开销的。

运维提示：调小db_slow_query_ms（例如改为 1）可强制让每条 SQL 都产出db_query明细行，便于排查；生产环境则建议保持阈值，用db_query_agg承载全局 QPS / P95 计算。

八、契约变更约束

• 新增字段：向后兼容，可直接加；
• 重命名 / 删除字段、改domain、改 markerBS_METRIC：破坏性变更，必须通知监控团队同步解析/告警规则；
• 字段单位固定：*_ms恒为毫秒；le_*桶恒为累积计数。

对外契约一旦发布即被监控层解析规则依赖（设计文档 §2 关键约束），这也是_DOMAIN_SWITCH对未知新 domain 默认放行、emit_metric对None字段省略不打的原因——契约演进以"加字段"为主路径。

九、埋点实现的工程约束（源码级剖析）

emit_metric（metric_log.py 第 79-98 行）是唯一的输出出口，其工程约束对理解契约行为至关重要：

• 零阻塞、绝不破坏业务流：整个函数包在 try/except 中，任何异常都只以 debug 级记录（非静默吞掉）后返回，埋点失败绝不影响调用方；
• 无隐藏时钟：时间一律由调用方传入（now参数），保证直方图/池窗口的确定性并便于测试；
• 不依赖请求 / 租户 ContextVar：Celery / Linsight worker 内同样可用；
• 不记录 SQL 原文与 bind 参数（安全约束 C6 / 坑 11）：sql_op（第 240-252 行）只取 SQL 前缀首个关键词（SELECT/INSERT/UPDATE/DELETE，否则OTHER），语句体与参数值从不保留——既防泄密（密码、PII），也避免超大IN(...)参数序列化拖慢主流程。

测试覆盖上，除了前文提到的emit_metric格式化单测，DbQueryHistogram的桶归类与窗口 flush 计数正确性、存储 result 分类（401/403 边界）也都有对应用例，分布在 src/backend/test/metric_log/ 目录的test_metric_log.py、test_db_metric.py、test_storage_metric.py、test_model_metric.py、test_eplus_metric.py、test_e2e_metric_log.py中。

十、本地验证方法

手动验证步骤（设计文档 §7）：

1. 本地启动后端：cd src/backend && uv run uvicorn bisheng.main:app；
2. 制造流量后过滤日志：grep 'BS_METRIC domain=' <stdout / 日志文件>；
3. 分域确认：
   • domain=db_query_agg：每db_agg_window_s秒一行、桶计数非零 → 可算 QPS / P95；
   • domain=db_pool：采样字段齐全；
   • 上传/下载文件看domain=obj_storage；
   • 跑一次模型调用看domain=model_invoke（与既有 ES telemetry 并存）；
   • 触发一次站内信看domain=eplus_notify；
4. 将db_slow_query_ms调小（如 1）可强制产出db_query慢查询明细行。

十一、已知边界与后续演进

从设计文档可以明确该方案的边界，避免误用：

• 不做：进程内/metrics端点、Grafana 看板、告警规则——这些全是监控层职责；
• 不做：存储 / 模型也走周期直方图汇总（当前低中频、原始样本足够）；
• 不做：连接池精确等待时长wait_ms全分布与"此刻几个在排队"的精确计数（当前饱和度 Gauge + 等待超时计数足够，不够时再子类化QueuePool._do_get）；
• 触发重写：若企业无集中日志管线、要求进程内暴露指标端点，则迁移到 Prometheus multiproc 方案；
• 采集盲区：任何绕过统一边界的调用路径（不经 engine 的裸 DBAPI 连接、不经get_minio_storage()直接 new minio client、旁路 wrapper 的模型调用）不会被采集，新增调用路径时需注意收敛回统一边界。

【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng