qihuaiflow/backend/python/app/core/otel.py

"""
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 嵌套）
    record_span_metrics(name, elapsed_ms, status)  记录 span 指标（耗时直方图 + 计数）
"""

import os
import logging

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


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。"""
    tr = get_tracer()
    if tr is None:
        return None
    try:
        return tr.start_as_current_span(name, attributes=attributes or {})
    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 出图）。失败安全。"""
    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")
        attrs = {"span": name, "status": status}
        _duration_hist.record(elapsed_ms, attrs)
        _span_counter.add(1, attrs)
    except Exception:  # pragma: no cover
        pass