AllForCompany/ProductionDataBaseSync_DataMacro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(health): 被动式健康检查 API（FastAPI，覆盖服务健康+数据健康）

Browse Source 
在同步进程内内置 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]。
This commit is contained in:
Misaka_Company
2026-08-05 12:13:14 +08:00
parent 7cef5153e6
commit 03d7bcea39
13 changed files with 1081 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
config.example.yaml
View File

@@ -21,6 +21,12 @@ logging:
level: INFO level: INFO
path: "<LOG_PATH>" path: "<LOG_PATH>"
health:
enabled: true # 开启被动式健康检查 HTTP endpoint
host: "0.0.0.0" # 监听地址，0.0.0.0 内网/FRP 均可访问
port: 8421 # 健康检查端口
log_level: "warning" # uvicorn 自身日志级别，避免刷屏
files: files:
- {file: "一车间.accdb", root: 2026, schema: "workshopOne", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "一车间每日催货落实记录_停"]} - {file: "一车间.accdb", root: 2026, schema: "workshopOne", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "一车间每日催货落实记录_停"]}
- {file: "二车间.accdb", root: 2026, schema: "workshopTwo", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]} - {file: "二车间.accdb", root: 2026, schema: "workshopTwo", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}

164
docs/plan-health-check-api.md Normal file
View File

@@ -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 小时"？我倾向"启动至今"（量可控，且能发现历史遗留）。
确认后即实施。

215
docs/plan-health-check-fastapi.md Normal file
View File

@@ -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 循环<br/>每轮更新 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 小时？倾向启动至今。
确认后实施。

2
requirements.txt
View File

@@ -1,4 +1,6 @@
pyodbc>=5.0.1 pyodbc>=5.0.1
PyYAML>=6.0.1 PyYAML>=6.0.1
pydantic>=2.6.0 pydantic>=2.6.0
fastapi>=0.110.0
uvicorn[standard]>=0.27.0
pytest>=8.0.0 pytest>=8.0.0

14
src/sync/config.py
View File

@@ -53,12 +53,26 @@ class FileMapping(BaseModel):
def target_table(self, access_table: str) -> str: def target_table(self, access_table: str) -> str:
return f"{access_table}{self.year_suffix}" 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): class SyncConfig(BaseModel):
sql_server: SqlServerConfig sql_server: SqlServerConfig
access: AccessConfig access: AccessConfig
runtime: RuntimeConfig runtime: RuntimeConfig
files: list[FileMapping] files: list[FileMapping]
logging: dict | None = None logging: dict | None = None
health: HealthConfig = HealthConfig()
def load_config(path: str) -> SyncConfig: def load_config(path: str) -> SyncConfig:
with open(path, "r", encoding="utf-8") as f: with open(path, "r", encoding="utf-8") as f:

308
src/sync/health.py Normal file
View File

@@ -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<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})"
r"|不一致\s*MISMATCH\s*:\s*(?P<mismatch>\d+)"
r"|一致\s*MATCH\s*:\s*(?P<match>\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_<YYYY-MM-DD>.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 "."

90
src/sync/service.py
View File

@@ -26,7 +26,10 @@ in a ``finally``. ``run(cfg)`` loops ``cycle`` with a sleep; ``main()``
loads the config from ``argv[1]`` (default ``config.yaml``). loads the config from ``argv[1]`` (default ``config.yaml``).
""" """
from __future__ import annotations from __future__ import annotations
import datetime as _dt
import os
import sys import sys
import threading
import time import time
import uuid import uuid
import logging import logging
@@ -36,30 +39,83 @@ from .access_reader import AccessReader
from .sql_writer import SqlWriter from .sql_writer import SqlWriter
from .capture import capture_file, CaptureStats from .capture import capture_file, CaptureStats
from .cleanup import cleanup_file from .cleanup import cleanup_file
from .health import ServiceState
from .logging_setup import setup_logging, set_cycle_id from .logging_setup import setup_logging, set_cycle_id
log = logging.getLogger("sync.service") log = logging.getLogger("sync.service")
def run(cfg): 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()`` Threading model: when ``cfg.health.enabled`` (default), uvicorn occupies
under the service host (e.g. NSSM). Not unit-tested (infinite loop); the main thread and the capture/apply/cleanup loop runs on a daemon
``cycle()`` is the testable unit. Emits an idle heartbeat every thread. NSSM sends its stop signal to the main (uvicorn) thread; on exit
``runtime.idle_heartbeat_seconds`` so a quiet log still proves liveness the daemon sync thread is terminated automatically. When health is
now that idle cycles log at DEBUG. disabled, the sync loop runs on the main thread (legacy behaviour).
""" """
setup_logging(cfg.logging) setup_logging(cfg.logging)
log.info( state = ServiceState(started_at=_dt.datetime.now(), pid=os.getpid())
"service started: files=%d poll_interval=%ds idle_heartbeat=%ds",
len(cfg.files), cfg.runtime.poll_interval_seconds, if cfg.health.enabled:
cfg.runtime.idle_heartbeat_seconds, # 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_since = None
idle_cycles = 0 idle_cycles = 0
while True: 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() now = time.monotonic()
if active: if active:
idle_since, idle_cycles = None, 0 idle_since, idle_cycles = None, 0
@@ -74,11 +130,15 @@ def run(cfg):
time.sleep(cfg.runtime.poll_interval_seconds) 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. """One capture -> apply -> cleanup pass over all files.
Returns True when the cycle did any work (captured / applied / cleaned / Returns ``(active, capture_stats)``: ``active`` is True when the cycle did
purged anything); ``run`` uses this for idle-heartbeat pacing. Per-file 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 capture/cleanup failures are logged and do not abort the cycle. Apply
failure does not block cleanup. The writer is always closed in a failure does not block cleanup. The writer is always closed in a
``finally``. Safe to call directly from tests (does not sleep or loop). ``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)( (log.info if activity else log.debug)(
"cycle finished in %.2fs", time.monotonic() - t0) "cycle finished in %.2fs", time.monotonic() - t0)
return activity return activity, total
finally: finally:
writer.close() writer.close()
set_cycle_id(None) set_cycle_id(None)

11
src/sync/web/__init__.py Normal file
View File

@@ -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"]

28
src/sync/web/app.py Normal file
View File

@@ -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/<name>.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

42
src/sync/web/deps.py Normal file
View File

@@ -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)

2
src/sync/web/routes/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
"""Route modules. Add new operational endpoints here and include their router
in ``app.py``."""

31
src/sync/web/routes/health.py Normal file
View File

@@ -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

183
tests/test_health.py Normal file
View File

@@ -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"