diff --git a/.env.example b/.env.example index 2ac80de..ff2224e 100644 --- a/.env.example +++ b/.env.example @@ -41,7 +41,8 @@ RAKUTEN_REQUEST_TIMEOUT_SECONDS=30 RAKUTEN_MAX_SITE_CONCURRENCY=8 # 单次抓取的最大尝试次数(含首次):1 次直发,2 次换 cookie 重试,3 次起动用浏览器兜底 RAKUTEN_HTTP_MAX_ATTEMPTS=3 -# Akamai cookie 最长复用时长(秒),超时后重新预热 +# Akamai cookie 最长复用时长(秒),超时后清空、由下一次请求的响应重新建立。 +# cookie 是随目标页响应下发的,正常路径不会额外访问站点首页。 RAKUTEN_SESSION_TTL_SECONDS=1800 # 浏览器兜底:纯 HTTP 被反爬拦截时用 Playwright 取回 cookie 再回灌重试。 diff --git a/README.md b/README.md index e091b35..aa89111 100644 --- a/README.md +++ b/README.md @@ -93,8 +93,8 @@ PC UA 在搜索页、详情页、店铺页上都能拿到完整模板。因此 - FastAPI 提供 HTTP API - Bearer Token 接口鉴权(`Authorization: Bearer `) -- Akamai cookie 自动预热与复用,过期自动重新预热(仅乐天链路) -- 失败逐级升级:重新预热 → Playwright 兜底取 cookie → 结构化报错 +- Akamai cookie 随目标页响应自动建立与复用,过期自动清空重建(仅乐天链路;正常路径不额外访问首页) +- 失败逐级升级:首页换一套 cookie → Playwright 兜底取 cookie → 结构化报错 - 搜索结果自动剔除混入的 CPC 广告位(乐天) - 识别站点深翻页「静默回绕到第 1 页」的行为,避免上游重复入库(乐天) - 商品详情按落地域名分派解析,覆盖楽天ブックス / Rakuten Fashion / ビックカメラ 等官方旗舰子站 @@ -660,6 +660,30 @@ trading 加购时的字段选择策略:多规格挑第一个非售罄的 varia (账号只读查询通道)**可以重试**——查询只读,重发没有副作用 (见 [docs/order-gateway.md §11](docs/order-gateway.md#11-账号只读查询通道))。 +## 链路追踪 + +可选,默认关闭。`RAKUTEN_OTEL_ENABLED=true` + `RAKUTEN_OTEL_ENDPOINT` 指向 OTLP/HTTP +端点后,三个服务分别以 `rakuten-scraping` / `rakuten-trading` / `rakuten-gateway` 上报。 + +FastAPI 与 httpx 走自动埋点,但那只覆盖「收到 HTTP 请求」与「发出 httpx 请求」两类 +边界。交易侧的实际工作两者都不是——站点交互走 Playwright,worker 主循环是后台任务, +所以这一侧手工埋点: + +- `order.task`:一笔下单任务的**根 span**(一个 `task_id` 一条 trace),带 + `order.route`(`execute` / `recovery` / `already_finished`)与终态 + `order.terminal_status`;被闸门或风控拦下时置 ERROR 并带 `error.code` +- `order.step.*`:清车 → 加购 → 校验 → 确认页 → 提交 → 付款,每步一个子 span, + 带 `order.evidence_ref`(可直接定位落盘证据) +- `site.*`:Playwright 站点交互(`site.add_to_cart`、`site.enter_checkout`、 + `site.submit_order`、`site.pay` 等) +- `account_query`:一张只读查询单的根 span,带 `query.outcome` + +worker 空转的长轮询(每 30 秒问一次网关有没有活干)刻意不埋点——它们没有信息量, +量却极大,会把观测后台刷满。 + +> 解析失败时页面 HTML 会作为 span event 上报(`RAKUTEN_OTEL_SNAPSHOT_MAX_BYTES` +> 控制上限,默认 2MB),用于事后复现「抓到的内容为什么解析不出预期字段」。 + ## 常用环境变量 完整列表见 [.env.example](.env.example)。配置项前缀统一为 `RAKUTEN_`,三个服务共用同一份 @@ -675,6 +699,7 @@ trading 加购时的字段选择策略:多规格挑第一个非售罄的 varia - 抓取:`RAKUTEN_MAX_SITE_CONCURRENCY`(默认 `8`,两站各自独立计数)、`RAKUTEN_HTTP_MAX_ATTEMPTS`(默认 `3`)、`RAKUTEN_SESSION_TTL_SECONDS`(默认 `1800`,仅乐天) - 浏览器兜底(仅乐天):`RAKUTEN_BROWSER_FALLBACK_ENABLED`、`RAKUTEN_BROWSER_HEADLESS`、`RAKUTEN_BROWSER_CHANNEL` - 代理(需日本 IP 时):`RAKUTEN_PROXY_SERVER`、`RAKUTEN_PROXY_USERNAME`、`RAKUTEN_PROXY_PASSWORD`、`RAKUTEN_PROXY_BYPASS` +- 链路追踪(可选,默认关闭;三个服务共用):`RAKUTEN_OTEL_ENABLED`、`RAKUTEN_OTEL_ENDPOINT`、`RAKUTEN_OTEL_HEADERS`、`RAKUTEN_OTEL_SNAPSHOT_MAX_BYTES` > 从中国大陆直连实测可用、无需代理;`RAKUTEN_PROXY_SERVER` 留空即可。设置后,所有面向 > 外部站点的 HTTPX 与 Playwright 流量都会经代理;本机和 Docker 服务间地址由 diff --git a/app/gateway/main.py b/app/gateway/main.py index 72f13d9..831fd5a 100644 --- a/app/gateway/main.py +++ b/app/gateway/main.py @@ -27,6 +27,7 @@ from app.gateway.task_queue import TaskQueue from app.shared.api import register_exception_handlers from app.shared.config import get_settings from app.shared.logging_setup import configure_logging +from app.shared.telemetry import instrument_app, setup_telemetry, shutdown_telemetry logger = logging.getLogger(__name__) @@ -96,6 +97,11 @@ async def lifespan(app: FastAPI): app.state.container = container configure_logging(container.settings) + # 网关这一侧原先完全没接 telemetry,导致 worker 出站请求带过来的 traceparent + # 没人接上:一条下单链路在网关这里断掉,观测后台上只看得到 worker 侧那半截。 + # CallbackNotifier 的 httpx 客户端是每次投递时才建的(不是 __init__ 里), + # 所以放在 build_container() 之后仍然赶在客户端创建之前。 + setup_telemetry(container.settings, service_name="rakuten-gateway") logger.info( "网关启动:%s:%s", container.settings.gateway_host, container.settings.gateway_port ) @@ -141,6 +147,7 @@ async def lifespan(app: FastAPI): # 等在途回调发完(各次发送有超时兜底),避免关停时静默丢通知 await container.notifier.aclose() await container.db.close() + shutdown_telemetry() def create_app() -> FastAPI: @@ -151,6 +158,7 @@ def create_app() -> FastAPI: app.include_router(queries_router) app.include_router(account_router) register_exception_handlers(app) + instrument_app(app) return app diff --git a/app/scraping/services/site_session.py b/app/scraping/services/site_session.py index 21d643e..44c5994 100644 --- a/app/scraping/services/site_session.py +++ b/app/scraping/services/site_session.py @@ -8,7 +8,13 @@ 画像(详情页只有手机 UA 才返回带 __INITIAL_STATE__ 的统一模板)。两条通道 各自持有独立 cookie 罐,避免把 PC 指纹拿到的 cookie 混用到手机请求上。 -抓取失败时的升级路径:重新预热 → 浏览器兜底取 cookie → 放弃。 +**不预热首页**:Akamai 的 cookie 是随任意一个页面响应下发的,目标页自己就会带回 +来,专门先打一次 `www.rakuten.co.jp/` 除了多一个出站请求(以及多一次被风控计数 +的机会)之外没有额外收益——首个请求无论打哪个 URL 都是冷的 ~11s,之后都复用 +cookie。首页只在**失败修复**路径上使用:目标页已经被挑战时,拿首页换一套干净 +cookie 比继续拿目标页去撞更安全(见 `_rewarm_on_home`)。 + +抓取失败时的升级路径:换 cookie(首页重新预热)→ 浏览器兜底取 cookie → 放弃。 """ from __future__ import annotations @@ -62,12 +68,18 @@ class _Profile: mobile: bool client: httpx.AsyncClient lock: asyncio.Lock = field(default_factory=asyncio.Lock) - warmed_at: float = 0.0 + # Akamai cookie 罐的建立时刻(monotonic)。0 表示当前没有可复用的 cookie。 + # 超过 session_ttl_seconds 就主动清空:拿着过期 cookie 去撞反而更容易被挑战。 + cookies_at: float = 0.0 @property def cookie_names(self) -> set[str]: return {cookie.name for cookie in self.client.cookies.jar} + @property + def akamai_cookies(self) -> set[str]: + return self.cookie_names & set(site.AKAMAI_COOKIE_NAMES) + class SiteSession: """乐天站点抓取会话,管理 cookie 预热、并发限流与失败升级""" @@ -115,13 +127,17 @@ class SiteSession: # ---- 状态 ---- def profile_status(self) -> dict[str, dict[str, Any]]: - """各通道的预热状态,供健康检查展示""" + """各通道的 cookie 状态,供健康检查展示 + + `warmed` 保留原字段名(上游健康检查看板在用),语义是「当前有可复用的 + Akamai cookie」——不再代表「已专门预热过首页」,因为正常路径不打首页了。 + """ now = time.monotonic() return { name: { - "warmed": profile.warmed_at > 0, - "age_seconds": round(now - profile.warmed_at, 1) if profile.warmed_at else None, - "cookies": sorted(profile.cookie_names & set(site.AKAMAI_COOKIE_NAMES)), + "warmed": profile.cookies_at > 0, + "age_seconds": round(now - profile.cookies_at, 1) if profile.cookies_at else None, + "cookies": sorted(profile.akamai_cookies), } for name, profile in self._profiles.items() } @@ -142,7 +158,8 @@ class SiteSession: ) -> FetchedPage: """抓取页面,返回 HTML 与最终落地地址 - 失败时按 重新预热 → 浏览器兜底 的顺序逐级升级重试。 + 直接打目标页(不先访问首页),失败时按 换 cookie → 浏览器兜底 的顺序 + 逐级升级重试。 Args: validator: 页面校验器,默认要求页面含 __INITIAL_STATE__。跨站抓取时 @@ -181,7 +198,8 @@ class SiteSession: try: for attempt in range(1, max_attempts + 1): span.set_attribute("scrape.attempts", attempt) - await self._ensure_warm(profile) + # 过期 cookie 主动丢掉:带着它去撞比裸请求更容易吃挑战页 + await self._drop_expired_cookies(profile) try: response = await profile.client.get(url) except httpx.HTTPError as exc: @@ -192,6 +210,10 @@ class SiteSession: ) continue + # 任何响应都可能带 set-cookie(Akamai 不保证每次下发),拿到就 + # 记下时刻,后续请求复用到 TTL 为止。 + self._note_cookies(profile) + if response.status_code == 404: err = ItemNotFoundError(f"Page not found: {url}") span.record_exception(err) @@ -223,9 +245,11 @@ class SiteSession: if attempt >= max_attempts: break - # 第一次失败先便宜地换一套 cookie;仍失败才动用浏览器 + # 第一次失败先便宜地换一套 cookie(丢掉旧的,用首页换新的); + # 仍失败才动用浏览器 if attempt == 1: - await self._invalidate(profile) + await self._rewarm_on_home(profile) + span.set_attribute("scrape.rewarmed_on_home", True) else: page = await self._escalate_to_browser(profile, url, validate) if page is not None: @@ -265,44 +289,53 @@ class SiteSession: def _is_server_error(reason: str) -> bool: return reason.startswith("upstream status") or reason.startswith("httpx") or "Error:" in reason - async def _ensure_warm(self, profile: _Profile) -> None: - """确保通道有一次新鲜的首页预热;过期时重新访问首页""" - if self._is_warm(profile): + def _note_cookies(self, profile: _Profile) -> None: + """目标页响应带回 Akamai cookie 时记下时刻,作为 TTL 起点 + + 已经在计时的不重置:TTL 要从「这套 cookie 第一次出现」算起,每次响应都 + 刷新会让一套 cookie 被无限续命,反而绕过了 session_ttl_seconds 的本意。 + """ + if profile.cookies_at: return + if profile.akamai_cookies: + profile.cookies_at = time.monotonic() - async with profile.lock: - if self._is_warm(profile): - return - try: - response = await profile.client.get(self._settings.home_url) - # Akamai 不保证每次都下发 cookie;首页探测成功本身就是可复用的 - # 预热结果,cookie 只用于观测和失败升级时的回灌。 - if response.status_code < 400: - profile.warmed_at = time.monotonic() - logger.info( - "会话预热完成:profile=%s status=%s cookies=%s", - profile.name, - response.status_code, - sorted(profile.cookie_names & set(site.AKAMAI_COOKIE_NAMES)), - ) - except httpx.HTTPError as exc: - # 预热失败不阻断本次抓取:直连目标页仍可能成功,只是慢 - logger.warning("会话预热失败:profile=%s err=%s", profile.name, exc) - profile.warmed_at = 0.0 - - def _is_warm(self, profile: _Profile) -> bool: - if not profile.warmed_at: - return False - if time.monotonic() - profile.warmed_at > self._settings.session_ttl_seconds: - return False - return True - - async def _invalidate(self, profile: _Profile) -> None: - """清空通道 cookie 并强制下次重新预热""" + async def _drop_expired_cookies(self, profile: _Profile) -> None: + """cookie 罐超过 TTL 时清空,让下一次请求裸奔换一套新的""" + if not profile.cookies_at: + return + if time.monotonic() - profile.cookies_at <= self._settings.session_ttl_seconds: + return async with profile.lock: profile.client.cookies.clear() - profile.warmed_at = 0.0 - logger.info("已清空会话 cookie,将重新预热:profile=%s", profile.name) + profile.cookies_at = 0.0 + logger.info("会话 cookie 已过期,已清空:profile=%s", profile.name) + + async def _rewarm_on_home(self, profile: _Profile) -> None: + """首次失败后的修复:丢掉旧 cookie,用首页换一套新的 + + 这是首页 URL 唯一的用途。目标页已经吃了挑战页,继续拿同一个 URL 去撞 + 只会把挑战坐实;首页是站点最"无害"的入口,换 cookie 的成功率更高。 + + 失败不阻断本次抓取(下一次 attempt 会裸请求目标页,只是慢),所以这里 + 只记日志。 + """ + async with profile.lock: + profile.client.cookies.clear() + profile.cookies_at = 0.0 + try: + response = await profile.client.get(self._settings.home_url) + except httpx.HTTPError as exc: + logger.warning("首页换 cookie 失败:profile=%s err=%s", profile.name, exc) + return + if profile.akamai_cookies: + profile.cookies_at = time.monotonic() + logger.info( + "已用首页换一套新 cookie:profile=%s status=%s cookies=%s", + profile.name, + response.status_code, + sorted(profile.akamai_cookies), + ) async def _escalate_to_browser( self, profile: _Profile, url: str, validate: PageValidator @@ -332,7 +365,7 @@ class SiteSession: domain=cookie.get("domain") or "", path=cookie.get("path") or "/", ) - profile.warmed_at = time.monotonic() + profile.cookies_at = time.monotonic() # 浏览器不回报最终 URL,这里以请求地址为准;跨站跳转场景下 HTTP 通道已先行 # 报错,走不到这一步。 diff --git a/app/shared/config.py b/app/shared/config.py index 268ad78..dd9f9db 100644 --- a/app/shared/config.py +++ b/app/shared/config.py @@ -73,7 +73,9 @@ class Settings(BaseSettings): request_timeout_seconds: float = 30.0 max_site_concurrency: int = 8 # 对站点的最大并发请求数 http_max_attempts: int = 3 # 单次抓取的最大尝试次数(含首次) - session_ttl_seconds: float = 1800.0 # Akamai cookie 会话最长复用时长,超时后重新预热 + # Akamai cookie 最长复用时长;超时后清空 cookie 罐,由下一次响应重新建立。 + # cookie 随目标页响应下发,正常路径不额外访问站点首页(见 site_session.py)。 + session_ttl_seconds: float = 1800.0 # ---- 浏览器兜底配置(仅抓取服务)---- # 纯 HTTP 被 Akamai 拦截时,用 Playwright 打开页面取回 cookie 再回灌给 diff --git a/app/shared/telemetry.py b/app/shared/telemetry.py index adbf54b..357a268 100644 --- a/app/shared/telemetry.py +++ b/app/shared/telemetry.py @@ -11,28 +11,45 @@ instrumentation(FastAPI、httpx)。失败时(如 endpoint 不可达)不 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 typing import TYPE_CHECKING +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 +from opentelemetry.trace import Span, SpanKind, Status, StatusCode from opentelemetry.util.types import AttributeValue -from app.shared.config import Settings +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 时判断当前是否已初始化。 @@ -88,8 +105,24 @@ def setup_telemetry(settings: Settings, *, service_name: str) -> None: def instrument_app(app: "FastAPI") -> None: - """FastAPI 应用打桩;未初始化时 noop,调用顺序无要求。""" - if _provider is 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) @@ -127,6 +160,74 @@ def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None: 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(): diff --git a/app/trading/worker/query_runner.py b/app/trading/worker/query_runner.py index 8a989be..a3532e2 100644 --- a/app/trading/worker/query_runner.py +++ b/app/trading/worker/query_runner.py @@ -21,8 +21,12 @@ import logging from datetime import datetime, timezone from typing import TYPE_CHECKING, Any +from opentelemetry import trace +from opentelemetry.trace import SpanKind + from app.shared.errors import AppError, InvalidRequestError from app.shared.task_state import AccountQueryKind +from app.shared.telemetry import record_error, set_attributes, suppressed from app.trading.worker.client import GatewayClient from app.trading.worker.models import QueryTask from app.trading.worker.site_interact import SiteInteractor @@ -32,6 +36,8 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +tracer = trace.get_tracer(__name__) + # 没给 since 时的窗口下界。用 epoch 而不是 None,是为了让翻页终止条件 # (`order_date < since`)与「不设下界」共用同一条代码路径,不多一个分支。 _EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) @@ -100,7 +106,9 @@ class QueryRunner: logger.info("账号查询 worker 启动:worker_id=%s", self.worker_id) while self._running: try: - query = await self._gateway.lease_query(self.worker_id, wait=30) + # 与下单主循环同样:空转的长轮询不埋点,见 runner.py::run + with suppressed(): + query = await self._gateway.lease_query(self.worker_id, wait=30) except AppError as exc: logger.warning("查询 lease 失败:%s (err=%s)", exc.message, exc.err_code) await asyncio.sleep(5) @@ -122,7 +130,27 @@ class QueryRunner: # ---- 单张查询单 ---- async def handle(self, query: QueryTask) -> None: - """执行一张查询单并回报结果。任何失败都转成一次「失败回报」,不抛出去""" + """执行一张查询单并回报结果。任何失败都转成一次「失败回报」,不抛出去 + + 这里开的 span 是一张查询单的根:底下挂着站点读取(`site.*`)与回报网关的 + HTTP 调用,一个 query_id 对应一条 trace。查询失败不抛出去(都转成失败 + 回报),所以失败信息由各分支显式记到 span 上——否则 trace 里会显示成功。 + """ + with tracer.start_as_current_span("account_query", kind=SpanKind.CONSUMER) as span: + set_attributes( + span, + { + "query.query_id": query.query_id, + "query.kind": query.kind, + "query.site": query.site, + "query.attempt": query.attempt, + "query.worker_id": self.worker_id, + }, + ) + await self._handle_traced(query, span) + + async def _handle_traced(self, query: QueryTask, span: "trace.Span") -> None: + """handle() 的实际执行体,拆出来只为让根 span 的 with 块保持一层缩进""" logger.info( "领到查询单:query_id=%s kind=%s attempt=%s", query.query_id, query.kind, query.attempt, @@ -132,11 +160,13 @@ class QueryRunner: self.execute(query), timeout=self._settings.account_query_timeout_seconds, ) - except asyncio.TimeoutError: + except asyncio.TimeoutError as exc: logger.warning( "查询超时(%s 秒,多半是下单任务正占着账号锁):query_id=%s", self._settings.account_query_timeout_seconds, query.query_id, ) + span.set_attribute("query.outcome", "timeout") + record_error(span, exc) await self._report_safe( query, success=False, @@ -152,12 +182,16 @@ class QueryRunner: "查询失败:query_id=%s code=%s msg=%s", query.query_id, exc.err_code, exc.message, ) + span.set_attribute("query.outcome", "failed") + record_error(span, exc) await self._report_safe( query, success=False, error_code=exc.err_code, error_message=exc.message ) return except Exception as exc: # noqa: BLE001 logger.exception("查询未预期异常:query_id=%s", query.query_id) + span.set_attribute("query.outcome", "unexpected_error") + record_error(span, exc) await self._report_safe( query, success=False, @@ -168,10 +202,12 @@ class QueryRunner: payload, oversized = self._enforce_result_size(result) if oversized is not None: + span.set_attribute("query.outcome", "oversized") await self._report_safe( query, success=False, error_code=1003, error_message=oversized ) return + span.set_attribute("query.outcome", "succeeded") await self._report_safe(query, success=True, result=payload) async def execute(self, query: QueryTask) -> dict[str, Any]: diff --git a/app/trading/worker/runner.py b/app/trading/worker/runner.py index 9843e4b..f34b2c8 100644 --- a/app/trading/worker/runner.py +++ b/app/trading/worker/runner.py @@ -28,6 +28,9 @@ import contextlib import logging from typing import TYPE_CHECKING +from opentelemetry import trace +from opentelemetry.trace import SpanKind, Status, StatusCode + from app.shared.errors import ( AppError, BrowserDeadError, @@ -35,6 +38,7 @@ from app.shared.errors import ( OrderGuardError, ) from app.shared.task_state import OrderState, TaskStatus +from app.shared.telemetry import record_error, set_attributes, suppressed from app.trading.worker import verify from app.trading.worker.client import GatewayClient from app.trading.worker.evidence import EvidenceStore @@ -47,6 +51,8 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +tracer = trace.get_tracer(__name__) + def _coerce_state(value: str | None) -> OrderState: """把网关返回的 state 字符串安全地包成 OrderState;None 或非法值回落到 CREATED""" @@ -115,7 +121,11 @@ class WorkerRunner: logger.info("worker 启动:worker_id=%s", self.worker_id) while self._running: try: - task = await self._gateway.lease(self.worker_id, wait=30) + # 空转的长轮询不埋点:30 秒一次、绝大多数返回空,每次都会变成一条 + # 孤立 trace 把观测后台刷满。领到任务后的调用都在 handle() 的根 + # span 底下,不受这里影响。 + with suppressed(): + task = await self._gateway.lease(self.worker_id, wait=30) except AppError as exc: logger.warning("lease 失败:%s (err=%s)", exc.message, exc.err_code) await asyncio.sleep(5) @@ -137,13 +147,42 @@ class WorkerRunner: # ---- 单任务调度 ---- async def handle(self, task: LeaseTask) -> None: - """单任务调度入口:本地幂等闸门 → 恢复核对 → 执行""" + """单任务调度入口:本地幂等闸门 → 恢复核对 → 执行 + + 这里开的 span 是**整条下单链路的根**:往下的每一步站点交互、每一次回报 + 网关都挂在它底下,一个 task_id 对应观测后台里的一条 trace。worker 是后台 + asyncio 任务,没有 HTTP 入口,不开这个根 span 的话下游 httpx 调用会各自 + 散成孤立 trace(这正是「只有请求记录、没有链路」的原因)。 + """ + with tracer.start_as_current_span("order.task", kind=SpanKind.CONSUMER) as span: + intent = task.intent or {} + set_attributes( + span, + { + "order.task_id": task.task_id, + "order.site": task.site, + "order.worker_id": self.worker_id, + "order.lease_count": task.lease_count, + "order.known_state": task.known_state, + "order.item_url": intent.get("item_url"), + "order.quantity": intent.get("quantity"), + }, + ) + try: + await self._dispatch(task, span) + except Exception as exc: + record_error(span, exc) + raise + + async def _dispatch(self, task: LeaseTask, span: "trace.Span") -> None: + """handle() 的实际分支逻辑,拆出来只为让根 span 的 with 块保持一层缩进""" # 本地幂等闸门:之前已完成过的任务不再执行 if await self._db.has_finished(task.task_id): final_state = await self._db.final_state(task.task_id) logger.info( "本地已完成,补报终态:task_id=%s state=%s", task.task_id, final_state ) + span.set_attribute("order.route", "already_finished") await self._report_safe( task, state=_coerce_state(final_state), @@ -154,10 +193,12 @@ class WorkerRunner: # 恢复领取:lease_count > 1 表示 stale → reclaim,必须先核对站点订单 if task.lease_count > 1: + span.set_attribute("order.route", "recovery") await self._handle_recovery(task) return # 常规执行 + span.set_attribute("order.route", "execute") await self._execute_with_renewal(task) async def _handle_recovery(self, task: LeaseTask) -> None: @@ -197,11 +238,24 @@ class WorkerRunner: # ---- 常规执行 ---- + async def _execute_recording_errors(self, task: LeaseTask) -> None: + """execute() 外面加一层:异常先记到任务根 span 上,再原样抛出 + + 下面那一串 except 分支会把异常**吞掉**转成 needs_human / failed 回报, + 异常冒不到 handle() 的根 span。在这里统一记一次,比每个分支各写一遍省事, + 也不会漏掉新增的分支。 + """ + try: + await self.execute(task) + except Exception as exc: + record_error(trace.get_current_span(), exc) + raise + async def _execute_with_renewal(self, task: LeaseTask) -> None: """在租约自动续期的上下文里执行任务""" async with self._renew_lease_every(task, interval=60): try: - await self.execute(task) + await self._execute_recording_errors(task) except NotImplementedError as exc: # 站点交互未实现(规格 §10):上报 needs_human,不视为 worker 失败 logger.warning( @@ -521,38 +575,65 @@ class WorkerRunner: 证据载体)时,其 html / screenshot 即本步骤要落盘的页面与整页截图; 显式传入的 `html` / `png` 优先级更高,供 step3/step4 等已经单独拿到 页面的调用点使用。 + + 每步一个 span,挂在 handle() 的任务根 span 底下:一条 trace 就是一单的 + 完整流水(清车 → 加购 → 校验 → 确认 → 提交 → 付款),卡在哪一步、每步 + 耗时多少、证据落在哪个 evidence_ref 都能直接读出来。 """ - result = await action() - if isinstance(result, PageSnapshot): - if html is None: - html = result.html or None - if png is None: - png = result.screenshot or None - elif html is None and isinstance(result, str): - # 旧契约兼容:站点方法返回字符串时视为页面 HTML - html = result - meta = { - "step": step_name, - "state": state.value, - **(evidence_meta or {}), - } - evidence_ref = self._evidence.write_step( - task.task_id, step_no, step_name, html=html, png=png, meta=meta - ) - await self._db.index_evidence(task.task_id, step_no, step_name, evidence_ref) - await self._db.record_event( - task.task_id, state.value, detail=detail, evidence_ref=evidence_ref - ) - await self._gateway.report( - task.task_id, - self.worker_id, - state=state, - payable_yen=payable_yen, - pay_deadline=pay_deadline, - site_order_id=site_order_id, - evidence_ref=evidence_ref, - detail=detail, - ) + with tracer.start_as_current_span(f"order.step.{step_name}") as span: + set_attributes( + span, + { + "order.task_id": task.task_id, + "order.step_no": step_no, + "order.step_name": step_name, + "order.state": state.value, + }, + ) + try: + result = await action() + except Exception as exc: + record_error(span, exc) + raise + + if isinstance(result, PageSnapshot): + if html is None: + html = result.html or None + if png is None: + png = result.screenshot or None + elif html is None and isinstance(result, str): + # 旧契约兼容:站点方法返回字符串时视为页面 HTML + html = result + meta = { + "step": step_name, + "state": state.value, + **(evidence_meta or {}), + } + evidence_ref = self._evidence.write_step( + task.task_id, step_no, step_name, html=html, png=png, meta=meta + ) + set_attributes( + span, + { + "order.evidence_ref": evidence_ref, + "order.site_order_id": site_order_id, + "order.payable_yen": payable_yen, + }, + ) + await self._db.index_evidence(task.task_id, step_no, step_name, evidence_ref) + await self._db.record_event( + task.task_id, state.value, detail=detail, evidence_ref=evidence_ref + ) + await self._gateway.report( + task.task_id, + self.worker_id, + state=state, + payable_yen=payable_yen, + pay_deadline=pay_deadline, + site_order_id=site_order_id, + evidence_ref=evidence_ref, + detail=detail, + ) async def _report_safe( self, @@ -566,7 +647,25 @@ class WorkerRunner: payable_yen: int | None = None, pay_deadline: str | None = None, ) -> None: - """回报 gateway,失败只记日志不抛——主循环不能因为回报失败退出""" + """回报 gateway,失败只记日志不抛——主循环不能因为回报失败退出 + + 顺带把终态标到当前 span 上。这一步是必要的:`_execute_with_renewal` 会把 + 闸门拦截、风控拦截、浏览器掉线等异常**吞掉**转成 needs_human 回报,异常 + 不会冒到 handle() 的根 span,于是一笔被拦下的单在 trace 里看起来跟成功 + 下单一模一样。所有这些分支都汇到这个方法,标在这里最省事也最不容易漏。 + + `is_recording()` 那道判断不是多余的:付款后监控(_monitor_order)是 + `create_task` 起的后台任务,而 asyncio 在创建时就把当时的 context 复制了 + 进去——等它真正跑起来,任务根 span 早已结束,但 `get_current_span()` 拿到 + 的仍是那个**已结束**的 span(不是 INVALID_SPAN)。往上写属性会打 + "Setting attribute on ended span" 警告。当前监控路径不传 terminal_status + 走不到这里,但这道判断保证以后传了也不会污染已完成的任务 span。 + """ + span = trace.get_current_span() + if terminal_status is not None and span.is_recording(): + span.set_attribute("order.terminal_status", terminal_status.value) + if terminal_status in (TaskStatus.NEEDS_HUMAN, TaskStatus.FAILED): + span.set_status(Status(StatusCode.ERROR, detail)) try: await self._gateway.report( task.task_id, diff --git a/app/trading/worker/site_interact.py b/app/trading/worker/site_interact.py index 187d27e..cbd11ea 100644 --- a/app/trading/worker/site_interact.py +++ b/app/trading/worker/site_interact.py @@ -96,6 +96,8 @@ from dataclasses import dataclass, field from datetime import datetime from typing import TYPE_CHECKING, TypeVar +from opentelemetry.trace import SpanKind + from app.shared.errors import ( BrowserDeadError, CartOperationError, @@ -113,6 +115,8 @@ from app.shared.purchase_contract import ( ) from app.shared.proxy import playwright_launch_proxy from app.shared.task_state import OrderState +from app.shared.telemetry import traced + from app.trading.core import auth_site from app.trading.worker.models import LeaseTask @@ -973,6 +977,7 @@ class SiteInteractor: # ---- 已实现:add_to_cart / verify_cart / cart_status ---- + @traced("site.add_to_cart", kind=SpanKind.CLIENT) async def add_to_cart(self, task: LeaseTask) -> PageSnapshot: """加购(worker 入口):从 task.intent 取字段,调 _add_to_cart_with_fields @@ -1015,6 +1020,7 @@ class SiteInteractor: screenshot=result.get("screenshot") or b"", ) + @traced("site.add_to_cart_payload", kind=SpanKind.CLIENT) async def add_to_cart_payload( self, *, @@ -1163,6 +1169,7 @@ class SiteInteractor: "screenshot": screenshot, } + @traced("site.verify_cart", kind=SpanKind.CLIENT) async def verify_cart(self, task: LeaseTask) -> PageSnapshot: """校验购物车里有没有刚加的商品 @@ -1196,6 +1203,7 @@ class SiteInteractor: # 2. 渲染 cart 页确认 item_id 在里面 return await self._verify_item_in_cart_html(item_id, label=f"task_id={task.task_id}") + @traced("site.cart_status", kind=SpanKind.CLIENT) async def cart_status(self) -> dict: """轻量查询购物车状态:调 cart count JSONP API,不渲染整页 @@ -1216,6 +1224,7 @@ class SiteInteractor: # ---- 已实现:clear_cart / remove_item(Playwright UI 点击)---- + @traced("site.clear_cart", kind=SpanKind.CLIENT) async def clear_cart(self) -> dict: """清空购物车:渲染 cart SPA → 反复点第一个「削除」按钮 → count API 校验 @@ -1304,6 +1313,7 @@ class SiteInteractor: "screenshot": screenshot, } + @traced("site.remove_item", kind=SpanKind.CLIENT) async def remove_item(self, item_id: str) -> dict: """删除购物车里指定 item_id 的商品 @@ -1506,6 +1516,7 @@ class SiteInteractor: # ---- 已实现:enter_checkout(到下单确认页,中间步骤未经真实 HTML 验证)---- + @traced("site.enter_checkout", kind=SpanKind.CLIENT) async def enter_checkout(self, task: LeaseTask) -> PageSnapshot: """进入下单确认页:购物车 → 点「購入手続き」→ 依次处理中间步骤 → 落地确认页 @@ -2017,6 +2028,7 @@ class SiteInteractor: """ return _parse_checkout_summary(html) + @traced("site.submit_order", kind=SpanKind.CLIENT) async def submit_order(self, task: LeaseTask) -> SubmitOutcome: """点击下单确认页的最终确认按钮,提交订单 @@ -2118,6 +2130,7 @@ class SiteInteractor: evidence=PageSnapshot(html=html, screenshot=screenshot), ) + @traced("site.pay", kind=SpanKind.CLIENT) async def pay(self, task: LeaseTask, site_order_id: str) -> PageSnapshot: """检查提交下单后是否已完成付款 / 是否触发了需要人工介入的验证环节 @@ -2181,6 +2194,7 @@ class SiteInteractor: except Exception: logger.debug("pay 关闭确认页失败", exc_info=True) + @traced("site.check_order_status", kind=SpanKind.CLIENT) async def check_order_status(self, site_order_id: str) -> OrderStatusSnapshot: """付款后监控的单次探测:查一次订单详情页的配送阶段,不循环 @@ -2198,6 +2212,7 @@ class SiteInteractor: """ return (await self.fetch_order_detail(site_order_id)).status + @traced("site.fetch_order_detail", kind=SpanKind.CLIENT) async def fetch_order_detail(self, site_order_id: str) -> OrderDetailSnapshot: """读一次订单详情页:配送阶段 + 页面原始 __INITIAL_STATE__ @@ -2269,6 +2284,7 @@ class SiteInteractor: f"check_order_status site_order_id={site_order_id}", read ) + @traced("site.list_recent_orders", kind=SpanKind.CLIENT) async def list_recent_orders( self, *, since: datetime, max_pages: int | None = None ) -> OrderListWindow: diff --git a/tests/test_site_session.py b/tests/test_site_session.py index d650fdc..b6feb2f 100644 --- a/tests/test_site_session.py +++ b/tests/test_site_session.py @@ -67,15 +67,17 @@ async def build_session(handler, *, browser=None, settings=None) -> SiteSession: return session -async def test_warmup_happens_before_first_fetch_and_is_reused(): +async def test_happy_path_never_touches_the_home_page(): + """目标页自己会带回 Akamai cookie,正常路径不该多打一次首页。""" seen: list[str] = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(str(request.url)) - headers = {} - if request.url.host == "www.rakuten.co.jp": - headers["set-cookie"] = "ak_bmsc=abc; Domain=.rakuten.co.jp; Path=/" - return httpx.Response(200, text=GOOD_PAGE, headers=headers) + return httpx.Response( + 200, + text=GOOD_PAGE, + headers={"set-cookie": "ak_bmsc=abc; Domain=.rakuten.co.jp; Path=/"}, + ) session = await build_session(handler) try: @@ -84,18 +86,20 @@ async def test_warmup_happens_before_first_fetch_and_is_reused(): finally: await session.close() - # 首页预热只做一次,第二次抓取直接复用 cookie - assert seen.count("https://www.rakuten.co.jp/") == 1 - assert seen.count(TARGET) == 2 + assert seen == [TARGET, TARGET] -async def test_successful_warmup_is_reused_when_upstream_sets_no_cookie(): - """首页可能返回 200 但不下发 Akamai cookie,仍不得重复预热。""" - seen: list[str] = [] +async def test_cookies_from_target_page_are_reused_across_fetches(): + """第一次响应下发的 cookie 要带到后续请求上,不必再走首页。""" + cookie_headers: list[str | None] = [] def handler(request: httpx.Request) -> httpx.Response: - seen.append(str(request.url)) - return httpx.Response(200, text=GOOD_PAGE) + cookie_headers.append(request.headers.get("cookie")) + return httpx.Response( + 200, + text=GOOD_PAGE, + headers={"set-cookie": "ak_bmsc=abc; Domain=.rakuten.co.jp; Path=/"}, + ) session = await build_session(handler) try: @@ -104,8 +108,30 @@ async def test_successful_warmup_is_reused_when_upstream_sets_no_cookie(): finally: await session.close() - assert seen.count("https://www.rakuten.co.jp/") == 1 - assert seen.count(TARGET) == 2 + assert cookie_headers[0] is None # 首个请求是冷的 + assert "ak_bmsc=abc" in (cookie_headers[1] or "") + + +async def test_expired_cookies_are_dropped_before_next_fetch(): + """cookie 罐超过 session_ttl_seconds 后要清空,不能带着过期 cookie 去撞。""" + cookie_headers: list[str | None] = [] + + def handler(request: httpx.Request) -> httpx.Response: + cookie_headers.append(request.headers.get("cookie")) + return httpx.Response( + 200, + text=GOOD_PAGE, + headers={"set-cookie": "ak_bmsc=abc; Domain=.rakuten.co.jp; Path=/"}, + ) + + session = await build_session(handler, settings=make_settings(session_ttl_seconds=0.0)) + try: + await session.fetch_html(TARGET, mobile=False) + await session.fetch_html(TARGET, mobile=False) + finally: + await session.close() + + assert cookie_headers == [None, None] # 每次都从干净状态起步 async def test_missing_state_marker_is_treated_as_blocked(): @@ -139,6 +165,32 @@ async def test_retry_recovers_when_a_later_attempt_succeeds(): assert attempts["n"] == 2 +async def test_first_failure_swaps_cookies_via_home_page(): + """首次失败时用首页换一套 cookie——这是首页 URL 唯一的用途。""" + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(str(request.url)) + if request.url.host == "www.rakuten.co.jp": + return httpx.Response( + 200, + text="home", + headers={"set-cookie": "ak_bmsc=fresh; Domain=.rakuten.co.jp; Path=/"}, + ) + if seen.count(TARGET) == 1: + return httpx.Response(200, text=BLOCK_PAGE) + return httpx.Response(200, text=GOOD_PAGE) + + session = await build_session(handler) + try: + assert await session.fetch_html(TARGET, mobile=False) == GOOD_PAGE + finally: + await session.close() + + # 首页只在失败之后出现一次,且排在两次目标页请求中间 + assert seen == [TARGET, "https://www.rakuten.co.jp/", TARGET] + + async def test_browser_fallback_supplies_cookies_and_page_on_persistent_block(): """浏览器已经取到页面时应直接采用,不再多打一次 HTTP""" browser = FakeBrowser( @@ -264,9 +316,9 @@ async def test_search_and_detail_use_separate_cookie_jars(): def handler(request: httpx.Request) -> httpx.Response: mobile = request.headers.get("sec-ch-ua-mobile") == "?1" - headers = {} - if request.url.host == "www.rakuten.co.jp": - headers["set-cookie"] = f"ak_bmsc={'sp' if mobile else 'pc'}; Domain=.rakuten.co.jp; Path=/" + headers = { + "set-cookie": f"ak_bmsc={'sp' if mobile else 'pc'}; Domain=.rakuten.co.jp; Path=/" + } return httpx.Response(200, text=GOOD_PAGE, headers=headers) session = await build_session(handler) @@ -282,7 +334,9 @@ async def test_search_and_detail_use_separate_cookie_jars(): assert sp_cookie == "sp" -async def test_profile_status_reports_warmup_state(): +async def test_profile_status_reports_cookie_state(): + """`warmed` 现在的语义是「当前有可复用的 Akamai cookie」,字段名为兼容保留。""" + def handler(request: httpx.Request) -> httpx.Response: headers = {"set-cookie": "ak_bmsc=abc; Domain=.rakuten.co.jp; Path=/"} return httpx.Response(200, text=GOOD_PAGE, headers=headers) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 3cf449d..9574fb0 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -1,22 +1,47 @@ """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 的行为。 不打真实网络:OTLPSpanExporter 创建时不发请求,BatchSpanProcessor 异步批量 -上报在没有 span 产生时也不会触发。 +上报在没有 span 产生时也不会触发。span 断言用独立的 InMemory provider,不碰 +全局 provider——OTel 的全局 provider 只允许设置一次,测试间共享会互相污染。 """ from __future__ import annotations +import pytest +from fastapi import FastAPI from opentelemetry import trace +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 app.shared import telemetry from app.shared.config import Settings +from app.shared.errors import OrderGuardError 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 @@ -24,6 +49,90 @@ def test_disabled_is_noop(): 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 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_enabled_initializes_and_shutdown_releases(): """enabled=true 时 setup 注册 TracerProvider,shutdown 后 _provider 复位""" settings = Settings( diff --git a/tests/test_worker_runner.py b/tests/test_worker_runner.py index 04e2a0a..191b503 100644 --- a/tests/test_worker_runner.py +++ b/tests/test_worker_runner.py @@ -20,8 +20,14 @@ from typing import Any import pytest +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 app.shared.errors import BrowserDeadError, CheckoutBlockedError, OrderGuardError from app.shared.task_state import OrderState, TaskStatus +from app.trading.worker import runner as runner_module from app.trading.worker import verify from app.trading.worker.evidence import EvidenceStore from app.trading.worker.local_db import LocalDB @@ -900,6 +906,130 @@ async def test_monitor_order_stops_after_max_checks_without_finding_order( assert gateway.reports == [] +# ---- 链路追踪:一个任务一条 trace ---- + + +@pytest.fixture +def spans(monkeypatch) -> InMemorySpanExporter: + """把 runner 模块级 tracer 换成写内存的 + + 不动全局 provider:OTel 的全局 provider 只允许设置一次,测试间共享会互相污染。 + """ + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(runner_module, "tracer", provider.get_tracer("test")) + return exporter + + +def _stub_happy_path(runner: WorkerRunner) -> None: + """把整条下单流程替换成桩,让 execute() 一路走到付款完成""" + + async def _clear(): + return {"removed_count": 0, "cart_count": 0, "html": "", "screenshot": b"p"} + + async def _snapshot(*args, **kwargs): + return PageSnapshot(html="", screenshot=b"p") + + async def _parse(html: str): + return CheckoutSummary(payable_yen=297) + + async def _submit(task): + return SubmitOutcome( + site_order_id="ord-1", evidence=PageSnapshot(html="", screenshot=b"p") + ) + + async def _unchanged(order_id: str): + return OrderStatusSnapshot(found=True, order_state=None) + + runner._site.clear_cart = _clear # type: ignore[assignment] + runner._site.add_to_cart = _snapshot # type: ignore[assignment] + runner._site.verify_cart = _snapshot # type: ignore[assignment] + runner._site.enter_checkout = _snapshot # type: ignore[assignment] + runner._site.parse_checkout = _parse # type: ignore[assignment] + runner._site.submit_order = _submit # type: ignore[assignment] + runner._site.pay = _snapshot # type: ignore[assignment] + runner._site.check_order_status = _unchanged # type: ignore[assignment] + + +async def test_order_task_emits_one_trace_with_a_span_per_step( + runner: WorkerRunner, spans: InMemorySpanExporter +): + """一笔下单 = 一条 trace,每个步骤一个子 span,全部挂在任务根 span 底下 + + worker 是后台 asyncio 任务,没有 HTTP 入口。不开这个根 span 的话,下游每次 + 出站调用都各自成为孤立 trace——观测后台上就只剩「一堆请求记录」,看不出 + 这是同一笔单、也看不出卡在哪一步。这条用例把链路结构钉住。 + """ + _stub_happy_path(runner) + + await runner.handle(_make_task(task_id="t1", intent={"item_url": "https://x/y/", "quantity": 2})) + + finished = spans.get_finished_spans() + root = next(s for s in finished if s.name == "order.task") + steps = [s for s in finished if s.name.startswith("order.step.")] + + # 步骤 span 覆盖整条流水,且顺序与执行顺序一致 + assert [s.name for s in steps] == [ + "order.step.cart-add", + "order.step.cart-check", + "order.step.order-confirm", + "order.step.order-submit", + "order.step.payment", + ] + # 每个步骤都挂在任务根底下,同一条 trace + for step in steps: + assert step.parent.span_id == root.context.span_id + assert step.context.trace_id == root.context.trace_id + + assert root.attributes["order.task_id"] == "t1" + assert root.attributes["order.route"] == "execute" + assert root.attributes["order.item_url"] == "https://x/y/" + # 证据路径落在 span 上:看到失败步骤能直接找到现场 + assert steps[0].attributes["order.evidence_ref"] + assert steps[3].attributes["order.site_order_id"] == "ord-1" + + +async def test_guard_failure_marks_task_span_as_error( + runner: WorkerRunner, spans: InMemorySpanExporter +): + """金额守卫拦下时任务根 span 要是 ERROR,并带上对外错误码 + + 闸门拦截会被 _execute_with_renewal 吞掉转 needs_human(不抛出 handle), + 所以根 span 很容易显示成功——这里确认它没有。 + """ + _stub_happy_path(runner) + + async def _pricey(html: str): + return CheckoutSummary(payable_yen=999_999) + + runner._site.parse_checkout = _pricey # type: ignore[assignment] + + await runner.handle(_make_task(task_id="t1")) + + root = next(s for s in spans.get_finished_spans() if s.name == "order.task") + assert root.status.status_code is StatusCode.ERROR + assert root.attributes["error.type"] == "OrderGuardError" + # 提交步骤没有发生:拦在确认页之后、提交之前 + names = [s.name for s in spans.get_finished_spans()] + assert "order.step.order-submit" not in names + + +async def test_idle_lease_does_not_emit_spans(runner: WorkerRunner, spans: InMemorySpanExporter): + """空转的长轮询不产生 span——否则每 30 秒一条孤立 trace 会把观测后台刷满""" + runner._running = True + + async def _lease_then_stop(worker_id: str, *, wait: int = 30, site: str | None = None): + runner.stop() # 领到空就停,避免死循环 + return None + + runner._gateway.lease = _lease_then_stop # type: ignore[assignment] + + await runner.run() + + assert spans.get_finished_spans() == () + + async def test_monitor_order_swallows_check_errors_and_keeps_polling( runner: WorkerRunner, ):