rakuten-api/tests/test_telemetry.py

"""telemetry 模块测试

验证：
1. 默认配置（otel_enabled=False）下 setup 是 noop，不初始化任何 provider。
2. enabled=true + endpoint 时 setup 注册真实 TracerProvider；shutdown 复位。
3. instrument_app 在 setup 之前调用也要真的装上中间件（三个服务都是导入期打桩）。
4. traced / set_attributes / record_error 的行为。
5. 「采集返回结果」这一侧：record_envelope（信封失败在 HTTP 层看不出来）、
   record_parse_failure（失败在 fetch 还是 parse）、snapshot（失败页面快照），
   以及 span_unless_suppressed 与 suppressed() 的配套关系。

不打真实网络：OTLPSpanExporter 创建时不发请求，BatchSpanProcessor 异步批量
上报在没有 span 产生时也不会触发。span 断言用独立的 InMemory provider，不碰
全局 provider——OTel 的全局 provider 只允许设置一次，测试间共享会互相污染。
"""
from __future__ import annotations

import pytest
from fastapi import FastAPI
from opentelemetry import trace
from opentelemetry.instrumentation import fastapi as otel_fastapi
from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import StatusCode
from opentelemetry.util.http import parse_excluded_urls

from app.shared import telemetry
from app.shared.config import Settings
from app.shared.errors import OrderGuardError, ScrapeParseError, UpstreamBlockedError
from app.shared.telemetry import is_initialized, setup_telemetry, shutdown_telemetry


@pytest.fixture
def spans(monkeypatch) -> InMemorySpanExporter:
    """把 telemetry 内部取到的 tracer 换成写内存的，用于断言 span

    不用全局 provider：`trace.set_tracer_provider` 只生效一次，一旦别的用例先
    设过，这里再设就被忽略（OTel 只打一条 warning），断言会莫名其妙拿不到 span。
    """
    exporter = InMemorySpanExporter()
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    monkeypatch.setattr(telemetry.trace, "get_tracer", provider.get_tracer)
    return exporter


def test_disabled_is_noop():
    """otel_enabled=False 时 setup/instrument/shutdown 都不初始化 provider"""
    settings = Settings(_env_file=None)  # 默认 otel_enabled=False
    setup_telemetry(settings, service_name="test-disabled")
    assert not is_initialized()


def test_instrument_app_works_before_setup_telemetry(monkeypatch):
    """回归：instrument_app 必须在 setup_telemetry 之前也能真的装上中间件

    三个服务都在模块导入时执行 `app = create_app()`（内部调 instrument_app），
    而 setup_telemetry 要等 lifespan 才跑。曾经这里用 `_provider is None` 做前置
    判断，于是导入期一律 return，FastAPI 从来没被打桩过——一条 server span 都
    没有。这条用例把那个顺序钉住。
    """
    monkeypatch.setattr(
        telemetry, "get_settings", lambda: Settings(_env_file=None, otel_enabled=True)
    )
    assert not is_initialized()  # 尚未 setup，正是导入期的状态

    app = FastAPI()
    try:
        telemetry.instrument_app(app)
        assert app._is_instrumented_by_opentelemetry
    finally:
        FastAPIInstrumentor.uninstrument_app(app)


def _otel_middleware(app: FastAPI) -> OpenTelemetryMiddleware:
    """从构建好的中间件栈里挖出 ASGI 打桩中间件，用于断言它的排除规则"""
    node = app.build_middleware_stack()
    while node is not None:
        if isinstance(node, OpenTelemetryMiddleware):
            return node
        node = getattr(node, "app", None)
    raise AssertionError("中间件栈里没有 OpenTelemetryMiddleware")


def test_health_excluded_from_server_spans(monkeypatch):
    """/health 不产生 server span，其它路径照常

    容器 HEALTHCHECK 每 30 秒探一次、上游也在轮询，这些请求各自是一条孤立 trace，
    量大且没有信息量。排除规则按 search 匹配完整 URL，所以要钉住两件事：/health
    真的被挡掉，且 `$` 锚点没有顺手把 /api/* 一起挡掉。
    """
    monkeypatch.setattr(
        telemetry, "get_settings", lambda: Settings(_env_file=None, otel_enabled=True)
    )
    app = FastAPI()
    try:
        telemetry.instrument_app(app)
        excluded = _otel_middleware(app).excluded_urls
        assert excluded.url_disabled("http://127.0.0.1:31108/health")
        assert not excluded.url_disabled("http://127.0.0.1:31108/api/cart/add")
        # 前缀匹配会误伤的反例：锚点保证只有 /health 本身被排除
        assert not excluded.url_disabled("http://127.0.0.1:31108/api/health-detail")
    finally:
        FastAPIInstrumentor.uninstrument_app(app)


def test_excluded_urls_empty_falls_back_to_env(monkeypatch):
    """配置留空时传 None，让 OTel 回落到它自己的环境变量而不是排除空字符串"""
    monkeypatch.setattr(
        telemetry,
        "get_settings",
        lambda: Settings(_env_file=None, otel_enabled=True, otel_excluded_urls="  "),
    )
    # OTel 那份环境变量在模块导入时就解析成常量了，setenv 已经晚了，只能直接替常量
    monkeypatch.setattr(
        otel_fastapi, "_excluded_urls_from_env", parse_excluded_urls("/metrics")
    )
    app = FastAPI()
    try:
        telemetry.instrument_app(app)
        excluded = _otel_middleware(app).excluded_urls
        assert excluded.url_disabled("http://127.0.0.1:31108/metrics")
        assert not excluded.url_disabled("http://127.0.0.1:31108/health")
    finally:
        FastAPIInstrumentor.uninstrument_app(app)


def test_instrument_app_skipped_when_otel_disabled(monkeypatch):
    """otel 关闭时不装中间件，省掉一层用不上的开销"""
    monkeypatch.setattr(
        telemetry, "get_settings", lambda: Settings(_env_file=None, otel_enabled=False)
    )
    app = FastAPI()
    telemetry.instrument_app(app)
    assert not getattr(app, "_is_instrumented_by_opentelemetry", False)


async def test_traced_records_span_and_reraises(spans):
    """traced 成功时留一个 span；异常时记 ERROR 状态并原样抛出"""

    @telemetry.traced("unit.ok")
    async def ok() -> str:
        return "done"

    @telemetry.traced("unit.boom")
    async def boom() -> None:
        raise OrderGuardError("金额超限")

    assert await ok() == "done"
    with pytest.raises(OrderGuardError):
        await boom()

    finished = {s.name: s for s in spans.get_finished_spans()}
    assert finished["unit.ok"].status.status_code is not StatusCode.ERROR
    failed = finished["unit.boom"]
    assert failed.status.status_code is StatusCode.ERROR
    # AppError 的对外错误码要落在 span 上：排查时按码筛比按异常类名筛更贴近上游
    assert failed.attributes["error.type"] == "OrderGuardError"
    assert failed.attributes["error.code"] == OrderGuardError("x").err_code


async def test_traced_nests_under_caller_span(spans):
    """traced 出来的 span 要挂在调用方的 span 底下，而不是各自成为孤立 trace"""

    @telemetry.traced("unit.child")
    async def child() -> None:
        return None

    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.root"):
        await child()

    by_name = {s.name: s for s in spans.get_finished_spans()}
    assert by_name["unit.child"].parent.span_id == by_name["unit.root"].context.span_id
    # 同一条 trace 才能在观测后台里连成一条链路
    assert by_name["unit.child"].context.trace_id == by_name["unit.root"].context.trace_id


def test_set_attributes_skips_none(spans):
    """可选字段为 None 时不落属性，避免一堆 None 噪声"""
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.attrs") as span:
        telemetry.set_attributes(span, {"a": 1, "b": None, "c": "x"})

    attributes = spans.get_finished_spans()[0].attributes
    assert attributes["a"] == 1
    assert attributes["c"] == "x"
    assert "b" not in attributes


def test_record_error_keeps_app_error_fields(spans):
    """AppError 的排查字段（错误码/可重试/状态码/消息）都要落到属性上

    只落 error.type 不够：上游看到的是错误码，「该不该重试」看 retryable，
    而 record_exception 记的 event 在多数观测后台里要展开才看得到、列表页筛不出来。
    """
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.err") as span:
        telemetry.record_error(span, UpstreamBlockedError("Akamai 挑战页"))

    attributes = spans.get_finished_spans()[0].attributes
    assert attributes["error.type"] == "UpstreamBlockedError"
    assert attributes["error.code"] == UpstreamBlockedError().err_code
    assert attributes["error.retryable"] is True
    assert attributes["error.status_code"] == 400
    assert "Akamai 挑战页" in attributes["error.message"]


def test_record_error_on_plain_exception_omits_app_error_fields(spans):
    """非 AppError 不应凭空长出 error.code / error.retryable 属性"""
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.plain") as span:
        telemetry.record_error(span, RuntimeError("boom"))

    attributes = spans.get_finished_spans()[0].attributes
    assert attributes["error.type"] == "RuntimeError"
    assert "error.code" not in attributes
    assert "error.retryable" not in attributes


def test_record_envelope_failure_marks_span_error(spans):
    """信封失败要置 ERROR 并落错误码——HTTP 层看不出这次调用失败了"""
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.envelope") as span:
        telemetry.record_envelope(
            span, success=False, err_code=6002, msg="租约无效", status_code=409
        )

    finished = spans.get_finished_spans()[0]
    assert finished.status.status_code is StatusCode.ERROR
    assert finished.attributes["api.success"] is False
    assert finished.attributes["api.code"] == 6002
    assert finished.attributes["api.status_code"] == 409
    assert finished.attributes["api.msg"] == "租约无效"


def test_record_envelope_success_leaves_status_ok(spans):
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.envelope_ok") as span:
        telemetry.record_envelope(span, success=True, status_code=200)

    finished = spans.get_finished_spans()[0]
    assert finished.status.status_code is not StatusCode.ERROR
    assert finished.attributes["api.success"] is True


def test_snapshot_truncates_and_carries_extra(spans):
    """超限 HTML 截断并标注，附带的来源信息（URL 等）也要落在同一条 event 上"""
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.snapshot") as span:
        telemetry.snapshot(
            span, "parse.failed_html", "x" * 100, 10,
            extra={"parse.url": "https://example.com/a", "parse.ignored": None},
        )

    event = spans.get_finished_spans()[0].events[0]
    assert event.name == "parse.failed_html"
    assert event.attributes["snapshot.html"] == "x" * 10
    assert event.attributes["snapshot.original_bytes"] == 100
    assert event.attributes["snapshot.truncated"] is True
    assert event.attributes["parse.url"] == "https://example.com/a"
    assert "parse.ignored" not in event.attributes


def test_snapshot_skips_empty_html(spans):
    """页面根本没取回来时不记空 event"""
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.snapshot_empty") as span:
        telemetry.snapshot(span, "parse.failed_html", None, 100)
        telemetry.snapshot(span, "parse.failed_html", "", 100)

    assert spans.get_finished_spans()[0].events == ()


def test_record_parse_failure_distinguishes_fetch_from_parse(spans):
    """失败阶段按「HTML 有没有拿到」区分：两者排查方向相反

    html 为空＝页面没取回来（通道/反爬/上游 5xx）；非空＝取回了但解析不出
    （多半站点改版）。fail_reason 要给真实异常类名，不能一律写死 parse_error。
    """
    tracer = telemetry.trace.get_tracer(__name__)
    with tracer.start_as_current_span("unit.fetch_fail") as span:
        telemetry.record_parse_failure(
            span, UpstreamBlockedError("挑战页"), html=None, max_bytes=100,
            url="https://example.com/a",
        )
    with tracer.start_as_current_span("unit.parse_fail") as span:
        telemetry.record_parse_failure(
            span, ScrapeParseError("没有 state"), html="<html/>", max_bytes=100,
            url="https://example.com/b",
        )
    with tracer.start_as_current_span("unit.delegate_fail") as span:
        telemetry.record_parse_failure(
            span, ScrapeParseError("转调失败"), stage="delegate",
        )

    by_name = {s.name: s for s in spans.get_finished_spans()}
    fetch = by_name["unit.fetch_fail"]
    assert fetch.attributes["parse.stage"] == "fetch"
    assert fetch.attributes["parse.fail_reason"] == "UpstreamBlockedError"
    # 页面没取回来，没有快照可落
    assert [e.name for e in fetch.events] == ["exception"]

    parsed = by_name["unit.parse_fail"]
    assert parsed.attributes["parse.stage"] == "parse"
    assert parsed.attributes["parse.fail_reason"] == "ScrapeParseError"
    snapshot_event = next(e for e in parsed.events if e.name == "parse.failed_html")
    assert snapshot_event.attributes["snapshot.html"] == "<html/>"
    assert snapshot_event.attributes["parse.url"] == "https://example.com/b"

    # 显式 stage 覆盖推断：编排方法的失败既不在自己的 fetch 也不在自己的 parse
    assert by_name["unit.delegate_fail"].attributes["parse.stage"] == "delegate"


def test_span_unless_suppressed_is_noop_inside_suppressed(spans):
    """`suppressed()` 里手工埋点必须退化成 noop，否则空转长轮询绕开抑制刷满后台

    `suppress_instrumentation` 只被 instrumentation 库尊重，手工
    `start_as_current_span` 不看它——worker 的 lease 正是在 suppressed() 里调
    GatewayClient._request 的。
    """
    tracer = telemetry.trace.get_tracer(__name__)

    with telemetry.suppressed():
        with telemetry.span_unless_suppressed(tracer, "unit.suppressed") as span:
            # 属性/异常写在 noop span 上不能报错，调用方不必分支
            telemetry.set_attributes(span, {"a": 1})
            telemetry.record_error(span, RuntimeError("boom"))
            assert not span.is_recording()

    assert spans.get_finished_spans() == ()

    with telemetry.span_unless_suppressed(tracer, "unit.not_suppressed") as span:
        assert span.is_recording()
    assert [s.name for s in spans.get_finished_spans()] == ["unit.not_suppressed"]


def test_enabled_initializes_and_shutdown_releases():
    """enabled=true 时 setup 注册 TracerProvider，shutdown 后 _provider 复位"""
    settings = Settings(
        _env_file=None,
        otel_enabled=True,
        otel_endpoint="http://localhost:4318/v1/traces",
    )
    try:
        setup_telemetry(settings, service_name="test-enabled")
        assert is_initialized()
        provider = trace.get_tracer_provider()
        # 初始化后是真实 SDK provider（class 名是 TracerProvider）；
        # SDK 启动前的占位 provider 不属于这个类型
        assert isinstance(provider, TracerProvider)
    finally:
        shutdown_telemetry()
        # 解除 httpx 全局 monkey-patch，避免污染其它测试
        try:
            HTTPXClientInstrumentor().uninstrument()
        except Exception:
            pass
    assert not is_initialized()