Python后端中间件专题22：Elasticsearch 宕机以后——读写边界与显式降级

Python后端中间件专题22Elasticsearch 宕机以后——读写边界与显式降级值班告警显示 Elasticsearch 连接拒绝。此时有人建议捕获异常后返回200 {items:[]}“页面至少不报错”。这是危险的伪降级用户看到的是“没有工单”系统实际知道的是“无法判断”。两者会驱动完全不同的业务决定。本课不追求搜索依赖永不失败而是把故障限制在搜索能力工单详情和创建仍由 PostgreSQL 服务搜索明确返回 503恢复后可以从 Outbox 或事实库补投影。我们会沿 HTTP adapter→FastAPI 路由看异常如何保持语义而不是把所有错误吞成空列表。上一课练习答案答案 EX-21-01 [code]保存为reindex_duplicate_probe.py从project/运行。源集合重复意味着我们的“预期 ID 集合”已经有歧义必须在创建候选索引前阻止。importasynciofromticketflow.search.reindeximportReindexMismatch,rebuild_and_switchclassRecordingBackend:def__init__(self):self.actions[]asyncdefcreate_index(self,name,definition):self.actions.append(create)asyncdefbulk_index(self,name,documents):self.actions.append(bulk)asyncdefdocument_versions(self,name):return{}asyncdefswitch_alias(self,alias,new_index):self.actions.append(switch)returnNoneasyncdefmain():backendRecordingBackend()documents[{id:ticket-1,version:1},{id:ticket-1,version:1},]try:awaitrebuild_and_switch(backend,aliasticketflow-tickets,new_indexticketflow-tickets-v2,source_readerlambda:documents,definition{},source_writes_frozenTrue,)exceptReindexMismatchaserror:assertstr(error)source contains duplicate document idselse:raiseAssertionError(duplicate source was accepted)assertbackend.actions[]print(duplicate source blocked before create)asyncio.run(main())从project/运行命令python reindex_duplicate_probe.py预期输出duplicate source blocked before create这里的关键不是 fake backend 多像 ES而是 production 函数在调用外部副作用前拒绝坏输入。真正的重建仍需第 21 课远端流程验收。答案 EX-21-02 [prose]回填取得的是某个读取时间窗里的字段值。随后工单 version 增长即使 v2 仍有相同 ID、总数也相同它的 title/status 可能旧于 PostgreSQL。可选方案包括记录回填水位再重放水位之后的 Outbox 事件直到追平回填期让投影事件同时写 v1/v2或在可接受窗口暂停变更并完成最后增量同步。TicketFlow 本次采用第三种从首份源快照直到 alias 切换都持有 PostgreSQL 写锁并校验候选 version 与第二份源快照。任何方案都必须以 version 比较拒绝乱序。alias 回滚只改变搜索读请求指向哪个投影不会也不应修改 PostgreSQL。工单更新已经是业务事实回滚到 v1 可能暂时读到旧投影因此操作员还要评估 v1 的落后程度并继续补事件而不是把搜索回滚宣传成业务数据回滚。把“无结果”和“无法搜索”分成两个 HTTP 契约ElasticsearchHTTP.search()有三类结果ES 成功响应解析 total、hits、highlight、sort 和两个聚合桶合法的零命中才返回空列表网络连接/超时或 ES 5xx抛SearchUnavailable(search service unavailable)ES 4xx说明请求/alias/权限等被拒绝抛带状态类别的SearchUnavailable供内部日志区分。FastAPI 的/tickets/search捕获该领域异常并返回 503GET /tickets/{id}根本不经过搜索 adapter仍查询 PostgreSQL。POST /tickets也不应同步要求 ES 成功它提交 ticketOutbox 后返回 201。于是故障矩阵是操作ES 503、PostgreSQL 正常时原因创建工单201事实和 Outbox 已提交投影稍后补按 ID 读详情200/404由当前租户下 PostgreSQL 事实决定全文搜索503无法证明结果完整不能伪装空结果若业务确实需要数据库兜底搜索也必须定义受限能力和性能边界例如只允许当前租户、近 7 天、前缀查询、严格限流并在响应里标出 degraded。无边界地将%keyword%打到主库不是高可用是把搜索故障扩大成数据库事故。TicketFlow 当前选择明确 503没有暗藏 SQL fallback。离线制造一次 503不用启动容器下面就是 23 会引用的本课练习形式也是本课核心实验。保存为search_503_probe.pyimportasyncioimporthttpxfromticketflow.search.http_adapterimportElasticsearchHTTP,SearchUnavailablefromticketflow.search.queryimportSearchFiltersasyncdefmain():asyncwithhttpx.AsyncClient(base_urlhttp://es.test:9200,transporthttpx.MockTransport(lambdarequest:httpx.Response(503)),)asclient:backendElasticsearchHTTP(client,aliasticketflow-tickets)try:awaitbackend.search(tenant_idtenant-a,queryvpn,filtersSearchFilters(),size10,)exceptSearchUnavailableaserror:assertstr(error)search service unavailableprint(explicit unavailable)else:raiseAssertionError(503 was presented as a successful empty search)asyncio.run(main())从project/运行命令python search_503_probe.py预期输出explicit unavailableMockTransport证明适配器翻译规则不证明 ES 已停止。远端 selectortests/integration/test_search_elasticsearch.py::test_real_elasticsearch_outage_preserves_ticket_get_and_search_returns_503先建立唯一测试工单再通过固定 harness 暂停本次隔离 Compose 的 elasticsearch核对已有详情200、搜索503 且不是空结果finally恢复服务并有界轮询搜索重新可用随后清理测试行。它不声称验证了故障期间创建工单或投影补齐。运行正式检查点从课程根运行python tools/verify_executable_revisions.py--lesson 22--python python此命令在全新环境运行 baseline 与 09–22 的累积本地合同。远端故障 selector 尚未运行状态为 PENDING不能把本地 fake 写成“ES 宕机实测通过”。故障分诊时先区分仅_search503、alias 404、Mapping 400、连接超时、集群只读/磁盘水位它们恢复动作不同。先保存请求 ID、alias、HTTP 状态和低基数 search outcome禁止把包含凭据的 ES URL 原样记日志。恢复后检查积压/版本必要时执行受控重建不以“容器 Up”作为恢复完成。本课练习练习 EX-22-01 [code]用httpx.MockTransport对ElasticsearchHTTP.search(tenant_idtenant-a, queryvpn, filtersSearchFilters(), size10)模拟 ES 503断言抛SearchUnavailable(search service unavailable)打印explicit unavailable。要求从project/、PYTHONPATHsrc环境运行。第 23 课会给出完整答案。练习 EX-22-02 [prose]假设 ES 503 而 PostgreSQL 可用分别说明工单详情、创建、搜索应返回什么解释为什么不能返回空搜索结果作为降级。再判断第 23 课增加 Nginx 统一入口后它是否能修复这个搜索故障。下一篇预告依赖降级说清之后第 23 课把两个 API 实例放到一个 Nginx 入口后面验证两个 upstream 都服务过请求、一台停机后仍能读取工单并划清可信转发头的责任。代理能切换实例却不能替 Elasticsearch 恢复数据。本课完整核心模块search/http_adapter.py以下是lesson-22的完整 HTTP adapter。它完整展示请求、异常翻译和响应塑形FastAPI 将SearchUnavailable映射为 503 的路由在api/app.py不是本模块的隐藏片段。from__future__importannotationsfromdataclassesimportdataclassimporthttpxfrom.queryimportSearchFilters,build_search_body,encode_cursorclassSearchUnavailable(RuntimeError):Search projection cannot serve this request; PostgreSQL remains authoritative.dataclass(frozenTrue,slotsTrue)classSearchHit:ticket_id:strscore:float|Nonesource:dict[str,object]highlight:dict[str,list[str]]sort:list[object]dataclass(frozenTrue,slotsTrue)classFacetBucket:One API-safe aggregation bucket, detached from Elasticsearch JSON.key:strcount:intdataclass(frozenTrue,slotsTrue)classSearchFacets:statuses:list[FacetBucket]priorities:list[FacetBucket]dataclass(frozenTrue,slotsTrue)classSearchResult:total:inthits:list[SearchHit]next_cursor:str|Nonefacets:SearchFacetsdef_facet_buckets(payload:object,name:str)-list[FacetBucket]:Reject malformed buckets instead of publishing plausible zero counts.ifnotisinstance(payload,dict)ornotisinstance(payload.get(buckets),list):raiseSearchUnavailable(fmalformed aggregation:{name})result:list[FacetBucket][]forrawinpayload[buckets]:if(notisinstance(raw,dict)orkeynotinrawornotisinstance(raw.get(doc_count),int)orisinstance(raw.get(doc_count),bool)orint(raw[doc_count])0):raiseSearchUnavailable(fmalformed aggregation:{name})result.append(FacetBucket(str(raw[key]),int(raw[doc_count])))returnresultclassElasticsearchHTTP:Small HTTP adapter; it does not leak ES response shapes into API code.def__init__(self,client:httpx.AsyncClient,*,alias:str)-None:self._clientclient self._aliasaliasasyncdefindex(self,**kwargs:object)-None:responseawaitself._client.put(f/{kwargs[index]}/_doc/{kwargs[id]},params{version:kwargs[version],version_type:kwargs[version_type]},jsonkwargs[document],)ifresponse.status_codenotin(200,201,409):raiseSearchUnavailable(findex service returned HTTP{response.status_code})asyncdefsearch(self,*,tenant_id:str,query:str,filters:SearchFilters,size:int,cursor:str|NoneNone,)-SearchResult:try:responseawaitself._client.post(f/{self._alias}/_search,jsonbuild_search_body(tenant_idtenant_id,queryquery,filtersfilters,sizesize,cursorcursor,),)excepthttpx.TransportErroraserror:raiseSearchUnavailable(search service unavailable)fromerrorelse:ifresponse.status_code500:raiseSearchUnavailable(search service unavailable)ifresponse.status_code400:raiseSearchUnavailable(fsearch request rejected with HTTP{response.status_code})payloadresponse.json()raw_hitspayload.get(hits,{}).get(hits,[])hits[SearchHit(ticket_idstr(raw[_id]),scoreraw.get(_score),sourcedict(raw.get(_source,{})),highlight{key:list(value)forkey,valueinraw.get(highlight,{}).items()},sortlist(raw.get(sort,[])))forrawinraw_hits]total_rawpayload.get(hits,{}).get(total,0)totalint(total_raw.get(value,0)ifisinstance(total_raw,dict)elsetotal_raw)next_cursorencode_cursor(hits[-1].sort)iflen(hits)sizeandhits[-1].sortelseNoneaggregationspayload.get(aggregations)ifnotisinstance(aggregations,dict):raiseSearchUnavailable(malformed aggregation: aggregations)facetsSearchFacets(statuses_facet_buckets(aggregations.get(status_counts),status_counts),priorities_facet_buckets(aggregations.get(priority_counts),priority_counts),)returnSearchResult(totaltotal,hitshits,next_cursornext_cursor,facetsfacets)