2 Commits

Author SHA1 Message Date
Misaka_Company
79a95465e7 refactor(web): API 路由统一 /api 前缀 + 完善 OpenAPI 文档
- 路由挂 APIRouter(prefix="/api"),健康检查路径 /health -> /api/health,
  后期加接口(/api/compare、/api/queue 等)自动共享前缀。
- 文档路径归到 /api 下:/api/docs、/api/redoc、/api/openapi.json
  (原 /docs、/redoc、/openapi.json 已 404)。
- 新增 Pydantic 响应模型 schemas.py,并在 /api/health 挂 response_model,
  使 /api/docs 展示完整响应结构。
- 为接口和每个响应字段补详细英文 description(覆盖范围、status 三态判定、
  HTTP 码含义、各字段语义如 cycle_lag 阈值/dead 含义等)。

不动业务逻辑与同步主循环,仅 Web 层。
2026-08-05 12:55:45 +08:00
Misaka_Company
03d7bcea39 feat(health): 被动式健康检查 API(FastAPI,覆盖服务健康+数据健康)
在同步进程内内置 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]。
2026-08-05 12:13:14 +08:00
14 changed files with 1338 additions and 15 deletions

View File

@@ -21,6 +21,12 @@ logging:
level: INFO
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:
- {file: "一车间.accdb", root: 2026, schema: "workshopOne", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "一车间每日催货落实记录_停"]}
- {file: "二车间.accdb", root: 2026, schema: "workshopTwo", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}

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()` 结束时更新 statelast_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` | 新增 HealthConfigSyncConfig 加 health 字段 |
| `src/sync/service.py` | run() 起 health 线程 + 传 statecycle() 更新 state |
| `config.example.yaml` | 补 health 段示例 |
| `tests/test_health.py` | **新增** 覆盖 status 判定、snapshot 结构、各状态组合 |
**零 SQL 改动、零 schema 改动、零新第三方依赖(标准库 http.server**
---
## 八、待你确认的决策点
1. **端口** `8421` 是否合适?(避开 114 上已有服务端口)
2. **unhealthy 的 HTTP 状态码**:始终 200状态在 bodyvs 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 小时"?我倾向"启动至今"(量可控,且能发现历史遗留)。
确认后即实施。

View File

@@ -0,0 +1,215 @@
# 方案 v2FastAPI 健康检查(面向多接口扩展)
> 基于"后期会引入更多运维接口"的前提,采用 FastAPI。
> **与 v1标准库 http.server的核心差异在部署模型**——FastAPI/uvicorn 阻塞主线程,必须重新设计进程结构。本方案先解决这个架构问题,再展开实现。
---
## 一、核心架构决策:同步循环放哪个线程?
FastAPI 的标准运行方式 `uvicorn.run(app)` 会**阻塞主线程**。而现有项目里 `service.run()`(同步循环)是主线程。两者都要"常驻",必须有一个让出主线程。这是用 FastAPI 唯一的硬约束,两条路径:
### 方案 AFastAPI 主线程 + 同步循环后台线程(✅ 推荐)
```
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` | 实时查 SyncQueuepending/error/dead/cleaned |
| `capture_since_start` | 内存累计enqueued/deferred/aged_out |
### 数据健康Data Health
| 指标 | 来源 |
|------|------|
| `last_compare` | 解析最近 `compare_ids_*.log` 报告头 |
| `drift_tables` | 实时查 SyncLogArchiveAgedOut、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 小时?倾向启动至今。
确认后实施。

View File

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

View File

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

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

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``).
"""
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)

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

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

@@ -0,0 +1,33 @@
"""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",
# Keep all auto-generated docs under /api alongside the operational
# routes, so the whole HTTP surface shares one prefix.
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json",
)
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)

View File

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

View File

@@ -0,0 +1,79 @@
"""GET /api/health -- passive health-check endpoint."""
from __future__ import annotations
from fastapi import APIRouter, Depends, Response
from ..deps import get_checker
from ..schemas import HealthResponse
from ...health import HealthChecker
# All operational routes mount under /api (see app.py). The health route lives
# at /api/health so future endpoints (/api/compare, /api/queue, ...) share one
# prefix without per-router configuration.
router = APIRouter(prefix="/api")
@router.get(
"/health",
response_model=HealthResponse,
summary="Get the sync service health snapshot",
description=(
"Returns a structured snapshot of the Access → SQL Server incremental "
"sync service. Use this for passive health probing (uptime monitors, "
"load-balancer checks, on-call dashboards).\n\n"
"**What it covers**\n\n"
"Two layers are reported:\n\n"
"1. **Service health** — whether the sync loop is alive and making "
"progress: the time since the last capture/apply/cleanup cycle "
"(`cycle_lag_seconds`), the current SyncQueue row counts by status, "
"and cumulative capture counters since process start.\n"
"2. **Data health** — lightweight signals about Access ↔ SQL Server "
"consistency: the result of the most recent scheduled full compare "
"(run daily ~05:08, parsed from its report file — NOT re-run on every "
"request), plus any abnormal audit traces (rows that aged out, "
"downgraded operations) and a sample of dead/error queue rows.\n\n"
"**Status derivation**\n\n"
"The top-level `status` field is derived from the signals above:\n\n"
"- `healthy` — the loop is pacing normally and no attention signals "
"are present.\n"
"- `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.\n"
"- `unhealthy` — the sync loop appears stalled (cycle lag exceeds "
"3× the poll interval, e.g. wedged on an Access lock wait).\n\n"
"**HTTP status codes**\n\n"
"The HTTP status mirrors `status` so off-the-shelf monitors that alert "
"on status code work without parsing JSON:\n\n"
"- `200 OK` — healthy.\n"
"- `503 Service Unavailable` — degraded or unhealthy. This keeps "
"\"service unreachable\" (a network error / timeout, no response at "
"all) distinguishable from \"service up but needs attention\".\n\n"
"**Cost**\n\n"
"Each request performs one short SQL query (queue + audit) and reads "
"an in-memory state snapshot updated by the sync loop; it does not run "
"a full compare. Safe to poll at second-scale intervals."
),
responses={
200: {"description": "Service is healthy."},
503: {
"description": (
"Service is degraded or unhealthy. The body still contains the "
"full snapshot; inspect `status`, `service_health` and "
"`data_health` for the cause."
),
"model": HealthResponse,
},
},
)
def health(response: Response,
checker: HealthChecker = Depends(get_checker)) -> dict:
"""Compute and return the health snapshot.
The heavy lifting lives in :class:`sync.health.HealthChecker`; this route
is a thin HTTP adapter that maps the resulting status to an HTTP code.
"""
snap = checker.snapshot()
# Map body status to HTTP code for code-based alerting.
if snap["status"] != "healthy":
response.status_code = 503
return snap

204
src/sync/web/schemas.py Normal file
View File

@@ -0,0 +1,204 @@
"""Pydantic response models for the API.
Declaring these (and passing them as ``response_model`` on routes) makes the
auto-generated OpenAPI/Swagger docs at ``/api/docs`` show the full response
shape per field, rather than a bare ``object``. Keep them structurally aligned
with the dicts produced by :class:`sync.health.HealthChecker`.
"""
from __future__ import annotations
from pydantic import BaseModel, Field
class QueueStatus(BaseModel):
"""Live SyncQueue row counts grouped by Status."""
pending: int | None = Field(
default=None,
description="Rows staged but not yet applied by usp_SyncApply. "
"Briefly non-zero right after capture; should drain to 0 "
"each cycle.",
)
error: int | None = Field(
default=None,
description="Rows whose apply failed and are under the retry budget. "
"Non-zero indicates a transient apply failure; they are "
"re-queued automatically. Drives the `degraded` status.",
)
dead: int | None = Field(
default=None,
description="Rows that exhausted retries (or were parked as dead on "
"capture). These represent changes that exist in Access "
"but never reached SQL Server and need manual review. "
"Drives the `degraded` status.",
)
cleaned: int | None = Field(
default=None,
description="Rows successfully applied and whose Access change-log "
"entry has been removed. Retained for a short audit "
"window then purged; not an error signal.",
)
class CaptureSinceStart(BaseModel):
"""Cumulative capture counters since the service process started."""
enqueued: int = Field(
default=0,
description="Total change-log rows successfully staged into SyncQueue "
"since process start.",
)
deferred: int = Field(
default=0,
description="Total Insert/Update log rows whose source row was "
"momentarily unreadable (ACE visibility latency) and were "
"deferred for retry on a later cycle. Occasional non-zero "
"values are normal under bulk-insert bursts.",
)
aged_out: int = Field(
default=0,
description="Total Insert/Update log rows still unreadable past the "
"defer window and parked as dead for manual review. "
"Non-zero drives the `degraded` status — these rows never "
"reached SQL Server.",
)
class ServiceHealth(BaseModel):
"""Live process and sync-loop metrics."""
last_cycle_at: str | None = Field(
default=None,
description="ISO 8601 timestamp of the most recent capture→apply→"
"cleanup cycle completion. Null until the first cycle "
"finishes.",
)
last_cycle_duration_s: float | None = Field(
default=None,
description="Wall-clock duration of the last cycle, in seconds.",
)
cycle_lag_seconds: float | None = Field(
default=None,
description="Seconds elapsed since `last_cycle_at`. The primary "
"liveness indicator: a healthy idle loop paces at "
"poll_interval (default 10s); exceeding 3× poll_interval "
"flips the status to `unhealthy`.",
)
last_cycle_active: bool = Field(
default=False,
description="True if the last cycle actually processed changes "
"(captured/applied/cleaned anything), False if it was an "
"idle pass.",
)
last_cycle_error: str | None = Field(
default=None,
description="Error message if the last cycle raised unexpectedly at "
"the top level (per-file failures are isolated and do not "
"set this). Null when the cycle completed normally.",
)
queue: QueueStatus = Field(
description="Live SyncQueue row counts by status.",
)
capture_since_start: CaptureSinceStart = Field(
description="Cumulative capture counters since process start.",
)
class LastCompare(BaseModel):
"""Result of the most recent scheduled full compare (parsed, not re-run)."""
available: bool = Field(
default=False,
description="Whether a compare report file was found. False when no "
"report has been generated yet (e.g. before the first "
"daily run) or the logs directory is unreadable.",
)
report: str | None = Field(
default=None,
description="File name of the parsed report (e.g. "
"compare_ids_2026-08-05.log).",
)
at: str | None = Field(
default=None,
description="ISO 8601 timestamp the compare report was generated.",
)
tables_match: int | None = Field(
default=None,
description="Number of compared tables whose Access and SQL row-ID "
"sets matched.",
)
tables_mismatch: int | None = Field(
default=None,
description="Number of compared tables that diverged. Non-zero drives "
"the `degraded` status via `mismatch`.",
)
mismatch: bool = Field(
default=False,
description="True if any table diverged in the last compare. Drives "
"the `degraded` status. Note this reflects the last "
"scheduled compare (daily), not a real-time check.",
)
class DriftTable(BaseModel):
"""A table with abnormal audit traces since process start."""
schema: str = Field(description="Target SQL Server schema name.")
table: str = Field(description="Target table name (with year suffix).")
kind: str = Field(
description="Trace kind: 'AgedOut' (Insert unreadable past defer "
"window) or the processed operate type when it diverged "
"from the original.",
)
count: int = Field(description="Number of abnormal audit rows for this table.")
class DataHealth(BaseModel):
"""Lightweight Access ↔ SQL Server consistency signals."""
last_compare: LastCompare = Field(
description="Result of the most recent scheduled full compare, parsed "
"from its report file. A full compare is NOT executed on "
"every health request (too heavy).",
)
drift_tables: list[DriftTable] = Field(
default_factory=list,
description="Tables with abnormal audit traces (aged-out or "
"downgraded operations) recorded since process start. "
"Empty under normal operation.",
)
dead_rows_sample: list[dict] = Field(
default_factory=list,
description="Up to 5 sample dead/error queue rows for triage "
"(table, record id, error message, ...). Empty when the "
"queue has no stuck rows.",
)
class HealthResponse(BaseModel):
"""Top-level health snapshot returned by GET /api/health."""
service: str = Field(
description="Service identifier (always 'DataMacroSync').",
)
status: str = Field(
description="Derived overall status: 'healthy', 'degraded', or "
"'unhealthy'. Mirrored to the HTTP status code "
"(200 / 503).",
)
checked_at: str = Field(
description="ISO 8601 timestamp the snapshot was generated.",
)
pid: int = Field(description="OS process id of the sync service.")
started_at: str = Field(
description="ISO 8601 timestamp the service process started.",
)
uptime_seconds: int = Field(
description="Seconds since the service process started.",
)
service_health: ServiceHealth = Field(
description="Live process and sync-loop metrics.",
)
data_health: DataHealth = Field(
description="Lightweight Access ↔ SQL Server consistency signals.",
)

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("/api/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("/api/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("/api/health")
assert r.status_code == 503
assert r.json()["status"] == "degraded"