上周出现了一次故障，recovery的过程比较慢，然后发现Shard 在做恢复的过程一般都是卡在TRANSLOG阶段，所以好奇这块是怎么完成的，于是有了这篇文章

这是一篇源码分析类的文章，大家需要先建立一个整体的概念，建议参看这篇文章

另外你可能还需要了解下 Recovery 阶段迁移过程：

INIT -> INDEX -> VERIFY_INDEX -> TRANSLOG -> FINALIZE -> DONE

概览

Recovery 其实有两种：

1. Primary的迁移/Replication的生成和迁移
2. Primary的恢复

org.elasticsearch.indices.cluster.IndicesClusterStateService.clusterChanged 被触发后，会触发applyNewOrUpdatedShards 函数的调用，这里是我们整个分析的起点。大家可以跑进去看看，然后跟着文章打开对应的源码浏览。

阅读完这篇文章，我们能够得到：

1. 熟悉整个recovery 流程
2. 了解translog机制
3. 掌握对应的代码体系结构

Primary的恢复

这个是一般出现故障集群重启的时候可能遇到的。首先需要从Store里进行恢复。

if (isPeerRecovery(shardRouting)) {
   ......
}
else {
  //走的这个分支
  indexService.shard(shardId).recoverFromStore(shardRouting, 
  new StoreRecoveryService.RecoveryListener() {
}

Primary 进行自我恢复，所以并不需要其他节点的支持。所以判定的函数叫做isPeerRecovery 其实还是挺合适的。

indexService.shard(shardId).recoverFromStore 调用的是 org.elasticsearch.index.shard.IndexShard的方法。

  public void recoverFromStore(ShardRouting shard, StoreRecoveryService.RecoveryListener recoveryListener) {
    ......
    final boolean shouldExist = shard.allocatedPostIndexCreate();
    storeRecoveryService.recover(this, shouldExist, recoveryListener);
    }

逻辑还是很清晰的，判断分片必须存在，接着将任务委托给 org.elasticsearch.index.shard.StoreRecoveryService.recover 方法，该方法有个细节需要了解下：

 if (indexShard.routingEntry().restoreSource() != null) {
       indexShard.recovering("from snapshot", 
           RecoveryState.Type.SNAPSHOT, 
           indexShard.routingEntry().restoreSource());
            } else {
       indexShard.recovering("from store", 
           RecoveryState.Type.STORE, 
           clusterService.localNode());
            }

ES会根据restoreSource 决定是从SNAPSHOT或者从Store里进行恢复。这里的indexShard.recovering并没有执行真正的recovering 操作，而是返回了一个recover的信息对象，里面包含了譬如节点之类的信息。

之后就将其作为一个任务提交出去了：

threadPool.generic().execute(new Runnable() {
            @Override
  public void run() {
  try {
  final RecoveryState recoveryState = indexShard.recoveryState();
  if (indexShard.routingEntry().restoreSource() != null) {
     restore(indexShard, recoveryState);
     } else {              
     recoverFromStore(indexShard, indexShouldExists, recoveryState);
    }

这里我们只走一条线，也就是进入 recoverFromStore 方法，该方法会执行索引文件的恢复动作，本质上是进入了INDEX Stage.

接着进行TranslogRecovery了

typesToUpdate = indexShard.performTranslogRecovery(indexShouldExists);
indexShard.finalizeRecovery();

继续进入 indexShard.performTranslogRecovery 方法：

  public Map<String, Mapping> performTranslogRecovery(boolean indexExists) {
        if (indexExists == false) {            
            final RecoveryState.Translog translogStats = recoveryState().getTranslog();
            translogStats.totalOperations(0);
            translogStats.totalOperationsOnStart(0);
        }
        final Map<String, Mapping> recoveredTypes = internalPerformTranslogRecovery(false, indexExists);     
        return recoveredTypes;
    }

这个方法里面，最核心的是 internalPerformTranslogRecovery方法，进入该方法后先进入 VERIFY_INDEX Stage,进行索引的校验，校验如果没有问题，就会进入我们期待的TRANSLOG 状态了。

进入TRANSLOG 后，先进行一些设置：

engineConfig.setEnableGcDeletes(false);
engineConfig.setCreate(indexExists == false);

这里的GC 指的是tranlog日志的删除问题，也就是不允许删除translog，接着会创建一个新的InternalEngine了，然后返回调用org.elasticsearch.index.shard.TranslogRecoveryPerformer.getRecoveredTypes

不过你看这个代码会比较疑惑，其实我一开始看也觉得纳闷：

  if (skipTranslogRecovery == false) {            
       markLastWrite();
        }
 createNewEngine(skipTranslogRecovery, engineConfig);
 return  engineConfig.getTranslogRecoveryPerformer().
                                  getRecoveredTypes();

我们并没有看到做translog replay的地方，而从上层的调用方来看：

typesToUpdate = indexShard.performTranslogRecovery(indexShouldExists);
indexShard.finalizeRecovery();

performTranslogRecovery 返回后，就立马进入扫尾(finalizeRecovery)阶段。 里面唯一的动作是createNewEngine，并且传递了skipTranslogRecovery 参数。 也就说，真正的translog replay动作是在createNewEngine里完成，我们经过探索，发现是在InternalEngine 的初始化过程完成的，具体代码如下：

try {
      if (skipInitialTranslogRecovery) {
          commitIndexWriter(writer,
                                       translog, 
                                       lastCommittedSegmentInfos.
                                       getUserData().
                                       get(SYNC_COMMIT_ID));
                } else {
                    recoverFromTranslog(engineConfig, translogGeneration);
                }
            } catch (IOException | EngineException ex) {
              .......
            }

里面有个recoverFromTranslog，我们进去瞅瞅：

   final TranslogRecoveryPerformer handler = engineConfig.getTranslogRecoveryPerformer();
        try (Translog.Snapshot snapshot = translog.newSnapshot()) {
            opsRecovered = handler.recoveryFromSnapshot(this, snapshot);
        } catch (Throwable e) {
            throw new EngineException(shardId, "failed to recover from translog", e);
        }

目前来看，所有的Translog recovery 动作其实都是由 TranslogRecoveryPerformer 来完成的。当然这个名字也比较好，翻译过来就是 TranslogRecovery 执行者。先对translog 做一个snapshot,然后根据这个snapshot开始进行恢复，进入 recoveryFromSnapshot 方法我们查看细节，然后会引导你进入
下面的方法：

 public void performRecoveryOperation(Engine engine, Translog.Operation operation, boolean allowMappingUpdates) {
        try {
            switch (operation.opType()) {
                case CREATE:
                    Translog.Create create = (Translog.Create) operation;
                    Engine.Create engineCreate = IndexShard.prepareCreate(docMapper(create.type()),
                            source(create.source()).index(shardId.getIndex()).type(create.type()).id(create.id())
                                    .routing(create.routing()).parent(create.parent()).timestamp(create.timestamp()).ttl(create.ttl()),
                            create.version(), create.versionType().versionTypeForReplicationAndRecovery(), Engine.Operation.Origin.RECOVERY, true, false);
                    maybeAddMappingUpdate(engineCreate.type(), engineCreate.parsedDoc().dynamicMappingsUpdate(), engineCreate.id(), allowMappingUpdates);
                    if (logger.isTraceEnabled()) {
                        logger.trace("[translog] recover [create] op of [{}][{}]", create.type(), create.id());
                    }
                    engine.create(engineCreate);
                    break;

终于看到了实际的translog replay 逻辑了。这里调用了标准的InternalEngine.create 等方法进行日志的恢复。其实比较有意思的是，我们在日志回放的过程中，依然会继续写translog。这里就会导致一个问题，如果我在做日志回放的过程中，服务器由当掉了(或者ES instance 重启了)，那么就会导致translog 变多了。这个地方是否可以再优化下？

假设我们完成了Translog 回放后，如果确实有重放，那么就行flush动作，删除translog,否则就commit Index。具体逻辑由如下的代码来完成：

if (opsRecovered > 0) {
     opsRecovered, translogGeneration == null ? null : translogGeneration.translogFileGeneration, translog
                            .currentFileGeneration());
            flush(true, true);
        } else if (translog.isCurrent(translogGeneration) == false) {
            commitIndexWriter(indexWriter, translog, lastCommittedSegmentInfos.getUserData().get(Engine.SYNC_COMMIT_ID));
        }

接着就进入了finalizeRecovery，然后，就没然后了。

 indexShard.finalizeRecovery();
            String indexName = indexShard.shardId().index().name();
            for (Map.Entry<String, Mapping> entry : typesToUpdate.entrySet()) {
                validateMappingUpdate(indexName, entry.getKey(), entry.getValue());
            }
            indexShard.postRecovery("post recovery from shard_store");

Primary的迁移/Replication的生成和迁移

一般这种recovery其实就是发生relocation或者调整副本的时候发生的。所以集群是在正常状态，一定有健康的primary shard存在，所以我们也把这种recovery叫做Peer Recovery。 入口和前面的Primary恢复是一样的，代码如下：

if (isPeerRecovery(shardRouting)) {
 //走的这个分支
.....
RecoveryState.Type type = shardRouting.primary() ? RecoveryState.Type.RELOCATION : RecoveryState.Type.REPLICA;
                recoveryTarget.startRecovery(indexShard, type, sourceNode, new PeerRecoveryListener(shardRouting, indexService, indexMetaData));
......           
}
else {
 ......
}

核心代码自然是 recoveryTarget.startRecovery。这里的recoveryTarget的类型是： org.elasticsearch.indices.recovery.RecoveryTarget

startRecovery方法的核心代码是：

threadPool.generic().execute(new RecoveryRunner(recoveryId));

也是启动一个县城异步执行的。RecoveryRunner调用的是RecoveryTarget的 doRecovery方法，在该方法里，会发出一个RPC请求：

final StartRecoveryRequest request = new StartRecoveryRequest(recoveryStatus.shardId(), recoveryStatus.sourceNode(), clusterService.localNode(),        false, metadataSnapshot, recoveryStatus.state().getType(), recoveryStatus.recoveryId());

recoveryStatus.indexShard().prepareForIndexRecovery();
            recoveryStatus.CancellableThreads().execute(new CancellableThreads.Interruptable() {
                @Override
                public void run() throws InterruptedException {
                    responseHolder.set(transportService.submitRequest(request.sourceNode(), RecoverySource.Actions.START_RECOVERY, request, new FutureTransportResponseHandler<RecoveryResponse>() {
                        @Override
                        public RecoveryResponse newInstance() {
                            return new RecoveryResponse();
                        }
                    }).txGet());
                }
            });

这个时候进入 INDEX Stage。 那谁接受处理的呢？ 我们先看看现在的类名叫啥？ RecoveryTarget。 我们想当然的想，是不是有RecoverySource呢？ 发现确实有，而且该类确实也有一个处理类：

 class StartRecoveryTransportRequestHandler extends TransportRequestHandler<StartRecoveryRequest> {
        @Override
        public void messageReceived(final StartRecoveryRequest request, final TransportChannel channel) throws Exception {
            RecoveryResponse response = recover(request);
            channel.sendResponse(response);
        }
    }

ES里这种通过Netty进行交互的方式，大家可以看看我之前写文章ElasticSearch Rest/RPC 接口解析。

这里我们进入RecoverSource对象的recover方法：

 private RecoveryResponse recover(final StartRecoveryRequest request) {
      .....
      if (IndexMetaData.isOnSharedFilesystem(shard.indexSettings())) {
            handler = new SharedFSRecoverySourceHandler(shard, request, recoverySettings, transportService, logger);
        } else {
            handler = new RecoverySourceHandler(shard, request, recoverySettings, transportService, logger);
        }
        ongoingRecoveries.add(shard, handler);
        try {
            return handler.recoverToTarget();
        } finally {
            ongoingRecoveries.remove(shard, handler);
        }
 }

我们看到具体负责处理的类是RecoverySourceHandler，之后调用该类的recoverToTarget方法。我对下面的代码做了精简，方便大家看清楚。

public RecoveryResponse recoverToTarget() {
        final Engine engine = shard.engine();
        assert engine.getTranslog() != null : "translog must not be null";
        try (Translog.View translogView = engine.getTranslog().newView()) {
     
            final SnapshotIndexCommit phase1Snapshot;
            phase1Snapshot = shard.snapshotIndex(false);
            phase1(phase1Snapshot, translogView);
                      
            try (Translog.Snapshot phase2Snapshot = translogView.snapshot()) {
                phase2(phase2Snapshot);
            } catch (Throwable e) {
                throw new RecoveryEngineException(shard.shardId(), 2, "phase2 failed", e);
            }

            finalizeRecovery();
        }
        return response;
    }

首先创建一个Translog的视图（创建视图的细节我现在也还没研究），接着的话对当前的索引进行snapshot。 然后进入phase1阶段，该阶段是把索引文件和请求的进行对比，然后得出有差异的部分，主动将数据推送给请求方。之后进入文件清理阶段，然后就进入translog 阶段：

protected void prepareTargetForTranslog(final Translog.View translogView) {

接着进入第二阶段：

try (Translog.Snapshot phase2Snapshot = translogView.snapshot()) {
                phase2(phase2Snapshot);           
            }

对当前的translogView 进行一次snapshot,然后进行translog发送:

int totalOperations = sendSnapshot(snapshot);

具体的发送逻辑如下：

 cancellableThreads.execute(new Interruptable() {
                    @Override
                    public void run() throws InterruptedException {
                        final RecoveryTranslogOperationsRequest translogOperationsRequest = new RecoveryTranslogOperationsRequest(
                                request.recoveryId(), request.shardId(), operations, snapshot.estimatedTotalOperations());
                        transportService.submitRequest(request.targetNode(), RecoveryTarget.Actions.TRANSLOG_OPS, translogOperationsRequest,
                                recoveryOptions, EmptyTransportResponseHandler.INSTANCE_SAME).txGet();
                    }
                });

这里发的请求，都是被 RecoveryTarget的TranslogOperationsRequestHandler 处理器来完成的，具体代码是：

 @Override
        public void messageReceived(final RecoveryTranslogOperationsRequest request, final TransportChannel channel) throws Exception {
            try (RecoveriesCollection.StatusRef statusRef = onGoingRecoveries.getStatusSafe(request.recoveryId(), request.shardId())) {
                final ClusterStateObserver observer = new ClusterStateObserver(clusterService, null, logger);
                final RecoveryStatus recoveryStatus = statusRef.status();
                final RecoveryState.Translog translog = recoveryStatus.state().getTranslog();
                translog.totalOperations(request.totalTranslogOps());
                assert recoveryStatus.indexShard().recoveryState() == recoveryStatus.state();
                try {
                    recoveryStatus.indexShard().performBatchRecovery(request.operations());

这里调用IndexShard.performBatchRecovery进行translog 的回放。

最后发送一个finalizeRecovery给target 节点，完成recovering操作。

关于Recovery translog 配置相关

在如下的类里有：

//org.elasticsearch.index.translog.TranslogService
INDEX_TRANSLOG_FLUSH_INTERVAL = "index.translog.interval";
INDEX_TRANSLOG_FLUSH_THRESHOLD_OPS = "index.translog.flush_threshold_ops";
INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE = "index.translog.flush_threshold_size";
INDEX_TRANSLOG_FLUSH_THRESHOLD_PERIOD = "index.translog.flush_threshold_period";
INDEX_TRANSLOG_DISABLE_FLUSH = "index.translog.disable_flush";

当服务器恢复时发现有存在的translog日志，就会进入TRANSLOG 阶段进行replay。translog 的recovery 是走的标准的InternalEngine.create/update等方法，并且还会再写translog，同时还有一个影响性能的地方是很多数据可能已经存在，会走update操作，所以性能还是非常差的。这个目前能够想到的解决办法是调整flush日志的频率，保证存在的translog 尽量的少。 上面的话可以看出有三个控制选项：

//每隔interval的时间，就去检查下面三个条件决定是不是要进行flush，
//默认5s。时间过长，会超出下面阈值比较大。
index.translog.interval 

//超过多少条日志后需要flush,默认Int的最大值
index.translog.flush_threshold_ops 

//定时flush，默认30m 可动态设置
index.translog.flush_threshold_period

//translog 大小超过多少后flush,默认512m  
index.translog.flush_threshold_size

本质上translog的恢复速度和条数的影响关系更大些，所以建议大家设置下 index.translog.flush_threshold_ops，比如多少条就一定要flush,否则积累的太多，
出现故障，恢复就慢了。这些参数都可以动态设置，但建议放到配置文件。

Elasticsearch Recovery详解

Elasticsearch Recovery详解

基础知识点

在Eleasticsearch中recovery指的就是一个索引的分片分配到另外一个节点的过程；一般在快照恢复、索引副本数变更、节点故障、节点重启时发生。由于master保存整个集群的状态信息，因此可以判断出哪些shard需要做再分配，以及分配到哪个结点，例如:

1. 如果某个shard主分片在，副分片所在结点挂了，那么选择另外一个可用结点，将副分片分配(allocate)上去，然后进行主从分片的复制。
2. 如果某个shard的主分片所在结点挂了，副分片还在，那么将副分片升级为主分片，然后做主从分片复制。
3. 如果某个shard的主副分片所在结点都挂了，则暂时无法恢复，等待持有相关数据的结点重新加入集群后，从该结点上恢复主分片，再选择另外的结点复制副分片。

 
正常情况下，我们可以通过ES的health的API接口，查看整个集群的健康状态和整个集群数据的完整性：

状态及含义如下：

• green: 所有的shard主副分片都是正常的；
• yellow: 所有shard的主分片都完好，部分副分片没有或者不完整，数据完整性依然完好；
• red: 某些shard的主副分片都没有了，对应的索引数据不完整。

 
recovery过程要消耗额外的资源，CPU、内存、结点之间的网络带宽等等。 这些额外的资源消耗，有可能会导致集群的服务性能下降，或者一部分功能暂时不可用。了解一些recovery的过程和相关的配置参数，对于减小recovery带来的资源消耗，加快集群恢复过程都是很有帮助的。
 

减少集群Full Restart造成的数据来回拷贝

ES集群可能会有整体重启的情况，比如需要升级硬件、升级操作系统或者升级ES大版本。重启所有结点可能带来的一个问题: 某些结点可能先于其他结点加入集群， 先加入集群的结点可能已经可以选举好master，并立即启动了recovery的过程，由于这个时候整个集群数据还不完整，master会指示一些结点之间相互开始复制数据。 那些晚到的结点，一旦发现本地的数据已经被复制到其他结点，则直接删除掉本地“失效”的数据。 当整个集群恢复完毕后，数据分布不均衡，显然是不均衡的，master会触发rebalance过程，将数据在节点之间挪动。整个过程无谓消耗了大量的网络流量；合理设置recovery相关参数则可以防范这种问题的发生。

gateway.expected_nodes
gateway.expected_master_nodes
gateway.expected_data_nodes

以上三个参数是说集群里一旦有多少个节点就立即开始recovery过程。 不同之处在于，第一个参数指的是master或者data都算在内，而后面两个参数则分指master和data node。
 
在期待的节点数条件满足之前, recovery过程会等待gateway.recover_after_time (默认5分钟) 这么长时间，一旦等待超时，则会根据以下条件判断是否启动:

gateway.recover_after_nodes
gateway.recover_after_master_nodes
gateway.recover_after_data_nodes

 
举例来说，对于一个有10个data node的集群，如果有以下的设置:

gateway.expected_data_nodes: 10
gateway.recover_after_time: 5m
gateway.recover_after_data_nodes: 8

那么集群5分钟以内10个data node都加入了，或者5分钟以后8个以上的data node加入了，都会立即启动recovery过程。
 

减少主副本之间的数据复制

如果不是full restart，而是重启单个data node，仍然会造成数据在不同结点之间来回复制。为避免这个问题，可以在重启之前，先关闭集群的shard allocation:

然后在节点重启完成加入集群后，再重新打开:

这样在节点重启完成后，尽量多的从本地直接恢复数据。

但是在ES1.6版本之前，即使做了以上措施，仍然会发现有大量主副本之间的数据拷贝。从表面去看，这点很让人不能理解。 主副本数据完全一致，ES应该直接从副本本地恢复数据就好了，为什么要重新从主片再复制一遍呢？ 原因在于recovery是简单对比主副本的segment file来判断哪些数据一致可以本地恢复，哪些不一致需要远端拷贝的。而不同节点的segment merge是完全独立运行的，可能导致主副本merge的深度不完全一样，从而造成即使文档集完全一样，产生的segment file却不完全一样。
 
为了解决这个问题，ES1.6版本以后加入了synced flush的新特性。 对于5分钟没有更新过的shard，会自动synced flush一下，实质是为对应的shard加了一个synced flush ID。这样当重启节点的时候，先对比一下shard的synced flush ID，就可以知道两个shard是否完全相同，避免了不必要的segment file拷贝，极大加快了冷索引的恢复速度。
 
需要注意的是synced flush只对冷索引有效，对于热索引（5分钟内有更新的索引）没有作用。 如果重启的结点包含有热索引，那么还是免不了大量的文件拷贝。因此在重启一个结点之前，最好按照以下步骤执行，recovery几乎可以瞬间完成:

1. 暂停数据写入程序
2. 关闭集群shard allocation
3. 手动执行POST /_flush/synced
4. 重启节点
5. 重新开启集群shard allocation 
6. 等待recovery完成，集群health status变成green
7. 重新开启数据写入程序

 

特大热索引为何恢复慢

对于冷索引，由于数据不再更新，利用synced flush特性，可以快速直接从本地恢复数据。 而对于热索引，特别是shard很大的热索引，除了synced flush派不上用场需要大量跨节点拷贝segment file以外，translog recovery是导致慢的更重要的原因。
 
从主片恢复数据到副片需要经历3个阶段:

1. 对主片上的segment file做一个快照，然后拷贝到复制片分配到的结点。数据拷贝期间，不会阻塞索引请求，新增索引操作记录到translog里。
2. 对translog做一个快照，此快照包含第一阶段新增的索引请求，然后重放快照里的索引操作。此阶段仍然不阻塞索引请求，新增索引操作记录到translog里。
3. 为了能达到主副片完全同步，阻塞掉新索引请求，然后重放阶段二新增的translog操作。

 
可见，在recovery完成之前，translog是不能够被清除掉的（禁用掉正常运作期间后台的flush操作）。如果shard比较大，第一阶段耗时很长，会导致此阶段产生的translog很大。重放translog比起简单的文件拷贝耗时要长得多，因此第二阶段的translog耗时也会显著增加。等到第三阶段，需要重放的translog可能会比第二阶段还要多。 而第三阶段是会阻塞新索引写入的，在对写入实时性要求很高的场合，就会非常影响用户体验。 因此，要加快大的热索引恢复速度，最好的方式是遵从上一节提到的方法: 暂停新数据写入，手动sync flush，等待数据恢复完成后，重新开启数据写入，这样可以将数据延迟影响可以降到最低。
 
万一遇到Recovery慢，想知道进度怎么办呢？ CAT Recovery API可以显示详细的recovery各个阶段的状态。 这个API怎么用就不在这里赘述了，参考:  CAT Recovery。
 

其他Recovery相关的专家级设置

还有其他一些专家级的设置（参见：  recovery）可以影响recovery的速度，但提升速度的代价是更多的资源消耗，因此在生产集群上调整这些参数需要结合实际情况谨慎调整，一旦影响应用要立即调整回来。 对于搜索并发量要求高，延迟要求低的场合，默认设置一般就不要去动了。 对于日志实时分析类对于搜索延迟要求不高，但对于数据写入延迟期望比较低的场合，可以适当调大indices.recovery.max_bytes_per_sec，提升recovery速度，减少数据写入被阻塞的时长。
 
最后要说的一点是ES的版本迭代很快，对于Recovery的机制也在不断的优化中。 其中有一些版本甚至引入了一些bug，比如在ES1.4.x有严重的translog recovery bug，导致大的索引trans log recovery几乎无法完成 （ issue #9226）  。因此实际使用中如果遇到问题，最好在Github的issue list里搜索一下，看是否使用的版本有其他人反映同样的问题。

本文转自张昺华-sky博客园博客，原文链接：http://www.cnblogs.com/bonelee/p/8135878.html ，如需转载请自行联系原作者