rakuten-api/app/shared/telemetry.py

"""OpenTelemetry traces 接入：仅在 otel_enabled=true 时初始化，否则全 noop

抓取与交易两个进程在 lifespan 启动时各自调用 `setup_telemetry(settings,
service_name=...)`：注册 TracerProvider + OTLP/HTTP exporter + 自动
instrumentation（FastAPI、httpx）。失败时（如 endpoint 不可达）不阻断主流程，
仅打日志；traces 是辅助观测，不应让进程起不来。

`shutdown_telemetry` 在 lifespan 关闭时 force_flush 后再 shutdown，确保缓冲区
里的 span 都已上报。

OTel SDK 默认的 ProxyTracerProvider 在 setup 之前就能用（noop span），所以
其它代码里直接 `trace.get_tracer(__name__)` + `start_as_current_span` 即可，
不必关心 telemetry 是否启用——禁用时 span 不会真正产生与上报。

自动 instrumentation（FastAPI + httpx）只覆盖「进程收到 HTTP 请求」与「进程发出
httpx 请求」两类边界。交易侧的实际工作两者都不是：站点交互走 Playwright（不经
httpx），worker 主循环是后台 asyncio 任务（没有 HTTP 入口）。所以那一侧必须手工
埋点，否则 trace 里只剩 worker 与网关之间的往返记录，看不到任何业务链路。本模块
为此提供三件东西：

- `traced`：给 async 方法套一层 span，异常自动记录（站点交互各步骤在用）
- `set_attributes` / `record_error`：批量写属性、统一记异常
- `suppressed`：屏蔽空转长轮询产生的孤立 trace
"""
from __future__ import annotations

import functools
import logging
from collections.abc import Awaitable, Callable, Iterator, Mapping
from contextlib import contextmanager
from typing import TYPE_CHECKING, ParamSpec, TypeVar

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.utils import suppress_instrumentation
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.trace.sampling import ALWAYS_ON
from opentelemetry.trace import Span, SpanKind, Status, StatusCode
from opentelemetry.util.types import AttributeValue

from app.shared.config import Settings, get_settings

if TYPE_CHECKING:
    from fastapi import FastAPI

P = ParamSpec("P")
R = TypeVar("R")

logger = logging.getLogger(__name__)

# 全局 provider 引用，用于 instrument_app / shutdown 时判断当前是否已初始化。
# 显式持有比依赖 trace.get_tracer_provider() 的类型判断更稳——后者在测试场景
# 下可能被其它用例改动全局状态。
_provider: TracerProvider | None = None


def setup_telemetry(settings: Settings, *, service_name: str) -> None:
    """初始化 OTel：TracerProvider + OTLP exporter + httpx 自动 instrumentation。

    - `otel_enabled=False` 或 endpoint 未配时仅打日志，不做任何事。
    - 必须在创建任何 httpx.AsyncClient 之前调用，否则 httpx 不会被打桩。
      两侧 main.py 的 lifespan 已把 setup 放在 site_session.start() 之前。
    - 重复调用安全（_provider 已设时直接返回）。
    """
    global _provider
    if _provider is not None:
        return
    if not settings.otel_enabled or not settings.otel_endpoint:
        logger.info("OpenTelemetry 未启用（endpoint 或 otel_enabled 未配置）")
        return

    resource = Resource.create({SERVICE_NAME: service_name})
    provider = TracerProvider(resource=resource, sampler=ALWAYS_ON)

    exporter = OTLPSpanExporter(
        endpoint=settings.otel_endpoint,
        headers=_parse_headers(settings.otel_headers),
        timeout=10,
    )
    provider.add_span_processor(
        BatchSpanProcessor(
            exporter,
            schedule_delay_millis=settings.otel_export_interval_ms,
        )
    )
    trace.set_tracer_provider(provider)
    _provider = provider

    # httpx 是抓取/交易两侧唯一的外部 HTTP 客户端，打桩后所有 AsyncClient 请求
    # 自动产生 CLIENT span。失败不影响主链路：进程仍可运行，只是看不到 span。
    try:
        HTTPXClientInstrumentor().instrument()
    except Exception:
        logger.warning("httpx 自动 instrumentation 失败", exc_info=True)

    logger.info(
        "OpenTelemetry 已启用：endpoint=%s service=%s",
        settings.otel_endpoint,
        service_name,
    )


def instrument_app(app: "FastAPI") -> None:
    """FastAPI 应用打桩。必须在应用开始服务之前调用，与 setup_telemetry 的先后无关。

    **不能用 `_provider is None` 做前置判断**：三个服务都在模块导入时执行
    `app = create_app()`，而 `setup_telemetry` 要等 lifespan 启动才跑，那时
    `_provider` 还是 None——照着判断就会直接 return，FastAPI 永远没被打桩，
    观测后台里一条 server span 都不会有。

    反过来「等 lifespan 里再打桩」也不行：instrument_app 是往应用上加中间件，
    应用一旦开始服务，加进去的中间件不生效（实测 lifespan 内调用后 server span
    为空）。所以只能在这里、在导入期就装上。

    provider 尚未设置时拿到的是 ProxyTracer，它在 `set_tracer_provider` 之后会
    自动委托到真实 provider（实测：导入期打桩 + lifespan 内设 provider，请求
    照样产生 span），所以顺序不构成问题。

    otel 关闭时跳过：省掉一层用不上的中间件。
    """
    if not get_settings().otel_enabled:
        return
    FastAPIInstrumentor.instrument_app(app)


def shutdown_telemetry() -> None:
    """flush + shutdown；幂等，未初始化时直接返回。"""
    global _provider
    if _provider is None:
        return
    try:
        _provider.force_flush()
        _provider.shutdown()
    except Exception:
        logger.debug("关闭 OpenTelemetry provider 失败", exc_info=True)
    _provider = None


def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None:
    """把 HTML 作为 span event 上报，超 max_bytes 截断并标注。

    用于解析失败时复现页面：span 自身只放结构化指标（items 数、source 等），
    完整 HTML 体量大、含商品/价格内容，仅在失败分支通过 event 携带。
    """
    if html is None or not html:
        return
    original_bytes = len(html)
    truncated = original_bytes > max_bytes
    payload = html if not truncated else html[:max_bytes]
    attributes: dict[str, AttributeValue] = {
        "snapshot.html": payload,
        "snapshot.original_bytes": original_bytes,
    }
    if truncated:
        attributes["snapshot.truncated"] = True
    span.add_event(name, attributes=attributes)


@contextmanager
def suppressed() -> Iterator[None]:
    """在这个上下文里不产生任何自动 instrumentation span。

    给「空转的长轮询」用：worker 每 30 秒问一次网关有没有活干，绝大多数时候
    返回空。这些请求各自成为一条孤立 trace，量大且没有信息量——把观测后台刷满
    的正是它们。领到任务后的每一次网关调用都在任务根 span 底下，不受影响。
    """
    with suppress_instrumentation():
        yield


def record_error(span: Span, exc: BaseException) -> None:
    """把异常记到 span 上并置 ERROR 状态。

    单独抽出来是因为 AppError 带 `err_code`（对外错误码），排查时按码筛比按
    异常类名筛更贴近上游看到的东西，值得单独落一个属性。
    """
    span.record_exception(exc)
    span.set_attribute("error.type", type(exc).__name__)
    err_code = getattr(exc, "err_code", None)
    if isinstance(err_code, int):
        span.set_attribute("error.code", err_code)
    span.set_status(Status(StatusCode.ERROR, f"{type(exc).__name__}: {exc}"))


def traced(
    name: str,
    *,
    kind: SpanKind = SpanKind.INTERNAL,
) -> Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]:
    """给 async 方法套一层 span，异常自动记录后原样抛出。

    交易侧的实际工作是 Playwright 页面操作，httpx 自动 instrumentation 完全看不到
    （浏览器请求不走 httpx），所以这些步骤必须手工埋点，否则 trace 里只剩 worker
    与网关之间的 HTTP 往返。用装饰器而不是在每个方法里写 with 块，是因为这些方法
    的函数体都已经很长，再加一层缩进不利于阅读。

    未启用 telemetry 时 tracer 是 noop，装饰器只多一次函数调用，可以无条件套。
    """

    def decorate(fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        @functools.wraps(fn)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            tracer = trace.get_tracer(fn.__module__)
            with tracer.start_as_current_span(name, kind=kind) as span:
                try:
                    return await fn(*args, **kwargs)
                except Exception as exc:
                    record_error(span, exc)
                    raise

        return wrapper

    return decorate


def set_attributes(span: Span, attributes: Mapping[str, AttributeValue | None]) -> None:
    """批量设置属性，跳过 None 值。

    站点交互里大量字段是可选的（site_order_id 要到提交后才有、payable_yen 只在
    确认页解析后才有），逐个 if 判断会把埋点代码写得比业务逻辑还长。
    """
    for key, value in attributes.items():
        if value is not None:
            span.set_attribute(key, value)


def _parse_headers(raw: str | None) -> list[tuple[str, str]] | None:
    """解析 "k1=v1,k2=v2" 形式的 header 配置；空输入返回 None。"""
    if not raw or not raw.strip():
        return None
    pairs: list[tuple[str, str]] = []
    for chunk in raw.split(","):
        key, sep, value = chunk.partition("=")
        if not sep or not key.strip():
            continue
        pairs.append((key.strip(), value.strip()))
    return pairs or None


def is_initialized() -> bool:
    """供测试断言使用。"""
    return _provider is not None