MySQL备份恢复方案（支持全量 + 增量 + PITR + dry-run + 报告生成）

书接上文：上回说到《企业级MySQL数据库逻辑备份方案》，接下来怎么能没有企业级备份恢复方案呢？

目录结构：

/mysql_restore/
├── scripts/
│   ├── restore_single.sh          # 核心恢复脚本
│   ├── restore_jobs.txt           # 实例配置
│   └── backup_binlog.sh           # 模拟备份脚本（测试用）
├── web/
│   ├── web_console.py             # Web 控制台
│   └── templates/
│       └── index.html             # 前端页面
├── logs/                          # 日志目录
├── reports/                       # 报告目录
└── binlogs/                       # binlog 存储

以上代码目录中提到的web版控制台，因我没有实测，就不贴出来了。

接下来是核心恢复脚本：

#!/bin/bash

# =============================================================================
# 文件：restore_all.sh
# 名称：MySQL PITR 恢复脚本
# 功能：支持全量 + 增量 + PITR + dry-run + 报告生成
# 作者：Bear
# 日期：2025-09-03
# =============================================================================

#set -euo pipefail

# 将 /usr/local/bin、/usr/bin、/data/mysql/mysql84/bin 添加到 PATH 里面
export PATH="/usr/local/bin:/usr/bin:/data/mysql/mysql84/bin:$PATH"

# ============ 配置 ============
LOG_DIR="/mysql_restore/logs"
REPORT_DIR="/mysql_restore/reports"
BINLOG_DIR="/mysql_restore/binlogs"

PROMETHEUS_PORT=9091

# ============ 全局变量 ============
TARGET_DATABASE=""
TARGET_TABLES=""
INSTANCE=""
USE_PROXY=""
TARGET_HOST=""
TARGET_PORT=""
TARGET_USER=""
TARGET_PASS=""
BACKUP_DIR=""
BINLOG_SOURCE_DIR=""
TARGET_RECOVERY_TIME=""
SINCE_POSITION=""
TAIL_MINUTES=""
DRY_RUN=0
DRY_RUN_REPORT=0
AUTO_BINLOG_POS=""
LOG_FILE=""
START_TIME=0
PITR=0

# ============ 工具函数 ============
info() { echo -e "[$(date '+%Y-%m-%d %H:%M:%S')] \033[1;32mINFO\033[0m $1"; }
warn() { echo -e "[$(date '+%Y-%m-%d %H:%M:%S')] \033[1;33mWARN\033[0m $1"; }
error() { echo -e "[$(date '+%Y-%m-%d %H:%M:%S')] \033[1;31mERROR\033[0m $1"; exit 1; }
dry_log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] DRY-RUN: $1" >> "$LOG_FILE"; }

exec_cmd() {
  local cmd="$1"
  local desc="$2"

  if [[ $DRY_RUN_REPORT -eq 1 ]]; then
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] [DRY-REPORT] $desc" >> "$LOG_FILE"
    return 0
  elif [[ $DRY_RUN -eq 1 ]]; then
    dry_log "$desc => $cmd"
  else
    info "$desc"
    eval "$cmd"
  fi
}

# 提取 binlog.position
extract_binlog_position() {
  info " 提取全备时的 binlog 位置..."
  local pos_file="$BACKUP_DIR/binlog.position"
  if [[ -f "$pos_file" ]]; then
    #AUTO_BINLOG_POS=$(cat "$pos_file" | tr -d ' \n' | sed 's/^"//;s/"$//')
    AUTO_BINLOG_POS=$(cat "$pos_file")
    if [[ -n "$AUTO_BINLOG_POS" && "$AUTO_BINLOG_POS" != ":" ]]; then
      info " 从 $pos_file 提取 pos: $AUTO_BINLOG_POS"
      return 0
    fi
  fi

  if [[ -f "$BACKUP_DIR/metadata" ]]; then
    local meta_pos=$(grep -E "Binlog position|Log position" "$BACKUP_DIR/metadata" | head -1 | sed -E 's/.*: //;s/^"//;s/"$//' | tr -d ' ')
    if [[ "$meta_pos" =~ ^[a-zA-Z0-9._-]+:[0-9]+$ ]]; then
      AUTO_BINLOG_POS="$meta_pos"
      info " 从 metadata 提取 pos: $AUTO_BINLOG_POS"
      return 0
    fi
  fi

  warn " 未找到 binlog position（$pos_file 不存在或格式错误）"
  return 1
}

# 全量恢复
do_full_restore() {
  info " 开始全量恢复: $BACKUP_DIR"

  # 记录恢复成功的目录
  SUCCESS_COUNT=0
  FAILED_DIRS=""

  # 遍历 /backup 下的每个子目录
  for DIR in "$BACKUP_DIR"/*/; do
	# 提取目录名（用于显示）
	BASE_DIR=$(basename "$DIR")
	echo "检查目录: $BASE_DIR"

	# 判断是否为目录且非空
	if [ -d "$DIR" ] && [ -n "$(ls -A "$DIR")" ]; then
		echo "正在恢复: $BASE_DIR"

#Usage:
#  myloader [OPTION?] multi-threaded MySQL loader
#
#Help Options:
#  -?, --help                        Show help options
#
#Application Options:
#  -d, --directory                   Directory of the dump to import
#  -q, --queries-per-transaction     Number of queries per transaction, default 1000
#  -o, --overwrite-tables            Drop tables if they already exist
#  -B, --database                    An alternative database to restore into
#  -s, --source-db                   Database to restore
#  -e, --enable-binlog               Enable binary logging of the restore data
#  -h, --host                        The host to connect to
#  -u, --user                        Username with the necessary privileges
#  -p, --password                    User password
#  -a, --ask-password                Prompt For User password
#  -P, --port                        TCP/IP port to connect to
#  -S, --socket                      UNIX domain socket file to use for connection
#  -t, --threads                     Number of threads to use, default 4
#  -C, --compress-protocol           Use compression on the MySQL connection
#  -V, --version                     Show the program version and exit
#  -v, --verbose                     Verbosity of output, 0 = silent, 1 = errors, 2 = warnings, 3 = info, default 2
#  --defaults-file                   Use a specific defaults file
#  --ssl                             Connect using SSL
#  --key                             The path name to the key file
#  --cert                            The path name to the certificate file
#  --ca                              The path name to the certificate authority file
#  --capath                          The path name to a directory that contains trusted SSL CA certificates in PEM format
#  --cipher                          A list of permissible ciphers to use for SSL encryption

		local myloader_cmd="myloader --directory='$DIR' --host='$TARGET_HOST' --port='$TARGET_PORT' --user='$TARGET_USER' --password='$TARGET_PASS' --threads=4 -C --overwrite-tables"

		#  添加数据库/表过滤
		if [[ -n "$TARGET_DATABASE" ]]; then
			local dbs=$(echo "$TARGET_DATABASE" | sed 's/,/ /g')
			myloader_cmd="$myloader_cmd --source-db $dbs"
		fi

		if [[ -n "$TARGET_TABLES" ]]; then
			myloader_cmd="$myloader_cmd --tables-list '$TARGET_TABLES'"
		fi

		if [[ "$USE_PROXY" == "proxy" ]]; then
			myloader_cmd="proxychains4 -q $myloader_cmd"
		fi

		exec_cmd "$myloader_cmd" "全量恢复 (过滤: db=$TARGET_DATABASE, tables=$TARGET_TABLES)"

		if [ $? -eq 0 ]; then
			echo "成功恢复: $BASE_DIR"
			((SUCCESS_COUNT++))
		else
			echo "恢复失败: $BASE_DIR"
			FAILED_DIRS="$FAILED_DIRS $BASE_DIR"
		fi
		echo "----------------------------------------"
	else
		echo "跳过空目录或非目录: $BASE_DIR"
	fi
  done
  
  # 输出最终结果
  echo "恢复完成！"
  echo "成功恢复: $SUCCESS_COUNT 个数据库"
  if [ -n "$FAILED_DIRS" ]; then
	echo "失败列表: $FAILED_DIRS"
  else
	echo "所有备份库恢复成功！"
  fi
  echo "结束时间: $(date)"

  info " 全量恢复完成"
  
}
  
# 安全提取编号
extract_num() {
    local f=$(basename "$1")
    printf "$((10#${f##*.}))"	 
}

# PITR 时间点恢复（从全量备份的binlog位置开始）
do_incremental() {
  #echo "do_incremental"
  local start_pos="${SINCE_POSITION:-$AUTO_BINLOG_POS}"
  [[ -z "$start_pos" ]] && info "  无增量源（--since 或 binlog.position），跳过增量恢复" && return 0

  #local binlog_file=$(echo "$start_pos" | cut -d: -f1)
  #local start_pos_val=$(echo "$start_pos" | cut -d: -f2)
  local binlog_file=$(echo "$start_pos" | awk '{print $1}')
  local start_pos_val=$(echo "$start_pos" | awk '{print $2}')


  if ! [[ "$start_pos_val" =~ ^[0-9]+$ ]]; then
    error " binlog 位置格式错误: $start_pos"
  fi

  #  使用传入的 --binlog-dir，否则 fallback 到全局 BINLOG_DIR
  local search_binlog_dir="${BINLOG_SOURCE_DIR:-$BINLOG_DIR}"
  local instance_dir="${TARGET_HOST}_${TARGET_PORT}"
  local binlog_base_dir="$search_binlog_dir/$instance_dir"

  if [[ ! -d "$binlog_base_dir" ]]; then
    #error " binlog 目录不存在: $binlog_base_dir"
        binlog_base_dir=$search_binlog_dir
  fi

  info " 增量恢复: 从 $binlog_file:$start_pos_val 开始应用 binlog"
  info " 搜索路径: $binlog_base_dir"

  local binlogs=($(find "$binlog_base_dir" -name "mysql-bin.*" | sort))

  local found_start=0
  local applied=0
  local bins=""
  local apply_cmd="mysqlbinlog"
  if [[ -n "$TARGET_DATABASE" ]]; then
	local db_filter=$(echo "$TARGET_DATABASE" | cut -d',' -f1)  # 只支持一个库过滤
	apply_cmd="$apply_cmd --database='$db_filter' --start-position=$start_pos_val --stop-datetime='$TARGET_RECOVERY_TIME'"
  else
	apply_cmd="$apply_cmd  --start-position=$start_pos_val --stop-datetime='$TARGET_RECOVERY_TIME'"
  fi
		
  target_num=$(extract_num "$binlog_file")
  applied_bins=()

  for bin in "${binlogs[@]}"; do	
	num=$(extract_num "$bin")
	echo "num:$num,target_num:$target_num"
	if (( num < target_num )); then
		continue; 
	fi
	
	if (( num >= target_num )); then
		applied_bins+=("$bin")
		applied=1
    fi
	
	if (( num == target_num )); then
		found_start=1
    fi
  done
  

  # 确保按顺序
  IFS=$'\n' applied_bins=($(sort <<<"${applied_bins[*]}"))
  unset IFS

  
  if [[ $found_start -eq 0 ]]; then
    error " 起始 binlog 文件未找到: $binlog_file"
	return 0
  fi
  
  if [[ $applied -eq 0 ]]; then
    warn " 无后续 binlog 可应用"
	return 0
  fi
  
  apply_cmd="$apply_cmd ${applied_bins[@]} | "
  
  if [[ "$USE_PROXY" == "proxy" ]]; then
    apply_cmd="proxychains4 -q $apply_cmd"
  fi
  
  apply_cmd="$apply_cmd mysql --force -h'$TARGET_HOST' -P'$TARGET_PORT' -u'$TARGET_USER' -p'$TARGET_PASS'"

  echo $apply_cmd
  exec_cmd "$apply_cmd" "增量: 应用 $filename"
  
  info " PITR增量恢复完成"
}


# 生成报告
generate_report() {
  local status="$1"
  local start_ts="$2"
  local end_ts="$3"
  local duration=$((end_ts - start_ts))
  local size=$(du -sh "$BACKUP_DIR" 2>/dev/null | awk '{print $1}' || echo "N/A")

  local report_file="$REPORT_DIR/restore_${INSTANCE}_$(date +%Y%m%d_%H%M%S).md"
  cat > "$report_file" << EOF
# MySQL 恢复报告

- **实例**: $INSTANCE
- **目标**: $TARGET_HOST:$TARGET_PORT
- **全备目录**: $BACKUP_DIR
- **状态**: $status
- **开始时间**: $(date -d "@$start_ts" "+%Y-%m-%d %H:%M:%S")
- **结束时间**: $(date -d "@$end_ts" "+%Y-%m-%d %H:%M:%S")
- **耗时**: ${duration}s
- **数据大小**: $size

## 执行模式
EOF

  if [[ $DRY_RUN_REPORT -eq 1 ]]; then
    echo "- 模式: \`--dry-run-report\` (只生成报告，不执行)" >> "$report_file"
    echo "- 用途: 审批、审计、恢复预演" >> "$report_file"
  elif [[ $DRY_RUN -eq 1 ]]; then
    echo "- 模式: \`--dry-run\` (模拟执行)" >> "$report_file"
  else
    echo "- 模式: 实际执行" >> "$report_file"
  fi

  echo -e "\n## 恢复摘要\n" >> "$report_file"
  echo " 全量恢复: 完成" >> "$report_file"
  [[ -n "$SINCE_POSITION" || -n "$AUTO_BINLOG_POS" ]] && echo " 增量恢复: 从 $(echo "$SINCE_POSITION|$AUTO_BINLOG_POS" | tr '|' ' ')" >> "$report_file"
  [[ -n "$TARGET_RECOVERY_TIME" ]] && echo " PITR: 恢复到 $TARGET_RECOVERY_TIME" >> "$report_file"

  echo -e "\n---\nGenerated by mysql_restore" >> "$report_file"
  info "报告已生成: $report_file"
}

# Prometheus 指标
start_prometheus_server() {
  local status="$1"
  local duration="$2"
  local size_kb="$3"

  cat > /tmp/mysql_restore.prom << EOF
# HELP mysql_restore_status 恢复状态 (1=成功, 0=失败, 2=报告模式)
# TYPE mysql_restore_status gauge
mysql_restore_status{instance="$INSTANCE",status="$status"} $( [[ "$status" == "SUCCESS" ]] && echo 1 || echo 0 )

# HELP mysql_restore_duration_seconds 恢复耗时
# TYPE mysql_restore_duration_seconds gauge
mysql_restore_duration_seconds{instance="$INSTANCE"} $duration

# HELP mysql_restore_data_size_kb 数据大小
# TYPE mysql_restore_data_size_kb gauge
mysql_restore_data_size_kb{instance="$INSTANCE"} $size_kb
EOF

  # 简单 HTTP 服务器（Python）
  nohup python3 -m http.server "$PROMETHEUS_PORT" --directory /tmp > /dev/null 2>&1 &
  info " Prometheus 指标已暴露: http://localhost:$PROMETHEUS_PORT/metrics"
}

# 主函数
main() {
  local args=()
  for arg in "$@"; do
    case "$arg" in
      --dry-run)
        DRY_RUN=1
        ;;
      --dry-run-report)
        DRY_RUN_REPORT=1
        DRY_RUN=1
        ;;
      --tail-minutes=*)
        TAIL_MINUTES="${arg#--tail-minutes=}"
        [[ ! "$TAIL_MINUTES" =~ ^[0-9]+$ ]] && error " --tail-minutes 必须是数字"
        ;;
      --since=*)
        SINCE_POSITION="${arg#--since=}"
        ;;
      --binlog-dir=*)
        BINLOG_SOURCE_DIR="${arg#--binlog-dir=}"
        info "BINLOG_SOURCE_DIR:$BINLOG_SOURCE_DIR"
        [[ ! -d "$BINLOG_SOURCE_DIR" ]] && error " binlog目录不存在: $BINLOG_SOURCE_DIR"
        ;;
      --database=*)
        TARGET_DATABASE="${arg#--database=}"
        ;;
      --tables=*)
        TARGET_TABLES="${arg#--tables=}"
        ;;
	  --pitr-only)
        PITR=2 #2-仅运行PITR按时间增量
        ;;
	  --pitr)
        PITR=1 #0：默认仅全量恢复；1：全量恢复之后，运行按时间增量
        ;;
      *)
        args+=("$arg")
        ;;
    esac
  done


  if [[ ${#args[@]} -lt 7 ]]; then
    error "用法: $0 <实例> <proxy|direct> <host> <port> <user> <pass> <backup_dir> [pitr_time] [--since=...] [--tail-minutes=N] [--dry-run|--dry-run-report]"
  fi

  INSTANCE="${args[0]}"
  USE_PROXY="${args[1]}"
  TARGET_HOST="${args[2]}"
  TARGET_PORT="${args[3]}"
  TARGET_USER="${args[4]}"
  TARGET_PASS="${args[5]}"
  BACKUP_DIR="${args[6]}"

  if [[ ${#args[@]} -ge 8 ]]; then
    TARGET_RECOVERY_TIME="${args[7]}"
    info "恢复时间点: $TARGET_RECOVERY_TIME"
  fi

  # --tail-minutes 优先级高于手动时间
  if [[ -n "$TAIL_MINUTES" ]]; then
    local now=$(date +%s)
    local target_ts=$((now - TAIL_MINUTES * 60))
    TARGET_RECOVERY_TIME=$(date -d "@$target_ts" "+%Y-%m-%d %H:%M:%S")
    info "计算恢复时间点: $TARGET_RECOVERY_TIME (前 $TAIL_MINUTES 分钟)"
  fi

  LOG_FILE="$LOG_DIR/restore_${INSTANCE}_$(date +%Y%m%d_%H%M%S).log"
  mkdir -p "$LOG_DIR" "$REPORT_DIR"
  START_TIME=$(date +%s)

  # 日志头
  {
    echo "=== MySQL 恢复任务 ==="
    echo "实例: $INSTANCE"
    echo "模式: $USE_PROXY"
    echo "目标: $TARGET_HOST:$TARGET_PORT"
    echo "用户: $TARGET_USER"
    echo "备份目录: $BACKUP_DIR"
    echo "PITR时间: $TARGET_RECOVERY_TIME"
    echo "执行模式: $( [[ $DRY_RUN_REPORT -eq 1 ]] && echo REPORT_ONLY || [[ $DRY_RUN -eq 1 ]] && echo DRY_RUN || echo EXECUTE )"
    echo "开始时间: $(date)"
    echo "========================"
  } >> "$LOG_FILE"

  info " 恢复模式: $( [[ $DRY_RUN_REPORT -eq 1 ]] && echo 'DRY-RUN-REPORT (只生成报告)' || [[ $DRY_RUN -eq 1 ]] && echo 'DRY-RUN (模拟执行)' || echo 'EXECUTE (真实执行)' )"

  # 提取 binlog 位置
  if [[ -z "$SINCE_POSITION" ]]; then
    extract_binlog_position || true
  fi

  echo "PITR:$PITR"

  # 执行流程
  if [[ $PITR -eq 1 ]]; then #全量之后再增量
    do_full_restore
	do_incremental
  elif [[ $PITR -eq 2 ]]; then #仅运行增量
    do_incremental
  else
	do_full_restore
  fi
  

  local end_time=$(date +%s)
  finalize_report "SUCCESS" "$end_time"
}

finalize_report() {
  local status="$1"
  local end_time="$2"
  generate_report "$status" "$START_TIME" "$end_time"
  start_prometheus_server "$status" "$((end_time - START_TIME))" "0"
}

# 入口
if [[ "${BASH_SOURCE[0]}" == "${0}" ]]; then
  main "$@"
fi

调用示例：

sudo /bin/bash /mysql_restore/scripts/restore_all.sh prod-*** proxy 172.16.100.*** 3306 backup_user ****** /data/mydumper_backup/backups/mysql-***-headless.***-data.svc.cluster.local_3306/mysql-***-headless.**-***-data.svc.cluster.local_3306_20250903_162833 "2025-09-05 16:55:00" --binlog-dir=/data/mysql_binlog_backup/mysql-***-headless.******-data.svc.cluster.local_3306  --tail-minutes=15  --pitr-only --dry-run-report