From 03d7bcea39b22aebd0416a52e0255558361bb7de Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 5 Aug 2026 12:13:14 +0800 Subject: [PATCH] =?UTF-8?q?feat(health):=20=E8=A2=AB=E5=8A=A8=E5=BC=8F?= =?UTF-8?q?=E5=81=A5=E5=BA=B7=E6=A3=80=E6=9F=A5=20API=EF=BC=88FastAPI?= =?UTF-8?q?=EF=BC=8C=E8=A6=86=E7=9B=96=E6=9C=8D=E5=8A=A1=E5=81=A5=E5=BA=B7?= =?UTF-8?q?+=E6=95=B0=E6=8D=AE=E5=81=A5=E5=BA=B7=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在同步进程内内置 FastAPI/uvicorn HTTP endpoint (GET /health)，返回结构化 健康快照，便于运维/监控被动探活： - 服务健康：cycle_lag（核心指标，超过 3×poll_interval 判 unhealthy）、 队列状态（pending/error/dead/cleaned 实时查）、capture 累计（enqueued/ deferred/aged_out）、最近 cycle 时间/耗时/错误。 - 数据健康：解析最近一次定时 compare_ids 报告（不跑全量 compare 太重）、 实时查 SyncLogArchive 异常痕迹（AgedOut/降级）、SyncQueue dead 行样本。 status 三态：healthy(200) / degraded(503, 有dead/error/aged_out/compare 不一致) / unhealthy(503, 主循环停滞)。状态码映射便于按码告警。 线程模型：uvicorn 占主线程，capture/apply/cleanup 循环跑后台 daemon 线程， NSSM 停服务时主线程退出、daemon 自动终止。health.enabled=false 时退化为 旧行为（同步循环占主线程）。Web 层(routes/)与业务逻辑(HealthChecker) 解耦，后期加运维接口（compare触发/死信管理/metrics）零结构改动。 零 SQL/零 schema 改动。新增依赖 fastapi/uvicorn[standard]。 --- config.example.yaml | 6 + docs/plan-health-check-api.md | 164 ++++++++++++++++ docs/plan-health-check-fastapi.md | 215 +++++++++++++++++++++ requirements.txt | 2 + src/sync/config.py | 14 ++ src/sync/health.py | 308 ++++++++++++++++++++++++++++++ src/sync/service.py | 90 +++++++-- src/sync/web/__init__.py | 11 ++ src/sync/web/app.py | 28 +++ src/sync/web/deps.py | 42 ++++ src/sync/web/routes/__init__.py | 2 + src/sync/web/routes/health.py | 31 +++ tests/test_health.py | 183 ++++++++++++++++++ 13 files changed, 1081 insertions(+), 15 deletions(-) create mode 100644 docs/plan-health-check-api.md create mode 100644 docs/plan-health-check-fastapi.md create mode 100644 src/sync/health.py create mode 100644 src/sync/web/__init__.py create mode 100644 src/sync/web/app.py create mode 100644 src/sync/web/deps.py create mode 100644 src/sync/web/routes/__init__.py create mode 100644 src/sync/web/routes/health.py create mode 100644 tests/test_health.py diff --git a/config.example.yaml b/config.example.yaml index cd0a87a..f467b3b 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -21,6 +21,12 @@ logging: level: INFO path: "" +health: + enabled: true # 开启被动式健康检查 HTTP endpoint + host: "0.0.0.0" # 监听地址，0.0.0.0 内网/FRP 均可访问 + port: 8421 # 健康检查端口 + log_level: "warning" # uvicorn 自身日志级别，避免刷屏 + files: - {file: "一车间.accdb", root: 2026, schema: "workshopOne", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "一车间每日催货落实记录_停"]} - {file: "二车间.accdb", root: 2026, schema: "workshopTwo", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]} diff --git a/docs/plan-health-check-api.md b/docs/plan-health-check-api.md new file mode 100644 index 0000000..d695ff9 --- /dev/null +++ b/docs/plan-health-check-api.md @@ -0,0 +1,164 @@ +# 方案：被动式健康检查 API（数据健康 + 服务健康） + +> 目标：在增量同步进程内内置一个轻量 HTTP endpoint，对外暴露结构化健康快照，覆盖「服务健康」与「数据健康」两层。 +> 形态：单进程内置（不另起服务、不新增 NSSM 配置），HTTP server 跑在后台线程，主同步循环零侵入。 + +--- + +## 一、设计原则 + +1. **单进程内置**：HTTP server 用标准库 `http.server` + 后台线程，跑在同步进程内。不另起进程、不新增 NSSM 服务、不引入 Flask 等第三方依赖。 +2. **主循环零侵入**：健康检查线程只读共享状态、只读 SQL，不触碰 capture/apply/cleanup 任何路径。HTTP server 崩溃不影响同步。 +3. **数据实时查**：服务健康（最后 cycle 时间/队列状态）从进程内存的状态对象读；数据健康（Access↔SQL 漂移、dead 行）实时查 SQL Server 和最近 compare 报告。 +4. **配置可选**：通过 `config.yaml` 的 `health` 段控制开关/端口/路径，默认开启但端口可配，关闭时不起线程。 + +--- + +## 二、健康检查覆盖的两层 + +### 服务健康（Service Health） +回答："同步进程在跑吗？最近正常工作吗？队列有没有卡死？" + +| 指标 | 来源 | 含义 | +|------|------|------| +| `status` | 综合判定 | healthy / degraded / unhealthy | +| `pid`, `started_at`, `uptime_seconds` | 进程 | 进程存活与运行时长 | +| `last_cycle_at`, `last_cycle_duration_s` | 内存状态 | 最近一次 cycle 时间与耗时 | +| `cycle_lag_seconds` | 实时计算 = now - last_cycle_at | **核心指标**：距上次 cycle 间隔，超过阈值→unhealthy | +| `last_cycle_active` | 内存状态 | 上轮是否处理了变更 | +| `queue` | 实时查 SyncQueue | pending/error/dead/cleaned 计数（dead>0→degraded） | +| `capture_since_start` | 内存状态累计 | 服务启动至今的 deferred/aged_out/enqueued 累计（aged_out>0→degraded） | + +### 数据健康（Data Health） +回答："Access 和 SQL 数据一致吗？最近核对结果如何？" + +| 指标 | 来源 | 含义 | +|------|------|------| +| `last_compare` | 解析最近的 `compare_ids_*.log` 报告头 | 最近一次 compare 的时间、mismatch 表数、是否一致 | +| `drift_tables` | 实时查 SyncLogArchive | 启动至今出现过的降级/异常记录涉及的表（AgedOut、Original≠Processed） | +| `dead_rows_sample` | 实时查 SyncQueue dead 行 | 卡死行的样本（表名/RecordID/错误），辅助定位 | + +> 注：数据健康**不**在每次请求时跑全量 compare（太重），而是读"最近一次定时 compare 的结果" + "归档表/队列里的异常痕迹"。全量漂移检测仍由每日 05:08 的 `compare_ids` 计划任务兜底。 + +--- + +## 三、status 综合判定逻辑 + +``` +unhealthy : cycle_lag_seconds > 3 × poll_interval (主循环停滞) + 或 进程内存状态长时间未更新（疑似卡死） + +degraded : queue.dead > 0 (有放弃的变更，数据可能缺) + 或 queue.error > 0 (有失败待重试) + 或 capture.aged_out > 0 (有读不到的 Insert 被判死) + 或 last_compare.mismatch (最近核对不一致) + +healthy : 其余情况 +``` + +`cycle_lag` 用 `3 × poll_interval`（默认 poll=10s → 30s）作阈值：正常空闲 cycle 间隔就是 10s 左右，超过 3 倍说明主循环被卡（比如 Access 锁等待）。 + +--- + +## 四、接口契约 + +**请求**：`GET /health`（也可配 `GET /` 简化） + +**响应**：HTTP 200 + JSON（无论 healthy/degraded/unhealthy 都返回 200，状态在 body 里；这样探活失败和网络故障可区分——网络故障是连不上/超时，服务不健康是 200 但 status 字段非 healthy）。 + +> 可选：对 unhealthy 同时返回 HTTP 503，便于直接接入按状态码告警的监控平台。这点待定（见决策点）。 + +**响应示例**： +```json +{ + "service": "DataMacroSync", + "status": "healthy", + "checked_at": "2026-08-05T11:30:00", + "pid": 1234, + "started_at": "2026-08-05T11:01:51", + "uptime_seconds": 1689, + "service_health": { + "last_cycle_at": "2026-08-05T11:29:50", + "last_cycle_duration_s": 1.05, + "cycle_lag_seconds": 10.0, + "last_cycle_active": false, + "queue": {"pending": 0, "error": 0, "dead": 0, "cleaned": 3829}, + "capture_since_start": {"enqueued": 103, "deferred": 0, "aged_out": 0} + }, + "data_health": { + "last_compare": { + "at": "2026-08-05T05:08:21", + "granularity": "ids", + "tables_compared": 81, + "mismatch": false, + "mismatch_tables": 0 + }, + "drift_tables": [], + "dead_rows_sample": [] + } +} +``` + +--- + +## 五、实现拆解 + +### 5.1 新增 `src/sync/health.py`（核心，约 200 行） +- `ServiceState` 数据类：进程内全局状态对象，由 `cycle()` 每轮更新（last_cycle_at / duration / active / 累计计数）。 +- `HealthChecker` 类：持有 `ServiceState` + `SyncConfig`，方法 `snapshot()` 返回上面 JSON 对应的 dict。实时查 SQL 用一次性 SqlWriter（查完即关）。 +- `run_health_server(state, cfg, host, port)`：起 `http.server.ThreadingHTTPServer`，handler 调 `HealthChecker.snapshot()` 序列化返回。 + +### 5.2 改 `src/sync/service.py` +- `run()` 入口创建 `ServiceState` 实例，传给 `cycle()`；`cycle()` 结束时更新 state（last_cycle_at 等）。 +- `run()` 起健康检查后台线程（`threading.Thread(target=run_health_server, daemon=True)`），主循环照常。 +- 新增累计：把每轮 `CaptureStats` 的 enqueued/deferred/aged_out 累加进 `ServiceState`。 + +### 5.3 改 `src/sync/config.py` +RuntimeConfig 或新增 HealthConfig： +```python +class HealthConfig(BaseModel): + enabled: bool = True + host: str = "0.0.0.0" # 监听地址，内网可访问 + port: int = 8421 # 健康检查端口 + path: str = "/health" # URL 路径 +``` +SyncConfig 增 `health: HealthConfig = HealthConfig()`（默认值，老 config.yaml 无需改动）。 + +### 5.4 改 `config.example.yaml` +补 `health` 段示例。 + +--- + +## 六、不改动的地方 + +| 模块 | 是否改动 | 原因 | +|------|---------|------| +| capture/apply/cleanup | ❌ | 主循环逻辑零侵入，只由 cycle 更新一个内存 state | +| sql_writer 的写方法 | ❌ | 健康检查只用现有的只读查询方法（queue_status_summary 等），不新增写操作 | +| NSSM 配置 | ❌ | 单进程内置，端口由进程自己起 | +| main.py | 小改 | `incremental --loop` 分支照常走 `service.run`（健康线程在 run 内起） | + +--- + +## 七、改动文件清单 + +| 文件 | 改动 | +|------|------| +| `src/sync/health.py` | **新增** ServiceState + HealthChecker + run_health_server | +| `src/sync/config.py` | 新增 HealthConfig，SyncConfig 加 health 字段 | +| `src/sync/service.py` | run() 起 health 线程 + 传 state；cycle() 更新 state | +| `config.example.yaml` | 补 health 段示例 | +| `tests/test_health.py` | **新增** 覆盖 status 判定、snapshot 结构、各状态组合 | + +**零 SQL 改动、零 schema 改动、零新第三方依赖（标准库 http.server）。** + +--- + +## 八、待你确认的决策点 + +1. **端口** `8421` 是否合适？（避开 114 上已有服务端口） +2. **unhealthy 的 HTTP 状态码**：始终 200（状态在 body）vs unhealthy/degraded 返回 503（便于按码告警）？我倾向后者。 +3. **host 监听地址**：`0.0.0.0`（内网/FRP 都可访问）vs `127.0.0.1`（仅本机，需配合 FRP 转发）？我倾向 `0.0.0.0`。 +4. **drift_tables 查询范围**：查"启动至今"的异常归档 vs 查"最近 N 小时"？我倾向"启动至今"（量可控，且能发现历史遗留）。 + +确认后即实施。 diff --git a/docs/plan-health-check-fastapi.md b/docs/plan-health-check-fastapi.md new file mode 100644 index 0000000..e5d01ca --- /dev/null +++ b/docs/plan-health-check-fastapi.md @@ -0,0 +1,215 @@ +# 方案 v2：FastAPI 健康检查（面向多接口扩展） + +> 基于"后期会引入更多运维接口"的前提，采用 FastAPI。 +> **与 v1（标准库 http.server）的核心差异在部署模型**——FastAPI/uvicorn 阻塞主线程，必须重新设计进程结构。本方案先解决这个架构问题，再展开实现。 + +--- + +## 一、核心架构决策：同步循环放哪个线程？ + +FastAPI 的标准运行方式 `uvicorn.run(app)` 会**阻塞主线程**。而现有项目里 `service.run()`（同步循环）是主线程。两者都要"常驻"，必须有一个让出主线程。这是用 FastAPI 唯一的硬约束，两条路径： + +### 方案 A：FastAPI 主线程 + 同步循环后台线程（✅ 推荐） + +``` +NSSM 启动 → main.py incremental --loop + ├─ 主线程: uvicorn.run(app) ← FastAPI 常驻 + └─ 后台线程(daemon): service cycle ← 同步循环搬到后台 +``` + +- **优点**：FastAPI 在主线程，signal handling、uvicorn 内部机制都按官方推荐姿势跑，最稳；未来加接口、加中间件、接 Prometheus 都顺畅。 +- **代价**：同步循环从主线程移到后台线程。但 `cycle()` 本身是纯函数式的（每轮独立、用完即关 writer），搬到后台线程风险可控——它本来就是为"被反复调用"设计的。 + +### 方案 B：同步循环主线程 + FastAPI 后台线程（❌ 不推荐） + +- **缺点**：uvicorn 官方明确不推荐嵌入非主线程，signal handler、事件循环绑定有边角问题；且 NSSM 的进程身份含糊（管的是"同步服务"还是"Web服务"？）。 + +**采用方案 A。** 下面所有实现都基于 A。 + +--- + +## 二、进程结构与生命周期 + +```mermaid +flowchart TD + NSSM[NSSM 启动 main.py incremental --loop] --> MAIN[主线程] + MAIN -->|"启动顺序"| S1[1. setup_logging] + S1 --> S2[2. 创建 ServiceState 全局状态] + S2 --> S3[3. 启动同步后台线程 daemon] + S3 --> S4[4. uvicorn.run 主线程阻塞] + S3 -.->|daemon 线程| CYC[cycle 循环
每轮更新 ServiceState] + S4 -.->|HTTP 请求到来| APP[FastAPI app] + APP -->|读| S2 + APP -->|实时查 SQL| DB[(SQL Server)] + CYC -->|写| S2 + CYC -->|读写| AC[(Access)] & DB + + style S4 fill:#e3f2fd,stroke:#1976d2 + style CYC fill:#fff3e0,stroke:#f57c00 + style APP fill:#e8f5e9,stroke:#388e3c +``` + +关键点： +- **ServiceState 是两线程间的唯一桥梁**：同步线程写、HTTP 线程读。用 `threading.Lock` 保护（或用简单的不可变快照替换，避免锁）。 +- **daemon 线程**：同步循环线程设 `daemon=True`，主进程退出时自动终止，不留孤儿。 +- **uvicorn 退出即进程退出**：NSSM stop → uvicorn 收到信号 → 主线程结束 → daemon 同步线程随之终止。 + +--- + +## 三、健康检查覆盖的两层（同 v1） + +### 服务健康（Service Health） +| 指标 | 来源 | +|------|------| +| `status` | 综合判定 healthy/degraded/unhealthy | +| `pid`, `started_at`, `uptime_seconds` | 进程 | +| `last_cycle_at`, `last_cycle_duration_s` | 内存状态（同步线程每轮更新） | +| `cycle_lag_seconds` | now - last_cycle_at（**核心指标**） | +| `queue` | 实时查 SyncQueue（pending/error/dead/cleaned） | +| `capture_since_start` | 内存累计（enqueued/deferred/aged_out） | + +### 数据健康（Data Health） +| 指标 | 来源 | +|------|------| +| `last_compare` | 解析最近 `compare_ids_*.log` 报告头 | +| `drift_tables` | 实时查 SyncLogArchive（AgedOut、Original≠Processed） | +| `dead_rows_sample` | 实时查 SyncQueue dead 行样本 | + +**数据健康不跑全量 compare**（太重），靠"最近定时 compare 结果 + 异常痕迹实时查"。全量漂移仍由每日 05:08 compare 兜底。 + +--- + +## 四、status 综合判定（同 v1） + +``` +unhealthy : cycle_lag_seconds > 3 × poll_interval (主循环停滞) +degraded : queue.dead > 0 或 queue.error > 0 + 或 capture.aged_out > 0 + 或 last_compare.mismatch +healthy : 其余 +``` + +unhealthy/degraded 时 HTTP 返回 **503**，healthy 返回 200，便于按状态码告警。 + +--- + +## 五、FastAPI 应用结构（为后期扩展铺路） + +``` +src/sync/ + web/ + __init__.py + app.py # FastAPI 实例 + 全局依赖（state/cfg 注入） + deps.py # Depends(): 取 ServiceState / SyncConfig + schemas.py # Pydantic 响应模型（HealthResponse 等） + routes/ + __init__.py + health.py # GET /health（本次实现） + # 后期: compare.py (触发/查询核对)、queue.py (死信管理)、metrics.py... +``` + +- 后期加接口只需在 `routes/` 加文件 + 在 `app.py` `include_router`，结构清晰。 +- 用 FastAPI 的 `Depends` 注入共享的 `ServiceState`，避免全局变量。 +- 响应用 Pydantic 模型（`schemas.py`），自动生成 `/docs` 给运维查阅。 + +**本次只实现 `routes/health.py`**，但目录结构一步到位，后期加接口零结构改动。 + +--- + +## 六、配置（config.py 新增） + +```python +class HealthConfig(BaseModel): + enabled: bool = True + host: str = "0.0.0.0" # 监听地址 + port: int = 8421 # 健康检查端口 + # uvicorn 运行参数 + log_level: str = "warning" # uvicorn 自身日志级别（避免刷屏） + +class SyncConfig(BaseModel): + # ... 既有字段 ... + health: HealthConfig = HealthConfig() # 默认开启，老 config.yaml 无需改 +``` + +--- + +## 七、service.py 改造（线程模型变更） + +`run()` 从"主线程跑循环"变为"启动后台同步线程 + 主线程跑 uvicorn"： + +```python +def run(cfg): + setup_logging(cfg.logging) + state = ServiceState(started_at=datetime.now(), pid=os.getpid()) + # 后台同步线程 + sync_thread = threading.Thread( + target=_sync_loop, args=(cfg, state), daemon=True, name="sync-cycle" + ) + sync_thread.start() + # 主线程跑 FastAPI（阻塞） + if cfg.health.enabled: + from sync.web.app import create_app + import uvicorn + app = create_app(state, cfg) + uvicorn.run(app, host=cfg.health.host, port=cfg.health.port, + log_level=cfg.health.log_level) + else: + # 健康检查关闭：主线程直接跑同步循环（兼容旧行为） + _sync_loop(cfg, state) + +def _sync_loop(cfg, state): + """同步循环（原 run() 的 while True 主体，抽出供后台线程调用）。""" + idle_since = None + while True: + t0 = time.monotonic() + active = cycle(cfg, state) # cycle 增加 state 参数用于更新状态 + state.update_cycle(active, duration=time.monotonic()-t0) # 新增 + # idle heartbeat 逻辑保留... + time.sleep(cfg.runtime.poll_interval_seconds) +``` + +`cycle(cfg, state)` 末尾新增 `state.update_cycle(...)`，并把每轮 `CaptureStats` 累计进 state。 + +--- + +## 八、依赖更新 + +`requirements.txt` 增加： +``` +fastapi>=0.110.0 +uvicorn[standard]>=0.27.0 +``` +（`uvicorn[standard]` 含 uvloop/httptools，性能更好；纯 Windows 下 uvloop 不可用会自动降级，无影响。） + +测试依赖（dev，不入 requirements）：`httpx`（FastAPI 测试用 `TestClient` 依赖它）。 + +--- + +## 九、改动文件清单 + +| 文件 | 改动 | +|------|------| +| `src/sync/web/app.py` | **新增** create_app + FastAPI 实例 | +| `src/sync/web/deps.py` | **新增** Depends 注入 | +| `src/sync/web/schemas.py` | **新增** Pydantic 响应模型 | +| `src/sync/web/routes/health.py` | **新增** GET /health | +| `src/sync/health.py` | **新增** ServiceState + HealthChecker（业务逻辑，与 Web 解耦） | +| `src/sync/service.py` | 改 run() 线程模型 + cycle() 更新 state | +| `src/sync/config.py` | 新增 HealthConfig | +| `requirements.txt` | 加 fastapi / uvicorn | +| `config.example.yaml` | 补 health 段示例 | +| `tests/test_health.py` | **新增** 覆盖 status 判定/snapshot/TestClient | + +**零 SQL 改动、零 schema 改动。Web 层与业务逻辑（HealthChecker）解耦，后期加接口只动 web/routes/。** + +--- + +## 十、待确认决策点 + +1. **端口 8421**？ +2. **unhealthy/degraded 返回 503**（按码告警友好）确认？ +3. **host 0.0.0.0**（内网+FRP 可访问）确认？ +4. **同步循环放后台 daemon 线程**（方案A）确认？这是与 v1 最大的结构差异。 +5. **drift_tables 查询范围**：启动至今 vs 最近 N 小时？倾向启动至今。 + +确认后实施。 diff --git a/requirements.txt b/requirements.txt index 2b00cfb..cd73127 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,6 @@ pyodbc>=5.0.1 PyYAML>=6.0.1 pydantic>=2.6.0 +fastapi>=0.110.0 +uvicorn[standard]>=0.27.0 pytest>=8.0.0 diff --git a/src/sync/config.py b/src/sync/config.py index 2436438..532ec66 100644 --- a/src/sync/config.py +++ b/src/sync/config.py @@ -53,12 +53,26 @@ class FileMapping(BaseModel): def target_table(self, access_table: str) -> str: return f"{access_table}{self.year_suffix}" + +class HealthConfig(BaseModel): + """被动式健康检查 HTTP endpoint 配置（FastAPI/uvicorn）。 + + 开启时 uvicorn 占主线程、同步循环跑后台 daemon 线程；关闭时退化为 + 旧行为（同步循环占主线程）。默认开启，老 config.yaml 无需改动。 + """ + enabled: bool = True + host: str = "0.0.0.0" # 监听地址，0.0.0.0 内网/FRP 均可访问 + port: int = 8421 # 健康检查端口 + log_level: str = "warning" # uvicorn 自身日志级别，避免刷屏 + + class SyncConfig(BaseModel): sql_server: SqlServerConfig access: AccessConfig runtime: RuntimeConfig files: list[FileMapping] logging: dict | None = None + health: HealthConfig = HealthConfig() def load_config(path: str) -> SyncConfig: with open(path, "r", encoding="utf-8") as f: diff --git a/src/sync/health.py b/src/sync/health.py new file mode 100644 index 0000000..5dfdb1f --- /dev/null +++ b/src/sync/health.py @@ -0,0 +1,308 @@ +"""Health-check business logic, decoupled from the web layer. + +Two pieces: + +* :class:`ServiceState` -- the in-memory state shared between the sync thread + (writer) and the FastAPI thread (reader). Updated by ``service.cycle`` once + per pass. Lock-free reads are safe because the snapshot is built by copying + the small set of scalar fields under a short lock; the sync loop never blocks + on this lock. + +* :class:`HealthChecker` -- builds the structured snapshot dict that the + ``/health`` endpoint returns. It reads ``ServiceState`` for the live + process/cycle metrics, then queries SQL Server once for queue health and the + audit/archive trail for data-health signals. It also parses the most recent + ``compare_ids_*.log`` report header so data health reflects the last daily + compare without re-running a full compare on every request (which would be + far too heavy). + +``status`` is derived: + +* ``unhealthy`` -- the sync loop appears stalled (cycle lag exceeds a multiple + of the poll interval, i.e. capture/apply is wedged, typically on an Access + lock wait); +* ``degraded`` -- the loop is running but something needs attention: queue + rows stuck in ``error``/``dead``, an Insert that aged out to dead, or the + last compare found mismatches; +* ``healthy`` -- otherwise. +""" +from __future__ import annotations + +import datetime as _dt +import logging +import os +import re +import threading + +from .config import SyncConfig +from .sql_writer import SqlWriter + +log = logging.getLogger(__name__) + +# How stale the last cycle may be (in poll intervals) before status flips to +# unhealthy. A normal idle cycle paces at poll_interval; 3x tolerates one slow +# pass (e.g. a large capture batch) without a false alarm. +CYCLE_LAG_MULTIPLIER = 3 + +_COMPARE_HEADER_RE = re.compile( + r"生成时间\s*:\s*(?P\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})" + r"|不一致\s*MISMATCH\s*:\s*(?P\d+)" + r"|一致\s*MATCH\s*:\s*(?P\d+)" +) + + +class ServiceState: + """In-memory snapshot of the running sync service, updated each cycle. + + Written by the sync thread (in :func:`sync.service.cycle`) and read by the + health endpoint thread. A single short lock guards the mutable fields; the + health reader copies them out atomically so it never sees a torn update. + """ + + def __init__(self, started_at: _dt.datetime, pid: int): + self.started_at = started_at + self.pid = pid + self._lock = threading.Lock() + # Mutable fields (updated each cycle): + self.last_cycle_at: _dt.datetime | None = None + self.last_cycle_duration_s: float | None = None + self.last_cycle_active: bool = False + self.last_cycle_error: str | None = None + # Cumulative since process start: + self.capture_enqueued: int = 0 + self.capture_deferred: int = 0 + self.capture_aged_out: int = 0 + + def update_cycle(self, active: bool, duration_s: float, + capture=None, error: str | None = None) -> None: + """Record the outcome of one cycle (called by the sync thread). + + ``capture`` is an optional :class:`sync.capture.CaptureStats`; when + given its counters are folded into the running totals. + """ + with self._lock: + self.last_cycle_at = _dt.datetime.now() + self.last_cycle_duration_s = duration_s + self.last_cycle_active = active + self.last_cycle_error = error + if capture is not None: + self.capture_enqueued += capture.enqueued + self.capture_deferred += capture.deferred + self.capture_aged_out += capture.aged_out + + def snapshot(self) -> dict: + """Return a thread-safe copy of the mutable fields (called by reader).""" + with self._lock: + return { + "started_at": self.started_at, + "pid": self.pid, + "last_cycle_at": self.last_cycle_at, + "last_cycle_duration_s": self.last_cycle_duration_s, + "last_cycle_active": self.last_cycle_active, + "last_cycle_error": self.last_cycle_error, + "capture_enqueued": self.capture_enqueued, + "capture_deferred": self.capture_deferred, + "capture_aged_out": self.capture_aged_out, + } + + +class HealthChecker: + """Builds the structured health snapshot for the ``/health`` endpoint.""" + + def __init__(self, state: ServiceState, cfg: SyncConfig): + self.state = state + self.cfg = cfg + + def snapshot(self) -> dict: + """Return the full health dict (service_health + data_health + status).""" + snap = self.state.snapshot() + now = _dt.datetime.now() + service = self._service_health(snap, now) + data = self._data_health() + status = self._derive_status(service, data) + return { + "service": "DataMacroSync", + "status": status, + "checked_at": now.isoformat(timespec="seconds"), + "pid": snap["pid"], + "started_at": snap["started_at"].isoformat(timespec="seconds"), + "uptime_seconds": int((now - snap["started_at"]).total_seconds()), + "service_health": service, + "data_health": data, + } + + # ------------------------------------------------------------------ service + + def _service_health(self, snap: dict, now: _dt.datetime) -> dict: + last = snap["last_cycle_at"] + cycle_lag = (now - last).total_seconds() if last else None + return { + "last_cycle_at": last.isoformat(timespec="seconds") if last else None, + "last_cycle_duration_s": snap["last_cycle_duration_s"], + "cycle_lag_seconds": cycle_lag, + "last_cycle_active": snap["last_cycle_active"], + "last_cycle_error": snap["last_cycle_error"], + "queue": self._queue_status(), + "capture_since_start": { + "enqueued": snap["capture_enqueued"], + "deferred": snap["capture_deferred"], + "aged_out": snap["capture_aged_out"], + }, + } + + def _queue_status(self) -> dict: + """Live SyncQueue row counts per status (best-effort).""" + try: + w = SqlWriter( + self.cfg.sql_server.conn_str, + self.cfg.sql_server.sync_queue_table, + self.cfg.sql_server.archive_table, + self.cfg.sql_server.apply_proc, + self.cfg.sql_server.apply_runlog_table, + ) + try: + return w.queue_status_summary() + finally: + w.close() + except Exception as e: + log.warning("health: queue status query failed: %s", e) + return {"error": str(e)} + + # -------------------------------------------------------------------- data + + def _data_health(self) -> dict: + return { + "last_compare": self._last_compare(), + "drift_tables": self._drift_tables(), + "dead_rows_sample": self._dead_rows_sample(), + } + + def _last_compare(self) -> dict: + """Parse the most recent compare_ids report header. + + The compare task writes ``logs/compare_ids_.log`` once a + day; reading its header is far cheaper than re-running a full compare + on every health request. + """ + log_dir = self._log_dir() + try: + candidates = sorted( + (f for f in os.listdir(log_dir) + if f.startswith("compare_ids_") and f.endswith(".log")), + reverse=True, + ) + except OSError: + return {"available": False} + if not candidates: + return {"available": False} + path = os.path.join(log_dir, candidates[0]) + return self._parse_compare_report(path) + + @staticmethod + def _parse_compare_report(path: str) -> dict: + ts = match_count = mismatch_count = None + try: + with open(path, "r", encoding="utf-8") as f: + for line in f: + if "生成时间" in line: + m = re.search(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", line) + if m: + ts = m.group(0) + elif "一致 MATCH" in line: + m = re.search(r":\s*(\d+)", line) + if m: + match_count = int(m.group(1)) + elif "不一致 MISMATCH" in line: + m = re.search(r":\s*(\d+)", line) + if m: + mismatch_count = int(m.group(1)) + except OSError as e: + return {"available": False, "error": str(e)} + return { + "available": True, + "report": os.path.basename(path), + "at": ts, + "tables_match": match_count, + "tables_mismatch": mismatch_count or 0, + "mismatch": (mismatch_count or 0) > 0, + } + + def _drift_tables(self) -> list: + """Tables with abnormal audit records (AgedOut / Original!=Processed). + + Reads SyncLogArchive for rows recorded since this process started -- + a bounded window that surfaces both fresh incidents and any historical + residue without scanning the whole table. + """ + try: + w = SqlWriter( + self.cfg.sql_server.conn_str, + self.cfg.sql_server.sync_queue_table, + self.cfg.sql_server.archive_table, + self.cfg.sql_server.apply_proc, + self.cfg.sql_server.apply_runlog_table, + ) + try: + cur = w._conn.cursor() + cur.execute( + f"SELECT TargetSchema, TargetTable, ProcessedOperateType, " + f"COUNT(*) AS cnt FROM {w.archive_table} " + f"WHERE CapturedAt >= ? " + f"AND (ProcessedOperateType <> OriginalOperateType " + f" OR ProcessedOperateType = 'AgedOut') " + f"GROUP BY TargetSchema, TargetTable, ProcessedOperateType " + f"ORDER BY cnt DESC", + self.state.started_at, + ) + return [ + {"schema": r[0], "table": r[1], "kind": r[2], "count": r[3]} + for r in cur.fetchall() + ] + finally: + w.close() + except Exception as e: + log.warning("health: drift tables query failed: %s", e) + return [] + + def _dead_rows_sample(self) -> list: + """A few sample dead/error queue rows for triage.""" + try: + w = SqlWriter( + self.cfg.sql_server.conn_str, + self.cfg.sql_server.sync_queue_table, + self.cfg.sql_server.archive_table, + self.cfg.sql_server.apply_proc, + self.cfg.sql_server.apply_runlog_table, + ) + try: + return w.queue_error_samples(5) + finally: + w.close() + except Exception as e: + log.warning("health: dead rows query failed: %s", e) + return [] + + # ------------------------------------------------------------------ status + + def _derive_status(self, service: dict, data: dict) -> str: + poll = self.cfg.runtime.poll_interval_seconds + lag = service.get("cycle_lag_seconds") + # unhealthy: the sync loop is stalled. + if lag is not None and lag > CYCLE_LAG_MULTIPLIER * poll: + return "unhealthy" + queue = service.get("queue") or {} + # degraded: running but needs attention. + if queue.get("dead") or queue.get("error"): + return "degraded" + if service.get("capture_since_start", {}).get("aged_out"): + return "degraded" + cmp = data.get("last_compare") or {} + if cmp.get("mismatch"): + return "degraded" + return "healthy" + + # ------------------------------------------------------------------ utils + + def _log_dir(self) -> str: + path = (self.cfg.logging or {}).get("path", "sync.log") + return os.path.dirname(os.path.abspath(path)) or "." diff --git a/src/sync/service.py b/src/sync/service.py index 3063b51..2a3f566 100644 --- a/src/sync/service.py +++ b/src/sync/service.py @@ -26,7 +26,10 @@ in a ``finally``. ``run(cfg)`` loops ``cycle`` with a sleep; ``main()`` loads the config from ``argv[1]`` (default ``config.yaml``). """ from __future__ import annotations +import datetime as _dt +import os import sys +import threading import time import uuid import logging @@ -36,30 +39,83 @@ from .access_reader import AccessReader from .sql_writer import SqlWriter from .capture import capture_file, CaptureStats from .cleanup import cleanup_file +from .health import ServiceState from .logging_setup import setup_logging, set_cycle_id log = logging.getLogger("sync.service") def run(cfg): - """Run ``cycle`` forever, sleeping ``poll_interval_seconds`` between passes. + """Run the sync loop, optionally behind a health-check HTTP server. - Configures logging once on entry. Intended to be started by ``main()`` - under the service host (e.g. NSSM). Not unit-tested (infinite loop); - ``cycle()`` is the testable unit. Emits an idle heartbeat every - ``runtime.idle_heartbeat_seconds`` so a quiet log still proves liveness - now that idle cycles log at DEBUG. + Threading model: when ``cfg.health.enabled`` (default), uvicorn occupies + the main thread and the capture/apply/cleanup loop runs on a daemon + thread. NSSM sends its stop signal to the main (uvicorn) thread; on exit + the daemon sync thread is terminated automatically. When health is + disabled, the sync loop runs on the main thread (legacy behaviour). """ setup_logging(cfg.logging) - log.info( - "service started: files=%d poll_interval=%ds idle_heartbeat=%ds", - len(cfg.files), cfg.runtime.poll_interval_seconds, - cfg.runtime.idle_heartbeat_seconds, + state = ServiceState(started_at=_dt.datetime.now(), pid=os.getpid()) + + if cfg.health.enabled: + # Sync loop on a daemon thread; uvicorn on main. + sync_thread = threading.Thread( + target=_sync_loop, args=(cfg, state), daemon=True, name="sync-cycle", + ) + sync_thread.start() + log.info( + "service started: files=%d poll_interval=%ds idle_heartbeat=%ds " + "(sync loop on daemon thread; health API on %s:%d)", + len(cfg.files), cfg.runtime.poll_interval_seconds, + cfg.runtime.idle_heartbeat_seconds, cfg.health.host, cfg.health.port, + ) + _run_health_server(cfg, state) + else: + log.info( + "service started: files=%d poll_interval=%ds idle_heartbeat=%ds " + "(health API disabled)", + len(cfg.files), cfg.runtime.poll_interval_seconds, + cfg.runtime.idle_heartbeat_seconds, + ) + _sync_loop(cfg, state) + + +def _run_health_server(cfg, state): + """Start uvicorn on the main thread (blocks until shutdown).""" + import uvicorn + from .web.app import create_app + app = create_app(state, cfg) + uvicorn.run( + app, + host=cfg.health.host, + port=cfg.health.port, + log_level=cfg.health.log_level, ) + + +def _sync_loop(cfg, state): + """The capture -> apply -> cleanup loop, paced by poll_interval. + + Extracted from the legacy ``run`` so it can run on either the main thread + (health disabled) or a daemon thread (health enabled). Updates ``state`` + each pass so the health endpoint can observe liveness. + """ idle_since = None idle_cycles = 0 while True: - active = cycle(cfg) + t0 = time.monotonic() + try: + active, capture = cycle(cfg, state) + except Exception: + # cycle() already isolates per-file failures; a top-level exception + # here means something unexpected -- record it on the state so the + # health endpoint surfaces it rather than masking a silent stall. + log.exception("cycle raised unexpectedly") + state.update_cycle(False, time.monotonic() - t0, + error="cycle raised unexpectedly") + active, capture = False, None + else: + state.update_cycle(active, time.monotonic() - t0, capture=capture) now = time.monotonic() if active: idle_since, idle_cycles = None, 0 @@ -74,11 +130,15 @@ def run(cfg): time.sleep(cfg.runtime.poll_interval_seconds) -def cycle(cfg) -> bool: +def cycle(cfg, state: ServiceState | None = None) -> tuple[bool, CaptureStats | None]: """One capture -> apply -> cleanup pass over all files. - Returns True when the cycle did any work (captured / applied / cleaned / - purged anything); ``run`` uses this for idle-heartbeat pacing. Per-file + Returns ``(active, capture_stats)``: ``active`` is True when the cycle did + any work (captured / applied / cleaned / purged); ``_sync_loop`` uses this + for idle-heartbeat pacing and ``state.update_cycle`` for health reporting. + ``state`` is optional (tests call without it); when given it is NOT updated + here -- the caller (_sync_loop) updates it once with the final timing, to + keep the cycle itself free of cross-cutting concerns. Per-file capture/cleanup failures are logged and do not abort the cycle. Apply failure does not block cleanup. The writer is always closed in a ``finally``. Safe to call directly from tests (does not sleep or loop). @@ -214,7 +274,7 @@ def cycle(cfg) -> bool: (log.info if activity else log.debug)( "cycle finished in %.2fs", time.monotonic() - t0) - return activity + return activity, total finally: writer.close() set_cycle_id(None) diff --git a/src/sync/web/__init__.py b/src/sync/web/__init__.py new file mode 100644 index 0000000..63bd242 --- /dev/null +++ b/src/sync/web/__init__.py @@ -0,0 +1,11 @@ +"""FastAPI web layer for the sync service. + +Hosts the health-check endpoint and -- later -- additional operational routes +(compare trigger, dead-letter management, Prometheus metrics, ...). The web +layer only wires HTTP concerns; all business logic lives in +:mod:`sync.health` (:class:`HealthChecker`) so it stays testable without a +running server. +""" +from .app import create_app + +__all__ = ["create_app"] diff --git a/src/sync/web/app.py b/src/sync/web/app.py new file mode 100644 index 0000000..e96dc07 --- /dev/null +++ b/src/sync/web/app.py @@ -0,0 +1,28 @@ +"""FastAPI application factory. + +``create_app(state, cfg)`` builds the app, binds the shared singletons (the +process-wide :class:`ServiceState` and :class:`SyncConfig`) into the dependency +graph, and mounts route modules. New operational endpoints are added by +creating a ``routes/.py`` with an ``APIRouter`` and ``include_router``- +ing it here. +""" +from __future__ import annotations + +from fastapi import FastAPI + +from ..config import SyncConfig +from ..health import ServiceState +from . import deps +from .routes import health as health_routes + + +def create_app(state: ServiceState, cfg: SyncConfig) -> FastAPI: + """Build the FastAPI app with health state/config bound for injection.""" + app = FastAPI( + title="DataMacroSync", + description="Access -> SQL Server incremental sync operational API", + version="1.0.0", + ) + deps.bind(state, cfg) + app.include_router(health_routes.router) + return app diff --git a/src/sync/web/deps.py b/src/sync/web/deps.py new file mode 100644 index 0000000..663b11c --- /dev/null +++ b/src/sync/web/deps.py @@ -0,0 +1,42 @@ +"""Shared FastAPI dependencies. + +The :class:`ServiceState` and :class:`SyncConfig` are created once in +``service.run`` and injected into routes via ``Depends``, avoiding global +singletons. Routes get a ready-to-use :class:`HealthChecker` rather than the +raw state, so each endpoint declares only what it needs. +""" +from __future__ import annotations + +from fastapi import Depends + +from ..config import SyncConfig +from ..health import HealthChecker, ServiceState + +# Bound at app creation time (create_app). Module-level holders are acceptable +# here because there is exactly one app per process and they are write-once. +_state: ServiceState | None = None +_cfg: SyncConfig | None = None + + +def bind(state: ServiceState, cfg: SyncConfig) -> None: + """Stash the singletons for the lifetime of this app (called once).""" + global _state, _cfg + _state = state + _cfg = cfg + + +def get_state() -> ServiceState: + assert _state is not None, "ServiceState not bound -- call bind() first" + return _state + + +def get_config() -> SyncConfig: + assert _cfg is not None, "SyncConfig not bound -- call bind() first" + return _cfg + + +def get_checker( + state: ServiceState = Depends(get_state), + cfg: SyncConfig = Depends(get_config), +) -> HealthChecker: + return HealthChecker(state, cfg) diff --git a/src/sync/web/routes/__init__.py b/src/sync/web/routes/__init__.py new file mode 100644 index 0000000..1f5e1f6 --- /dev/null +++ b/src/sync/web/routes/__init__.py @@ -0,0 +1,2 @@ +"""Route modules. Add new operational endpoints here and include their router +in ``app.py``.""" diff --git a/src/sync/web/routes/health.py b/src/sync/web/routes/health.py new file mode 100644 index 0000000..7615f36 --- /dev/null +++ b/src/sync/web/routes/health.py @@ -0,0 +1,31 @@ +"""GET /health -- passive health-check endpoint. + +Returns a structured snapshot covering service health (process, cycle lag, +queue state) and data health (last compare result, drift traces, dead rows). + +HTTP status follows the body's ``status`` field so off-the-shelf monitors that +alert on status code work without parsing JSON: + +* ``200`` -- healthy +* ``503`` -- degraded or unhealthy + +This keeps "service unreachable" (network/down) distinguishable from "service +up but needs attention" (200 would conflate them). +""" +from __future__ import annotations + +from fastapi import APIRouter, Depends, Response + +from ..deps import get_checker +from ...health import HealthChecker + +router = APIRouter() + + +@router.get("/health", summary="Sync service health check") +def health(response: Response, checker: HealthChecker = Depends(get_checker)) -> dict: + snap = checker.snapshot() + # Map body status to HTTP code for code-based alerting. + if snap["status"] != "healthy": + response.status_code = 503 + return snap diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..43a3760 --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,183 @@ +"""Tests for the health-check layer: ServiceState, HealthChecker status logic, +and the /health endpoint via FastAPI TestClient.""" +import datetime as _dt +from unittest.mock import MagicMock, patch + +from fastapi.testclient import TestClient + +from sync.config import ( + AccessConfig, FileMapping, HealthConfig, RuntimeConfig, + SqlServerConfig, SyncConfig, +) +from sync.health import CYCLE_LAG_MULTIPLIER, HealthChecker, ServiceState +from sync.web.app import create_app + + +def _cfg(**health_kw): + return SyncConfig( + sql_server=SqlServerConfig(conn_str="x"), + access=AccessConfig(driver="d", roots={"2026": "r"}), + runtime=RuntimeConfig(poll_interval_seconds=10), + files=[FileMapping(file="f.accdb", root="2026", schema="s", year_suffix="_Y")], + health=HealthConfig(**health_kw), + ) + + +def _state(**kw): + s = ServiceState(started_at=_dt.datetime.now(), pid=1) + # Force last_cycle_at to "recent" by default so cycle_lag is small. + s.last_cycle_at = _dt.datetime.now() + for k, v in kw.items(): + setattr(s, k, v) + return s + + +# ---------------------------------------------------------------- ServiceState + +def test_state_snapshot_is_independent_copy(): + s = ServiceState(started_at=_dt.datetime.now(), pid=1) + s.update_cycle(active=True, duration_s=1.5) + snap = s.snapshot() + snap["capture_enqueued"] = 999 # mutate the copy + assert s.snapshot()["capture_enqueued"] == 0 # original untouched + + +def test_state_update_cycle_accumulates_capture_totals(): + from sync.capture import CaptureStats + s = ServiceState(started_at=_dt.datetime.now(), pid=1) + s.update_cycle(True, 1.0, capture=CaptureStats(enqueued=5, deferred=2, aged_out=1)) + s.update_cycle(True, 1.0, capture=CaptureStats(enqueued=3, deferred=1, aged_out=0)) + snap = s.snapshot() + assert snap["capture_enqueued"] == 8 + assert snap["capture_deferred"] == 3 + assert snap["capture_aged_out"] == 1 + + +# --------------------------------------------------------------- status logic + +def _checker_with(state, cfg, queue=None, compare_mismatch=False): + """Build a checker whose SQL queries are stubbed (no real DB).""" + c = HealthChecker(state, cfg) + c._queue_status = lambda: queue or {} + c._drift_tables = lambda: [] + c._dead_rows_sample = lambda: [] + c._last_compare = lambda: {"mismatch": compare_mismatch} + return c + + +def test_status_healthy_when_recent_cycle_clean_queue(): + cfg = _cfg() + state = _state() + c = _checker_with(state, cfg, queue={"pending": 0, "dead": 0, "error": 0}) + assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()), + c._data_health()) == "healthy" + + +def test_status_unhealthy_when_cycle_lag_exceeds_threshold(): + cfg = _cfg() # poll_interval=10 -> threshold = 30s + # last cycle was 60s ago + state = ServiceState(started_at=_dt.datetime.now() - _dt.timedelta(seconds=120), pid=1) + state.last_cycle_at = _dt.datetime.now() - _dt.timedelta(seconds=60) + c = _checker_with(state, cfg) + assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()), + c._data_health()) == "unhealthy" + + +def test_status_degraded_when_dead_rows(): + cfg = _cfg() + state = _state() + c = _checker_with(state, cfg, queue={"dead": 2}) + assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()), + c._data_health()) == "degraded" + + +def test_status_degraded_when_error_rows(): + cfg = _cfg() + state = _state() + c = _checker_with(state, cfg, queue={"error": 1}) + assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()), + c._data_health()) == "degraded" + + +def test_status_degraded_when_aged_out_accumulated(): + cfg = _cfg() + state = _state() + state.update_cycle(True, 1.0) # no capture arg + state.capture_aged_out = 3 # simulate prior aged-out incidents + c = _checker_with(state, cfg, queue={}) + snap = c.snapshot() + assert snap["status"] == "degraded" + + +def test_status_degraded_when_compare_mismatch(): + cfg = _cfg() + state = _state() + c = _checker_with(state, cfg, queue={}, compare_mismatch=True) + assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()), + c._data_health()) == "degraded" + + +# -------------------------------------------------------------- snapshot shape + +def test_snapshot_has_expected_top_level_keys(): + cfg = _cfg() + state = _state() + c = _checker_with(state, cfg, queue={"pending": 0}) + snap = c.snapshot() + for k in ("service", "status", "checked_at", "pid", "started_at", + "uptime_seconds", "service_health", "data_health"): + assert k in snap + assert "cycle_lag_seconds" in snap["service_health"] + assert "last_compare" in snap["data_health"] + + +# ------------------------------------------------------------- HTTP endpoint + +def test_health_endpoint_returns_200_when_healthy(): + cfg = _cfg() + state = _state() + with patch.object(HealthChecker, "_queue_status", return_value={}), \ + patch.object(HealthChecker, "_drift_tables", return_value=[]), \ + patch.object(HealthChecker, "_dead_rows_sample", return_value=[]), \ + patch.object(HealthChecker, "_last_compare", + return_value={"available": False}): + app = create_app(state, cfg) + client = TestClient(app) + r = client.get("/health") + assert r.status_code == 200 + body = r.json() + assert body["status"] == "healthy" + assert body["service"] == "DataMacroSync" + + +def test_health_endpoint_returns_503_when_unhealthy(): + cfg = _cfg() + # stale cycle -> unhealthy + state = ServiceState(started_at=_dt.datetime.now() - _dt.timedelta(seconds=120), pid=1) + state.last_cycle_at = _dt.datetime.now() - _dt.timedelta(seconds=60) + with patch.object(HealthChecker, "_queue_status", return_value={}), \ + patch.object(HealthChecker, "_drift_tables", return_value=[]), \ + patch.object(HealthChecker, "_dead_rows_sample", return_value=[]), \ + patch.object(HealthChecker, "_last_compare", + return_value={"available": False}): + app = create_app(state, cfg) + client = TestClient(app) + r = client.get("/health") + assert r.status_code == 503 + assert r.json()["status"] == "unhealthy" + + +def test_health_endpoint_returns_503_when_degraded(): + cfg = _cfg() + state = _state() + with patch.object(HealthChecker, "_queue_status", + return_value={"dead": 1}), \ + patch.object(HealthChecker, "_drift_tables", return_value=[]), \ + patch.object(HealthChecker, "_dead_rows_sample", return_value=[]), \ + patch.object(HealthChecker, "_last_compare", + return_value={"available": False}): + app = create_app(state, cfg) + client = TestClient(app) + r = client.get("/health") + assert r.status_code == 503 + assert r.json()["status"] == "degraded"