""" OpenTelemetry 桥接 (Phase 1.I 标准化) ===================================== 设计基准(标准对照表见 STANDARDS_AUDIT_2026-07-24.md §I): - 标准:OpenTelemetry SDK + OTLP gRPC exporter —— 云原生可观测事实标准。 - traces:span 经 observability.span() 嵌套导出(见 observability.py)。 - metrics:MeterProvider 经 OTLP 导出 span 耗时直方图 + 调用计数,供 Grafana 出图。 - 失败安全(生产红线):若 OTLP 端点不可达 / 初始化异常,自动降级为 no-op, 绝不阻断业务请求、不产生未捕获异常。 - 门控:OTEL_ENABLED=1 默认开启;置 0 则零开销、不走 OTel。 - 发射目标:OTEL_EXPORTER_OTLP_ENDPOINT,默认 http://qihu-aiflow-alloy:4317 (经 Grafana Alloy 桥接 OTLP→Prometheus,再由 Prometheus 喂给 Grafana 出图)。 对外 API: init_otel() 首次调用时构建 TracerProvider + MeterProvider(幂等) get_tracer() 返回 tracer(未启用/失败则 None) get_meter() 返回 meter(未启用/失败则 None) 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__) _ENABLED = os.getenv("OTEL_ENABLED", "1") not in ("0", "false", "False", "") _ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://qihu-aiflow-alloy:4317").rstrip("/") if not _ENDPOINT.startswith("http://") and not _ENDPOINT.startswith("https://"): _ENDPOINT = "http://" + _ENDPOINT _tracer = None _meter = None _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。""" global _tracer, _meter, _initialized if _initialized: return _initialized = True if not _ENABLED: logger.info("[otel] disabled (OTEL_ENABLED=0)") return try: from opentelemetry import trace, metrics from opentelemetry.sdk.resources import Resource from opentelemetry.semconv.resource import ResourceAttributes from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( OTLPSpanExporter, ) from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( OTLPMetricExporter, ) resource = Resource.create({ResourceAttributes.SERVICE_NAME: "qihu-aiflow"}) # --- traces --- trace_provider = TracerProvider(resource=resource) trace_provider.add_span_processor( BatchSpanProcessor(OTLPSpanExporter(endpoint=_ENDPOINT, insecure=True)) ) trace.set_tracer_provider(trace_provider) _tracer = trace.get_tracer("qihu-aiflow") # --- metrics --- metric_reader = PeriodicExportingMetricReader( OTLPMetricExporter(endpoint=_ENDPOINT, insecure=True) ) meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader]) metrics.set_meter_provider(meter_provider) _meter = metrics.get_meter("qihu-aiflow") logger.info(f"[otel] tracer+meter ready -> {_ENDPOINT}") except Exception as e: # pragma: no cover - 降级保护 logger.warning(f"[otel] init failed, degraded to no-op: {e}") _tracer = None _meter = None def get_tracer(): """返回 tracer;首次调用触发 init。未启用/失败返回 None。""" if _tracer is None and _ENABLED: init_otel() return _tracer def get_meter(): """返回 meter;首次调用触发 init。未启用/失败返回 None。""" if _meter is None and _ENABLED: init_otel() return _meter def current_span_cm(name: str, attributes: dict | None = None): """返回一个 OTel span 的上下文管理器(同步/异步通用)。失败时返回 None。 自动注入双 Claw 强制标签(#876 §D):tenant_id + claw_type + trace_id。 setdefault 语义——业务已显式传入的字段不会被覆盖。 """ tr = get_tracer() if tr is None: return None try: 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 出图)。失败安全。 指标维度带 tenant_id + claw_type(trace_id 因高基数不计入指标标签, 仅作为 span 属性存在)。双 Claw 对比看板据此按 claw_type 分组。 """ m = get_meter() if m is None: return global _duration_hist, _span_counter try: if _duration_hist is None: _duration_hist = m.create_histogram( "aiflow_span_duration_ms", "Span duration in milliseconds" ) _span_counter = m.create_counter("aiflow_span_count", "Span invocation count") 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 pass