Compare commits
3 Commits
master
...
feat/healt
| Author | SHA1 | Date | |
|---|---|---|---|
|
79a95465e7 | ||
|
|
03d7bcea39 | ||
|
|
7cef5153e6 |
@@ -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"]}
|
||||
|
||||
281
docs/incremental-sync-flow.md
Normal file
281
docs/incremental-sync-flow.md
Normal file
@@ -0,0 +1,281 @@
|
||||
# 增量同步流程详解（Incremental Sync Flow）
|
||||
|
||||
> 本文档梳理 Access → SQL Server 增量同步的完整流程，逐节点说明「发生了什么、读了/写了什么、状态如何流转」，作为重构「Insert 降级 Delete」逻辑的决策依据。
|
||||
>
|
||||
> 涉及代码：`src/sync/service.py`（主循环 cycle）、`src/sync/capture.py`（捕获）、`sql/02_sync_apply.sql`（应用）、`src/sync/cleanup.py`（清理）、`src/sync/sql_writer.py`（SQL 端读写）、`src/sync/access_reader.py`（Access 端读取）。
|
||||
|
||||
---
|
||||
|
||||
## 一、整体架构：一轮 cycle 的三段流水线
|
||||
|
||||
每个 cycle（默认间隔 `poll_interval_seconds=10s`）跑一遍三段，顺序固定、不可调换：
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
START([cycle 开始<br/>分配 cycle_id]) --> CAP
|
||||
|
||||
CAP["【1. Capture 捕获】<br/>逐库读 TableChangeLog → 回读整行 → 入队 SyncQueue<br/><i>src/sync/capture.py</i>"]
|
||||
CAP --> APPLY
|
||||
|
||||
APPLY["【2. Apply 应用】<br/>调 usp_SyncApply 把 SyncQueue pending 行落到镜像表<br/><i>sql/02_sync_apply.sql</i>"]
|
||||
APPLY --> HEALTH
|
||||
|
||||
HEALTH["【2.5 队列健康检查】<br/>查 error/dead 卡死行并告警"]
|
||||
HEALTH --> CLEAN
|
||||
|
||||
CLEAN["【3. Cleanup 清理】<br/>删 Access 已应用日志 → SyncQueue 行标 cleaned<br/><i>src/sync/cleanup.py</i>"]
|
||||
CLEAN --> PURGE
|
||||
|
||||
PURGE["【3.5 Purge 回收】<br/>删 SyncQueue 中超保留期的 cleaned 行"]
|
||||
PURGE --> DONE([cycle 结束<br/>休眠 poll_interval])
|
||||
|
||||
style CAP fill:#e3f2fd,stroke:#1976d2
|
||||
style APPLY fill:#fff3e0,stroke:#f57c00
|
||||
style HEALTH fill:#fce4ec,stroke:#c2185b
|
||||
style CLEAN fill:#e8f5e9,stroke:#388e3c
|
||||
style PURGE fill:#f3e5f5,stroke:#7b1fa2
|
||||
```
|
||||
|
||||
**关键设计约束**（决定重构可行性的红线）：
|
||||
- **Access 的 `TableChangeLog` 是 append-only，无状态字段**——它只是个待处理队列，无法在上面记录「已重试几次」。
|
||||
- **日志清除的唯一依据是 SyncQueue 的 `Status='applied'`**——只要一条日志对应的队列行不是 applied，cleanup 就不会删它（详见第三节）。
|
||||
- **SyncQueue 有唯一索引 `(SourceFile,SourceTable,SourceLogID)` 去重**——同一日志第二次入队会被静默跳过，不会覆盖原行、不会自增计数。
|
||||
|
||||
---
|
||||
|
||||
## 二、Capture 阶段（数据捕获）—— ❗重构的核心战场
|
||||
|
||||
逐库处理，每个 Access 文件独立隔离（单库失败不影响其它）。
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A([开始 capture 某个文件]) --> B["读 Access TableChangeLog<br/>最旧 N 条（按 ID 升序）<br/>N = capture_batch_size=500"]
|
||||
B --> C{有日志行?}
|
||||
C -- 否 --> Z([capture 结束])
|
||||
C -- 是 --> D[逐行处理]
|
||||
|
||||
D --> E{"表是否在同步范围内?<br/>is_synced_table"}
|
||||
E -- 否 --> F["跳过（out_of_scope）<br/>不计入，下轮仍会读到"]
|
||||
F --> D
|
||||
E -- 是 --> G{"OperateType?"}
|
||||
|
||||
G -- Insert/Update --> H["🔑 回读整行<br/>read_row(table, record_id)<br/>SELECT * FROM 表 WHERE ID=?"]
|
||||
H --> I{读到行?}
|
||||
I -- 是 --> J["row_data = JSON 序列化<br/>op 保持 Insert/Update"]
|
||||
I -- ❌否 --> K["⚠️ 降级 op = Delete<br/>row_data = None<br/>写 DOWNGRADE 警告日志"]
|
||||
|
||||
G -- Delete --> L["op = Delete<br/>row_data = None<br/>（不回读，Delete 无需数据）"]
|
||||
|
||||
G -- 其它未知 --> M["跳过（unknown_op）<br/>下轮仍会读到"]
|
||||
|
||||
J --> N["写 SyncLogArchive（永久审计）<br/>记录 OriginalOperateType + ProcessedOperateType"]
|
||||
K --> N
|
||||
L --> N
|
||||
|
||||
N --> O["入队 SyncQueue（去重插入）<br/>INSERT...WHERE NOT EXISTS"]
|
||||
O --> P{插入成功?}
|
||||
P -- 是 --> Q["enqueued +1"]
|
||||
P -- 否（去重命中）--> R["dedup_skipped +1<br/>说明上轮 apply 失败残留"]
|
||||
|
||||
Q --> S{还有下一行?}
|
||||
R --> S
|
||||
F --> S
|
||||
M --> S
|
||||
S -- 是 --> D
|
||||
S -- 否 --> T["打印 capture summary<br/>read/enqueued/downgraded/..."]
|
||||
T --> Z
|
||||
|
||||
style K fill:#ffcdd2,stroke:#c62828,stroke-width:3px
|
||||
style H fill:#fff9c4,stroke:#f9a825
|
||||
style I fill:#fff9c4,stroke:#f9a825
|
||||
```
|
||||
|
||||
### 🔴 问题节点：降级 Delete（`capture.py:93-108`）
|
||||
|
||||
```python
|
||||
if op in ("Insert", "Update"):
|
||||
d = reader.read_row(lr.table_name, lr.record_id)
|
||||
if d is None:
|
||||
op = "Delete" # ← 问题根源：读不到就降级
|
||||
st.downgraded += 1
|
||||
log.warning("capture DOWNGRADE %s->Delete ...")
|
||||
```
|
||||
|
||||
**为什么读不到？两个场景无法区分：**
|
||||
1. **真删除**：行被 Insert 后又 Delete（Access 客户端先插后删）→ 这时降级 Delete 是「碰巧正确」。
|
||||
2. **可见性延迟**（本次事件的根因）：行已插入但 ACE 引擎尚未对其他 ODBC 连接可见（批量插入时窗口可达 11 秒）→ 这时降级 Delete 是**有害的误伤**。
|
||||
|
||||
**代码当前无法区分这两种情况**，统一降级为 Delete。这就是要重构的核心。
|
||||
|
||||
---
|
||||
|
||||
## 三、Apply 阶段（数据应用）
|
||||
|
||||
调用存储过程 `usp_SyncApply`，**按表分组、集合化处理**所有 pending 行。
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A([call_apply<br/>EXEC usp_SyncApply]) --> B["重置 error 行<br/>RetryCount < max_retries 的 → pending"]
|
||||
B --> C["按 TargetSchema+TargetTable 分组<br/>遍历每个目标表"]
|
||||
|
||||
C --> D{该表有 pending 行?}
|
||||
D -- 否 --> C
|
||||
D -- 是 --> E["统计 pending 数 / distinct RecordID 数"]
|
||||
|
||||
E --> F["BEGIN TRAN"]
|
||||
|
||||
F --> G["🔑 构建列清单<br/>从 sys.columns 读目标表所有列<br/>（排除 ID/computed/identity/timestamp）"]
|
||||
|
||||
G --> H["构建动态 SQL"]
|
||||
|
||||
H --> I["分支1: Upsert（MERGE）<br/>ranked CTE: 按 RecordID 分区,<br/>SourceLogID DESC 取 rn=1<br/>仅 OperateType∈Insert/Update 且 RowData 非空"]
|
||||
I --> J["SET IDENTITY_INSERT ON<br/>MERGE 目标表<br/>匹配则 UPDATE, 不匹配则 INSERT<br/>@merged = @@ROWCOUNT"]
|
||||
|
||||
J --> K["分支2: Delete<br/>同一 ranked CTE 的 rn=1 行<br/>仅 OperateType=Delete"]
|
||||
K --> L["DELETE 目标表 WHERE ID IN (...)<br/>@deleted = @@ROWCOUNT"]
|
||||
|
||||
L --> M["把该表所有 pending 行<br/>Status → applied, AppliedAt = now<br/>@applied = @@ROWCOUNT"]
|
||||
|
||||
M --> N[COMMIT]
|
||||
|
||||
N --> O["写 SyncApplyRunLog 审计<br/>pending/merged/deleted/applied/<br/>error/dead + CycleID + 耗时"]
|
||||
O --> C
|
||||
|
||||
N -.失败.-> X["ROLLBACK"]
|
||||
X --> Y["超 max_retries → dead<br/>否则 → error（下轮重试）"]
|
||||
Y --> O
|
||||
|
||||
style I fill:#e3f2fd,stroke:#1976d2
|
||||
style K fill:#ffcdd2,stroke:#c62828
|
||||
style M fill:#fff3e0,stroke:#f57c00
|
||||
```
|
||||
|
||||
### 关键：保序「最后操作胜」（`02_sync_apply.sql:95-135`）
|
||||
|
||||
单个 `ranked` CTE 同时供 Upsert 和 Delete 两个分支使用：
|
||||
```sql
|
||||
ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn
|
||||
```
|
||||
- `rn=1` 是该 RecordID **真正的最后一条日志**。
|
||||
- Upsert 分支：`rn=1 AND OperateType IN ('Insert','Update')`
|
||||
- Delete 分支：`rn=1 AND OperateType='Delete'`
|
||||
|
||||
所以**降级成 Delete 的行，在这里会真的去 SQL 端执行 DELETE**。本次事件中 5 行 Delete 的 `@deleted=0`（SQL 里本就没这些行，空打），但 `@applied=5`（队列行照样被标 applied）。
|
||||
|
||||
---
|
||||
|
||||
## 四、Cleanup 阶段（日志清除）—— ❗决定「重试」能否成立的命脉
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A([cleanup 某个文件]) --> B["查 SyncQueue 中<br/>Status=applied 的 SourceLogID 列表<br/>applied_log_ids(file)"]
|
||||
B --> C{有 applied 行?}
|
||||
C -- 否 --> Z([cleanup 结束, 返回 0])
|
||||
C -- 是 --> D["DELETE FROM Access.TableChangeLog<br/>WHERE ID IN (上述列表)<br/>分批 + 锁重试"]
|
||||
D --> E{删除数 == 预期?}
|
||||
E -- 否 --> F["WARNING: 部分日志已不在<br/>（被外部或中断的运行删过）"]
|
||||
E -- 是 --> G["mark_cleaned:<br/>这些队列行 Status → cleaned<br/>CleanedAt = now"]
|
||||
F --> G
|
||||
G --> H["INFO: 删除了 N 条 Access 日志"]
|
||||
H --> Z
|
||||
|
||||
style D fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
|
||||
style B fill:#fff9c4,stroke:#f9a825
|
||||
```
|
||||
|
||||
### 🔑 cleanup 的判定条件是重构的支点
|
||||
|
||||
**cleanup 只删 `Status='applied'` 的日志**（`sql_writer.py:264-272` 的 `applied_log_ids`）。这意味着：
|
||||
|
||||
| capture 对该日志的处理 | SyncQueue 行状态 | cleanup 是否删 Access 日志 | 后果 |
|
||||
|------------------------|------------------|----------------------------|------|
|
||||
| 降级 Delete（现状） | applied | **删除** | ❌ 日志消失，再无重试机会 |
|
||||
| **跳过不入队（重构后）** | （无对应行） | **不删** | ✅ 日志保留，下轮重读 |
|
||||
|
||||
**结论：重构只要做到「读不到 → 不入队」，cleanup 这一段天然会把日志保留下来，无需改动 cleanup.py。** 这是「跨 cycle 重试」能够成立的根基。
|
||||
|
||||
---
|
||||
|
||||
## 五、数据流转全景：一条日志的完整生命周期
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
subgraph Access["Access 端（.accdb）"]
|
||||
T1[(业务表<br/>如 接收)]
|
||||
TCL[(TableChangeLog<br/>变更日志 append-only)]
|
||||
end
|
||||
|
||||
subgraph SQL["SQL Server 端（CompanyDB）"]
|
||||
SQ[(SyncQueue<br/>待处理队列)]
|
||||
ARCH[(SyncLogArchive<br/>永久审计)]
|
||||
RL[(SyncApplyRunLog<br/>apply 运行日志)]
|
||||
MIRROR[(镜像表<br/>如 接收_YEAR2026)]
|
||||
end
|
||||
|
||||
T1 -- "数据宏 After I/U/D<br/>写入一行日志" --> TCL
|
||||
TCL -- "① capture 读取" --> CAP[Capture]
|
||||
CAP -- "回读整行" --> T1
|
||||
CAP -- "② 入队（去重）" --> SQ
|
||||
CAP -- "② 审计存档" --> ARCH
|
||||
SQ -- "③ apply 处理" --> APPLY[usp_SyncApply]
|
||||
APPLY -- "MERGE/DELETE" --> MIRROR
|
||||
APPLY -- "pending→applied" --> SQ
|
||||
APPLY -- "记录运行结果" --> RL
|
||||
SQ -- "④ cleanup 查 applied" --> CLEAN[Cleanup]
|
||||
CLEAN -- "⑤ 删已应用日志" --> TCL
|
||||
CLEAN -- "applied→cleaned" --> SQ
|
||||
|
||||
style CAP fill:#e3f2fd,stroke:#1976d2
|
||||
style APPLY fill:#fff3e0,stroke:#f57c00
|
||||
style CLEAN fill:#e8f5e9,stroke:#388e3c
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 六、本次事件的重放（在上述流程中的路径）
|
||||
|
||||
5 条 Insert 日志（RecordID 16255-16259）在一轮 cycle 中的遭遇：
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A["8/3 09:01 Access 批量插入 5 行<br/>数据宏写 5 条 Insert 日志<br/>SourceLogID 15533-15537"] --> B
|
||||
|
||||
B["09:00:50 Capture 读取这 5 条日志<br/>（注: CapturedAt 比 OriginalTime 早 11s<br/> = ACE 可见性窗口）"] --> C
|
||||
|
||||
C["read_row 回读 5 行<br/>SELECT * FROM 接收 WHERE ID=16255..16259"] --> D
|
||||
|
||||
D["❌ 全部返回 None<br/>（行尚未对其他连接可见）"] --> E
|
||||
|
||||
E["🔴 降级为 Delete<br/>入队 SyncQueue, OperateType=Delete<br/>RowData=null"] --> F
|
||||
|
||||
F["09:00:51 Apply: Delete 分支<br/>DELETE 接收_YEAR2026 WHERE ID IN(16255..16259)<br/>@deleted=0（SQL 本就没这5行）<br/>@applied=5（队列行标 applied）"] --> G
|
||||
|
||||
G["队列行 Status = applied"] --> H
|
||||
|
||||
H["Cleanup: 查到这5条 applied<br/>DELETE Access.TableChangeLog ID=15533-15537"] --> I
|
||||
|
||||
I["🔴 日志被删，5 条 Insert 证据消失<br/>这5行从此再无机会被同步进 SQL"] --> J
|
||||
|
||||
J["8/4 05:08 compare: missing_in_sql=5<br/>ID 16255-16259"]
|
||||
|
||||
style D fill:#ffcdd2,stroke:#c62828
|
||||
style E fill:#ffcdd2,stroke:#c62828,stroke-width:3px
|
||||
style H fill:#e8f5e9,stroke:#388e3c
|
||||
style I fill:#ffcdd2,stroke:#c62828
|
||||
style J fill:#fff9c4,stroke:#f9a825
|
||||
```
|
||||
|
||||
**链式灾难**：降级 Delete（capture）→ 空打 Delete 但标 applied（apply）→ 删 Access 日志（cleanup）。三段配合下，这 5 行被「合法地」从同步链路中抹除。只要在 capture 段断开第一环（不降级、不入队），后面 apply/cleanup 就不会碰它们，日志保留，下轮自然重试。
|
||||
|
||||
---
|
||||
|
||||
## 七、重构决策点（待定）
|
||||
|
||||
基于上述流程，重构的核心是改造 capture 阶段的「降级」分支。需要决策的问题：
|
||||
|
||||
1. **读不到时的动作**：跳过不入队（日志保留，下轮重读）vs 入队但标记 defer 状态？
|
||||
2. **重试上限的判定依据**：用「重试次数」（需要存储计数）vs 用「日志年龄时间窗」（无需存储，靠 `now - OriginalTime > 阈值`）？
|
||||
3. **终态处理**：超限后该 Access 日志保留（占队列头持续重读）还是删除（丢证据）？
|
||||
4. **Update 日志**：是否套用同一套 defer 逻辑？
|
||||
|
||||
我倾向的方案：**读不到 → 不入队 + 写 defer 审计；用日志年龄（如 120s）作终态判据；超限则入队标 dead（保留 Access 日志不删，待人工）**。零 schema 改动、零状态存储。等你审完这份流程图确认方向后，我再动手。
|
||||
211
docs/plan-capture-defer-by-age.md
Normal file
211
docs/plan-capture-defer-by-age.md
Normal file
@@ -0,0 +1,211 @@
|
||||
# 重构方案：capture 读不到行时按「日志年龄」延迟重试
|
||||
|
||||
> 目标：根治「Insert 日志回读不到行 → 降级 Delete → 数据丢失」的缺陷（8/4 事件根因）。
|
||||
> 核心改动：读不到 → **不入队、不降级**，按 Access 日志的年龄决定「下轮重试」还是「判死待人工」。
|
||||
> 改动面：`capture.py`（主）、`config.py`（新增2个配置项）、`capture.py` 的 CaptureStats（新增计数器）。**零 schema 改动、零 SQL 改动、不动 apply/cleanup。**
|
||||
|
||||
---
|
||||
|
||||
## 一、当前问题行为 vs 重构后行为
|
||||
|
||||
| 场景 | 当前行为（缺陷） | 重构后行为 |
|
||||
|------|------------------|-----------|
|
||||
| Insert/Update 读不到行（可见性延迟，本次事件） | 降级 Delete → 入队 → apply 空打 Delete + 标 applied → cleanup 删 Access 日志 → **数据永久丢失** | 年龄 < 阈值：跳过不入队，日志保留，下轮重读 → 读到后正常 Insert ✅ |
|
||||
| Insert/Update 读不到行（真删除：先插后删） | 降级 Delete（碰巧正确） | 年龄 < 阈值：跳过；后续 Delete 日志会兜底正确清理；超龄判死待人工 ✅ |
|
||||
| Insert/Update 一直读不到（真异常/数据损坏） | 降级 Delete（错误） | 年龄 ≥ 阈值：入队标 dead，保留 Access 日志，**告警待人工** ✅ |
|
||||
|
||||
---
|
||||
|
||||
## 二、配置项新增（`src/sync/config.py` 的 RuntimeConfig）
|
||||
|
||||
```python
|
||||
class RuntimeConfig(BaseModel):
|
||||
# ... 既有字段 ...
|
||||
# Insert/Update 日志回读不到行时，按日志年龄延迟重试：年龄小于此秒数则
|
||||
# 跳过不入队（保留 Access 日志，下一轮 cycle 重新捕获），度过 ACE 引擎
|
||||
# 的可见性窗口；超过此年龄仍读不到则判定为真删除/异常，入队标 dead 待人工。
|
||||
# 设为 0 可关闭延迟重试（退化为旧的"立即判死"语义，但不再降级 Delete）。
|
||||
capture_defer_seconds: int = 120
|
||||
```
|
||||
|
||||
- **默认值 120 秒**：覆盖本次事件观察到的 ~11 秒可见性窗口，并留足 10 倍余量；对 `poll_interval=10s` 意味着最多重试约 12 轮。
|
||||
- 单一配置项，`config.yaml` 无需改动即可生效（用默认值）。`defer_dead_op` 不暴露为配置（实现细节，固定为 `dead` 状态，见下）。
|
||||
|
||||
---
|
||||
|
||||
## 三、capture.py 改动（核心）
|
||||
|
||||
### 3.1 CaptureStats 新增两个计数器
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class CaptureStats:
|
||||
# ... 既有字段 ...
|
||||
read: int = 0
|
||||
enqueued: int = 0
|
||||
dedup_skipped: int = 0
|
||||
downgraded: int = 0 # 保留字段，重构后恒为 0（兼容旧日志解析）
|
||||
deferred: int = 0 # 【新】读不到行但年龄 < 阈值，跳过待下轮重试
|
||||
aged_out: int = 0 # 【新】读不到行且年龄 ≥ 阈值，判死待人工
|
||||
out_of_scope: int = 0
|
||||
unknown_op: int = 0
|
||||
# ... merge() / ops_str() 同步更新 ...
|
||||
```
|
||||
|
||||
### 3.2 降级分支重构（`capture_file` 中 Insert/Update 回读逻辑）
|
||||
|
||||
**替换** 现有的第 93-108 行（整段 `if op in ("Insert", "Update"):` 块）：
|
||||
|
||||
```python
|
||||
if op in ("Insert", "Update"):
|
||||
d = reader.read_row(lr.table_name, lr.record_id)
|
||||
if d is None:
|
||||
# 读不到行：不再降级 Delete。按 Access 日志年龄决定延迟重试还是判死。
|
||||
# 必须计算日志年龄（原始设计无此逻辑）。
|
||||
age_seconds = _log_age_seconds(lr.time)
|
||||
if age_seconds < cfg.runtime.capture_defer_seconds:
|
||||
# 暂态读不到（ACE 可见性窗口）：跳过，不入队、不清理。
|
||||
# Access 日志因无对应 applied 队列行，cleanup 不会删除，下轮重试。
|
||||
st.deferred += 1
|
||||
log.warning(
|
||||
"capture DEFER %s file=%s table=%s record_id=%s log_id=%s "
|
||||
"log_time=%s age=%ds (source row unreadable; will retry next "
|
||||
"cycle while log age < %ds)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age_seconds),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
continue # ← 关键：跳过本行，archive/queue 都不写
|
||||
else:
|
||||
# 超龄仍读不到：真删除或真异常。入队标 dead，不降级、不执行任何
|
||||
# 破坏性操作。保留 Access 日志（无 applied 行 → cleanup 不删），
|
||||
# 队列健康检查会告警，等待人工介入。
|
||||
st.aged_out += 1
|
||||
log.warning(
|
||||
"capture AGED-OUT %s file=%s table=%s record_id=%s log_id=%s "
|
||||
"log_time=%s age=%ds >= %ds -- enqueuing as dead for manual "
|
||||
"review (source row still unreadable after defer window)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age_seconds),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
op = "dead" # 仅用于入队时的状态标记，见下
|
||||
else:
|
||||
row_data = json.dumps(d, ensure_ascii=False)
|
||||
```
|
||||
|
||||
### 3.3 超龄行的入队方式（`aged_out` 分支）
|
||||
|
||||
超龄行需要进 SyncQueue 但**不能被 apply 执行任何 SQL 操作**（没数据可插，也不能 Delete）。两种实现可选（我倾向 A）：
|
||||
|
||||
**方案 A（推荐）：入队后直接标 dead，OperateType 保留原 Insert/Update 真相**
|
||||
- 入队时 `OperateType` 仍写 `Insert`/`Update`（保留原始意图，便于审计），`RowData=null`。
|
||||
- 入队后立即 `UPDATE ... SET Status='dead', ErrorMsg='source row unreadable after {age}s defer'`。
|
||||
- apply 的游标只选 `Status='pending'`，dead 行不会被处理 → 不会误删。
|
||||
- queue 健康检查已有 dead 告警（`service.py:166-184`），自动浮现。
|
||||
|
||||
**方案 B：新增 OperateType='Noop'** — 改动面更大（apply 存储过程需识别），不推荐。
|
||||
|
||||
> 需要在 `sql_writer.py` 新增一个方法 `insert_dead_row(row, error_msg)`，逻辑 = 先 `insert_queue_row`（去重插入）再 `UPDATE ... SET Status='dead', RetryCount=<对应>, ErrorMsg=?`。
|
||||
|
||||
### 3.4 日志年龄计算辅助函数 `_log_age_seconds`
|
||||
|
||||
```python
|
||||
import datetime as _dt
|
||||
|
||||
def _log_age_seconds(log_time: object) -> float:
|
||||
"""Access 日志行 Time 字段距今的秒数。log_time 是 pyodbc 返回的 datetime。
|
||||
异常时返回一个大数（视为已超龄），确保宁可判死也不无限重试。"""
|
||||
try:
|
||||
if isinstance(log_time, _dt.datetime):
|
||||
return (_dt.datetime.now() - log_time).total_seconds()
|
||||
# Access via ODBC 通常返回 datetime；兜底处理 naive/其它类型
|
||||
return float("inf")
|
||||
except Exception:
|
||||
return float("inf")
|
||||
```
|
||||
|
||||
> ⚠️ **时区/时钟注意**：`lr.time` 是 Access 端写入的本地时间，`datetime.now()` 也是本机本地时间，两者同在 114 主机同一时区，可直接相减。本次事件中 OriginalTime/CapturedAt 的"倒挂"现象（差11秒）不影响此逻辑——因为按年龄判断，即便 lr.time 偏早，age 只会被算得更大，倾向判死而非误伤，方向安全。
|
||||
|
||||
---
|
||||
|
||||
## 四、归档表（SyncLogArchive）的处理
|
||||
|
||||
| 分支 | 是否写 archive | 理由 |
|
||||
|------|---------------|------|
|
||||
| DEFER（跳过） | **不写** | 日志保留在 Access，下轮 capture 会重新读到并正常归档；此时写 archive 反而会在去重表里留下"读不到"的半成品记录 |
|
||||
| AGED-OUT（判死） | **写** | 超龄是终态，需留永久审计（OriginalOperateType=Insert/Update, ProcessedOperateType='AgedOut', RowData=null, OriginalTime=lr.time） |
|
||||
|
||||
> ProcessedOperateType 新增值 `'AgedOut'`（仅 archive 表用，varchar(10) 放得下 7 字符）。这是纯审计标记，不影响任何执行逻辑。
|
||||
|
||||
---
|
||||
|
||||
## 五、不改动的地方（明确边界）
|
||||
|
||||
| 模块 | 是否改动 | 原因 |
|
||||
|------|---------|------|
|
||||
| `cleanup.py` | ❌ 不改 | 只删 `Status='applied'` 的日志；DEFER 行不入队无 applied 行 → 日志保留；AGED-OUT 标 dead 非 applied → 日志也保留。天然自洽。 |
|
||||
| `sql/02_sync_apply.sql` | ❌ 不改 | apply 游标只选 `pending`，dead 行天然跳过；DEFER 行根本不入队。 |
|
||||
| `sql/01_sync_queue.sql` | ❌ 不改 | 不新增列，不改索引。 |
|
||||
| `service.py` | ❌ 不改 | 队列健康检查已有 error/dead 告警（`queue_error_samples`），AGED-OUT 的 dead 行会自动被它捕获并 WARNING。capture summary 日志格式已包含新计数器（由 CaptureStats.ops_str/merge 驱动）。 |
|
||||
| `access_reader.py` | ❌ 不改 | `read_row` 行为不变。 |
|
||||
| `config.yaml` | ❌ 不改 | 用默认值 120s 即可。 |
|
||||
|
||||
---
|
||||
|
||||
## 六、本次事件 5 行的重放（重构后）
|
||||
|
||||
```
|
||||
cycle N (09:00:50): capture 读到 5 条 Insert 日志
|
||||
→ read_row 返回 None
|
||||
→ age = now(09:00:50) - log_time(09:01:01) → 注: 因 Access 时间戳特性 age 可能算成负或小
|
||||
→ 即便按最保守计算，age 远 < 120s
|
||||
→ DEFER：跳过不入队，写 WARNING，Access 日志保留
|
||||
|
||||
cycle N+1 (09:01:00): capture 再次读到这 5 条日志
|
||||
→ read_row 此刻可见性窗口已过（11s > 窗口）→ 读到行 ✅
|
||||
→ 正常入队 OperateType=Insert，带完整 RowData
|
||||
→ apply MERGE → SQL 正确写入 5 行 ✅
|
||||
→ cleanup 删 Access 日志（这次是 applied，合理）
|
||||
```
|
||||
|
||||
**结果：8/4 compare 不再出现 missing_in_sql=5。**
|
||||
|
||||
---
|
||||
|
||||
## 七、验证计划
|
||||
|
||||
1. **单元测试**（`tests/test_capture.py`）：
|
||||
- mock `read_row` 返回 None + `lr.time` 为近时 → 断言 `deferred=1, enqueued=0, aged_out=0`，不调用 `insert_queue_row` / `insert_archive_row`。
|
||||
- mock `read_row` 返回 None + `lr.time` 为 200s 前 → 断言 `aged_out=1`，调用 `insert_dead_row`，Status=dead。
|
||||
- mock `read_row` 返回 dict + 任意时间 → 断言正常入队（回归测试）。
|
||||
- `capture_defer_seconds=0` → 任何读不到都立即判死（边界）。
|
||||
2. **现有测试回归**：`pytest` 全绿（确保去重/正常 Insert/真 Delete 路径不受影响）。
|
||||
3. **集成验证**（部署后观察 1-2 天）：
|
||||
- 关注 capture summary 日志的 `deferred=` 计数，确认批量插入场景下有 defer 发生且下轮 enqueued。
|
||||
- 关注 `queue health` WARNING，确认 dead 行（如有）被正确告警。
|
||||
- 跑 `compare --granularity ids`，确认无 missing_in_sql。
|
||||
|
||||
---
|
||||
|
||||
## 八、改动文件清单
|
||||
|
||||
| 文件 | 改动类型 | 说明 |
|
||||
|------|---------|------|
|
||||
| `src/sync/config.py` | 新增字段 | RuntimeConfig 加 `capture_defer_seconds: int = 120` |
|
||||
| `src/sync/capture.py` | 核心重构 | 降级分支 → defer/aged_out 分支；CaptureStats 加 2 计数器；新增 `_log_age_seconds` |
|
||||
| `src/sync/sql_writer.py` | 新增方法 | `insert_dead_row(row, error_msg)`：去重插入后立即标 dead |
|
||||
| `tests/test_capture.py` | 新增用例 | 覆盖 DEFER / AGED-OUT / 正常 / 边界 4 种情况 |
|
||||
|
||||
**总计 4 个文件，零 SQL/零 schema 改动。**
|
||||
|
||||
---
|
||||
|
||||
## 九、待你确认的决策点
|
||||
|
||||
1. **`capture_defer_seconds` 默认值 120s** 是否合适？（覆盖11s窗口×10倍余量）
|
||||
2. **AGED-OUT 行的处理**：入队标 dead（方案A，推荐）vs 其它？
|
||||
3. **archive 表 ProcessedOperateType 新增值 `'AgedOut'`** 是否可接受？
|
||||
4. **downgraded 计数器**：保留为恒0（向后兼容旧日志解析）还是直接删除？
|
||||
|
||||
确认后我即按此方案执行。
|
||||
164
docs/plan-health-check-api.md
Normal file
164
docs/plan-health-check-api.md
Normal 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
215
docs/plan-health-check-fastapi.md
Normal 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 小时？倾向启动至今。
|
||||
|
||||
确认后实施。
|
||||
@@ -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
|
||||
|
||||
@@ -2,23 +2,32 @@
|
||||
|
||||
Every consumed ``TableChangeLog`` row is appended to the permanent audit store
|
||||
(``SyncLogArchive``) BEFORE it is enqueued, so the original evidence survives
|
||||
cleanup. On top of that, this module now emits a detailed audit trail to the
|
||||
cleanup. On top of that, this module emits a detailed audit trail to the
|
||||
service log:
|
||||
|
||||
- WARNING for every operate-type downgrade (Insert/Update -> Delete because
|
||||
the source row was unreadable at capture time), with record id, log id and
|
||||
the Access log timestamp -- the exact event class behind the 14287 incident,
|
||||
previously invisible in the text log;
|
||||
- WARNING for every Insert/Update that could not be read back from the source
|
||||
table at capture time. Two outcomes, both logged by identity (record id,
|
||||
log id, log time, age):
|
||||
* DEFER -- the log row is younger than ``capture_defer_seconds``: treated
|
||||
as an ACE visibility-latency window. The row is NOT enqueued, the Access
|
||||
log is left intact (cleanup only deletes rows whose queue status is
|
||||
``applied``), and the next cycle re-reads it. This is the fix for the
|
||||
data-loss incident where an Insert downgraded to Delete silently dropped
|
||||
rows from SQL Server (see docs/plan-capture-defer-by-age.md).
|
||||
* AGED-OUT -- still unreadable past the defer window: enqueued directly as
|
||||
``dead`` (usp_SyncApply never selects dead rows, so no destructive SQL
|
||||
runs) and left for manual review; the Access log is also preserved.
|
||||
- WARNING for unknown operate types, with enough identity (log id / record id
|
||||
/ time) to locate and repair the offending log row manually;
|
||||
- a per-file INFO summary: rows read, newly enqueued, dedup-skipped
|
||||
(re-capture after a failed apply -- a symptom worth noticing), downgraded,
|
||||
out-of-scope, unknown ops, the processed log-ID range and a per-operation
|
||||
breakdown;
|
||||
(re-capture after a failed apply -- a symptom worth noticing), deferred,
|
||||
aged-out, out-of-scope, unknown ops, the processed log-ID range and a
|
||||
per-operation breakdown;
|
||||
- DEBUG detail for individual out-of-scope and dedup skips.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
@@ -31,6 +40,23 @@ from .targets import is_synced_table
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _log_age_seconds(log_time: object) -> float:
|
||||
"""Seconds elapsed since the Access log row's ``Time`` value.
|
||||
|
||||
``log_time`` is the raw pyodbc datetime from ``TableChangeLog.Time`` (set
|
||||
by the Access data macro). Both it and ``datetime.now()`` run on the same
|
||||
host/clock, so direct subtraction is valid. On any anomaly (missing /
|
||||
non-datetime value) returns ``inf`` so the caller errs toward ageing out
|
||||
rather than retrying forever.
|
||||
"""
|
||||
try:
|
||||
if isinstance(log_time, _dt.datetime):
|
||||
return (_dt.datetime.now() - log_time).total_seconds()
|
||||
return float("inf")
|
||||
except Exception:
|
||||
return float("inf")
|
||||
|
||||
|
||||
@dataclass
|
||||
class CaptureStats:
|
||||
"""Counters for one capture pass (per file, or aggregated per cycle)."""
|
||||
@@ -38,7 +64,8 @@ class CaptureStats:
|
||||
read: int = 0 # change-log rows read from Access
|
||||
enqueued: int = 0 # rows newly inserted into SyncQueue
|
||||
dedup_skipped: int = 0 # already queued (re-capture after a failed apply)
|
||||
downgraded: int = 0 # Insert/Update downgraded to Delete
|
||||
deferred: int = 0 # Insert/Update unreadable but young -> retry next cycle
|
||||
aged_out: int = 0 # Insert/Update still unreadable past defer window -> dead
|
||||
out_of_scope: int = 0 # log rows for tables outside the sync scope
|
||||
unknown_op: int = 0 # log rows with an unrecognised OperateType
|
||||
min_log_id: int | None = None
|
||||
@@ -53,7 +80,8 @@ class CaptureStats:
|
||||
self.read += other.read
|
||||
self.enqueued += other.enqueued
|
||||
self.dedup_skipped += other.dedup_skipped
|
||||
self.downgraded += other.downgraded
|
||||
self.deferred += other.deferred
|
||||
self.aged_out += other.aged_out
|
||||
self.out_of_scope += other.out_of_scope
|
||||
self.unknown_op += other.unknown_op
|
||||
for k, v in other.ops.items():
|
||||
@@ -93,19 +121,66 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
|
||||
if op in ("Insert", "Update"):
|
||||
d = reader.read_row(lr.table_name, lr.record_id)
|
||||
if d is None:
|
||||
# 行已删（或此刻不可读），降级为 Delete。这是数据差异排查的
|
||||
# 头号嫌疑事件（参见 14287 事件），必须在文本日志显式留痕，
|
||||
# 而不只是写入 SyncLogArchive。
|
||||
op = "Delete"
|
||||
st.downgraded += 1
|
||||
# 回读不到整行：不再降级 Delete。按 Access 日志年龄决定动作，
|
||||
# 根治"Insert 降级 Delete 致数据丢失"缺陷。
|
||||
age = _log_age_seconds(lr.time)
|
||||
if age < cfg.runtime.capture_defer_seconds:
|
||||
# 暂态（ACE 可见性延迟）：跳过，不入队、不写 archive。
|
||||
# Access 日志因无 applied 队列行，cleanup 不会删除，
|
||||
# 下一轮 cycle 会重新读到。
|
||||
st.deferred += 1
|
||||
log.warning(
|
||||
"capture DOWNGRADE %s->Delete file=%s table=%s "
|
||||
"record_id=%s log_id=%s log_time=%s "
|
||||
"(source row unreadable at capture time; original intent "
|
||||
"preserved in SyncLogArchive.OriginalOperateType)",
|
||||
lr.operate_type, fm.file, lr.table_name,
|
||||
lr.record_id, lr.id, lr.time,
|
||||
"capture DEFER %s file=%s table=%s record_id=%s "
|
||||
"log_id=%s log_time=%s age=%ds (source row unreadable; "
|
||||
"will retry next cycle while age < %ds)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
continue
|
||||
# 超龄仍读不到：真删除或真异常。入队标 dead，不降级、不执行
|
||||
# 任何破坏性 SQL。保留 Access 日志（无 applied 行 → cleanup
|
||||
# 不删），队列健康检查会告警，等待人工介入。
|
||||
st.aged_out += 1
|
||||
log.warning(
|
||||
"capture AGED-OUT %s file=%s table=%s record_id=%s "
|
||||
"log_id=%s log_time=%s age=%ds >= %ds -- enqueuing as "
|
||||
"dead for manual review (source row still unreadable "
|
||||
"after defer window)",
|
||||
lr.operate_type, fm.file, lr.table_name, lr.record_id,
|
||||
lr.id, lr.time, int(age),
|
||||
cfg.runtime.capture_defer_seconds,
|
||||
)
|
||||
target_schema = fm.schema
|
||||
target_table = fm.target_table(lr.table_name)
|
||||
writer.insert_archive_row(ArchiveRow(
|
||||
source_file=fm.file,
|
||||
source_table=lr.table_name,
|
||||
source_log_id=lr.id,
|
||||
record_id=lr.record_id,
|
||||
target_schema=target_schema,
|
||||
target_table=target_table,
|
||||
original_operate_type=lr.operate_type,
|
||||
processed_operate_type="AgedOut",
|
||||
row_data=None,
|
||||
original_time=lr.time,
|
||||
))
|
||||
qr = QueueRow(
|
||||
source_file=fm.file,
|
||||
source_table=lr.table_name,
|
||||
record_id=lr.record_id,
|
||||
target_schema=target_schema,
|
||||
target_table=target_table,
|
||||
source_log_id=lr.id,
|
||||
operate_type=op, # 保留原始 Insert/Update，便于审计
|
||||
row_data=None,
|
||||
)
|
||||
writer.insert_dead_row(
|
||||
qr,
|
||||
f"source row unreadable after {int(age)}s defer window "
|
||||
f"(capture_defer_seconds={cfg.runtime.capture_defer_seconds})",
|
||||
)
|
||||
continue
|
||||
else:
|
||||
row_data = json.dumps(d, ensure_ascii=False)
|
||||
elif op != "Delete":
|
||||
@@ -122,7 +197,7 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
|
||||
# Persist the ORIGINAL log entry to the permanent audit store BEFORE the
|
||||
# queue insert (and long before cleanup deletes the Access log). This
|
||||
# keeps both the source operate type (lr.operate_type) and the processed
|
||||
# one (op) so a downgrade like Insert->Delete stays reconstructible.
|
||||
# one (op) so any divergence stays reconstructible.
|
||||
writer.insert_archive_row(ArchiveRow(
|
||||
source_file=fm.file,
|
||||
source_table=lr.table_name,
|
||||
@@ -164,11 +239,11 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
|
||||
if st.read:
|
||||
log.info(
|
||||
"capture file=%s read=%d enqueued=%d dedup_skipped=%d "
|
||||
"downgraded=%d out_of_scope=%d unknown_op=%d "
|
||||
"deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d "
|
||||
"log_ids=%s..%s ops={%s}",
|
||||
fm.file, st.read, st.enqueued, st.dedup_skipped, st.downgraded,
|
||||
st.out_of_scope, st.unknown_op, st.min_log_id, st.max_log_id,
|
||||
st.ops_str(),
|
||||
fm.file, st.read, st.enqueued, st.dedup_skipped, st.deferred,
|
||||
st.aged_out, st.out_of_scope, st.unknown_op, st.min_log_id,
|
||||
st.max_log_id, st.ops_str(),
|
||||
)
|
||||
else:
|
||||
log.debug("capture file=%s: change log empty", fm.file)
|
||||
|
||||
@@ -27,6 +27,12 @@ class RuntimeConfig(BaseModel):
|
||||
cleanup_batch_size: int = 200
|
||||
cleanup_lock_retries: int = 3
|
||||
cleaned_retention_hours: int = 24
|
||||
# Insert/Update 日志回读不到整行时，按 Access 日志的"年龄"（log_time 距今
|
||||
# 秒数）决定动作：年龄小于此阈值视为 ACE 引擎的可见性延迟，跳过不入队
|
||||
# （Access 日志保留，下一轮 cycle 重新捕获）；年龄达到此阈值仍读不到则
|
||||
# 判定为真删除/异常，入队标 dead 待人工。根治"降级 Delete 致数据丢失"
|
||||
# （见 docs/plan-capture-defer-by-age.md）。设为 0 关闭延迟重试。
|
||||
capture_defer_seconds: int = 60
|
||||
# Idle cycles now log at DEBUG; the service emits an INFO heartbeat at this
|
||||
# interval while idle so a quiet log still proves the service is alive.
|
||||
idle_heartbeat_seconds: int = 600
|
||||
@@ -47,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
308
src/sync/health.py
Normal 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 "."
|
||||
@@ -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)
|
||||
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",
|
||||
"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).
|
||||
@@ -109,10 +169,10 @@ def cycle(cfg) -> bool:
|
||||
activity = True
|
||||
log.info(
|
||||
"capture summary: read=%d enqueued=%d dedup_skipped=%d "
|
||||
"downgraded=%d out_of_scope=%d unknown_op=%d ops={%s}",
|
||||
"deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d ops={%s}",
|
||||
total.read, total.enqueued, total.dedup_skipped,
|
||||
total.downgraded, total.out_of_scope, total.unknown_op,
|
||||
total.ops_str(),
|
||||
total.deferred, total.aged_out, total.out_of_scope,
|
||||
total.unknown_op, total.ops_str(),
|
||||
)
|
||||
else:
|
||||
log.debug("capture summary: no new change-log rows in any file")
|
||||
@@ -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)
|
||||
|
||||
@@ -132,6 +132,35 @@ class SqlWriter:
|
||||
# autocommit: statement already committed.
|
||||
return cur.rowcount > 0
|
||||
|
||||
def insert_dead_row(self, row: QueueRow, error_msg: str) -> bool:
|
||||
"""Enqueue ``row`` then immediately mark it ``dead`` (never applied).
|
||||
|
||||
Used by capture for an Insert/Update log whose source row stays
|
||||
unreadable past the defer window (``capture_defer_seconds``): the row
|
||||
is recorded for audit/health-alerting purposes but never executed by
|
||||
``usp_SyncApply`` (which only selects ``Status='pending'``), so no
|
||||
destructive SQL ever runs against the mirror table. The OperateType is
|
||||
preserved as the original Insert/Update so the intent stays visible,
|
||||
while ErrorMsg explains why it was parked.
|
||||
|
||||
Dedup semantics match ``insert_queue_row``: if the (SourceFile,
|
||||
SourceTable, SourceLogID) row already exists, the insert is skipped
|
||||
and the existing row is re-marked dead. Returns True when a new row
|
||||
was inserted, False on a dedup hit (the re-mark still happens).
|
||||
"""
|
||||
inserted = self.insert_queue_row(row)
|
||||
cur = self._conn.cursor()
|
||||
cur.execute(
|
||||
f"UPDATE {self.queue_table} SET Status='dead', ErrorMsg=? "
|
||||
"WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?",
|
||||
error_msg,
|
||||
row.source_file,
|
||||
row.source_table,
|
||||
row.source_log_id,
|
||||
)
|
||||
# autocommit: statement already committed.
|
||||
return inserted
|
||||
|
||||
def insert_archive_row(self, row: ArchiveRow) -> bool:
|
||||
"""Append ``row`` to the permanent audit store (dedup on source keys).
|
||||
|
||||
|
||||
11
src/sync/web/__init__.py
Normal file
11
src/sync/web/__init__.py
Normal 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
33
src/sync/web/app.py
Normal 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
42
src/sync/web/deps.py
Normal 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
2
src/sync/web/routes/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
"""Route modules. Add new operational endpoints here and include their router
|
||||
in ``app.py``."""
|
||||
79
src/sync/web/routes/health.py
Normal file
79
src/sync/web/routes/health.py
Normal 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
204
src/sync/web/schemas.py
Normal 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.",
|
||||
)
|
||||
@@ -1,23 +1,34 @@
|
||||
import datetime as _dt
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from sync.config import FileMapping, SyncConfig, AccessConfig, RuntimeConfig, SqlServerConfig
|
||||
from sync.access_reader import LogRow
|
||||
from sync.capture import capture_file
|
||||
|
||||
def _cfg():
|
||||
|
||||
def _cfg(defer=60):
|
||||
return SyncConfig(sql_server=SqlServerConfig(conn_str="x"),
|
||||
access=AccessConfig(driver="d", roots={"2026": "r"}),
|
||||
runtime=RuntimeConfig(),
|
||||
runtime=RuntimeConfig(capture_defer_seconds=defer),
|
||||
files=[])
|
||||
|
||||
|
||||
def _fm():
|
||||
return FileMapping(file="x.accdb", root="2026", schema="s",
|
||||
year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
|
||||
|
||||
def test_capture_insert_reads_row_and_queues():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding",
|
||||
year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", None)]
|
||||
reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = {"ID": 34041, "订单号": "X1"}
|
||||
writer = MagicMock()
|
||||
n = capture_file(fm, reader, writer, cfg)
|
||||
assert n == 1
|
||||
st = capture_file(fm, reader, writer, cfg)
|
||||
assert st.enqueued == 1
|
||||
assert st.deferred == 0 and st.aged_out == 0
|
||||
args = writer.insert_queue_row.call_args[0][0]
|
||||
assert args.target_schema == "TIGWelding"
|
||||
assert args.target_table == "表壳焊接记录_YEAR2026"
|
||||
@@ -25,37 +36,106 @@ def test_capture_insert_reads_row_and_queues():
|
||||
assert '"订单号": "X1"' in args.row_data
|
||||
assert args.source_log_id == 10
|
||||
|
||||
def test_capture_update_missing_row_downgrades_to_delete():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
||||
|
||||
def test_capture_unreadable_young_row_is_deferred():
|
||||
# Insert 日志年龄 < capture_defer_seconds → 跳过，不入队、不写 archive
|
||||
cfg = _cfg(defer=60)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(11, "T", "5", "Update", None)]
|
||||
reader.read_row.return_value = None # 行已删
|
||||
reader.read_log.return_value = [LogRow(11, "T", "5", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = None # 回读不到
|
||||
writer = MagicMock()
|
||||
n = capture_file(fm, reader, writer, cfg)
|
||||
assert n == 1
|
||||
args = writer.insert_queue_row.call_args[0][0]
|
||||
assert args.operate_type == "Delete"
|
||||
assert args.row_data is None
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.deferred == 1
|
||||
assert st.enqueued == 0 and st.aged_out == 0
|
||||
writer.insert_queue_row.assert_not_called()
|
||||
writer.insert_dead_row.assert_not_called()
|
||||
writer.insert_archive_row.assert_not_called()
|
||||
|
||||
|
||||
def test_capture_unreadable_young_update_also_deferred():
|
||||
# Update 同样走 defer 路径（不降级 Delete）
|
||||
cfg = _cfg(defer=60)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(12, "T", "6", "Update", _dt.datetime.now())]
|
||||
reader.read_row.return_value = None
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.deferred == 1
|
||||
writer.insert_queue_row.assert_not_called()
|
||||
writer.insert_dead_row.assert_not_called()
|
||||
|
||||
|
||||
def test_capture_unreadable_aged_row_marked_dead():
|
||||
# Insert 日志年龄 >= capture_defer_seconds → 入队标 dead，不执行破坏性 SQL
|
||||
cfg = _cfg(defer=60)
|
||||
old_time = _dt.datetime.now() - _dt.timedelta(seconds=200)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(13, "T", "7", "Insert", old_time)]
|
||||
reader.read_row.return_value = None # 仍读不到
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.aged_out == 1
|
||||
assert st.enqueued == 0 and st.deferred == 0
|
||||
# archive 留痕（ProcessedOperateType=AgedOut）
|
||||
arch = writer.insert_archive_row.call_args[0][0]
|
||||
assert arch.original_operate_type == "Insert"
|
||||
assert arch.processed_operate_type == "AgedOut"
|
||||
assert arch.row_data is None
|
||||
# 入队标 dead，OperateType 保留原始 Insert
|
||||
qr, err = writer.insert_dead_row.call_args[0]
|
||||
assert qr.operate_type == "Insert" # 不降级
|
||||
assert qr.row_data is None
|
||||
assert "200s" in err
|
||||
|
||||
|
||||
def test_capture_defer_zero_disables_retry():
|
||||
# capture_defer_seconds=0 → 任何读不到都立即判死（边界）
|
||||
cfg = _cfg(defer=0)
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(14, "T", "8", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = None
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.aged_out == 1
|
||||
assert st.deferred == 0
|
||||
|
||||
|
||||
def test_capture_delete_never_reads_row():
|
||||
# Delete 日志无需回读，直接入队
|
||||
cfg = _cfg()
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(15, "T", "9", "Delete", _dt.datetime.now())]
|
||||
writer = MagicMock()
|
||||
st = capture_file(_fm(), reader, writer, cfg)
|
||||
assert st.enqueued == 1
|
||||
reader.read_row.assert_not_called()
|
||||
qr = writer.insert_queue_row.call_args[0][0]
|
||||
assert qr.operate_type == "Delete"
|
||||
assert qr.row_data is None
|
||||
|
||||
|
||||
def test_capture_skips_excluded_tables():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"])
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026",
|
||||
exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"])
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", None),
|
||||
LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", None)]
|
||||
reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", _dt.datetime.now()),
|
||||
LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", _dt.datetime.now())]
|
||||
writer = MagicMock()
|
||||
assert capture_file(fm, reader, writer, cfg) == 0
|
||||
st = capture_file(fm, reader, writer, cfg)
|
||||
assert st.enqueued == 0 and st.out_of_scope == 2
|
||||
writer.insert_queue_row.assert_not_called()
|
||||
|
||||
|
||||
def test_capture_include_tables_filter():
|
||||
cfg = _cfg()
|
||||
fm = FileMapping(file="x.accdb", root="2026", schema="inspectionRecords", year_suffix="_YEAR2026",
|
||||
exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
|
||||
reader = MagicMock()
|
||||
reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", None),
|
||||
LogRow(2, "其它表", "1", "Insert", None)]
|
||||
reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", _dt.datetime.now()),
|
||||
LogRow(2, "其它表", "1", "Insert", _dt.datetime.now())]
|
||||
reader.read_row.return_value = {"ID": 1}
|
||||
writer = MagicMock()
|
||||
assert capture_file(fm, reader, writer, cfg) == 1
|
||||
st = capture_file(fm, reader, writer, cfg)
|
||||
assert st.enqueued == 1
|
||||
assert writer.insert_queue_row.call_args[0][0].target_table == "检验合格记录表_YEAR2026"
|
||||
|
||||
183
tests/test_health.py
Normal file
183
tests/test_health.py
Normal 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"
|
||||
Reference in New Issue
Block a user