《Elasticsearch 源码解析与优化实战》第13章：Snapshot 模块分析

curl -X PUT " localhost: 9200/_ snapshot/my_ backup" -H ' Content-Type: application/json' -d'
{
    "type": "fs",
    "settings": {
        "location": "/mnt/my_backup"
    }
}

curl -X GET "localhost:9200/_snapshot/my_backup"

{
    "my_ backup": {
        "type": "fs",
        "settings": {
            "location": "/mnt/my_backup"
        }
    }
}

curl -X GET "localhost:9200/_snapshot/repo*,*backup*"

curl -X GET "localhost:9200/_snapshot"
或
curl -X GET "localhost:9200/_snapshot/_all"

curl -X DELETE "localhost:9200/_snapshot/my_backup/snapshot_1"

curl -X DELETE "localhost:9200/_snapshot/my_backup"

path.repo: ["/mnt/my_backups"]

path.repo: ["\\\\MY_SERVER\\Snapshots"]

curl -X PUT "localhost:9200/_snapshot/my_backup/snapshot_1?wait_for_completion=true"

curl -X PUT "localhost:9200/_snapshot/my_backup/snapshot_2?wait_for_completion=true" -H 'Content-Type: application/json' -d'
{
    "indices": "index_1, index_2",
    "ignore_unavailable": true,
    "include_global_state": true
}

curl -X GET "localhost:9200/_snapshot/my_backup/snapshot_1"

{
    " snapshots": [
    {
       "snapshot": "snapshot_1",
       "version": "6.1.2",
        "indices": [
            "website"
        ],
        "state": " SUCCESS",
        "start_time": "2018-05-15T03:40:06.571Z",
        "end_time": "2018-05-15T07:53:40.977Z",
        "duration_in_millis": 15214406,
        " failures": [] ,
        "shards": {
            "total": б,
            "failed": O,
            "successful": 6
        }
    }
  ]
}

curl -X GET "localhost: 9200/_snapshot/my_backup/snapshot_*, other_snapshot"

curl -X GET "localhost:9200/_snapshot/my_backup/_all"

curl -X GET "localhost:9200/_snapshot/my_backup/_current"

curl -X GET "localhost:9200/_snapshot/_status"

"stats": {
    "number_of_files": 31,
    "processed_files": 31,
    "total_size_in_bytes": 33802,
    "processed_size_in_bytes": 33802,
    "start_time_in_millis": 1526355676967,
    "time_in_millis": 15144003
}

curl -X GET "localhost:9200/_snapshot/my_backup/_status"

curl -X GET "localhost:9200/_snapshot/my_backup/snapshot_1, snapshot_2/status"

curl -x DELETE "localhost:9200/_snapshot/my_backup/snapshot_1"

curl -X POST "localhost:9200/snapshot/my_backup/snapshot_1/_restore'

curl -X POST " localhost:9200/snapshot/mybackup/snapshot_1/_restore" -H 'Content-Type: application/json' -d'
{
    "indices": "index_1, index_2" ,
    "ignore_unavailable": true,
    "include_global_state": true,
    "rename_pattern": "index_(.+)",
    "rename_replacement": "restored_index_$1"
}

curl -X POST "localhost:9200/_snapshot/my_backup/snapshot_1/_restore" -H 'Content-Type: application/json' -d'
    "indices": "index_ 1",
    "index settings": {
        " index.number_of_replicas": 0
    }
    "ignore_index_settings": [
        "index.refresh_interval"
    ]
}

public RestChannelConsumer prepareRequest (final RestRequest request, final NodeClient client) throws IOException {
    //将请求封装为CreateSnapshotRequest结构
    CreateSnapshotRequest createSnaps hotRequest = createSnapshotRequest(request.param ("repository"), request.param("snapshot"));
    request.applyContentParser(p -> createSnapshotRequest.source(p.mapOrdered()));
    //设置超时wait_for_completion 等待参数信息
    createSnapshotRequest.masterNodeTimeout(request.paramAsTime("master_timeout", createSnapshotRequest.masterNodeTimeout()));
    createSnapshotRequest.waitForCompletion(request.par amAsBoolean("wait_for_completion", false));
    return channel -> client.admin().cluster().createSnapshot(createSnapshotRequest, new RestToXContentListener<> (channel));
}

代码摘要如下：
protected void doStart (ClusterState clusterState) {
    if (nodes. isLocalNodeElectedMaster () ll localExecute (request) ) {
        //本地是主节点，在新的线程池中处理请求
        threadPool.executor(executor).execute(new ActionRunnable (delegate) {
            protected void doRun() throws Exception {
                masterOperation(task, request, clusterState, delegate) ;
            }
        ]);
    } else { //转发到主节点。request为组装好的CreateSnapshotRequest结构
        transportService.sendRequest (masterNode, actionName, request, new ActionLi stenerResponseHandler <Response> (listener,
        TransportMasterNodeAction.this::newResponse)
    }
}

protected void masterOperation(.. .) {
    snapshotsService.createSnapshot (snapshotRequest, new SnapshotsService.CreateSnapshotListener () {
        public void onResponse () {
            //根据需要注册一个Listener, 快照执行完毕才返回响应给客户端.
            if (request.waitForCompletion()) {
            snapshotsService.addListener (new SnapshotsService.SnapshotCompletionListener () {
                public void onSnapshotCompletion (Snapshot snapshot, SnapshotInfo snapshotInfo) {
                    ......
                }
                public void onSnapshotFailure (Snapshot snapshot, Exception e) {
                    ......
                }
            ]) ;
        } else {//给客户端返回结果，快照任务后台执行
            listener.onResponse (new CreateSnapshotResponse());
        }
    }
    //对执行失败的处理
    public void onFailure (Exception е) {
        listener. onFailure (е) ;
    }
}) ;

public void createSnapshot (final SnapshotRequest request, final CreateSnapshotListener listener) {
    clusterService.submitStateUpdateTask (request.cause()，new ClusterStateUpdateTask() {
        //定义要执行的任务
        public ClusterState execute (ClusterState currentState) {
            //快照任务不能并行，同一时间只能有一个快照在执行
            if (snapshots == null || snapshots.entries().isEmpty()) {
                //要快照的索引列表
                List<IndexId> snapshotIndices = repositoryData.resolveNewIndices (indices);
                newSnapshot = new SnapshotsInProgress.Entry (new Snapshot(repositoryName，snapshotId),
                    request.includeGlobalState(), 
                    request.partial(),
                    State.INIT, //初始状态为INIT 
                    snapshotIndices,
                    System.currentTimeMillis()，
                    repositoryData.getGenId(),
                    null) ;
                snapshots = new SnapshotsInProgress (newSnapshot);
            } else {
                throw new ConcurrentSnapshotExecut ionException (repositoryName, snapshotName，" a snapshot is already running");
                //根据snapshots信息生成新的集群状态
                return ClusterState.builder (currentState).putCustom(SnapshotsInProgress.TYPE，snapshots).build();
            }
            //待state为INIT 集群状态处理成功后，发送State为START的集群状态
            public void clusterStateProcessed(String source, ClusterState oldState, final ClusterState newState) {
                if (newSnapshot != null) {
                    threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(() ->
                        beginSnapshot (newState, newSnapshot, request.partia (), listener)
                );
            }
        }
    }
}

public void initializeSnapshot (SnapshotId snapshotId, List<IndexId> indices, MetaData clusterMetaData) {
    try {
        //写全局元信息
        globalMetaDataFormat.write (clusterMetaData, snapshotsBlobContainer, snapshotId.getUUID());
        //为快照的每个索引写索引级元信息
        for (IndexId index : indices) {
            final IndexMetaData indexMetaData = clusterMetaData.index.(index.getName());
            final BlobPath indexPath = basePath().add ("indices").add(index.getId());
            final BlobContainer indexMetaDataBlobContainer = blobStore().blobContainer(indexPath) ;
            indexMetaDataFormat.write (indexMetaData, indexMetaDataBlobContainer, snapshotId.getUUID());
        } catch (IOException ех) {
            throw new SnapshotCreationException (metadata.name(), snapshotId, ex);
        }
    }
}

for (Obj ectObjectCursor<ShardId, ShardSnapshotStatus> shard : entry.shards ()) {
    //准备本节点要处理的分片
    if (localNodeId. equals (shard. value.nodeId())) {
        if (shard. value.state() == State. INIT && (snapshotShards == null || ! snapshotShard .shards.containsKey(shard.key))) {
            logger.trace("[(]] - Adding shard to the queue", shard.key);
            startedShards.put (shard.key, new IndexShardSnapshotStatus());
        }
    }
}
//本节点要处理的分片列表
newSnapshots.put(entry.snapshot(), startedshards) ;

if (newSnapshots. isEmpty() == false) {
    Executor executor.= threadPool.executor (ThreadPool.Names.SNAPSHOT);
    for (final Map. Entry<Snapshot, Map<ShardId, IndexShardSnapshotStatus>> entry : newSnapshots.entrySet()) {
        //shard级并发执行快照
        for (final Map.Entry<ShardId, IndexShardSnapshotstatus> shardEntry : entry.getValue().entrySet()) {
            executor.execute (new Abs tractRunnable () {
                public void doRun() {
                    //对特定分片执行快照
                    snapshot (indexShard, snapshot, indexId, shardEntry.getValue());
                        public void onAfter () {
                            //向Master节点发送请求以更新快照状态
                            final Exception exception = failure.get ();
                            if (exception != null) l
                                final String failure = ExceptionsHelper.detailedMessage(exception) ;
                                notifyFailedSnapshotShard (snapshot, shardId, localNodeId, failure, masterNode) ;
                            } else {
                                notifySuccess fulSnapshotShard (snapshot, shardId, localNodeId, masterNode);
                            }
                        }
                ]) ;
            }
       }
}

IndexCommitRef (SnapshotDeletionPolicy deletionPolicy) throws IOException {
    //调用Lucene接口创建快照，返回提交点
    indexCommit = deletionPolicy. snapshot() ;
    //处理完毕，释放快照资源
    onClose = () -> deletionPolicy. release (indexCommit) ;
}

this.deletionPolicy = new CombinedDeletionPolicy (
new SnapshotDeletionPolicy (new KeepOnlyLastCommi tDeletionPolicy()), ...);

//创建Lucene快照，快照前先执行flush操作，但不执行refresh操作（flush操作时，会执行一次fresh）
try (Engine.IndexCommitRef snapshotRef = indexShard.acquireIndexCommit (true)) {
    //根据Lucene提交点创建快照
    repository.snapshotShard (indexShard, snapshot. getSnapshotId(), indexId, snapshotRef.getIndexCommit()，snapshotStatus);
}

fileNames = snapshotIndexCommit.getFileNames () ;
//遍历Lucene提交点中的文件列表
for (String fileName : fileNames) {
    //检查是否需要中止快照
    if (snapshotStatus. aborted() ) {
        throw new IndexShardSnapshotFailedException (shardId, "Aborted") ;
    }
    BlobStoreIndexShardSnapshot.FileInfo existingFileInfo = null;
    //在仓库中查找这个文件是否存在，由于Lucene文件的不变性，只要文件名存在，文件就是相同的，不检测校验和
    List<BlobStore IndexShardSnapshot.FileInfo> filesInfo = snapshots.findPhysicalIndexFiles (fileName) ;
    if (filesInfo != null) {
        for (BlobStore IndexShardSnapshot. FileInfo fileInfo : filesInfo) {
            if (fileInfo.isSame (md) && snapshotFileExistsInBlobs (fileInfo, blobs)) {
                // 仓库中已存在
                existingFileInfo = fileInfo;
                break;
            }
        }
    }
    if (existingFileInfo == null) {
        BlobStoreIndexShardSnapshot. FileInfo snapshotFileInfo = new BlobStoreIndexShardSnapshot.FileInfo(.. .) ;
        indexCommitPointFiles.add (snapshotFileInfo) ;  //添加到本次快照的全部文件列表
        filesToSnapshot. add (snapshotFileInfo) ;//添加到新增文件列表
    } else {
        indexCommi tPointFiles . add (existingFileInfo) ;/ /添加到本次快照的全部文件列表
    }
}

for (BlobStoreIndexShardSnapshot.FileInfo snapshotFileInfo : filesToSnapshot) {
    snapshotFile (snapshotFileInfo) ;
}

private void snapshotFile (final BlobStoreIndexShardSnapshot.FileInfo fileInfo) throws IOException {
    final String file = fileInfo.physicalName () ;
    try (IndexInput indexInput = store. openVeri fyingInput (file, IOContext.READONCE, fileInfo.metadata() ) ) {
        for (int i = O; i < fileInfo.numberOfParts() ; i++) {
            InputStream inputStream = inputStreamIndexInput;
            //初始化限速模块
            if (snapshotRateLimiter != null) {
                inputStream = new RateLimitingInputStream(inputStreamIndexInput, snapshotRateLimiter, snapshotRateLimitingTime InNanos :: inc) ;
            }
            inputStream = new AbortableInputStream (inputStream, fileInfo.physicalName()) ;
            //复制文件
            blobContainer.writeBlob (fileInfo.partName(i), inputStream, partBytes) ;
        }
        //比较checksum是否正确。- -是Lucene元信息中存储的校验和，二是数据复制到目
        //的地址过程中，计算出来的校验和
        Store.verify (indexInput) ;
        snapshotStatus.addProcessedFile (fileInfo.length()) ;
    }
}

BlobStoreIndexShardSnapshot snapshot = new BlobStoreIndexShardSnapshot(snapshotId.getName(), snapshotIndexCommit.getGeneration(), indexCommitPointFiles, snapshotStatus.startTime (), System.currentTimeMillis() - snapshotStatus.startTime() , indexNumberOfFiles, indexTotalFilesSize) ;
//写快照文件，快照文件中描述了本次快照要使用的所有相关Lucene分段
indexSha rdSnapshotFormat.write (snapshot, blobContainer, snapshotId.getUUID());

protected void finalize (List<SnapshotFiles> snapshots, int fileListGeneration, Map<String，BlobMetaData> blobs) {
    BlobStore IndexShardSnapshots newSnapshots = new BlobStore IndexShardSnapshots (snapshots) ;
    //从当前分片的快照中删除索引index-*, 这个index-*文件是对快照列表的索引
    //blobName是仓库中分片下一个具体的文件，例如，_ 1、 index-*、 snap-*
    for (String blobName : blobs.keySet()) {
        if (indexShardSnapshotsFormat.isTempBl obName (blobName) || blobName.startsWith (SNAPSHOT INDEX_ PREFIX)) {
            blobContainer.deleteBlob (blobName);
        }
        //遍历当前分片仓库中的数据文件，删除快照中未使用的文件
        for (String blobName : blobs.keySet()) {
            //删除无用的文件，数据BLOB前缀DATA_BLOB_PREFIX值为
            if (blobName.startsWith (DATA_BLOB_PREFIX)) {
                if (newSnapshots.findNameFile (BlobStoreIndexShardSnapshot.FileInfo.canonicalName (blobName)) == null) {
                    blobContainer.deleteBlob (blobName) ;
                }
            }
        }
        //如果全部快照都删除，则不创建index-*索引文件
        if (snapshots.size() > 0) {
            //写index-*索引文件
            indexSha rdSnapshotsFormat.writeAtomic (newSnapshots, blobContainer, Integer.toString (fileListGeneration)) ;
        }
    }
}

-rw-rw-r--  1 avatar avatar   29 Aug  6 11:15 incompatible-snapshots
-rw-rw-r--  1 avatar avatar 3743 Aug 13 01:01 index-4
-rw-rw-r--  1 avatar avatar 2920 Aug 13 01:02 index-5
-rw-rw-r--  1 avatar avatar    8 Aug 13 01:02 index.latest
drwxrwxr-x 25 avatar avatar 4096 Aug 13 01:02 indices
-rw-rw-r--  1 avatar avatar  103 Aug  9 01:00 meta-GgFN1wAqQm65dDXZyEaJgg.dat
-rw-rw-r--  1 avatar avatar  103 Aug 13 01:00 meta-rPBlWtpkQgSp7eEAObC2UA.dat
-rw-rw-r--  1 avatar avatar  488 Aug  9 01:13 snap-GgFN1wAqQm65dDXZyEaJgg.dat
-rw-rw-r--  1 avatar avatar  493 Aug 13 01:01 snap-rPBlWtpkQgSp7eEAObC2UA.dat

private void deleteSnapshot(...) {
    //提交集群任务
    clusterService.submitStateUpdateTask ("delete snapshot", new ClusterStateUpdateTask (priority) {
        public ClusterState execute (ClusterState currentState) throws Exception {
            SnapshotsInProgress snapshots = currentState.custom(SnapshotsInProgress.TYPE);
            SnapshotsInProgress.Entry snapshotEntry = snapshots!= null ? snapshots.snapshot (snapshot) : null; 
            if (snapshotEntry == null) {
                //快照没有运行，将删除请求信息放到Custom字段的
                //SnapshotDeletionsInProgress 结构中
                clusterStateBuilder.putCustom (SnapshotDeletions InProgress.TYPE, deletionsInProgress) ;
            } else {
                //将删除快照请求的相关信息放到Customs的SnapshotsInProgress 中，
                //并将state设置为ABORTED
                SnapshotsInProgress . Entry newSnapshot = new SnapshotsInProgress.Entry (snapshotEntry, State.ABORTED, shards) ;
                snapshots = new SnapshotsInProgress (newSnapshot) ;
                clusterStateBuilder.putCustom (SnapshotsInProgress.TYPE, snapshots) ;
            }
            snapshots = new SnapshotsInProgress (newSnapshot) ;
            clusterStateBuilder.putCustom (Snapshots InProgress.TYPE，snapshots) ;
            return clusterStateBuilder.build() ;
        }
        //处理集群状态发布失败的情况
        public void onFailure (String source, Exception e) {}
        //集群状态发布完毕
        public void clusterStateProcessed (String source, ClusterState oldState，ClusterState newState) {
            if (waitForSnapshot) {
                //处理客户端等待完成的情况
            } else
                //删除快照文件
                deleteSnapshotFromRepository (snapshot, listener, repositoryStateId) ;
            }
        }
    }) ;
}

private void deleteSnapshotFromRepository(...) {
    threadPool.executor (ThreadPool.Names.SNAPSHOT).execute(() --> {
        //执行删除
        repository.deleteSnapshot (snapshot.getSnapshotId()，repositoryStateId) ;
        removeSnapshotDeletionFromClusterState (snapshot, null, listener) ;
    }) ;
}

public void deleteSnapshot (SnapshotId snapshotId, 1ong repositoryStateId) {
    MetaData metaData = readSnapshotMetaData (snapshotId, snapshot.version() , repositoryData.resolveIndices (indices)，true);
    try {
        //更新快照列表的索引文件index-*，将快照信息从快照列表中删除
        final RepositoryData updatedRepositoryData = repositoryData.removeSnapshot (snapshotId) ;
        writeIndexGen (updatedRepositoryData, repositoryStateId) ;
        //删除全局快照文件(仓库根目录下的) snap-*
        deleteSnapshotBlobIgnoringErrors (snapshot, snapshotId.getUUID()) ;
        //删除全局元信息文件meta-* .dat
        deleteGlobalMetaDataBlobIgnoringErrors (snapshot, snapshotId.getUUID()) ;
        //删除全部索引
        for (String index : indices) {
            //删除索引的元信息文件meta-* .dat
            indexMetaDataFormat.delete (indexMetaDataBlobContainer, snapshotId.getUUID()) ;
            if (metaData != null) {
                IndexMetaData indexMetaData = metaData.index (index) ;
                if (indexMetaData != null) {
                    for (int shardId= 0; shardId < indexMetaData.getNumberOfShards () ; shardId++) {
                        //删除分片快照，这里是删除实际的Lucene文件
                        delete (snapshotId, snapshot. version()，indexId, new ShardId (indexMetaData.getIndex()，shardId) ) ;
                    }
                }
            }
        }
        //删除仓库中已经不存在的索引目录
        for (final IndexId indexId : indicesToCleanUp) {
            indicesBlobContainer . deleteBlob(indexId.getId() ) ;
        }
    }
}

public void delete() {
    Tuple<BlobStoreIndexShardSnapshots, Integer> tuple = buildBlobStore IndexShardSnapshots (blobs) ;
    //仓库现有的快照列表
    BlobStore IndexShardSnapshots snapshots = tuple.v1() ;
    int fileListGeneration = tuple.v2 () ;
    //删除分片的snapshot文件
    indexShardSnapshotFormat (version).delete (blobContainer, snapshotId.getUUID()) ;
    //构建需要保留的快照列表
    List<SnapshotFiles> newSnapshotsList = new ArrayList<> () ;
    for (SnapshotFiles point : snapshots) {
        if (!point.snapshot ().equals (snapshotId.getName())) {
            newSnapshotsList.add (point) ;
        }
    }
    //传入需要保留的快照列表，而非要删除的快照列表。仓库中存在的, 但不被列表中快照使用的文件被删除
    finalize (newSnapshotsList, fileListGeneration + 1, blobs) ;
}

属于snapshot_1的文件有：
_0
_1
_2
_3

属于snapshot _2 的文件有:
_0
_1
_2
_3
_4
_5
_6
_7

private void process IndexShardSnapshots (ClusterChangedEvent event) {
    if (snapshotsInProgress !=null) {
        for (SnapshotsInProgress . Entry entry : snapshots InProgress.entries()) {
            if (entry.state() == State . STARTED) {
                //处理快照创建
            } else if (entry.state() == State. ABORTED) {
                //处理快照取消
                if (snapshotShards != null) {
                    for (Obj ectObjectCursor<ShardId, ShardSnapshotStatus> shard : entry.shards () ) {
                        if (snapshotStatus != null) {
                            switch (snapshotStatus.stage()) {
                                case INIT:
                                case STARTED:
                                    //设置中止标识
                                    snapshotStatus.abort() ;
                                case ....
                            }
                        }
                    }
                }
            }
        }
    }
}

public void abort() {
    this.aborted = true;
}

public int read (byte[] b，int off，int len) throws IOException {
    checkAborted() ;
    return in. read(b, off, len) ;
}

private void checkAborted() {
    if (snapshotStatus.aborted() ) {
        throw new IndexShardSnapshotFailedException (shardId, "Aborted") ;
    }
}