From 8381896eebbad5870b7d06f44a781f7917ea752b Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Fri, 28 Aug 2026 16:07:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(observability):=20=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E5=93=8D=E5=BA=94=E4=B8=8E=E7=BD=91=E5=85=B3=E4=BF=A1=E5=B0=81?= =?UTF-8?q?=E8=BF=9B=E9=93=BE=E8=B7=AF=EF=BC=8C=E6=89=8B=E5=B7=A5=E5=9F=8B?= =?UTF-8?q?=E7=82=B9=E5=B0=8A=E9=87=8D=20suppressed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 失败此前在 trace 里近乎不可见：异常处理器把异常吃掉换成信封响应，自动 instrumentation 只看到一个 HTTP 状态码，而 AppError 默认 400、信封里 success=false，跟正常返回分不出来。 - api.py：四个异常处理器（对外失败的唯一出口）各记一次 span；兜底处理器额外 record_error——对外只回一句无信息量的错误文案，异常类型与栈只在本地日志里 - telemetry.py：新增 record_envelope / record_parse_failure / span_unless_suppressed；record_error 补 error.message / retryable / status_code；snapshot 支持 extra 带上「这份 HTML 是哪来的」 - worker/client.py：_request 自建 span，活到解信封之后。httpx 那个 CLIENT span 在 request() 返回时就结束，此时信封还没解——success=false code=6002（租约 无效）在它看来是完成的 200 请求 - span_unless_suppressed：suppress_instrumentation 只被 instrumentation 库尊重， 手工 span 不看它，lease 空转长轮询会从这个口子把孤立 trace 放回来 - 两个 scraping client 的解析失败分支收拢到 record_parse_failure；shop_items 显式标 stage=delegate 且不落快照（它自己不抓页面，按 html 推断只会得出 「fetch 失败」的错误结论） Co-Authored-By: Claude Opus 5 (1M context) --- app/scraping/services/rakuma_client.py | 47 ++++---- app/scraping/services/rakuten_client.py | 46 +++---- app/shared/api.py | 47 +++++++- app/shared/telemetry.py | 152 ++++++++++++++++++++++-- app/trading/worker/client.py | 79 +++++++++--- tests/test_telemetry.py | 152 +++++++++++++++++++++++- 6 files changed, 451 insertions(+), 72 deletions(-) diff --git a/app/scraping/services/rakuma_client.py b/app/scraping/services/rakuma_client.py index e3bb2f9..221b1b6 100644 --- a/app/scraping/services/rakuma_client.py +++ b/app/scraping/services/rakuma_client.py @@ -37,7 +37,7 @@ from app.scraping.utils.rakuma_urls import ( split_item_url, split_shop_url, ) -from app.shared.telemetry import snapshot +from app.shared.telemetry import record_parse_failure logger = logging.getLogger(__name__) tracer = trace.get_tracer(__name__) @@ -77,10 +77,11 @@ class RakumaClient: url, len(result.items), result.total_count, ) return result - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise async def categories(self, payload: RakumaCategoryRequest) -> RakumaCategoryData: @@ -110,10 +111,11 @@ class RakumaClient: data.category_id, data.name, len(data.children), data.total_count, ) return data - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise async def item_detail(self, payload: RakumaItemDetailRequest) -> RakumaItemDetailData: @@ -139,10 +141,11 @@ class RakumaClient: url, detail.item_name[:40], detail.price, detail.is_sold_out, ) return detail - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise async def shop_detail(self, payload: RakumaShopDetailRequest) -> RakumaShopDetailData: @@ -177,10 +180,11 @@ class RakumaClient: shop_id, detail.shop_name, detail.item_count, detail.review_count, ) return detail - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise async def shop_items(self, payload: RakumaShopItemsRequest) -> RakumaShopItemsData: @@ -205,10 +209,11 @@ class RakumaClient: shop_id, len(result.items), result.total_count, ) return result - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise @staticmethod diff --git a/app/scraping/services/rakuten_client.py b/app/scraping/services/rakuten_client.py index cd2e6b9..e337c70 100644 --- a/app/scraping/services/rakuten_client.py +++ b/app/scraping/services/rakuten_client.py @@ -45,7 +45,7 @@ from app.scraping.utils.urls import ( split_item_url, split_shop_url, ) -from app.shared.telemetry import snapshot +from app.shared.telemetry import record_parse_failure logger = logging.getLogger(__name__) tracer = trace.get_tracer(__name__) @@ -87,10 +87,11 @@ class RakutenClient: url, len(result.items), result.ad_count, result.total_count, ) return result - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise async def genres(self, payload: GenreRequest) -> GenreData: @@ -113,10 +114,11 @@ class RakutenClient: result.genre_id or "root", result.name, len(result.children), ) return result - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise async def shop_detail(self, payload: ShopDetailRequest) -> ShopDetailData: @@ -142,10 +144,11 @@ class RakutenClient: result.shop_code, result.shop_id, result.shop_name, result.review_count, ) return result - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise async def shop_items(self, payload: ShopItemsRequest) -> SearchResultData: @@ -170,9 +173,11 @@ class RakutenClient: span.set_attribute("parse.items", len(result.items)) span.set_attribute("parse.total", result.total_count) return result - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") + except Exception as exc: + # 本方法自己不抓页面，失败一定发生在它转调的 shop_detail / search + # 里（那两个 span 已各自记了自己的失败页面），所以显式标 delegate、 + # 不落快照：按 html 推断只会得出「fetch 失败」的错误结论。 + record_parse_failure(span, exc, stage="delegate") raise async def item_detail(self, payload: ItemDetailRequest) -> ItemDetailData: @@ -226,8 +231,9 @@ class RakutenClient: url, detail.source, detail.item_name[:40], detail.price, detail.sku.variant_count, ) return detail - except Exception: - span.record_exception() - span.set_attribute("parse.fail_reason", "parse_error") - snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes) + except Exception as exc: + record_parse_failure( + span, exc, html=html, + max_bytes=self._settings.otel_snapshot_max_bytes, url=url, + ) raise diff --git a/app/shared/api.py b/app/shared/api.py index db6e30f..94fb162 100644 --- a/app/shared/api.py +++ b/app/shared/api.py @@ -16,10 +16,12 @@ from typing import Any, Generic, TypeVar from fastapi import Depends, FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse +from opentelemetry import trace from pydantic import BaseModel, Field, ValidationError from starlette.exceptions import HTTPException as StarletteHTTPException from app.shared.errors import AppError, AuthenticationError +from app.shared.telemetry import record_envelope, record_error logger = logging.getLogger(__name__) @@ -98,16 +100,43 @@ def jsonable_errors(errors: list[dict]) -> list[dict]: return [{key: value for key, value in error.items() if key != "ctx"} for error in errors] +def _trace_failure( + exc: BaseException | None, *, err_code: int, msg: str, status_code: int +) -> None: + """把一次失败响应记到当前 server span 上（FastAPI 自动 instrumentation 建的那个） + + 这些处理器是所有对外失败的**唯一出口**，也是链路上唯一还知道「异常长什么样」 + 的地方：它们把异常吃掉换成 200/4xx 的信封响应，异常不再向上冒，自动 + instrumentation 只看得到一个 HTTP 状态码。尤其 AppError 默认 status_code=400、 + 信封里 `success=false`，在 trace 里跟正常返回几乎分不出来——不在这里记一次， + 上游报「调用失败了」时链路里根本找不到对应的错误。 + + span 没在录（otel 关闭、或 /health 这类被 excluded_urls 排除的路径）时 + `set_attributes` / `record_error` 都作用在 NonRecordingSpan 上，是 noop， + 不必额外判断。 + """ + span = trace.get_current_span() + if exc is not None: + record_error(span, exc) + record_envelope(span, success=False, err_code=err_code, msg=msg, status_code=status_code) + + def register_exception_handlers(app: FastAPI) -> None: """给应用挂上全套异常处理器 两个入口都调用它，保证抓取失败与下单失败返回的错误结构完全一致， 上游只需要按 code 分支，不必区分是哪个服务回的。 + + 每个处理器除了构造响应，还把这次失败记到当前 server span 上（见 + `_trace_failure`）——处理器是失败的唯一出口，不记就等于链路里没有这次失败。 """ @app.exception_handler(AppError) async def app_error_handler(_: Request, exc: AppError) -> JSONResponse: """业务异常处理器：返回结构化的错误响应""" + _trace_failure( + exc, err_code=exc.err_code, msg=exc.message, status_code=exc.status_code + ) return JSONResponse( status_code=exc.status_code, content=ApiResponse[None]( @@ -123,11 +152,15 @@ def register_exception_handlers(app: FastAPI) -> None: async def validation_error_handler(_: Request, exc: RequestValidationError) -> JSONResponse: """请求参数校验异常处理器""" errors = exc.errors() + msg = _format_validation_msg(errors) + # 校验失败不记异常本身（pydantic 的 ValidationError 栈很长且没有诊断价值）， + # 只留错误码与整理后的字段消息——排查要看的是「哪个字段不合法」 + _trace_failure(None, err_code=1002, msg=msg, status_code=422) return JSONResponse( status_code=422, content=ApiResponse[object]( success=False, - msg=_format_validation_msg(errors), + msg=msg, data=jsonable_errors(errors), code=1002, ).model_dump(), @@ -137,11 +170,13 @@ def register_exception_handlers(app: FastAPI) -> None: async def pydantic_validation_error_handler(_: Request, exc: ValidationError) -> JSONResponse: """Pydantic 模型校验异常处理器""" errors = exc.errors() + msg = _format_validation_msg(errors) + _trace_failure(None, err_code=1002, msg=msg, status_code=422) return JSONResponse( status_code=422, content=ApiResponse[object]( success=False, - msg=_format_validation_msg(errors), + msg=msg, data=jsonable_errors(errors), code=1002, ).model_dump(), @@ -152,11 +187,13 @@ def register_exception_handlers(app: FastAPI) -> None: """HTTP 异常处理器（404、500 等）""" status_code = int(getattr(exc, "status_code", 500) or 500) err_code = 1404 if status_code == 404 else 1500 + detail = str(getattr(exc, "detail", "HTTP error")) + _trace_failure(None, err_code=err_code, msg=detail, status_code=status_code) return JSONResponse( status_code=status_code, content=ApiResponse[None]( success=False, - msg=str(getattr(exc, "detail", "HTTP error")), + msg=detail, data=None, code=err_code, ).model_dump(), @@ -167,6 +204,10 @@ def register_exception_handlers(app: FastAPI) -> None: async def unhandled_exception_handler(_: Request, exc: Exception) -> JSONResponse: """兜底异常处理器：捕获所有未处理的异常""" logger.exception("未处理异常：%s", exc) + # 这里最需要 record_error：对外只回一句无信息量的 "Internal server error"， + # 真正的异常类型与栈只在本进程日志里。记到 span 上，链路里就能直接看到 + # 是什么炸了，不必再去捞日志按时间对。 + _trace_failure(exc, err_code=1500, msg="Internal server error", status_code=500) return JSONResponse( status_code=500, content=ApiResponse[None]( diff --git a/app/shared/telemetry.py b/app/shared/telemetry.py index 18f3da0..42bd092 100644 --- a/app/shared/telemetry.py +++ b/app/shared/telemetry.py @@ -34,7 +34,10 @@ 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.instrumentation.utils import ( + is_instrumentation_enabled, + suppress_instrumentation, +) from opentelemetry.sdk.resources import SERVICE_NAME, Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor @@ -52,6 +55,10 @@ R = TypeVar("R") logger = logging.getLogger(__name__) +# 错误信息类属性的截断长度。站点错误页抽出来的 msg 可能很长（_extract_error_message +# 拼两条提示），而属性值过长会把 OTLP 请求撑大；排查看的是前半句，够了。 +_MSG_MAX_CHARS = 512 + # 全局 provider 引用，用于 instrument_app / shutdown 时判断当前是否已初始化。 # 显式持有比依赖 trace.get_tracer_provider() 的类型判断更稳——后者在测试场景 # 下可能被其它用例改动全局状态。 @@ -147,11 +154,21 @@ def shutdown_telemetry() -> None: _provider = None -def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None: +def snapshot( + span: Span, + name: str, + html: str | None, + max_bytes: int, + *, + extra: Mapping[str, AttributeValue | None] | None = None, +) -> None: """把 HTML 作为 span event 上报，超 max_bytes 截断并标注。 用于解析失败时复现页面：span 自身只放结构化指标（items 数、source 等）， 完整 HTML 体量大、含商品/价格内容，仅在失败分支通过 event 携带。 + + `extra` 用来带上「这份 HTML 是哪来的」——落地 URL、页面标题、证据文件路径 + 之类。光有一坨 HTML 还得自己回头对是哪一步的产物，附在同一条 event 上省事。 """ if html is None or not html: return @@ -164,6 +181,9 @@ def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None: } if truncated: attributes["snapshot.truncated"] = True + for key, value in (extra or {}).items(): + if value is not None: + attributes[key] = value span.add_event(name, attributes=attributes) @@ -174,25 +194,141 @@ def suppressed() -> Iterator[None]: 给「空转的长轮询」用：worker 每 30 秒问一次网关有没有活干，绝大多数时候 返回空。这些请求各自成为一条孤立 trace，量大且没有信息量——把观测后台刷满 的正是它们。领到任务后的每一次网关调用都在任务根 span 底下，不受影响。 + + **只挡自动 instrumentation**：OTel 那个上下文标记是给 instrumentation 库看的， + 手工 `start_as_current_span` 不看它，照样会建 span。所以在这个上下文里手工埋点 + 要走 `span_unless_suppressed`，否则空转长轮询会从另一个口子把孤立 trace 放回来。 """ with suppress_instrumentation(): yield +@contextmanager +def span_unless_suppressed( + tracer: trace.Tracer, name: str, *, kind: SpanKind = SpanKind.INTERNAL +) -> Iterator[Span]: + """同 `start_as_current_span`，但在 `suppressed()` 里退化成 noop span。 + + 手工埋点与 `suppressed()` 的配套件。`suppress_instrumentation` 只被 + instrumentation 库尊重，手工建的 span 不受它影响——worker 的 `lease` / + `lease_query` 正是在 `suppressed()` 里调 `GatewayClient._request` 的，那里若 + 无条件建 span，空转的长轮询就会每 30 秒产出一条孤立 trace，等于绕开了 + `suppressed()` 本来要解决的问题。 + + 退化时给的是 `INVALID_SPAN`（NonRecordingSpan）：`set_attributes` / + `record_error` / `record_envelope` 作用在它上面全是 noop，调用方不必分支。 + """ + if not is_instrumentation_enabled(): + yield trace.INVALID_SPAN + return + with tracer.start_as_current_span(name, kind=kind) as span: + yield span + + def record_error(span: Span, exc: BaseException) -> None: """把异常记到 span 上并置 ERROR 状态。 - 单独抽出来是因为 AppError 带 `err_code`（对外错误码），排查时按码筛比按 - 异常类名筛更贴近上游看到的东西，值得单独落一个属性。 + 单独抽出来是因为 AppError 带的那几个字段（对外错误码 `err_code`、是否可重试 + `retryable`、HTTP 状态码）正是排查时真正要看的东西：按码筛比按异常类名筛更 + 贴近上游看到的结果，而 `retryable` 直接决定这次失败该不该重来。异常消息也单独 + 落一个属性——`record_exception` 记的 event 在多数观测后台里要展开才看得到， + 列表页按 `error.message` 筛不出来。 """ 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) + retryable = getattr(exc, "retryable", None) + set_attributes( + span, + { + "error.type": type(exc).__name__, + "error.message": str(exc)[:_MSG_MAX_CHARS], + "error.code": err_code if isinstance(err_code := getattr(exc, "err_code", None), int) else None, + "error.retryable": retryable if isinstance(retryable, bool) else None, + "error.status_code": sc if isinstance(sc := getattr(exc, "status_code", None), int) else None, + }, + ) span.set_status(Status(StatusCode.ERROR, f"{type(exc).__name__}: {exc}")) +def record_envelope( + span: Span, + *, + success: bool, + err_code: int | None = None, + msg: str | None = None, + status_code: int | None = None, +) -> None: + """把 `ApiResponse` 信封的结果记到 span 上。 + + 自动 instrumentation 只看 HTTP 层，而本项目的失败**在信封里**：一次 + `success=false, code=6002` 的回报，HTTP 层跟成功的调用长得一模一样（很多还 + 是 200）。于是 trace 里只剩「调用发生过」，「这次到底成没成、错在哪个码」 + 全在 body 里，不显式记就永远看不到——这正是「只知道调用了、不知道异常怎么 + 来的」的直接原因。 + + 出入两侧都用它：服务端异常处理器往 server span 上记（见 `shared.api`）， + worker 解信封时往 client span 上记（见 `trading.worker.client`），同一套 + `api.*` 属性名，一条 trace 里两侧的结论可以直接对上。 + """ + set_attributes( + span, + { + "api.success": success, + "api.code": err_code, + "api.status_code": status_code, + "api.msg": msg[:_MSG_MAX_CHARS] if msg else None, + }, + ) + if not success: + span.set_status(Status(StatusCode.ERROR, msg or f"api.code={err_code}")) + + +def add_event( + span: Span, name: str, attributes: Mapping[str, AttributeValue | None] +) -> None: + """记一条 span event，跳过 None 值属性。 + + 「过程」不能用属性表达：同名属性后写覆盖先写，三次抓取尝试写完只剩最后一次 + 的状态码，中间那两次为什么失败、升级到哪一级全被盖掉了。每次尝试各记一条 + event，链路里才看得出升级路径。 + """ + span.add_event( + name, + attributes={key: value for key, value in attributes.items() if value is not None}, + ) + + +def record_parse_failure( + span: Span, + exc: BaseException, + *, + html: str | None = None, + max_bytes: int = 0, + url: str | None = None, + stage: str | None = None, +) -> None: + """页面类操作失败的统一记法：错误详情 + 失败阶段 + 失败页面快照。 + + 抓取侧与 ラクマ 侧一共十个接口的失败分支原本各写一遍同样三步，且都漏了最关键 + 的一件事——**失败在哪一步**。`html` 是否已拿到恰好就是判据：还是空说明页面根本 + 没取回来（通道 / 反爬 / 上游 5xx），非空说明取回了但解析不出（多半站点改版）。 + 两者的排查方向完全相反，所以作为属性直接落下来，不让人对着一坨 HTML 猜。 + + `stage` 可显式覆盖：像 `shop_items` 那种「转调另外两个接口」的编排方法，失败 + 既不在自己的 fetch 也不在自己的 parse，据 html 推断只会给出错的结论。 + """ + record_error(span, exc) + set_attributes( + span, + { + "parse.stage": stage or ("fetch" if not html else "parse"), + # 保留原有属性名（观测后台的既有筛选条件），但值改成真实异常类名—— + # 原先无论什么失败都写死 "parse_error"，把反爬阻断也说成解析失败。 + "parse.fail_reason": type(exc).__name__, + }, + ) + snapshot(span, "parse.failed_html", html, max_bytes, extra={"parse.url": url}) + + def traced( name: str, *, diff --git a/app/trading/worker/client.py b/app/trading/worker/client.py index e48880a..bc3086c 100644 --- a/app/trading/worker/client.py +++ b/app/trading/worker/client.py @@ -15,15 +15,25 @@ import logging from typing import Any import httpx +from opentelemetry import trace +from opentelemetry.trace import SpanKind from app.shared.config import Settings from app.shared.errors import AppError from app.shared.proxy import httpx_client_options from app.shared.task_state import OrderState, TaskStatus +from app.shared.telemetry import ( + record_envelope, + record_error, + set_attributes, + span_unless_suppressed, +) from app.trading.worker.models import LeaseTask, QueryTask logger = logging.getLogger(__name__) +tracer = trace.get_tracer(__name__) + class GatewayClient: """网关 HTTP 客户端 @@ -49,27 +59,58 @@ class GatewayClient: # ---- 基础封装 ---- async def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]: - """发起请求并解信封。失败（success=False）抛 AppError""" - response = await self._client.request(method, path, **kwargs) - try: - body = response.json() - except ValueError as exc: - raise AppError( - message=f"网关响应不是合法 JSON：HTTP {response.status_code}", - code="GATEWAY_BAD_BODY", - err_code=3001, - retryable=True, - ) from exc + """发起请求并解信封。失败（success=False）抛 AppError - if not body.get("success"): - raise AppError( - message=body.get("msg", "网关返回失败"), - code="GATEWAY_ERROR", - err_code=int(body.get("code", 1500)), - retryable=False, - status_code=response.status_code, + 整段套一个自己的 span，而不是依赖 httpx 自动 instrumentation 那个 CLIENT + span：**网关的失败在信封里，不在 HTTP 状态码上**。httpx 那个 span 在 + `request()` 返回时就结束了，此时信封还没解——一次 `success=false, code=6002` + （租约无效）的调用在它看来是完成的 200 请求，链路里跟成功毫无区别。 + 本 span 活到解信封之后，所以能把「这次调用的结论」记下来。 + """ + with span_unless_suppressed( + tracer, + f"gateway.{path.strip('/').replace('/', '.')}", + kind=SpanKind.CLIENT, + ) as span: + set_attributes(span, {"gateway.method": method, "gateway.path": path}) + response = await self._client.request(method, path, **kwargs) + try: + body = response.json() + except ValueError as exc: + err = AppError( + message=f"网关响应不是合法 JSON：HTTP {response.status_code}", + code="GATEWAY_BAD_BODY", + err_code=3001, + retryable=True, + ) + record_error(span, err) + span.set_attribute("gateway.status_code", response.status_code) + raise err from exc + + if not body.get("success"): + err_code = int(body.get("code", 1500)) + msg = body.get("msg", "网关返回失败") + # 记成信封结果而不是只抛异常：上报被网关拒时，链路里能直接按 + # api.code 筛出是哪一类拒绝（6002 租约无效 / 6003 状态不允许…）， + # 不必回头翻 worker 日志。 + record_envelope( + span, + success=False, + err_code=err_code, + msg=msg, + status_code=response.status_code, + ) + raise AppError( + message=msg, + code="GATEWAY_ERROR", + err_code=err_code, + retryable=False, + status_code=response.status_code, + ) + record_envelope( + span, success=True, status_code=response.status_code ) - return body + return body # ---- 接口 ---- diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index a409741..dd87590 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -5,6 +5,9 @@ 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，不碰 @@ -27,7 +30,7 @@ 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 +from app.shared.errors import OrderGuardError, ScrapeParseError, UpstreamBlockedError from app.shared.telemetry import is_initialized, setup_telemetry, shutdown_telemetry @@ -189,6 +192,153 @@ def test_set_attributes_skips_none(spans): 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="", 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"] == "" + 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(