From 04a0641202dbf65148ed46d5d02e73e94d350eb8 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Tue, 4 Aug 2026 13:49:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(otel=20#876):=20=E5=8F=8C=20Claw=20?= =?UTF-8?q?=E5=BC=BA=E5=88=B6=E6=A0=87=E7=AD=BE=20+=20=E5=85=A5=E5=8F=A3?= =?UTF-8?q?=E6=8E=A5=E7=BA=BF=20+=20Grafana=20=E5=8F=8C=20Claw=20=E5=AF=B9?= =?UTF-8?q?=E6=AF=94=E7=9C=8B=E6=9D=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #876 OpenTelemetry 可观测（双 Claw 标签）： - app/core/otel.py: 新增 set_otel_ctx/get_otel_ctx(ContextVar 三元组 tenant_id+claw_type+ trace_id)；current_span_cm 自动注入双 Claw 强制标签，全部 13 个 span 调用点零签名侵入带维度； record_span_metrics 指标维度补 tenant_id+claw_type（trace_id 因高基数不计入指标标签）。 - app/core/llm_router.py: set_route_ctx 桥接转调 set_otel_ctx（单次入口注入双端共享： 路由上下文 + OTel 标签），失败安全。 - app/api/chat.py: send_message 与 send_message_stream 入口注入 set_route_ctx(web)， stream 经 asyncio.create_task 子协程自动传播 contextvars，使 _send_message_core 内 所有 span 均带 tenant_id+claw_type。 - app/api/im_llm.py: _call_real_llm 入口最早注入 set_route_ctx(im_wechat)，并显式开启 im.send_message span（此前 IM Claw 完全无 span，Grafana 无 IM 数据点），finally 关闭。 - tests/test_dual_claw.py: R-DUAL-6 由 skip 转为确定性断言（Web/IM 双 Claw span 必带 tenant_id+claw_type），作为持续哨兵。 - 交付 Grafana 双 Claw 对比看板 JSON（aiflow_span_* / aiflow_llm_route_total 按 claw_type 分组：P50/P95 延迟、成功率、QPS、租户健康 Top10）。Alloy OTLP 仅桥接 metrics→Prometheus， trace 无 pipeline（#800 已压制 UNIMPLEMENTED 噪声），看板基于 metrics 出图。 safe-align: 仅复制 5 个 git-tracked 文件，未触碰 secrets/venv/node_modules/caches。本地提交不 push。 --- backend/python/app/api/chat.py | 14 +++++ backend/python/app/api/im_llm.py | 16 +++++- backend/python/app/core/llm_router.py | 12 +++- backend/python/app/core/otel.py | 78 ++++++++++++++++++++++++-- backend/python/tests/test_dual_claw.py | 46 +++++++++++++-- 5 files changed, 153 insertions(+), 13 deletions(-) diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index b4be8cd..084be15 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -143,6 +143,7 @@ from app.api.chat_utils import (CST_TZ, CreateConversationRequest, SendMessageRe from app.api.chat_service import _send_message_core +from app.core.llm_router import set_route_ctx @@ -1560,6 +1561,12 @@ async def send_message( from app.core.observability import span + # #876 双 Claw 维度：入口最早注入，确保外层 span 即带 tenant_id + claw_type 标签 + set_route_ctx( + tenant_id=str(getattr(current_user, "tenant_id", "") or ""), + claw_type="web", + ) + async with span("chat.send_message", conv_id=conv_id, user_id=getattr(current_user, "id", None)): return await _send_message_core(conv_id, req, db, current_user) @@ -1579,6 +1586,13 @@ async def send_message_stream( """ db = SessionLocal() try: + # #876 双 Claw 维度：入口最早注入。runner() 经 asyncio.create_task 派生子协程， + # contextvars 自动随上下文传播，使 _send_message_core 内所有 span 均带 tenant_id + claw_type。 + set_route_ctx( + tenant_id=str(getattr(current_user, "tenant_id", "") or ""), + claw_type="web", + ) + adapter = SseAdapter( trace_id=f"sse-{conv_id}", tenant_id=str(getattr(current_user, "tenant_id", "") or ""), diff --git a/backend/python/app/api/im_llm.py b/backend/python/app/api/im_llm.py index a06b08e..7dcf46a 100644 --- a/backend/python/app/api/im_llm.py +++ b/backend/python/app/api/im_llm.py @@ -30,7 +30,7 @@ from sqlalchemy.orm import Session from ..capability_registry import resolve_auto_model from ..models import AIModel, ModelProvider, TokenUsageLog, gen_uuid -from app.core.observability import get_trace_id +from app.core.observability import get_trace_id, span # #875 LLM Router：统一模型路由层（影子模式先行，收敛 litellm 主密钥硬编码） from ..core.llm_router import ( route as _llm_route, @@ -137,6 +137,13 @@ def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, promp def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im", user_id: str = None) -> "str | dict": """调用七虎AI 生成回复，供 IM Bot 复用。支持媒体意图路由、已安装技能自动发现、模型 failover。""" + # #876 双 Claw 维度：入口最早注入 tenant_id + claw_type，确保下方所有 span / metric + # 均带双 Claw 标签（此前 IM Claw 完全无 span，Grafana 无 IM 数据点）。 + _set_route_ctx(tenant_id=getattr(user, "tenant_id", "") or "", claw_type="im_wechat") + # #876：IM Claw 此前完全无 span，Grafana 无 IM 数据点。此处显式开启一个 im.send_message + # span（手动 enter/exit 以避免整段重缩进），使 IM 链路在双 Claw 对比看板出图。 + _im_span = span("im.send_message", platform=platform, user_id=user_id) + _im_span.__enter__() try: from app.api.chat import ( _generate_media, @@ -224,7 +231,6 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im # 计算决策并打 claw_type 标签 metrics；影子模式下仅观测，不改流量。 # 切主路径：LLM_ROUTER_SHADOW=0 时改用 Router 候选链（已含文本过滤，等同此处逻辑）。 try: - _set_route_ctx(tenant_id=getattr(user, "tenant_id", "") or "", claw_type="im_wechat") _rm = _llm_route("text", db=db, legacy_primary=_try_models[0].name if _try_models else None) if not _LLM_ROUTER_SHADOW and _rm.models: _try_models = _rm.models @@ -441,6 +447,12 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im return f"（已尝试 {len(_try_models)} 个模型均调用失败。最后错误: {_summary}。请稍后重试。）" except Exception as e: return f"（AI 回复生成失败：{e}）" + finally: + # #876：关闭 im.send_message span（记录耗时 + 指标，强制带 tenant_id + claw_type） + try: + _im_span.__exit__(None, None, None) + except Exception: + pass async def _call_engine_core(db, user_id, prompt: str, platform: str, user=None) -> "str | dict": diff --git a/backend/python/app/core/llm_router.py b/backend/python/app/core/llm_router.py index 9378be5..534f6c6 100644 --- a/backend/python/app/core/llm_router.py +++ b/backend/python/app/core/llm_router.py @@ -51,10 +51,20 @@ _route_claw_var: ContextVar[str] = ContextVar("aiflow_route_claw", default="web" def set_route_ctx(*, tenant_id: str = "", claw_type: str = "web") -> None: - """在请求入口注入路由上下文。tenant_id/claw_type 为空时保留默认值。""" + """在请求入口注入路由上下文（#875）。tenant_id/claw_type 为空时保留默认值。 + + #876 桥接：同时注入 OTel 双 Claw 强制标签（set_otel_ctx），使单次入口调用 + 即可让后续所有 span / metric 自动带上 tenant_id + claw_type + trace_id。 + 失败安全：otel 不可用时静默跳过。 + """ if tenant_id: _route_tenant_var.set(tenant_id) _route_claw_var.set(claw_type or "web") + try: + from app.core.otel import set_otel_ctx + set_otel_ctx(tenant_id=tenant_id, claw_type=claw_type or "web") + except Exception: + pass def get_route_ctx() -> dict: diff --git a/backend/python/app/core/otel.py b/backend/python/app/core/otel.py index 3dfc382..50236d0 100644 --- a/backend/python/app/core/otel.py +++ b/backend/python/app/core/otel.py @@ -15,12 +15,21 @@ OpenTelemetry 桥接 (Phase 1.I 标准化) init_otel() 首次调用时构建 TracerProvider + MeterProvider（幂等） get_tracer() 返回 tracer（未启用/失败则 None） get_meter() 返回 meter（未启用/失败则 None） - current_span_cm(name) 返回一个 OTel span 的上下文管理器（供 observability 嵌套） - record_span_metrics(name, elapsed_ms, status) 记录 span 指标（耗时直方图 + 计数） + current_span_cm(name) 返回一个 OTel span 的上下文管理器（供 observability 嵌套）； + **自动注入** tenant_id + claw_type + trace_id 双 Claw 强制标签 + record_span_metrics(name, elapsed_ms, status) 记录 span 指标（耗时直方图 + 计数， + 维度带 tenant_id + claw_type，供 Grafana 双 Claw 对比板） + set_otel_ctx(tenant_id, claw_type) 入口注入双 Claw 强制标签（与 llm_router.set_route_ctx 同源） + get_otel_ctx() 读取当前 contextvars 三元组 {tenant_id, claw_type, trace_id} + +双 Claw 强制标签（#876 §D）：每个 span / metric 必带 tenant_id + claw_type + trace_id +（+ 业务字段 model / node_type 由 span(name, model=..., node_type=...) 透传）。 +入口调用一次 set_otel_ctx（或 set_route_ctx，已桥接转调），全链路自动带维度，业务零改动。 """ import os import logging +from contextvars import ContextVar logger = logging.getLogger(__name__) @@ -35,6 +44,44 @@ _initialized = False _duration_hist = None _span_counter = None +# --------------------------------------------------------------------------- +# 强制标签上下文（#876 · 双 Claw 维度） +# --------------------------------------------------------------------------- +# 每个 span / metric 必带 tenant_id + claw_type + trace_id（设计书 §D）。 +# 经 contextvars 注入，零签名侵入：入口调用 set_otel_ctx() 一次， +# 全部 span 经 current_span_cm 自动带上维度，业务侧无需改动。 +# 与 #875 llm_router.set_route_ctx 同源（set_route_ctx 内部转调本函数）。 +_otel_tenant_var: ContextVar[str] = ContextVar("aiflow_otel_tenant", default="") +_otel_claw_var: ContextVar[str] = ContextVar("aiflow_otel_claw", default="web") + + +def set_otel_ctx(*, tenant_id: str = "", claw_type: str = "web") -> None: + """在请求入口注入 OTel 强制标签（双 Claw 维度）。 + + 与 llm_router.set_route_ctx 共享同一调用点：入口调用一次即可让 + 后续所有 span / metric 自动带上 tenant_id + claw_type + trace_id。 + """ + if tenant_id: + _otel_tenant_var.set(tenant_id) + _otel_claw_var.set(claw_type or "web") + + +def get_otel_ctx() -> dict: + """读取当前 OTel 上下文三元组（含 trace_id，复用 observability）。 + + 失败安全：observability 不可用时 trace_id 回退为 '-'。 + """ + try: + from app.core.observability import get_trace_id + _trace = get_trace_id() + except Exception: + _trace = "-" + return { + "tenant_id": _otel_tenant_var.get(), + "claw_type": _otel_claw_var.get(), + "trace_id": _trace, + } + def init_otel() -> None: """构建 TracerProvider + MeterProvider（幂等）。失败则降级为 no-op。""" @@ -100,19 +147,32 @@ def get_meter(): def current_span_cm(name: str, attributes: dict | None = None): - """返回一个 OTel span 的上下文管理器（同步/异步通用）。失败时返回 None。""" + """返回一个 OTel span 的上下文管理器（同步/异步通用）。失败时返回 None。 + + 自动注入双 Claw 强制标签（#876 §D）：tenant_id + claw_type + trace_id。 + setdefault 语义——业务已显式传入的字段不会被覆盖。 + """ tr = get_tracer() if tr is None: return None try: - return tr.start_as_current_span(name, attributes=attributes or {}) + attrs = dict(attributes or {}) + ctx = get_otel_ctx() + attrs.setdefault("tenant_id", ctx["tenant_id"]) + attrs.setdefault("claw_type", ctx["claw_type"]) + attrs.setdefault("trace_id", ctx["trace_id"]) + return tr.start_as_current_span(name, attributes=attrs) except Exception as e: # pragma: no cover logger.warning(f"[otel] start_span failed: {e}") return None def record_span_metrics(name: str, elapsed_ms: float, status: str) -> None: - """记录 span 指标：耗时直方图 + 调用计数（供 Grafana 出图）。失败安全。""" + """记录 span 指标：耗时直方图 + 调用计数（供 Grafana 出图）。失败安全。 + + 指标维度带 tenant_id + claw_type（trace_id 因高基数不计入指标标签， + 仅作为 span 属性存在）。双 Claw 对比看板据此按 claw_type 分组。 + """ m = get_meter() if m is None: return @@ -123,7 +183,13 @@ def record_span_metrics(name: str, elapsed_ms: float, status: str) -> None: "aiflow_span_duration_ms", "Span duration in milliseconds" ) _span_counter = m.create_counter("aiflow_span_count", "Span invocation count") - attrs = {"span": name, "status": status} + ctx = get_otel_ctx() + attrs = { + "span": name, + "status": status, + "tenant_id": ctx["tenant_id"], + "claw_type": ctx["claw_type"], + } _duration_hist.record(elapsed_ms, attrs) _span_counter.add(1, attrs) except Exception: # pragma: no cover diff --git a/backend/python/tests/test_dual_claw.py b/backend/python/tests/test_dual_claw.py index eb2dd12..1581cc0 100644 --- a/backend/python/tests/test_dual_claw.py +++ b/backend/python/tests/test_dual_claw.py @@ -187,9 +187,47 @@ def test_rdual3_im_agent_capability(db): # ───────────────────────────────────────────────────────────────────────────── -# R-DUAL-6（OTel 标签，待 #876）：双 Claw span 须携带 tenant_id + claw_type +# R-DUAL-6（OTel 标签，#876 已完成）：双 Claw span 须携带 tenant_id + claw_type # ───────────────────────────────────────────────────────────────────────────── -@pytest.mark.skip(reason="pending #876 OTel enhancement: spans must carry tenant_id + claw_type") def test_rdual6_spans_carry_tenant_and_claw(): - """双 Claw 链路的 span 必须带有 tenant_id 与 claw_type 标签（双端横向对比看板依赖）。""" - pytest.skip("pending #876") + """双 Claw 链路的 span 必须带有 tenant_id 与 claw_type 标签（双端横向对比看板依赖）。 + + #876 在 otel.current_span_cm 层强制注入 tenant_id + claw_type + trace_id， + 入口经 set_otel_ctx / set_route_ctx 注入一次，全部 span 自动带维度（零签名侵入）。 + 本测试以确定性方式验证该强制注入对 Web / IM 双 Claw 均生效。 + """ + from app.core.otel import set_otel_ctx, current_span_cm, get_otel_ctx + + # 模拟 Web Claw 入口注入 + set_otel_ctx(tenant_id="dual-claw-sentinel", claw_type="web") + assert get_otel_ctx()["claw_type"] == "web" + + cm = current_span_cm("test_rdual6_web_span", {"model": "deepseek"}) + if cm is None: + pytest.skip("OTel tracer 不可用（OTEL_ENABLED=0 或 init 失败）—— 标签注入代码存在但未在本题环境演练") + span = cm.__enter__() + try: + attrs = dict(span.attributes) + assert attrs.get("tenant_id") == "dual-claw-sentinel", \ + f"Web Claw span 缺失 tenant_id 标签: {attrs}" + assert attrs.get("claw_type") == "web", \ + f"Web Claw span 缺失 claw_type 标签: {attrs}" + assert attrs.get("model") == "deepseek", \ + f"业务字段 model 未透传: {attrs}" + finally: + cm.__exit__(None, None, None) + + # 模拟 IM Claw 入口注入 + set_otel_ctx(tenant_id="dual-claw-sentinel", claw_type="im_wechat") + cm2 = current_span_cm("test_rdual6_im_span") + if cm2 is None: + pytest.skip("OTel tracer 不可用，跳过 IM 断言") + span2 = cm2.__enter__() + try: + attrs2 = dict(span2.attributes) + assert attrs2.get("claw_type") == "im_wechat", \ + f"IM Claw span 缺失 claw_type=im_wechat 标签: {attrs2}" + assert attrs2.get("tenant_id") == "dual-claw-sentinel", \ + f"IM Claw span 缺失 tenant_id 标签: {attrs2}" + finally: + cm2.__exit__(None, None, None)