一套脚本搞定 Elasticsearch 跨集群数据搬迁，无需共享存储

文章目录

• • 一、方案概述
  • 二、前置准备
  • • 2.1 目标单节点 ES 配置
    • 2.2 创建快照目录
    • 2.3 网络连通性检查
    • 2.4 单节点环境建议
  • 三、手动操作步骤
  • • 步骤 1：从集群拉取数据（reindex）
    • 步骤 2：校验文档数
    • 步骤 3：注册快照仓库
    • 步骤 4：创建快照
    • 步骤 5：打包快照文件
  • 四、辅助脚本
  • • 4.1 Reindex 进度监控脚本
    • 4.2 等待索引健康变绿脚本
    • 4.3 完整迁移脚本（一键执行）
    • 4.4 快照恢复脚本（目标端）
  • 五、注意事项
  • 六、故障排查

一、方案概述

本手册描述将数据从现有 ES 集群迁移到单节点 ES，并通过快照仓库导出备份文件的完整流程。整体链路如下：

集群 → reindex remote → 单节点 ES → fs 快照仓库 → 快照导出

该方案绕开了"集群无共享存储导致快照失败"的问题——在单节点 ES 上使用本地 fs 仓库创建快照，无需 NFS/SMB 共享存储。

流程总览

二、前置准备

2.1 目标单节点 ES 配置

在目标单节点 ES 的 elasticsearch.yml 中追加以下配置：

reindex.remote.whitelist: ["源集群IP:9200"]
path.repo: ["/data/es_backup"]

修改后重启单节点 ES。

2.2 创建快照目录

mkdir -p /data/es_backup
chown -R elasticsearch:elasticsearch /data/es_backup

2.3 网络连通性检查

从单节点 ES 所在机器执行：

curl http://源集群IP:9200

如果源集群开了 TLS，使用 https；如果开了账号密码，后续 reindex 时带上 username、password。

2.4 单节点环境建议

单节点 ES 的 number_of_replicas 建议设为 0，避免索引健康状态变黄：

curl -X PUT "http://localhost:9200/_all/_settings" \\
-H 'Content-Type: application/json' \\
-d '{"index":{"number_of_replicas":0}}'

三、手动操作步骤

步骤 1：从集群拉取数据（reindex）

在单节点 ES 上执行：

curl -X POST http://localhost:9200/_reindex?wait_for_completion=false \\
-H 'Content-Type: application/json' \\
-d '{
"source": {
"remote": {
"host": "http://源集群IP:9200",
"username": "elastic",
"password": "你的密码",
"socket_timeout": "5m"
},
"index": "原索引名",
"size": 1000
},
"dest": {
"index": "原索引名",
"op_type": "create"
}
}'

如果源集群没开账号密码，去掉 username、password 即可。返回结果中包含 task 字段，记录该 task_id 用于后续进度查询。

步骤 2：校验文档数

reindex 完成后，对比源集群和目标单节点的文档数：

# 源集群
curl http://源集群IP:9200/_cat/indices/索引名?v

# 目标单节点
curl http://localhost:9200/_cat/indices/索引名?v

两边 docs.count 一致再进入下一步。

步骤 3：注册快照仓库

curl -X PUT http://localhost:9200/_snapshot/es_repo \\
-H 'Content-Type: application/json' \\
-d '{
"type": "fs",
"settings": {
"location": "/data/es_backup",
"compress": true
}
}'

单节点环境下，path.repo 配本地目录即可，不需要共享存储。

步骤 4：创建快照

curl -X PUT "http://localhost:9200/_snapshot/es_repo/snap1?wait_for_completion=true" \\
-H 'Content-Type: application/json' \\
-d '{
"indices": "需要导出的索引",
"include_global_state": false
}'

返回 "state": "SUCCESS" 表示快照完成。

步骤 5：打包快照文件

cd /data
tar czf es_backup_$(date +%Y%m%d).tar.gz es_backup

之后即可将该包拷贝到目标环境用于恢复。

四、辅助脚本

4.1 Reindex 进度监控脚本

文件名：monitor_reindex.sh

用途：轮询查询 reindex 任务进度，自动检测任务完成并退出。

#!/bin/bash

# ============ 配置区 ============
ES_HOST="http://localhost:9200"
# 如果 ES 开了账号密码，取消下面两行注释并填写
# ES_USER="elastic"
# ES_PASS="your_password"

# 如果指定了 task_id，直接查该任务；否则查所有正在运行的 reindex 任务
TASK_ID="${1:-}"
# =================================

AUTH=""
if [ -n "$ES_USER" ] && [ -n "$ES_PASS" ]; then
AUTH="-u ${ES_USER}:${ES_PASS}"
fi

echo "========================================"
echo " ES Reindex 进度监控"
echo " 目标: $ES_HOST"
echo "========================================"

while true; do
echo ""
echo "[$(date '+%Y-%m-%d %H:%M:%S')] 查询中…"

if [ -n "$TASK_ID" ]; then
# 查询指定任务
RESULT=$(curl -s ${AUTH} "${ES_HOST}/_tasks/${TASK_ID}?pretty")
else
# 查询所有 reindex 任务
RESULT=$(curl -s ${AUTH} "${ES_HOST}/_tasks?actions=*reindex&pretty")
fi

echo "$RESULT"

# 判断是否还有任务在运行
RUNNING=$(echo "$RESULT" | grep -c '"node"')

if [ "$RUNNING" -eq 0 ]; then
echo ""
echo "========================================"
echo " 所有 reindex 任务已完成！"
echo "========================================"
break
fi

# 提取进度信息（如果有）
echo "$RESULT" | grep -E '"total"|"updated"|"created"|"batches"'

echo ""
echo "30秒后刷新… (Ctrl+C 退出)"
sleep 30
done

用法：

• 不指定 task_id，自动监控所有 reindex 任务：./monitor_reindex.sh
• 指定 task_id，监控某个具体任务：./monitor_reindex.sh <task_id>

task_id 是 reindex 接口返回的值，格式类似 xxx:yyy。

脚本行为：

• 每 30 秒刷新一次进度
• 显示 total、created、updated、batches 等字段
• 当没有 reindex 任务在运行时，自动退出并提示完成
• 支持带账号密码的 ES，取消注释并填写即可

4.2 等待索引健康变绿脚本

文件名：wait_green.sh

用途：轮询检查索引健康状态，单节点环境下自动将未分配副本数设为 0，等待状态变绿。

#!/bin/bash

# ============ 配置区 ============
ES_HOST="http://localhost:9200"
# 如果 ES 开了账号密码，取消下面两行注释并填写
# ES_USER="elastic"
# ES_PASS="your_password"

# 要监控的索引，多个用逗号分隔；留空则监控所有索引
INDEXES="${1:-}"

# 最大等待时间（秒），0 表示不超时
MAX_WAIT=0

# 是否自动把副本数设为 0（单节点推荐开启）
AUTO_FIX_REPLICAS=true
# =================================

AUTH=""
if [ -n "$ES_USER" ] && [ -n "$ES_PASS" ]; then
AUTH="-u ${ES_USER}:${ES_PASS}"
fi

START_TIME=$(date +%s)

echo "========================================"
echo " ES 索引健康检查"
echo " 目标: $ES_HOST"
echo " 索引: ${INDEXES:-全部}"
echo " 自动修复副本: $AUTO_FIX_REPLICAS"
echo "========================================"

FIXED_INDICES=""

while true; do
# 检查超时
if [ "$MAX_WAIT" -gt 0 ]; then
NOW=$(date +%s)
ELAPSED=$((NOW – START_TIME))
if [ "$ELAPSED" -ge "$MAX_WAIT" ]; then
echo ""
echo "========================================"
echo " 等待超时（${MAX_WAIT}秒），退出"
echo "========================================"
exit 1
fi
fi

echo ""
echo "[$(date '+%Y-%m-%d %H:%M:%S')] 检查中…"

# 获取集群健康状态
if [ -n "$INDEXES" ]; then
HEALTH=$(curl -s ${AUTH} "${ES_HOST}/_cluster/health/${INDEXES}?pretty")
else
HEALTH=$(curl -s ${AUTH} "${ES_HOST}/_cluster/health?pretty")
fi

STATUS=$(echo "$HEALTH" | grep '"status"' | head -1 | awk -F'"' '{print $4}')

echo "$HEALTH" | grep -E '"status"|"number_of_nodes"|"active_shards"|"relocating_shards"|"initializing_shards"|"unassigned_shards"'

if [ "$STATUS" = "green" ]; then
echo ""
echo "========================================"
echo " 索引健康状态已变绿！"
echo "========================================"
exit 0
fi

# 单节点环境下，如果是 yellow 且开启了自动修复
if [ "$STATUS" = "yellow" ] && [ "$AUTO_FIX_REPLICAS" = true ]; then
# 获取未分配分片的索引列表
UNASSIGNED=$(curl -s ${AUTH} "${ES_HOST}/_cat/shards/${INDEXES}" 2>/dev/null | grep UNASSIGNED | awk '{print $1}' | sort -u)

for idx in $UNASSIGNED; do
# 跳过系统索引
if [[ "$idx" == .* ]]; then
continue
fi

# 检查是否已经修复过
if echo "$FIXED_INDICES" | grep -q "$idx"; then
continue
fi

echo ""
echo " 检测到索引 [$idx] 有未分配副本，自动设为 0…"
curl -s ${AUTH} -X PUT "${ES_HOST}/${idx}/_settings" \\
-H 'Content-Type: application/json' \\
-d '{"index": {"number_of_replicas": 0}}' | grep -o '"acknowledged":[a-z]*'

FIXED_INDICES="${FIXED_INDICES} ${idx}"
done
fi

echo ""
echo "10秒后刷新… (Ctrl+C 退出)"
sleep 10
done

用法：

• 监控所有索引：./wait_green.sh
• 监控指定索引：./wait_green.sh "index1,index2"

脚本行为：

• 每 10 秒检查一次集群健康状态
• 显示 status、active_shards、unassigned_shards 等关键指标
• 单节点环境下，如果状态是 yellow，会自动把有未分配副本的索引副本数设为 0
• 状态变绿后自动退出，返回码 0
• 支持设置最大等待时间，超时退出返回码 1
• 不想自动修复副本时，将 AUTO_FIX_REPLICAS=true 改为 false

4.3 完整迁移脚本（一键执行）

文件名：migrate_and_snapshot.sh

用途：将 reindex、进度监控、文档数校验、等待变绿、注册仓库、创建快照全部串联，一键执行完成完整迁移流程。

#!/bin/bash

# ============================================================
# ES 集群数据迁移到单节点 + 快照导出 一键脚本
# ============================================================

# ============ 配置区 ============

# 如果ES不用账号密码，保留空字符串即可
# 源集群地址
SRC_HOST="http://源集群IP:9200"
SRC_USER="elastic"
SRC_PASS="你的密码"

# 目标单节点地址
DST_HOST="http://localhost:9200"
DST_USER=""
DST_PASS=""

# 要迁移的索引，多个用逗号分隔
INDEXES="index1,index2,index3"

# 快照仓库路径（需与 elasticsearch.yml 中 path.repo 一致）
SNAPSHOT_PATH="/data/es_backup"
SNAPSHOT_REPO="es_repo"
SNAPSHOT_NAME="snap_$(date +%Y%m%d_%H%M%S)"

# 轮询间隔（秒）
POLL_INTERVAL=10

# 最大等待时间（秒），0 表示不超时
MAX_WAIT=0

# ================================

# 颜色输出
RED='\\033[0;31m'
GREEN='\\033[0;32m'
YELLOW='\\033[1;33m'
NC='\\033[0m'

log_info() { echo -e "${GREEN}[INFO]${NC} $(date '+%H:%M:%S') $1"; }
log_warn() { echo -e "${YELLOW}[WARN]${NC} $(date '+%H:%M:%S') $1"; }
log_error() { echo -e "${RED}[ERROR]${NC} $(date '+%H:%M:%S') $1"; }

# 构造 curl 认证参数
build_auth() {
local user="$1" pass="$2"
if [ -n "$user" ] && [ -n "$pass" ]; then
echo "-u ${user}:${pass}"
fi
}

SRC_AUTH=$(build_auth "$SRC_USER" "$SRC_PASS")
DST_AUTH=$(build_auth "$DST_USER" "$DST_PASS")

# 检查 ES 是否可达
check_es() {
local host="$1" auth="$2" label="$3"
if curl -s ${auth} "${host}/" > /dev/null 2>&1; then
log_info "${label} 可达"
return 0
else
log_error "${label} 不可达: ${host}"
return 1
fi
}

# 获取集群文档数（索引不存在时返回空）
get_doc_count() {
local host="$1" auth="$2" idx="$3"
local resp
resp=$(curl -s ${auth} "${host}/_cat/indices/${idx}?h=docs.count")
if echo "$resp" | grep -q '"index_not_found_exception"'; then
return 0
fi
echo "$resp" | tr -d '[:space:]'
}

# ============================================================
# 主流程
# ============================================================

echo "============================================================"
echo " ES 数据迁移 + 快照导出"
echo " 源集群: ${SRC_HOST}"
echo " 目标节点: ${DST_HOST}"
echo " 索引: ${INDEXES}"
echo " 快照名称: ${SNAPSHOT_NAME}"
echo "============================================================"
echo ""

# ———- 1. 检查连通性 ———-
log_info "步骤1: 检查 ES 连通性"
check_es "$SRC_HOST" "$SRC_AUTH" "源集群" || exit 1
check_es "$DST_HOST" "$DST_AUTH" "目标节点" || exit 1
echo ""

# 过滤源集群中不存在的索引（部分缺失时跳过缺失的，其余继续）
log_info "步骤1.5: 索引存在性检查"
WORK_INDEXES=""
for idx in $(echo "$INDEXES" | tr ',' ' '); do
if [ -n "$(get_doc_count "$SRC_HOST" "$SRC_AUTH" "$idx")" ]; then
WORK_INDEXES="${WORK_INDEXES:+${WORK_INDEXES},}${idx}"
else
log_warn " 索引 [${idx}] 在源集群中不存在，跳过该索引的迁移"
fi
done
if [ -z "$WORK_INDEXES" ]; then
log_error "配置的索引在源集群中均不存在，中止迁移"
exit 1
fi
if [ "$WORK_INDEXES" != "$INDEXES" ]; then
log_warn "实际迁移索引: ${WORK_INDEXES}"
fi
INDEXES="$WORK_INDEXES"
echo ""

# ———- 2. 执行 reindex ———-
log_info "步骤2: 开始 reindex"
TASK_IDS=""

for idx in $(echo "$INDEXES" | tr ',' ' '); do
log_info " 迁移索引: ${idx}"

SRC_COUNT=$(get_doc_count "$SRC_HOST" "$SRC_AUTH" "$idx")
log_info " 源集群文档数: ${SRC_COUNT}"

RESP=$(curl -s -X POST "${DST_HOST}/_reindex?wait_for_completion=false" \\
${DST_AUTH} \\
-H 'Content-Type: application/json' \\
-d "{
\\"source\\": {
\\"remote\\": {
\\"host\\": \\"${SRC_HOST}\\",
\\"username\\": \\"${SRC_USER}\\",
\\"password\\": \\"${SRC_PASS}\\",
\\"socket_timeout\\": \\"5m\\"
},
\\"index\\": \\"${idx}\\",
\\"size\\": 1000
},
\\"dest\\": {
\\"index\\": \\"${idx}\\",
\\"op_type\\": \\"create\\"
}
}")

TASK_ID=$(echo "$RESP" | grep -o '"task":"[^"]*"' | awk -F'"' '{print $4}')

if [ -n "$TASK_ID" ]; then
log_info " 任务ID: ${TASK_ID}"
TASK_IDS="${TASK_IDS} ${TASK_ID}"
else
log_error " reindex 提交失败: ${RESP}"
exit 1
fi
done

echo ""

# ———- 3. 轮询进度 ———-
log_info "步骤3: 监控 reindex 进度"

START_TIME=$(date +%s)
ALL_DONE=false

while [ "$ALL_DONE" = false ]; do
# 超时检查
if [ "$MAX_WAIT" -gt 0 ]; then
NOW=$(date +%s)
ELAPSED=$((NOW – START_TIME))
if [ "$ELAPSED" -ge "$MAX_WAIT" ]; then
log_error "等待超时（${MAX_WAIT}秒），退出"
exit 1
fi
fi

ALL_DONE=true

for tid in $TASK_IDS; do
RESULT=$(curl -s ${DST_AUTH} "${DST_HOST}/_tasks/${tid}")
COMPLETE=$(echo "$RESULT" | grep -c '"completed":true')

if [ "$COMPLETE" -eq 0 ]; then
ALL_DONE=false
# 提取进度
TOTAL=$(echo "$RESULT" | grep -o '"total":[0-9]*' | head -1)
CREATED=$(echo "$RESULT" | grep -o '"created":[0-9]*' | head -1)
echo -ne "\\r 任务 ${tid}: ${TOTAL}, ${CREATED} … "
fi
done

if [ "$ALL_DONE" = true ]; then
echo ""
log_info "所有 reindex 任务已完成"
break
fi

sleep "$POLL_INTERVAL"
done

echo ""

# ———- 3.5 补建空索引（源索引无文档时 reindex 不会在目标端建索引） ———-
log_info "步骤3.5: 检查并补建目标端索引"

for idx in $(echo "$INDEXES" | tr ',' ' '); do
if [ -z "$(get_doc_count "$DST_HOST" "$DST_AUTH" "$idx")" ]; then
CREATE_RESP=$(curl -s -X PUT "${DST_HOST}/${idx}")
log_info " 创建空索引 [${idx}]: $(echo "$CREATE_RESP" | grep -o '"acknowledged":[a-z]*')"
fi
done

echo ""

# ———- 4. 校验文档数 ———-
log_info "步骤4: 校验文档数"
MISMATCH=false

for idx in $(echo "$INDEXES" | tr ',' ' '); do
SRC_COUNT=$(get_doc_count "$SRC_HOST" "$SRC_AUTH" "$idx")
DST_COUNT=$(get_doc_count "$DST_HOST" "$DST_AUTH" "$idx")

if [ -z "$SRC_COUNT" ] || [ -z "$DST_COUNT" ]; then
log_warn " ${idx}: 源=[${SRC_COUNT:-索引缺失}], 目标=[${DST_COUNT:-索引缺失}] ✗ 索引缺失!"
MISMATCH=true
continue
fi

if [ "$SRC_COUNT" = "$DST_COUNT" ]; then
log_info " ${idx}: 源=${SRC_COUNT}, 目标=${DST_COUNT} ✓"
else
log_warn " ${idx}: 源=${SRC_COUNT}, 目标=${DST_COUNT} ✗ 不一致!"
MISMATCH=true
fi
done

if [ "$MISMATCH" = true ]; then
log_warn "部分索引文档数不一致，请检查"
fi
echo ""

# ———- 5. 等待索引变绿 ———-
log_info "步骤5: 等待索引健康变绿"

# 逐个查询，得到目标节点上实际存在的索引（合并查询时任一索引缺失会导致整次调用 404）
HEALTH_INDEXES=""
for idx in $(echo "$INDEXES" | tr ',' ' '); do
if [ -n "$(get_doc_count "$DST_HOST" "$DST_AUTH" "$idx")" ]; then
HEALTH_INDEXES="${HEALTH_INDEXES:+${HEALTH_INDEXES},}${idx}"
else
log_warn " 索引 [${idx}] 在目标节点不存在，不等待其变绿"
fi
done
if [ -z "$HEALTH_INDEXES" ]; then
log_error "目标节点上不存在任何待检查的索引，请检查 reindex 是否成功"
exit 1
fi

FIXED_INDICES=""
GREEN_OK=false
GREEN_START=$(date +%s)
NOGREEN_ROUNDS=0

while [ "$GREEN_OK" = false ]; do
HEALTH=$(curl -s ${DST_AUTH} "${DST_HOST}/_cluster/health/${HEALTH_INDEXES}")
STATUS=$(echo "$HEALTH" | grep -o '"status":"[^"]*"' | awk -F'"' '{print $4}')

# 拿不到状态（索引不存在/closed/API 报错等）时直接退出，避免无反馈死循环
if [ -z "$STATUS" ]; then
log_error "健康检查接口未返回状态，查询的索引: ${HEALTH_INDEXES}"
log_error "原始响应: [${HEALTH}]"
exit 1
fi

NOW=$(date +%s)
if [ "$STATUS" = "green" ]; then
log_info "索引健康状态已变绿"
GREEN_OK=true
break
fi

NOGREEN_ROUNDS=$((NOGREEN_ROUNDS + 1))
log_warn " 当前状态: ${STATUS}（已等待 $((NOW – GREEN_START))s）"

if [ "$NOGREEN_ROUNDS" -ge 30 ]; then
log_error "等待 ${NOGREEN_ROUNDS} 轮仍未变绿，退出。请手动排查:"
log_error " curl -s ${DST_HOST}/_cat/shards/${HEALTH_INDEXES}"
exit 1
fi

# 单节点自动修复副本（yellow：副本未分配；red：分片未分配，尝试清零副本）
if [ "$STATUS" = "yellow" ] || [ "$STATUS" = "red" ]; then
UNASSIGNED=$(curl -s ${DST_AUTH} "${DST_HOST}/_cat/shards/${HEALTH_INDEXES}" 2>/dev/null \\
| grep UNASSIGNED | awk '{print $1}' | sort -u)

for idx in $UNASSIGNED; do
if echo "$FIXED_INDICES" | grep -q "$idx"; then
continue
fi
log_info " 索引 [${idx}] 副本未分配，自动设为 0"
curl -s ${DST_AUTH} -X PUT "${DST_HOST}/${idx}/_settings" \\
-H 'Content-Type: application/json' \\
-d '{"index":{"number_of_replicas":0}}' > /dev/null
FIXED_INDICES="${FIXED_INDICES} ${idx}"
done
fi

sleep "$POLL_INTERVAL"
done

echo ""

# ———- 6. 创建快照 ———-
log_info "步骤6: 注册快照仓库"

curl -s -X PUT "${DST_HOST}/_snapshot/${SNAPSHOT_REPO}" \\
${DST_AUTH} \\
-H 'Content-Type: application/json' \\
-d "{
\\"type\\": \\"fs\\",
\\"settings\\": {
\\"location\\": \\"${SNAPSHOT_PATH}\\",
\\"compress\\": true
}
}" > /dev/null

log_info "步骤7: 创建快照 ${SNAPSHOT_NAME}"

SNAP_RESP=$(curl -s -X PUT "${DST_HOST}/_snapshot/${SNAPSHOT_REPO}/${SNAPSHOT_NAME}?wait_for_completion=true" \\
${DST_AUTH} \\
-H 'Content-Type: application/json' \\
-d "{
\\"indices\\": \\"${HEALTH_INDEXES}\\",
\\"include_global_state\\": false
}")

SNAP_STATE=$(echo "$SNAP_RESP" | grep -o '"state":"[^"]*"' | awk -F'"' '{print $4}')

if [ "$SNAP_STATE" = "SUCCESS" ]; then
log_info "快照创建成功: ${SNAPSHOT_NAME}"
echo ""
log_info "快照文件位于: ${SNAPSHOT_PATH}"
log_info "可执行打包: tar czf ${SNAPSHOT_NAME}.tar.gz -C ${SNAPSHOT_PATH} ."
else
log_error "快照创建失败: ${SNAP_RESP}"
exit 1
fi

echo ""
echo "============================================================"
echo " 迁移 + 快照导出 完成"
echo "============================================================"

使用前准备：

执行：

./migrate_and_snapshot.sh

脚本会自动完成：提交 reindex 任务 → 轮询进度直到完成 → 对比源/目标文档数 → 自动把副本设为 0 等待变绿 → 注册快照仓库并创建快照。最后提示快照文件位置，直接 tar 打包带走即可。

4.4 快照恢复脚本（目标端）

文件名：restore_snapshot.sh

用途：在目标环境解压快照包、注册仓库、执行恢复、等待变绿并校验文档数，与迁移脚本闭环。

#!/bin/bash

# ============================================================
# ES 快照恢复 一键脚本
# ============================================================

# ============ 配置区 ============

# 目标 ES 地址
ES_HOST="http://localhost:9200"
ES_USER=""
ES_PASS=""

# 快照包路径（迁移脚本打包出来的 tar.gz）
SNAPSHOT_TAR="/data/es_backup.tar.gz"

# 解压后的快照目录（需与 elasticsearch.yml 中 path.repo 一致）
SNAPSHOT_PATH="/data/es_backup"

# 快照仓库名
SNAPSHOT_REPO="es_repo"

# 快照名称（留空则自动选择最新的快照）
SNAPSHOT_NAME=""

# 要恢复的索引，多个用逗号分隔；留空则恢复快照中所有索引
RESTORE_INDEXES=""

# 是否自动把副本数设为 0（单节点推荐开启）
AUTO_FIX_REPLICAS=true

# 轮询间隔（秒）
POLL_INTERVAL=10

# 最大等待时间（秒），0 表示不超时
MAX_WAIT=0

# ================================

RED='\\033[0;31m'
GREEN='\\033[0;32m'
YELLOW='\\033[1;33m'
NC='\\033[0m'

log_info() { echo -e "${GREEN}[INFO]${NC} $(date '+%H:%M:%S') $1"; }
log_warn() { echo -e "${YELLOW}[WARN]${NC} $(date '+%H:%M:%S') $1"; }
log_error() { echo -e "${RED}[ERROR]${NC} $(date '+%H:%M:%S') $1"; }

AUTH=""
if [ -n "$ES_USER" ] && [ -n "$ES_PASS" ]; then
AUTH="-u ${ES_USER}:${ES_PASS}"
fi

# 检查 ES 是否可达
if ! curl -s ${AUTH} "${ES_HOST}/" > /dev/null 2>&1; then
log_error "ES 不可达: ${ES_HOST}"
exit 1
fi

echo "============================================================"
echo " ES 快照恢复"
echo " 目标节点: ${ES_HOST}"
echo " 快照包: ${SNAPSHOT_TAR}"
echo "============================================================"
echo ""

# ———- 1. 解压快照包 ———-
log_info "步骤1: 解压快照包"

if [ ! -f "$SNAPSHOT_TAR" ]; then
log_error "快照包不存在: ${SNAPSHOT_TAR}"
exit 1
fi

mkdir -p "$SNAPSHOT_PATH"
tar xzf "$SNAPSHOT_TAR" -C "$SNAPSHOT_PATH" –strip-components=0 2>/dev/null

# 如果 tar 里自带目录层级，尝试直接解压到 SNAPSHOT_PATH
if [ ! -f "${SNAPSHOT_PATH}/index-0" ] && [ ! -f "${SNAPSHOT_PATH}/meta-*.dat" ]; then
# 尝试去掉一层目录
SUB_DIR=$(find "$SNAPSHOT_PATH" -mindepth 1 -maxdepth 1 -type d | head -1)
if [ -n "$SUB_DIR" ]; then
mv "${SUB_DIR}"/* "$SNAPSHOT_PATH"/ 2>/dev/null
rmdir "$SUB_DIR" 2>/dev/null
fi
fi

# 验证快照目录
if [ ! -f "${SNAPSHOT_PATH}/index-0" ] && [ ! -f "${SNAPSHOT_PATH}/meta-*.dat" ]; then
log_warn "未在 ${SNAPSHOT_PATH} 中找到快照文件，但继续尝试注册仓库"
fi

chown -R elasticsearch:elasticsearch "$SNAPSHOT_PATH" 2>/dev/null
log_info "快照包已解压到: ${SNAPSHOT_PATH}"
echo ""

# ———- 2. 注册快照仓库 ———-
log_info "步骤2: 注册快照仓库"

REPO_RESP=$(curl -s -X PUT "${ES_HOST}/_snapshot/${SNAPSHOT_REPO}" \\
${AUTH} \\
-H 'Content-Type: application/json' \\
-d "{
\\"type\\": \\"fs\\",
\\"settings\\": {
\\"location\\": \\"${SNAPSHOT_PATH}\\",
\\"compress\\": true
}
}")

if ! echo "$REPO_RESP" | grep -q '"acknowledged":true'; then
log_error "仓库注册失败: ${REPO_RESP}"
exit 1
fi
log_info "仓库 [${SNAPSHOT_REPO}] 注册成功"
echo ""

# ———- 3. 确定快照名称 ———-
if [ -z "$SNAPSHOT_NAME" ]; then
log_info "步骤3: 自动查找最新快照"
SNAP_LIST=$(curl -s ${AUTH} "${ES_HOST}/_snapshot/${SNAPSHOT_REPO}/_all")
SNAP_NAME=$(echo "$SNAP_LIST" | grep -o '"snapshot":"[^"]*"' | tail -1 | awk -F'"' '{print $4}')
if [ -z "$SNAP_NAME" ]; then
log_error "未找到可用快照"
exit 1
fi
SNAPSHOT_NAME="$SNAP_NAME"
log_info "使用快照: ${SNAPSHOT_NAME}"
else
log_info "步骤3: 使用指定快照 ${SNAPSHOT_NAME}"
fi
echo ""

# ———- 4. 获取快照中的索引列表 ———-
log_info "步骤4: 查看快照内容"

SNAP_INFO=$(curl -s ${AUTH} "${ES_HOST}/_snapshot/${SNAPSHOT_REPO}/${SNAPSHOT_NAME}")
SNAP_STATE=$(echo "$SNAP_INFO" | grep -o '"state":"[^"]*"' | awk -F'"' '{print $4}')

if [ "$SNAP_STATE" != "SUCCESS" ]; then
log_error "快照状态不是 SUCCESS: ${SNAP_STATE}"
exit 1
fi
log_info "快照状态: SUCCESS"

SNAP_INDICES=$(echo "$SNAP_INFO" | grep -o '"indices":$$[^]]*$$' | head -1)
log_info "快照包含索引: ${SNAP_INDICES}"
echo ""

# ———- 5. 执行恢复 ———-
log_info "步骤5: 开始恢复快照"

# 构建恢复请求体
RESTORE_BODY="{
\\"ignore_unavailable\\": true,
\\"include_global_state\\": false"

if [ -n "$RESTORE_INDEXES" ]; then
RESTORE_BODY="${RESTORE_BODY},
\\"indices\\": \\"${RESTORE_INDEXES}\\""
fi

RESTORE_BODY="${RESTORE_BODY}
}"

RESTORE_RESP=$(curl -s -X POST \\
"${ES_HOST}/_snapshot/${SNAPSHOT_REPO}/${SNAPSHOT_NAME}/_restore?wait_for_completion=false" \\
${AUTH} \\
-H 'Content-Type: application/json' \\
-d "$RESTORE_BODY")

log_info "恢复请求已提交: ${RESTORE_RESP}"
echo ""

# ———- 6. 等待索引变绿 ———-
log_info "步骤6: 等待索引恢复完成并变绿"

TARGET_IDX="${RESTORE_INDEXES}"
if [ -z "$TARGET_IDX" ]; then
# 从快照信息中提取索引名
TARGET_IDX=$(echo "$SNAP_INFO" | grep -o '"indices":$$[^]]*$$' \\
| head -1 | tr -d '[]"' | sed 's/indices://;s/,/ /g')
fi

START_TIME=$(date +%s)
FIXED_INDICES=""

while true; do
# 超时检查
if [ "$MAX_WAIT" -gt 0 ]; then
NOW=$(date +%s)
ELAPSED=$((NOW – START_TIME))
if [ "$ELAPSED" -ge "$MAX_WAIT" ]; then
log_warn "等待超时（${MAX_WAIT}秒）"
break
fi
fi

# 获取健康状态
if [ -n "$TARGET_IDX" ]; then
HEALTH=$(curl -s ${AUTH} "${ES_HOST}/_cluster/health/${TARGET_IDX}")
else
HEALTH=$(curl -s ${AUTH} "${ES_HOST}/_cluster/health")
fi

STATUS=$(echo "$HEALTH" | grep -o '"status":"[^"]*"' | awk -F'"' '{print $4}')

if [ "$STATUS" = "green" ]; then
log_info "索引健康状态已变绿"
break
fi

ACTIVE=$(echo "$HEALTH" | grep -o '"active_shards":[0-9]*' | head -1)
RELOC=$(echo "$HEALTH" | grep -o '"relocating_shards":[0-9]*' | head -1)
INIT=$(echo "$HEALTH" | grep -o '"initializing_shards":[0-9]*' | head -1)
UNASSIGNED=$(echo "$HEALTH" | grep -o '"unassigned_shards":[0-9]*' | head -1)
echo -ne "\\r 状态: ${STATUS} | ${ACTIVE} | ${RELOC} | ${INIT} | ${UNASSIGNED} … "

# 单节点自动修复副本
if [ "$STATUS" = "yellow" ] && [ "$AUTO_FIX_REPLICAS" = true ]; then
UNASSIGNED_IDX=$(curl -s ${AUTH} "${ES_HOST}/_cat/shards" 2>/dev/null \\
| grep UNASSIGNED | awk '{print $1}' | sort -u)

for idx in $UNASSIGNED_IDX; do
if echo "$FIXED_INDICES" | grep -q "$idx"; then
continue
fi
log_info " 索引 [${idx}] 副本未分配，自动设为 0"
curl -s ${AUTH} -X PUT "${ES_HOST}/${idx}/_settings" \\
-H 'Content-Type: application/json' \\
-d '{"index":{"number_of_replicas":0}}' > /dev/null
FIXED_INDICES="${FIXED_INDICES} ${idx}"
done
fi

sleep "$POLL_INTERVAL"
done

echo ""

# ———- 7. 校验文档数 ———-
log_info "步骤7: 校验恢复结果"

# 从快照中获取原始文档数
SNAP_INDICES_LIST=$(echo "$SNAP_INFO" | grep -o '"indices":$$[^]]*$$' \\
| head -1 | tr -d '[]"' | sed 's/indices://;s/[" ]//g' | tr ',' '\\n')

for idx in $SNAP_INDICES_LIST; do
# 从快照信息中获取原始文档数
ORIG_COUNT=$(echo "$SNAP_INFO" | grep -o "\\"${idx}\\"[^}]*" \\
| grep -o '"total":[0-9]*' | head -1 | awk -F: '{print $2}')

# 获取恢复后的文档数
CUR_COUNT=$(curl -s ${AUTH} "${ES_HOST}/_cat/indices/${idx}?h=docs.count" \\
2>/dev/null | tr -d '[:space:]')

if [ -n "$CUR_COUNT" ] && [ -n "$ORIG_COUNT" ]; then
if [ "$CUR_COUNT" = "$ORIG_COUNT" ]; then
log_info " ${idx}: 原始=${ORIG_COUNT}, 恢复=${CUR_COUNT} ✓"
else
log_warn " ${idx}: 原始=${ORIG_COUNT}, 恢复=${CUR_COUNT} ✗"
fi
elif [ -n "$CUR_COUNT" ]; then
log_info " ${idx}: 恢复文档数=${CUR_COUNT}"
else
log_warn " ${idx}: 未找到该索引"
fi
done

echo ""
echo "============================================================"
echo " 快照恢复完成"
echo "============================================================"

使用前准备：

执行：

./restore_snapshot.sh

可选参数说明：

注意事项：

• 恢复时 include_global_state 必须设为 false，因为迁移脚本创建快照时没有包含全局状态，否则恢复会报错。
• 如果目标 ES 中已存在同名且处于 open 状态的索引，恢复会失败。需要先关闭或删除同名索引。
• 快照包解压后，脚本会自动调整目录层级，适配不同打包方式。

五、注意事项

六、故障排查