From ae12403c0b6bd3f03c8d82047fd58e0a6bca0e4b Mon Sep 17 00:00:00 2001 From: liuwenxin <1009266660@qq.com> Date: Tue, 3 Nov 2026 17:18:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E5=A4=87=E4=BB=BD):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E5=BF=AB=E7=85=A7=E4=BD=8D=E7=82=B9=E4=B8=8E=E8=BD=AE=E8=BD=AC?= =?UTF-8?q?=E6=9C=9F=E9=97=B4=E7=9A=84=E5=A2=9E=E9=87=8F=E9=81=97=E6=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 统一跨平台备份入口，校验快照位点、完整性及增量链后隔离恢复。 末个 binlog 按读取位点归档，避免轮转后并发事务被续接跳过。 修复 Windows 安装器的编码、默认路径和花括号路径参数。 --- backend/tests/unit/test_payroll_backup.py | 144 ++++++ docker/Dockerfile.backup | 7 + scripts/backup_full.sh | 63 +-- scripts/backup_incremental.sh | 88 +--- scripts/health_check.sh | 107 +--- scripts/payroll_backup.py | 489 ++++++++++++++++++ scripts/restore_drill.sh | 115 +--- .../windows/Install-PayrollBackupTasks.ps1 | 51 ++ 8 files changed, 700 insertions(+), 364 deletions(-) create mode 100644 backend/tests/unit/test_payroll_backup.py create mode 100644 docker/Dockerfile.backup create mode 100644 scripts/payroll_backup.py create mode 100644 scripts/windows/Install-PayrollBackupTasks.ps1 diff --git a/backend/tests/unit/test_payroll_backup.py b/backend/tests/unit/test_payroll_backup.py new file mode 100644 index 0000000..833649c --- /dev/null +++ b/backend/tests/unit/test_payroll_backup.py @@ -0,0 +1,144 @@ +"""备份失败不得伪装成功，损坏文件及断链不得进入恢复。""" + +import importlib.util +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +spec = importlib.util.spec_from_file_location( + "payroll_backup", Path(__file__).resolve().parents[3] / "scripts/payroll_backup.py" +) +backup = importlib.util.module_from_spec(spec) +spec.loader.exec_module(backup) + + +@pytest.fixture +def chain(tmp_path): + data = tmp_path / "full.sql.gz" + data.write_bytes(b"fixture") + baseline = {"kind": "FULL", "database": "payroll", "server_uuid": "server-a", + "point": {"file": "binlog.000001", "position": 100}, + "entries": [{"file": data.name, "sha256": backup.checksum(data)}]} + backup.json_write(tmp_path / "full.json", baseline) + state = {"full": "full.json", "point": baseline["point"], "increments": []} + backup.json_write(tmp_path / "state.json", state) + return SimpleNamespace(directory=tmp_path, container="test", database="payroll"), baseline, state + + +def test_备份篡改与路径越界必须拒绝(chain): + args, _, _ = chain + (args.directory / "full.sql.gz").write_bytes(b"corrupted") + with pytest.raises(backup.BackupError, match="SHA-256"): + backup.checked_chain(args) + with pytest.raises(backup.BackupError, match="文件名"): + backup.load_manifest(args.directory, "../external.json") + + +def test_跨全量与位点断链拒绝恢复(chain): + args, _, state = chain + change = {"full": "other.json", "start": state["point"], "end": state["point"], "entries": []} + backup.json_write(args.directory / "increment.json", change) + state["increments"] = ["increment.json"] + backup.json_write(args.directory / "state.json", state) + with pytest.raises(backup.BackupError, match="不连续"): + backup.checked_chain(args) + + +def test_无更新不轮转日志也不生成增量(chain, monkeypatch): + args, baseline, state = chain + calls = [] + def sql(*values): + calls.append(values) + return baseline["server_uuid"] + monkeypatch.setattr(backup, "mysql", sql) + monkeypatch.setattr(backup, "position", lambda _: state["point"]) + assert backup.incremental(args)["status"] == "UNCHANGED" + assert len(calls) == 1 + assert not list(args.directory.glob("increment*")) + + +def test_服务器更换后不续接旧链(chain, monkeypatch): + args, _, _ = chain + monkeypatch.setattr(backup, "mysql", lambda *a: "server-b") + with pytest.raises(backup.BackupError, match="身份已变化"): + backup.incremental(args) + + +def test_备份检查拒绝同服务器上的其他数据库(chain, monkeypatch): + args, _, _ = chain + args.database = "other_payroll" + calls = [] + monkeypatch.setattr(backup, "mysql", lambda *a: calls.append(a)) + with pytest.raises(backup.BackupError, match="检查目标不符"): + backup.check(args) + assert not calls + + +def test_轮转后的并发事务归档到末位点且续接不跳过(chain, monkeypatch): + args, baseline, state = chain + start = state["point"] + end = {"file": "binlog.000002", "position": 400} + monkeypatch.setattr(backup, "mysql", lambda *a: baseline["server_uuid"]) + monkeypatch.setattr(backup, "position", lambda _: end) + monkeypatch.setattr(backup, "logs", lambda _: ["binlog.000001", "binlog.000002"]) + commands = [] + + def archive(command, path): + commands.append(command) + path.write_bytes(b"archived") + return {"file": path.name, "sha256": backup.checksum(path)} + + monkeypatch.setattr(backup, "stream_gzip", archive) + first = backup.incremental(args) + manifest = backup.load_manifest(args.directory, first["manifest"]) + assert manifest["start"] == start + assert manifest["end"] == end + assert [entry["binlog"] for entry in manifest["entries"]] == [ + "binlog.000001", "binlog.000002" + ] + assert commands[-1][-4:] == ["head", "-c", "400", "/var/lib/mysql/binlog.000002"] + + # 下一轮仍从上一轮已归档的末位点续接，跨轮不会遗漏或重复计算事务。 + next_end = {"file": "binlog.000003", "position": 157} + monkeypatch.setattr(backup, "position", lambda _: next_end) + monkeypatch.setattr(backup, "logs", lambda _: ["binlog.000002", "binlog.000003"]) + second = backup.incremental(args) + next_manifest = backup.load_manifest(args.directory, second["manifest"]) + assert next_manifest["start"] == end + assert [entry["binlog"] for entry in next_manifest["entries"]] == [ + "binlog.000002", "binlog.000003" + ] + assert backup.checked_chain(args)[1] == [manifest, next_manifest] + + +def test_损坏清单不得启动任何恢复容器(chain, monkeypatch): + args, _, _ = chain + (args.directory / "full.sql.gz").write_bytes(b"broken") + commands = [] + monkeypatch.setattr(backup, "run", lambda command, **kw: commands.append(command)) + with pytest.raises(backup.BackupError): + backup.restore(args) + assert commands == [] + + +def test_清单原子落盘不留下半成品(tmp_path): + target = tmp_path / "manifest.json" + backup.json_write(target, {"status": "SUCCEEDED"}) + assert json.loads(target.read_text())["status"] == "SUCCEEDED" + assert not list(tmp_path.glob("*.partial")) + + +def test_MySQL探活使用TCP避开镜像初始化临时服务(monkeypatch): + commands = [] + + def run(command, **kwargs): + commands.append(command) + assert kwargs["data"] == b"SELECT 1" + return b"1\n" + + monkeypatch.setattr(backup, "run", run) + assert backup.mysql("isolated-test", "SELECT 1") == "1" + command = commands[0] + assert "--host=127.0.0.1 --protocol=TCP" in command[command.index("-c") + 1] diff --git a/docker/Dockerfile.backup b/docker/Dockerfile.backup new file mode 100644 index 0000000..d8c4555 --- /dev/null +++ b/docker/Dockerfile.backup @@ -0,0 +1,7 @@ +# 官方最小 MySQL 镜像缺少 mysqlbinlog；仅在独立恢复演练镜像内补齐同版本工具。 +ARG MYSQL_IMAGE=mysql:8.0.46 +FROM ${MYSQL_IMAGE} +ADD https://repo.mysql.com/yum/mysql-8.0-community/el/9/x86_64/mysql-community-client-8.0.46-1.el9.x86_64.rpm /tmp/mysql-client.rpm +RUN echo "decc2c06fac430b7987cfae4e4039cb49bdce56dfadfd0a28738bf0757f04f17 /tmp/mysql-client.rpm" | sha256sum -c - \ + && rpm -i --nodeps --replacefiles /tmp/mysql-client.rpm \ + && mysqlbinlog --version diff --git a/scripts/backup_full.sh b/scripts/backup_full.sh index 2030a4d..3fa7364 100644 --- a/scripts/backup_full.sh +++ b/scripts/backup_full.sh @@ -1,64 +1,5 @@ #!/usr/bin/env bash -# 衍水工资系统 —— MySQL 全量备份（⑨ 备份恢复与运维检查） -# -# 思路：在跑中的容器内执行 mysqldump，免暴露 3306 端口给宿主机工具， -# 也避免宿主机装 MySQL 客户端。凭据从 docker/.env 读取（不入脚本、不入版本库）。 -# -# 用法：在项目根目录执行 bash scripts/backup_full.sh -# 产物：db/backups/payroll_YYYYMMDD_HHMMSS.sql.gz，永久保留，不自动清理。 - +# 统一到跨平台备份工具；成功以完整性清单及退出状态为准。 set -euo pipefail - ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -cd "$ROOT" - -if [[ -f docker/.env ]]; then - # shellcheck disable=SC1091 - source docker/.env -else - echo "ERROR: docker/.env 不存在（请先创建 docker/.env 并填入 MySQL 凭据）" >&2 - exit 1 -fi - -: "${MYSQL_DATABASE:?docker/.env 缺 MYSQL_DATABASE}" -: "${MYSQL_ROOT_PASSWORD:?docker/.env 缺 MYSQL_ROOT_PASSWORD}" - -BACKUP_DIR="$ROOT/db/backups" -mkdir -p "$BACKUP_DIR" -STAMP="$(date +%Y%m%d_%H%M%S)" -OUT="$BACKUP_DIR/payroll_${STAMP}.sql.gz" - -# 容器名固定（compose 显式 container_name），与 docker/docker-compose.dev.yml 保持一致 -CONTAINER="ys-mysql-dev" -if ! docker inspect "$CONTAINER" >/dev/null 2>&1; then - echo "ERROR: 容器 $CONTAINER 未运行（先 docker compose up -d）" >&2 - exit 1 -fi - -echo "开始备份 $MYSQL_DATABASE → $(basename "$OUT")" -docker exec "$CONTAINER" sh -c \ - "mysqldump -uroot -p\"\$MYSQL_ROOT_PASSWORD\" --single-transaction --routines --triggers \"$MYSQL_DATABASE\"" \ - | gzip > "$OUT" - -if [[ ! -s "$OUT" ]]; then - echo "ERROR: 备份文件为空（数据库名或凭据可能不对）" >&2 - rm -f "$OUT" - exit 1 -fi - -SIZE="$(du -h "$OUT" | cut -f1)" -echo "备份完成：$(basename "$OUT") ($SIZE)" - -# 记录 binlog 位点：增量备份（backup_incremental.sh）与恢复重放的对齐锚点—— -# 从「此位点之后」的 binlog 开始重放即为完整增量链 -docker exec "$CONTAINER" sh -c \ - 'mysql -uroot -p"$MYSQL_ROOT_PASSWORD" -N -e "SHOW MASTER STATUS"' > "${OUT}.pos" 2>/dev/null -if [[ -s "${OUT}.pos" ]]; then - echo "binlog 位点：$(awk '{print $1, "pos="$2}' "${OUT}.pos") → $(basename "$OUT").pos" -fi - -echo "永久留存：未自动清理任何备份" - -echo "" -echo "最新备份：" -ls -lh "$BACKUP_DIR"/*.sql.gz 2>/dev/null | tail -3 +exec "${PYTHON:-python}" "$ROOT/scripts/payroll_backup.py" full "$@" diff --git a/scripts/backup_incremental.sh b/scripts/backup_incremental.sh index 088ab87..c24e5b5 100644 --- a/scripts/backup_incremental.sh +++ b/scripts/backup_incremental.sh @@ -1,89 +1,5 @@ #!/usr/bin/env bash -# 衍水工资系统 —— MySQL 增量备份（binlog 归档） -# -# 口径（澄清 #19 业务答复 2026-10-19）：每日全量 + 有更新时增量补充备份。 -# 增量 = 归档 MySQL binlog：自上次归档之后产生的新 binlog 文件逐个拷出压缩； -# 没有新变更时不产出文件（「有更新时」才补充）。 -# -# 恢复顺序：最新全量（backup_full.sh 产物）→ 依次重放其后的 binlog 增量包 -# （用 mysqlbinlog 重放，见 docs/备份恢复手册.md 增量章节）。 -# -# 实现说明：mysql:8.0 官方镜像不含 mysqlbinlog 客户端，故不走 -# --read-from-remote-server，直接从数据目录 tar 拷出原始 binlog 文件； -# 只归档「已写完」的文件（SHOW BINARY LOGS 列表最后一个为当前写入中， -# 留到下次收取，配合 30 天 binlog 过期窗口余量充足）。 -# -# 用法：在项目根目录执行 bash scripts/backup_incremental.sh -# 产物：db/backups/binlog_NNNNNN.gz（NNNNNN 为 binlog 序号），永久保留，不自动清理。 - +# 统一到跨平台备份工具；成功以完整性清单及退出状态为准。 set -euo pipefail - ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -cd "$ROOT" - -if [[ -f docker/.env ]]; then - # shellcheck disable=SC1091 - source docker/.env -else - echo "ERROR: docker/.env 不存在（请先创建 docker/.env 并填入 MySQL 凭据）" >&2 - exit 1 -fi -: "${MYSQL_DATABASE:?docker/.env 缺 MYSQL_DATABASE}" - -BACKUP_DIR="$ROOT/db/backups" -STATE_FILE="$BACKUP_DIR/.incremental_state" -mkdir -p "$BACKUP_DIR" - -# 容器名固定（compose 显式 container_name），与 docker/docker-compose.dev.yml 保持一致 -CONTAINER="ys-mysql-dev" -if ! docker inspect "$CONTAINER" >/dev/null 2>&1; then - echo "ERROR: 容器 $CONTAINER 未运行（先 docker compose up -d）" >&2 - exit 1 -fi - -# 列出 binlog：文件名 序号 大小（mysql -N 无表头，直接出数据行） -mapfile -t BINLOGS < <( - docker exec "$CONTAINER" sh -c \ - 'mysql -uroot -p"$MYSQL_ROOT_PASSWORD" -N -e "SHOW BINARY LOGS"' 2>/dev/null \ - | awk '{print $1}' -) -if [[ ${#BINLOGS[@]} -eq 0 ]]; then - echo "ERROR: SHOW BINARY LOGS 无输出（binlog 未开启或凭据不对）" >&2 - exit 1 -fi -if [[ ${#BINLOGS[@]} -eq 1 ]]; then - echo "仅有写入中的 binlog（${BINLOGS[0]}），无已写完文件可归档" - exit 0 -fi -COMPLETED=("${BINLOGS[@]:0:${#BINLOGS[@]}-1}") # 去掉写入中的最后一个 - -LAST="$(cat "$STATE_FILE" 2>/dev/null || echo "")" -NEW=() -for f in "${COMPLETED[@]}"; do - # 与状态文件比较序号：无状态（首次）全收；有状态只收更大的 - if [[ -z "$LAST" ]] || (( ${f##*.} > ${LAST##*.} )); then - NEW+=("$f") - fi -done - -if [[ ${#NEW[@]} -eq 0 ]]; then - echo "无新增变更（上次归档至 ${LAST}，无需补充增量）" - exit 0 -fi - -STAMP="$(date +%Y%m%d_%H%M%S)" -echo "归档 ${#NEW[@]} 个 binlog（${NEW[0]} ~ ${NEW[-1]}）" -for f in "${NEW[@]}"; do - OUT="$BACKUP_DIR/${f}.gz" - docker exec "$CONTAINER" sh -c "tar cf - -C /var/lib/mysql '$f'" | gzip > "$OUT" - if [[ ! -s "$OUT" ]]; then - echo "ERROR: $OUT 为空" >&2 - rm -f "$OUT" - exit 1 - fi - echo " $(basename "$OUT") ($(du -h "$OUT" | cut -f1))" -done -echo "$STAMP ${NEW[-1]}" > "$STATE_FILE" - -# 归档文件永久留存；服务器原始 binlog 过期与归档留存是不同配置，部署前验证归档链完整。 -echo "增量归档完成，最新至 ${NEW[-1]}" +exec "${PYTHON:-python}" "$ROOT/scripts/payroll_backup.py" incremental "$@" diff --git a/scripts/health_check.sh b/scripts/health_check.sh index 029a305..18df930 100644 --- a/scripts/health_check.sh +++ b/scripts/health_check.sh @@ -1,106 +1,5 @@ #!/usr/bin/env bash -# 运维健康检查 —— 容器/磁盘/备份新鲜度/登录异常（⑨ 输出一份可读报告） -# -# 用法：在项目根目录执行 bash scripts/health_check.sh -# 每项 PASS/WARN/FAIL，最后给出结论。WARN/FAIL 时退出码非零提示运维介入。 - -set -uo pipefail - +# 统一到跨平台备份工具；成功以完整性清单及退出状态为准。 +set -euo pipefail ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -cd "$ROOT" - -if [[ -f docker/.env ]]; then - # shellcheck disable=SC1091 - source docker/.env -fi - -ISSUES=0 -report() { printf "%-10s %s\n" "$1" "$2"; } -warn() { report "WARN" "$1"; ISSUES=$((ISSUES+1)); } -fail() { report "FAIL" "$1"; ISSUES=$((ISSUES+1)); } -pass() { report "PASS" "$1"; } - -echo "======== 衍水工资系统 健康检查 $(date '+%Y-%m-%d %H:%M:%S') ========" - -# 1. 容器健康 -for C in ys-mysql-dev ys-redis-dev; do - if ! docker inspect "$C" >/dev/null 2>&1; then - fail "容器 $C 不存在" - continue - fi - STATUS="$(docker inspect -f '{{.State.Health.Status}}' "$C" 2>/dev/null || echo none)" - RUNNING="$(docker inspect -f '{{.State.Running}}' "$C" 2>/dev/null)" - if [[ "$RUNNING" == "true" && "$STATUS" == "healthy" ]]; then - pass "容器 $C 运行中且健康" - elif [[ "$RUNNING" == "true" ]]; then - warn "容器 $C 运行中但健康检查=$STATUS" - else - fail "容器 $C 未运行" - fi -done - -# 2. 磁盘水位 -DISK_PCT="$(df -h "$ROOT" | awk 'NR==2 {gsub(/%/,"",$5); print $5}')" -if [[ -z "$DISK_PCT" ]]; then - warn "磁盘信息读取失败" -elif (( DISK_PCT < 80 )); then - pass "磁盘使用率 ${DISK_PCT}%（阈值 80%）" -elif (( DISK_PCT < 90 )); then - warn "磁盘使用率 ${DISK_PCT}%（建议清理）" -else - fail "磁盘使用率 ${DISK_PCT}%（高危，立即清理）" -fi - -# 3. 备份新鲜度 -BACKUP_DIR="$ROOT/db/backups" -LATEST="$(ls -t "$BACKUP_DIR"/payroll_*.sql.gz 2>/dev/null | head -1)" -if [[ -z "$LATEST" ]]; then - warn "未找到任何备份（应每日备份）" -else - AGE_H=$(( ($(date +%s) - $(stat -c %Y "$LATEST")) / 3600 )) - if (( AGE_H < 24 )); then - pass "最新备份 $(basename "$LATEST")，${AGE_H} 小时前" - elif (( AGE_H < 48 )); then - warn "最新备份已 ${AGE_H} 小时前（超过 24 小时）" - else - fail "最新备份已 ${AGE_H} 小时前（备份任务可能已停）" - fi -fi - -# 4. Redis 存活（密码经容器内环境变量展开，不出现在宿主进程命令行； -# 写法同 backup_full.sh 的 MYSQL_ROOT_PASSWORD） -if [[ -n "${REDIS_PASSWORD:-}" ]]; then - if docker exec ys-redis-dev sh -c 'redis-cli --no-auth-warning -a "$REDIS_PASSWORD" ping' 2>/dev/null | grep -q PONG; then - pass "Redis 响应 PONG" - else - warn "Redis ping 无响应（检查 REDIS_PASSWORD 或容器状态）" - fi -else - warn "docker/.env 缺 REDIS_PASSWORD，跳过 Redis 探活" -fi - -# 5. 登录日志异常（24h 失败次数） -if [[ -n "${MYSQL_ROOT_PASSWORD:-}" ]] && docker inspect ys-mysql-dev >/dev/null 2>&1; then - FAILS=$(docker exec ys-mysql-dev sh -c \ - "mysql -uroot -p\"\$MYSQL_ROOT_PASSWORD\" ${MYSQL_DATABASE:-payroll} -N -B -e \" - SELECT COUNT(*) FROM login_logs - WHERE success = 0 AND create_time > NOW() - INTERVAL 24 HOUR;\" 2>/dev/null" || echo "?") - if [[ "$FAILS" == "?" ]]; then - warn "登录日志查询失败" - elif (( FAILS < 10 )); then - pass "24h 登录失败 ${FAILS} 次（阈值 10）" - else - warn "24h 登录失败 ${FAILS} 次（可能有暴力破解，建议核查）" - fi -else - warn "跳过登录日志检查（无凭据或容器未运行）" -fi - -echo "==============================================================" -if (( ISSUES == 0 )); then - echo "结论：全部通过，系统健康。" - exit 0 -else - echo "结论：发现 $ISSUES 项需关注（见上）。" - exit 1 -fi +exec "${PYTHON:-python}" "$ROOT/scripts/payroll_backup.py" check "$@" diff --git a/scripts/payroll_backup.py b/scripts/payroll_backup.py new file mode 100644 index 0000000..000e0d6 --- /dev/null +++ b/scripts/payroll_backup.py @@ -0,0 +1,489 @@ +"""跨平台 MySQL 备份：一致位点、binlog 完整链、独立容器恢复、只读检查。""" + +import argparse +import gzip +import hashlib +import json +import os +import re +import secrets +import shutil +import subprocess +import sys +import time +from datetime import UTC, datetime +from itertools import pairwise +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +IDENT = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,63}$") +LOG = re.compile(r"^binlog\.\d{6,}$") + + +class BackupError(RuntimeError): + pass + + +def run(args, *, data=None, env=None): + result = subprocess.run(args, input=data, capture_output=True, env=env, check=False) + if result.returncode: + # 原始数据库错误可能含业务内容；只报告操作及退出状态。 + raise BackupError(f"{args[0]} 操作失败，退出码 {result.returncode}") + return result.stdout + + +def mysql(container, sql, database=None): + args = [ + "docker", + "exec", + "-i", + container, + "sh", + "-c", + # 官方镜像初始化期间会启动禁用 TCP 的临时服务，不能把它当作就绪实例。 + 'export MYSQL_PWD="$MYSQL_ROOT_PASSWORD"; ' + 'exec mysql -uroot --host=127.0.0.1 --protocol=TCP -N -B "$@"', + "mysql", + ] + if database: + args.append(database) + return run(args, data=sql.encode()).decode("utf-8").strip() + + +def json_write(path, value): + temporary = path.with_suffix(path.suffix + ".partial") + temporary.write_text( + json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8" + ) + temporary.replace(path) + + +def checksum(path): + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def stream_to(args, source): + """按块恢复大库和日志，避免将全部数据加载进内存。""" + with subprocess.Popen( + args, + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) as process: + shutil.copyfileobj(source, process.stdin, 1024 * 1024) + process.stdin.close() + if process.wait() != 0: + raise BackupError("恢复数据流执行失败") + + +def logs(container): + rows = [r.split("\t")[0] for r in mysql(container, "SHOW BINARY LOGS").splitlines()] + if not rows or any(not LOG.fullmatch(r) for r in rows): + raise BackupError("binlog 未开启或文件名不受支持") + return rows + + +def position(container): + parts = mysql(container, "SHOW MASTER STATUS").split("\t") + if len(parts) < 2 or not LOG.fullmatch(parts[0]): + raise BackupError("无法获得 binlog 位点") + return {"file": parts[0], "position": int(parts[1])} + + +def stream_gzip(args, out): + temporary = out.with_suffix(out.suffix + ".partial") + with subprocess.Popen( + args, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL + ) as process: + with gzip.open(temporary, "wb") as dest: + shutil.copyfileobj(process.stdout, dest, 1024 * 1024) + if process.wait() != 0: + raise BackupError("备份流失败，未发布成功清单，临时文件保留供核对") + temporary.replace(out) + return {"file": out.name, "sha256": checksum(out)} + + +def full(args): + stamp = ( + datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + + "_" + + secrets.token_hex(3) + ) + out = args.directory / f"full_{stamp}.sql.gz" + server = mysql(args.container, "SELECT @@server_uuid") + binary = logs(args.container) # 关闭 binlog 时不生成无法续接的全量。 + command = [ + "docker", + "exec", + args.container, + "sh", + "-c", + ( + 'export MYSQL_PWD="$MYSQL_ROOT_PASSWORD"; exec mysqldump -uroot ' + "--single-transaction --source-data=2 --set-gtid-purged=OFF " + '--routines --events --triggers --hex-blob --no-tablespaces "$1"' + ), + "backup", + args.database, + ] + entry = stream_gzip(command, out) + point, completed, tables = None, False, [] + with gzip.open(out, "rt", encoding="utf-8") as stream: + for line in stream: + match = re.search( + r"(?:SOURCE|MASTER)_LOG_FILE='([^']+)', (?:SOURCE|MASTER)_LOG_POS=(\d+)", + line, + ) + if match: + point = {"file": match[1], "position": int(match[2])} + table = re.match(r"CREATE TABLE `([^`]+)`", line) + if table: + tables.append(table[1]) + completed = completed or line.startswith("-- Dump completed on ") + if not completed or not point or not tables or not LOG.fullmatch(point["file"]): + raise BackupError("全量缺少完成标识、业务表或一致快照位点，不能作为恢复基线") + manifest = { + "kind": "FULL", + "database": args.database, + "server_uuid": server, + "created_at": datetime.now(UTC).isoformat(), + "image": args.image, + "point": point, + "entries": [entry], + "tables": sorted(tables), + "logs_seen": binary, + } + name = f"full_{stamp}.json" + json_write(args.directory / name, manifest) + json_write( + args.directory / "state.json", + { + "full": name, + "point": point, + "increments": [], + "checked_at": datetime.now(UTC).isoformat(), + }, + ) + return {"status": "SUCCEEDED", "manifest": name, "point": point} + + +def load_manifest(directory, name): + if Path(name).name != name: + raise BackupError("备份清单必须是备份目录内的文件名") + value = json.loads((directory / name).read_text(encoding="utf-8")) + for entry in value["entries"]: + filename = entry["file"] + if ( + Path(filename).name != filename + or checksum(directory / filename) != entry["sha256"] + ): + raise BackupError("备份文件缺失或 SHA-256 不符") + return value + + +def incremental(args): + state_path = args.directory / "state.json" + if not state_path.exists(): + raise BackupError("请先完成全量备份") + state = json.loads(state_path.read_text(encoding="utf-8")) + baseline = load_manifest(args.directory, state["full"]) + if baseline["database"] != args.database or baseline["server_uuid"] != mysql( + args.container, "SELECT @@server_uuid" + ): + raise BackupError("数据库或服务器身份已变化，必须先建立新的全量基线") + current = position(args.container) + if state["point"] == current: + state["checked_at"] = datetime.now(UTC).isoformat() + json_write(state_path, state) + return {"status": "UNCHANGED", "point": current} + available = logs(args.container) + if state["point"]["file"] not in available: + raise BackupError("增量链起点已被清理，必须重新全量备份") + mysql(args.container, "FLUSH BINARY LOGS") + end = position(args.container) + available = logs(args.container) + names = available[ + available.index(state["point"]["file"]) : available.index(end["file"]) + 1 + ] + numbers = [int(n.split(".")[1]) for n in names] + if any(b != a + 1 for a, b in pairwise(numbers)): + raise BackupError("binlog 序号存在缺口，拒绝发布增量成功") + stamp = secrets.token_hex(8) + entries = [] + for name in names: + # 轮转后读取位点前仍可能有事务写入新文件；末文件也要归档到该位点， + # 否则下一次从 end 续接时会永久跳过这些事务。固定字节边界避免重复重放。 + command = ["docker", "exec", args.container] + if name == end["file"]: + command.extend(["head", "-c", str(end["position"]), f"/var/lib/mysql/{name}"]) + else: + command.extend(["cat", f"/var/lib/mysql/{name}"]) + entry = stream_gzip( + command, + args.directory / f"{stamp}_{name}.gz", + ) + entry["binlog"] = name + entries.append(entry) + manifest_name = f"increment_{stamp}.json" + manifest = { + "kind": "INCREMENTAL", + "full": state["full"], + "start": state["point"], + "end": end, + "entries": entries, + "created_at": datetime.now(UTC).isoformat(), + } + json_write(args.directory / manifest_name, manifest) + state["increments"].append(manifest_name) + state["point"] = end + state["checked_at"] = datetime.now(UTC).isoformat() + json_write(state_path, state) + return {"status": "SUCCEEDED", "manifest": manifest_name, "files": len(entries)} + + +def checked_chain(args): + state = json.loads((args.directory / "state.json").read_text(encoding="utf-8")) + baseline = load_manifest(args.directory, state["full"]) + point, increments = baseline["point"], [] + for name in state["increments"]: + change = load_manifest(args.directory, name) + if change["full"] != state["full"] or change["start"] != point: + raise BackupError("增量清单不连续或跨全量基线") + increments.append(change) + point = change["end"] + if point != state["point"]: + raise BackupError("增量末位点与状态不一致") + return baseline, increments + + +def restore(args): + """只创建随机命名、独立可写层的临时 MySQL；不接触已有数据卷。""" + baseline, increments = checked_chain(args) + target = "payroll-drill-" + secrets.token_hex(8) + env = os.environ | {"MYSQL_ROOT_PASSWORD": secrets.token_urlsafe(32)} + volume = target + "-data" + run(["docker", "volume", "create", "--label", "payroll.restore-drill=true", volume]) + run( + [ + "docker", + "run", + "-d", + "--name", + target, + "--label", + "payroll.restore-drill=true", + "--mount", + f"type=volume,source={volume},target=/var/lib/mysql", + "--env", + "MYSQL_ROOT_PASSWORD", + args.image, + ], + env=env, + ) + try: + for _ in range(300): + try: + mysql(target, "SELECT 1") + break + except BackupError: + time.sleep(2) + else: + raise BackupError("隔离恢复容器未就绪") + database = baseline["database"] + if not IDENT.fullmatch(database): + raise BackupError("备份数据库名称无效") + mysql(target, f"CREATE DATABASE `{database}` CHARACTER SET utf8mb4") + command = [ + "docker", + "exec", + "-i", + target, + "sh", + "-c", + 'export MYSQL_PWD="$MYSQL_ROOT_PASSWORD"; exec mysql -uroot "$1"', + "restore", + database, + ] + with subprocess.Popen( + command, + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) as process: + with gzip.open( + args.directory / baseline["entries"][0]["file"], "rb" + ) as source: + shutil.copyfileobj(source, process.stdin, 1024 * 1024) + process.stdin.close() + if process.wait() != 0: + raise BackupError("隔离全量恢复失败") + for change in increments: + paths = [] + for entry in change["entries"]: + name = entry["binlog"] + if not LOG.fullmatch(name): + raise BackupError("增量文件名无效") + with gzip.open(args.directory / entry["file"], "rb") as source: + stream_to( + [ + "docker", + "exec", + "-i", + target, + "sh", + "-c", + 'cat > "/tmp/$1"', + "archive", + name, + ], + source, + ) + paths.append(f"/tmp/{name}") + if paths: + # mysqlbinlog 必须在同一次调用中按顺序消费多个文件，保留临时表上下文。 + decoder_command = [ + "docker", + "exec", + target, + "mysqlbinlog", + "--disable-log-bin", + f"--database={database}", + f"--start-position={change['start']['position']}", + *paths, + ] + with subprocess.Popen( + decoder_command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL + ) as decoder: + with subprocess.Popen( + command, + stdin=decoder.stdout, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) as consumer: + decoder.stdout.close() + consumed = consumer.wait() + if decoder.wait() or consumed: + raise BackupError("增量重放失败，不能将恢复标为成功") + tables = mysql(target, "SHOW TABLES", database).splitlines() + if not increments and sorted(tables) != baseline["tables"]: + raise BackupError("恢复后表清单不符") + counts = { + table: int(mysql(target, f"SELECT COUNT(*) FROM `{table}`", database)) + for table in tables + if IDENT.fullmatch(table) + } + result = { + "status": "SUCCEEDED", + "container": target, + "database": database, + "increments": len(increments), + "table_counts": counts, + "scope": "独立容器；生产库未覆盖", + } + json_write(args.directory / f"{target}.json", result) + return result + finally: + # 仅停止本次随机创建且带标签的容器，保留恢复内容供复核，不批量删除历史演练。 + run(["docker", "stop", target]) + + +def check(args): + baseline, increments = checked_chain(args) + if baseline["database"] != args.database: + raise BackupError("备份数据库与检查目标不符") + age = ( + datetime.now(UTC) - datetime.fromisoformat(baseline["created_at"]) + ).total_seconds() + if age > 26 * 3600: + raise BackupError("最新全量备份超过 26 小时") + uuid = mysql(args.container, "SELECT @@server_uuid") + if uuid != baseline["server_uuid"]: + raise BackupError("运行服务器与备份身份不符") + if mysql(args.container, "SELECT @@innodb_force_recovery") != "0": + raise BackupError("数据库仍处于强制恢复模式") + state = json.loads((args.directory / "state.json").read_text(encoding="utf-8")) + checked_at = state.get("checked_at", baseline["created_at"]) + if ( + datetime.now(UTC) - datetime.fromisoformat(checked_at) + ).total_seconds() > 30 * 60: + raise BackupError("增量采集检查超过 30 分钟未运行") + redis = getattr(args, "redis_container", None) + if redis: + pong = run( + [ + "docker", + "exec", + redis, + "sh", + "-c", + 'REDISCLI_AUTH="$REDIS_PASSWORD" redis-cli --no-auth-warning PING', + ] + ) + if pong.strip() != b"PONG": + raise BackupError("Redis 探活失败") + stat = shutil.disk_usage(args.directory) + if stat.free / stat.total < 0.1: + raise BackupError("备份盘可用空间低于 10%") + return { + "status": "SUCCEEDED", + "full_age_hours": round(age / 3600, 2), + "increment_count": len(increments), + "free_bytes": stat.free, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "action", choices=["full", "incremental", "check", "restore-drill"] + ) + parser.add_argument("--container", default="ys-mysql-dev") + parser.add_argument("--database", default="payroll") + parser.add_argument("--directory", type=Path, default=ROOT / "db/backups") + parser.add_argument("--image", default="payroll-mysql-backup:8.0.46") + parser.add_argument("--redis-container", default="ys-redis-dev") + args = parser.parse_args() + if not IDENT.fullmatch(args.database) or not re.fullmatch( + r"[A-Za-z0-9_.-]+", args.container + ): + parser.error("容器名或数据库名无效") + args.directory.mkdir(parents=True, exist_ok=True) + lock = args.directory / ".backup.lock" + try: + with lock.open("x", encoding="utf-8") as stream: + stream.write(str(os.getpid())) + except FileExistsError: + print("备份目录已有活动锁，请核实上次进程状态", file=sys.stderr) + return 1 + try: + result = { + "full": full, + "incremental": incremental, + "check": check, + "restore-drill": restore, + }[args.action](args) + json_write( + args.directory / f"last-{args.action}-run.json", + {"finished_at": datetime.now(UTC).isoformat(), **result}, + ) + print(json.dumps(result, ensure_ascii=False)) + return 0 + except (BackupError, OSError, ValueError, KeyError) as error: + json_write( + args.directory / "last-failure.json", + { + "finished_at": datetime.now(UTC).isoformat(), + "action": args.action, + "status": "FAILED", + "reason": str(error), + }, + ) + print(f"操作失败：{error}", file=sys.stderr) + return 1 + finally: + lock.unlink() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/restore_drill.sh b/scripts/restore_drill.sh index 51da158..37eb67c 100644 --- a/scripts/restore_drill.sh +++ b/scripts/restore_drill.sh @@ -1,116 +1,5 @@ #!/usr/bin/env bash -# 恢复演练 —— 把最近一份备份恢复到「临时库」并核对行数（⑨：备份不演练 = 没有备份） -# -# 演练绝不直接覆盖生产库；临时库名 payroll_drill_<时间戳>，核对完由 -# 调用方决定是否删除（本脚本默认不 DROP，安全优先）。 -# -# 用法：在项目根目录执行 bash scripts/restore_drill.sh -# 产物：临时库 + 终端行数对比报告；--drop-exec 参数在核对通过后删除临时库。 - +# 统一到跨平台备份工具；成功以完整性清单及退出状态为准。 set -euo pipefail - ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -cd "$ROOT" - -if [[ -f docker/.env ]]; then - # shellcheck disable=SC1091 - source docker/.env -else - echo "ERROR: docker/.env 不存在" >&2 - exit 1 -fi - -: "${MYSQL_DATABASE:?docker/.env 缺 MYSQL_DATABASE}" -: "${MYSQL_ROOT_PASSWORD:?docker/.env 缺 MYSQL_ROOT_PASSWORD}" - -CONTAINER="ys-mysql-dev" -if ! docker inspect "$CONTAINER" >/dev/null 2>&1; then - echo "ERROR: 容器 $CONTAINER 未运行" >&2 - exit 1 -fi - -BACKUP_DIR="$ROOT/db/backups" -LATEST="$(ls -t "$BACKUP_DIR"/payroll_*.sql.gz 2>/dev/null | head -1)" -if [[ -z "$LATEST" ]]; then - echo "ERROR: 未找到备份文件（先 bash scripts/backup_full.sh）" >&2 - exit 1 -fi - -DRILL_DB="payroll_drill_$(date +%Y%m%d_%H%M%S)" - -# 0. --drop-exec：演练前先清掉**全部**历史遗留 drill 库（含上一轮「保留供 -# 核对」的那份——否则每演练一轮净增一份整库拷贝，磁盘被无人认领的 -# drill 库吃满），演练完成后本轮临时库也随之删除 -if [[ "${1:-}" == "--drop-exec" ]]; then - OLD_DBS=$(docker exec "$CONTAINER" sh -c \ - "mysql -uroot -p\"\$MYSQL_ROOT_PASSWORD\" -N -B -e \"SHOW DATABASES LIKE 'payroll_drill_%';\"" 2>/dev/null || true) - for OLD in $OLD_DBS; do - docker exec "$CONTAINER" sh -c \ - "mysqladmin -uroot -p\"\$MYSQL_ROOT_PASSWORD\" -f drop \"$OLD\"" - echo "已清理历史演练库 $OLD" - done -fi - -echo "演练恢复：$(basename "$LATEST") → 临时库 $DRILL_DB" - -# 1. 创建临时库（mysqladmin create 无反引号转义问题） -docker exec "$CONTAINER" sh -c \ - "mysqladmin -uroot -p\"\$MYSQL_ROOT_PASSWORD\" create \"$DRILL_DB\"" - -# 2. 解压导入 -gunzip -c "$LATEST" | docker exec -i "$CONTAINER" sh -c \ - "mysql -uroot -p\"\$MYSQL_ROOT_PASSWORD\" \"$DRILL_DB\"" - -# 3. 行数核对：**全部业务表**（新增表必须同步补入本清单——备份漏表时 -# 演练要能发现；alembic_version 不列，行数恒 1 无核对价值） -TABLES=( - adjustment_orders approval_orders approval_steps business_items business_roles - business_secondary_allocations business_status_logs businesses customer_rating_grades - customers login_logs monthly_settlements operation_logs ownership_change_logs - param_change_batches param_definitions param_values performance_ledgers permissions - personnel personnel_plans personnel_regions products receipts role_permissions roles - settlement_items shipment_items shipments system_settings user_roles users -) - -echo "" -echo "======== 行数对比（源库 vs 演练库）========" -printf "%-24s %12s %12s %6s\n" "表名" "源库" "演练库" "结果" -ALL_OK=1 -for T in "${TABLES[@]}"; do - SRC=$(docker exec "$CONTAINER" sh -c \ - "mysql -uroot -p\"\$MYSQL_ROOT_PASSWORD\" -N -B -e \"SELECT COUNT(*) FROM $MYSQL_DATABASE.$T;\" 2>/dev/null" || echo "?") - DST=$(docker exec "$CONTAINER" sh -c \ - "mysql -uroot -p\"\$MYSQL_ROOT_PASSWORD\" -N -B -e \"SELECT COUNT(*) FROM $DRILL_DB.$T;\" 2>/dev/null" || echo "?") - if [[ "$SRC" == "?" || "$DST" == "?" ]]; then - # 查询失败（表缺失/权限/客户端故障）不可与「数据一致」混淆：? - # 与 ? 相等会误报可恢复——备份真的漏表时演练必须红 - MARK="✗" - ALL_OK=0 - elif [[ "$SRC" == "$DST" ]]; then - MARK="✓" - else - MARK="✗" - ALL_OK=0 - fi - printf "%-24s %12s %12s %6s\n" "$T" "$SRC" "$DST" "$MARK" -done -echo "==========================================" - -if [[ "$ALL_OK" == "1" ]]; then - echo "" - echo "结果：全部表行数一致，备份可恢复。" - if [[ "${1:-}" == "--drop-exec" ]]; then - docker exec "$CONTAINER" sh -c \ - "mysqladmin -uroot -p\"\$MYSQL_ROOT_PASSWORD\" -f drop \"$DRILL_DB\"" - echo "本轮临时库 $DRILL_DB 已删除（历史遗留亦在演练前清理）。" - else - echo "临时库 $DRILL_DB 保留（核对用）；确认无误后重跑本脚本加 --drop-exec，" - echo "会先清掉包括它在内的全部演练库再演练，本轮完成后同样即删。" - fi - exit 0 -else - echo "" - echo "结果：存在不一致，请检查备份文件与数据库结构。" >&2 - echo "临时库 $DRILL_DB 保留供排查。" >&2 - exit 1 -fi +exec "${PYTHON:-python}" "$ROOT/scripts/payroll_backup.py" restore-drill "$@" diff --git a/scripts/windows/Install-PayrollBackupTasks.ps1 b/scripts/windows/Install-PayrollBackupTasks.ps1 new file mode 100644 index 0000000..dc6494b --- /dev/null +++ b/scripts/windows/Install-PayrollBackupTasks.ps1 @@ -0,0 +1,51 @@ +[CmdletBinding(SupportsShouldProcess = $true)] +param( + [Parameter(Mandatory = $true)][string]$PythonPath, + [Parameter(Mandatory = $true)][string]$BackupDirectory, + [string]$ProjectRoot, + [string]$Container = 'ys-mysql-dev', + [string]$Database = 'payroll', + [switch]$Apply +) +$ErrorActionPreference = 'Stop' +if (-not $ProjectRoot) { + $ProjectRoot = Split-Path (Split-Path $PSScriptRoot -Parent) -Parent +} +$project = [IO.Path]::GetFullPath($ProjectRoot) +$python = [IO.Path]::GetFullPath($PythonPath) +$backups = [IO.Path]::GetFullPath($BackupDirectory) +$script = Join-Path $project 'scripts\payroll_backup.py' +if (-not (Test-Path -LiteralPath $python -PathType Leaf) -or -not (Test-Path -LiteralPath $script -PathType Leaf)) { + throw 'Python 或备份程序不存在。' +} +if ($Container -notmatch '^[A-Za-z0-9_.-]+$' -or $Database -notmatch '^[A-Za-z][A-Za-z0-9_]{0,63}$') { throw '容器或数据库标识无效。' } +if (@($python, $script, $backups) | Where-Object { $_.Contains('"') }) { throw '路径不能含双引号。' } +$taskPath = '\PayrollManagementSystem-Backup\' +$common = ' --directory "' + $backups + '" --container ' + $Container + ' --database ' + $Database +$definitions = @( + @{ Name = 'DailyFull'; Action = 'full'; Time = '02:30' }, + @{ Name = 'ChangedIncremental'; Action = 'incremental'; Time = '00:10' }, + @{ Name = 'BackupCheck'; Action = 'check'; Time = '07:30' } +) +foreach ($definition in $definitions) { + $arguments = '"' + $script + '" ' + $definition.Action + $common + [pscustomobject]@{ TaskPath = $taskPath; Name = $definition.Name; Python = $python; Arguments = $arguments; Enabled = [bool]$Apply } + if (-not $Apply) { continue } + $existing = Get-ScheduledTask -TaskPath $taskPath -TaskName $definition.Name -ErrorAction SilentlyContinue + if ($existing -and (@($existing.Actions).Count -ne 1 -or $existing.Actions[0].Execute -ne $python -or $existing.Actions[0].Arguments -ne $arguments)) { + throw '同名任务已有不同动作，拒绝覆盖。' + } + if ($PSCmdlet.ShouldProcess($definition.Name, '注册当前用户本地开发备份任务')) { + $scheduler = New-Object -ComObject 'Schedule.Service' + $scheduler.Connect() + try { $null = $scheduler.GetFolder($taskPath) } + catch { $null = $scheduler.GetFolder('\').CreateFolder('PayrollManagementSystem-Backup') } + $action = New-ScheduledTaskAction -Execute $python -Argument $arguments -WorkingDirectory $project + $trigger = if ($definition.Action -eq 'incremental') { + New-ScheduledTaskTrigger -Once -At ((Get-Date).Date.AddMinutes(10)) -RepetitionInterval (New-TimeSpan -Minutes 15) + } else { New-ScheduledTaskTrigger -Daily -At $definition.Time } + $principal = New-ScheduledTaskPrincipal -UserId ([Security.Principal.WindowsIdentity]::GetCurrent().Name) -LogonType Interactive -RunLevel Limited + $settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -MultipleInstances IgnoreNew -RestartCount 2 -RestartInterval (New-TimeSpan -Minutes 5) + Register-ScheduledTask -TaskPath $taskPath -TaskName $definition.Name -Action $action -Trigger $trigger -Principal $principal -Settings $settings -Force | Out-Null + } +}