背景

之前我们生产 ES 集群因为数据分片过大，导致集群重启无法选举，具体可以看这篇文章。当系统分片数据量越来越大，给生产集群造成一定压力，同时也会影响数据检索和查询效率。为了减轻集群压力，缩小集群分片数，减少集群故障，需要考虑数据归档方案，将查询频率低的数据从集群中归档到一个集中区域。

数据归档方案

ES 集群的数据归档有几种方案

1. 基于节点属性，对集群节点进行标签管理。将历史查询频率低的 cold 数据保存在集中的几个节点中，将实时数据放在 warm 节点上，单独对 cold 节点数据进行数据归档。
2. snapshot 数据快照。通过对集群数据打 snapshot 快照，同时结合数据索引生命周期管理 (ILM)，将历史数据从集群中删除，需要查询历史数据时，再将索引数据从快照中恢复。snapshot 快照又有多种存储介质，主要有 GlusterFS、AWS S3、Microsoft Azure、COS等。下面主要分析Snapshot 快照

Snapshot 数据快照

因为 ES 底层核心是基于 lucene 框架，一个分片即是一个 lucene 对象实例。ES snapshot 快照本质是对 lucene 物理文件的拷贝。我们先看下集群数据目录下面有哪些文件

为了避免集群数据目录冲突，node.lock 文件可以确保一次只能从一个数据目录读取/写入一个 ES 相关的安装启动信息。

global-1389.st 是一个包含元数据的状态文件，存储着集群状态以及集群分片映射等信息，global- 前缀表示这是一个全局状态文件，而 1389 是该文件的版本号，当集群状态发生变化时，会更新元数据文件。

indices 文件夹下的是我们具体索引的数据文件，这里的 index 文件夹由 lucene 写入，而 translog 文件夹和 \\_state 文件夹由 ES 写入。lucene 每次添加/修改的数据先放在内存中，并不是实时落盘的，而是直到缓冲区满了或者用户执行 commit api，数据才会落盘，形成一个 segement，保存在文件中。ES 节点实现了 translog 类， 即数据索引前，会先写入到日志文件中。translog 用于在节点机器突发故障(比如断电或者其他原因)导致节点宕机，重启节点时就会重放日志，这样相当于把用户的操作模拟了一遍。保证了数据的不丢失。这里的操作有点类似 MySQL 的 redo log 和 bin log，redo log 作为机器异常宕机或者存储介质发生故障后的数据恢复使用，而 binlog 作为 MySQL 恢复数据使用，一般用作主从复制集群搭建或者第三方插件数据同步(比如：canal)。

节点宕机重启后并非重放所有的 translog，而是最新没有提交索引的那一部分。所以必须有一个 checkpoint, 即 translog.ckp 文件，保证节点宕机重启后可以从最近已提交的文件处重放，类似 bin log 的 position.

index 文件是 lucene 框架维护的，主要是为写入的文档建立倒排索引，其具体文件格式和作用如下 :

具体可以查看官方链接

下面具体分析快照相关的操作，以下代码版本基于 elasticsearch 7.5 版本

快照请求

首先是创建快照，当我们通过curl 发送 PUT/POST /_snapshot 请求时，rest controller 接收请求 RestCreateSnapshotAction，并对请求进行预处理，生成 createSnapshotRequest 发送到集群任意节点。

因为 CreateSnapshotRequest 是必须由 master 节点处理的请求，所以其他节点收到请求后会先比较 localNodeId 是否等于 masterNodeId，如果是就会用线程池处理 action，不是就会找 master 节点处理(没找到 master 节点会报 no known master node, scheduling a retry )

创建快照

创造快照具体可以查看 createSnapshot

master 节点创建快照请求，会先校验快照命名信息(快照名为小写，不能有空格/逗号/#号/不能以_开头/及其他特殊字符)，同时通过 randomBase64UUID 生成 snapshotId (包含 name 和 uuid 以及 hashCode )。获取快照仓库信息，从快照仓库中获取快照信息、索引信息、快照状态。之后会创建一个更新集群状态的任务 submitStateUpdateTask，任务先校验集群中是否有 delete snapshot/cleanup repository 进程，同一时间不能有以上两种进程存在。接着会检查需要快照的索引(与仓库已有的索引比较)， 初始化 snapshot 进程，根据 snapshots 信息生成新的集群状态。

而 snapshot 请求更新集群状态的任务执行完之后会调用 clusterStateProcessed 函数，标记集群的状态已经更新。函数此时才会开始正式执行集群数据打快照操作。会调用线程池执行索引分片 shard 级别的 snapshot 任务，查找每个索引的所有主分片信息，最后会根据索引的 shard 信息生成新的集群状态。

集群中其他节点会监听集群状态变化事件 ，并对事件中的自定义 snapshots 事件进行处理。这里会先校验集群状态前后的变化，如果快照事件状态由无到有或者前后 snapshot 事件状态不相等，才会开始处理快照事件。这里会再对快照进行一轮校验，删除不再存在的快照，中止已被删除列表中正在运行的分片快照。

开始计算本节点需要处理的 shard，并将计算结果放到 startedShards 对象中，shardSnapshots 是一个 Map 对象，key 为 ShardId 对象(包含索引对象和 shardId 以及 hashCode)，value 为另一个 IndexShardSnapshotStatus 枚举对象(枚举值有INIT STARTED FINALIZE DONE FAILURE ABORTED ).

计算完后从 snapshot 线程池获取线程，根据 shard 循环执行 snapshot 函数，snapshot 线程池是在节点启动后就初始化好的。

在 snapshot 函数中会执行一次 flush ,获取 IndexCommit 的最新写入状态，返回当前 commit point 的 Engine.IndexCommitRef 类实例对象 snapshotRef。

而 snapshotRef 对象里包含 IndexCommit 类对象，IndexCommit 是由 lucene 引擎定义的，包含了底层 lucene 文件 segement 文件名，所有索引文件，索引所在的目录等信息

执行 flush 后续调用到下层的 repository.snapshotShard 函数，会先获取仓库中的 blobs 信息，比较仓库最新的快照数据，是否有新增文件，有新增文件执行 snapshotFile 函数，最终执行

shardContainer.writeBlob(fileInfo.partName(i), inputStream, partBytes, true); 

拷贝文件到共享文件系统，完成数据快照，而拷贝文件的 blobContainer.writeBlob 是一个虚方法，对于不同的仓库文件系统有不同的实现，对于共享文件系统（fs）来说，拷贝过程通过 Streams.copy 实现，并在拷贝完成后执行 IOUtils.fsync 刷盘。完成文件拷贝之后会生成本次BlobStoreIndexShardSnapshot 信息，用于下一次快照比对

这里以 cos 文件为例，最终生成的文件如下

删除快照

删除快照处理流程与上面流程大体类似，主要发送的请求是 deleteSnapshotRequest , 对请求的处理也是先构建 request，发送到任意节点，节点再将请求发送到 master 节点，master 节点会先获取仓库中快照信息，找到需要删除的仓库快照，提交更新集群状态任务，对于正在进行中的快照任务，将其标记为 ABORTED 状态，其他数据节点监听集群状态变更事件，最终调用底层的

shardContainer.deleteBlobsIgnoringIfNotExists(blobsToDelete); 

删除仓库中的快照数据

快照恢复

恢复快照则是另一个请求 restoreSnapshotRequest ，大体处理流程也类似。都是实现的同一个方法 masterOperation

同样提交一个更新集群状态的 task 任务，该任务会检查恢复的前提条件，例如索引别名恢复等，并且根据需要恢复的shard列表创建 RestoreInProgress 记录，设置 RecoverySource type 为 SNAPSHOT 并更新集群状态。

其他节点检测到需要创建 shard 的事件，会开始创建 shard，根据分片恢复的 type(EMPTY_STORE/EXISTING_STORE/PEER/SNAPSHOT/LOCAL_SHARDS) 选择 restoreSnapshot 方法，从远程仓库中读取快照和元数据，通过 fileInputStream 读取文件信息并恢复索引数据。并且在 recovery 过程中还可以更改index的设置，比如原来为1副本，调整为2 副本，恢复成功后，会执行 allocationService.reroute 对分片进行重新路由。

增量快照

增量快照的核心是比较 lucene 的 segements 不可变文件信息，每次创建快照时会建立一个 IndexCommit 提交点，包含 segmentsfilename (segment 是 lucene 的不可变对象)，在处理分片快照请求时会先查找分段文件是否存在，文件信息是存储在 List<FileInfo> 对象中，如果文件信息存在，会比较 checksum 及 hash 值，如果都相同会跳过 snapshot 备份。否则会将该分片信息添加到本次快照的需要处理的全部文件列表

elasticsearch-repository-cos 插件

目前腾讯云 ES 服务的数据快照基本上都是存储在 cos 文件资源服务上，这里文件传输就需要使用 elasticsearch-repository-cos 插件，插件实现了 ES 底层 block 块存储接口的方法，读写块的具体实现为调用 cos client 上传和下载文件

以创建快照为例，前面可以看到，通过 ES API 创建 snapshot 请求时，最终调用

blobContainer.writeBlob(fileInfo.partName(i), inputStream, partBytes); 

方法实现文件拷贝，而插件具体的实现方法是调用 cos client 上传文件到 cos 服务器。

同时，插件也支持大文件分块上传，默认文件最大块大小为 5GB

使用步骤

1. 这里使用插件首先需要在集群中所有节点执行安装命令，7.5.1 版本为例

${es_install_dir}/bin/elasticsearch-plugin install file:///${plugin_dir}/elasticsearch-cos-7.5.1.zip

1. 依次重启集群中节点
2. 注册仓库快照

PUT _snapshot/my_cos_backup
{
    "type": "cos",
    "settings": {
        "access_key_id": "xxxxxx",
        "access_key_secret": "xxxxxxx",
        "bucket": "xxxx-xxxx",
        "region": "ap-guangzhou",
        "compress": true,
        "chunk_size": "500mb",
        "base_path": ""
    }
}

1. 查看快照仓库

GET _snapshot

1. 创建快照

PUT _snapshot/my_cos_backup/snapshot_20221221?wait_for_completion=true

1. 查看快照

GET _snapshot/my_cos_backup/snapshot_20221221

1. 从快照恢复数据

# 恢复 2022.05月数据
POST _snapshot/my_cos_backup/snapshot_20221221/_restore
{
  "indices": "*2021.05*"
}

1. 定时快照

通过前面的分析，创建快照时涉及集群状态变更，以及大量的文件 IO 操作，所以一般都是在凌晨请求较少时执行快照操作

PUT /_slm/policy/nightly-snapshots
{
  "schedule": "0 0 19 * * ?", 
  "name": "<es-snap-{now/d}>", 
  "repository": "my_cos_backup", 
  "config": { 
    "indices": ["*"] 
  },
  "retention": { 
    "expire_after": "30d", 
    "min_count": 5, 
    "max_count": 50 
  }
}
# 每天国际时间19点(北京时间凌晨3点)执行 snapshot 任务，快照30天有效，最少5张，最多50张