《Elasticsearch 源码解析与优化实战》第10章：索引恢复流程分析

IndicesClusterStateService#applyClusterState
    -> createOrUpdateShards()
    -> createShard()
    -> IndicesService.createShard()
    -> IndexShard.startRecovery()

public void startRecovery(RecoveryState recoveryState, PeerRecoveryTargetService recoveryTargetService,
                              PeerRecoveryTargetService.RecoveryListener recoveryListener, RepositoriesService repositoriesService,
                              BiConsumer<String, MappingMetaData> mappingUpdateConsumer,
                              IndicesService indicesService) {
   
    assert recoveryState.getRecoverySource().equals(shardRouting.recoverySource());
    switch (recoveryState.getRecoverySource().getType()) {
        case EMPTY_STORE:
        case EXISTING_STORE:
            markAsRecovering("from store", recoveryState); // mark the shard as recovering on the cluster state thread
            threadPool.generic().execute(() -> {
                try {
                    //主分片从本地恢复
                    if (recoverFromStore()) {
                        recoveryListener.onRecoveryDone(recoveryState);
                    }
                } catch (Exception e) {
                    recoveryListener.onRecoveryFailure(recoveryState,
                            new RecoveryFailedException(recoveryState, null, e), true);
                }
            });
            break;
        case PEER:
            try {
                markAsRecovering("from " + recoveryState.getSourceNode(), recoveryState);
                //副分片从远程主分片恢复
                recoveryTargetService.startRecovery(this, recoveryState.getSourceNode(), recoveryListener);
            } catch (Exception e) {
                failShard("corrupted preexisting index", e);
                recoveryListener.onRecoveryFailure(recoveryState,
                        new RecoveryFailedException(recoveryState, null, e), true);
            }
            break;
        case SNAPSHOT:
            markAsRecovering("from snapshot", recoveryState); // mark the shard as recovering on the cluster state thread
            SnapshotRecoverySource recoverySource = (SnapshotRecoverySource) recoveryState.getRecoverySource();
            threadPool.generic().execute(() -> {
                try {
                    final Repository repository = repositoriesService.repository(recoverySource.snapshot().getRepository());
                    //从快照恢复
                    if (restoreFromRepository(repository)) {
                        recoveryListener.onRecoveryDone(recoveryState);
                    }
                } catch (Exception e) {
                    recoveryListener.onRecoveryFailure(recoveryState,
                            new RecoveryFailedException(recoveryState, null, e), true);
                }
            });
            break;
        case LOCAL_SHARDS:
            final IndexMetaData indexMetaData = indexSettings().getIndexMetaData();
            final Index resizeSourceIndex = indexMetaData.getResizeSourceIndex();
            final List<IndexShard> startedShards = new ArrayList<>();
            final IndexService sourceIndexService = indicesService.indexService(resizeSourceIndex);
            final Set<ShardId> requiredShards;
            final int numShards;
            if (sourceIndexService != null) {
                requiredShards = IndexMetaData.selectRecoverFromShards(shardId().id(),
                        sourceIndexService.getMetaData(), indexMetaData.getNumberOfShards());
                for (IndexShard shard : sourceIndexService) {
                    if (shard.state() == IndexShardState.STARTED && requiredShards.contains(shard.shardId())) {
                        startedShards.add(shard);
                    }
                }
                numShards = requiredShards.size();
            } else {
                numShards = -1;
                requiredShards = Collections.emptySet();
            }
           
            if (numShards == startedShards.size()) {
                assert requiredShards.isEmpty() == false;
                markAsRecovering("from local shards", recoveryState); // mark the shard as recovering on the cluster state thread
                threadPool.generic().execute(() -> {
                    try {
                        //从本节点其他分片恢复(shrink时)
                        if (recoverFromLocalShards(mappingUpdateConsumer, startedShards.stream()
                                .filter((s) -> requiredShards.contains(s.shardId())).collect(Collectors.toList()))) {
                            recoveryListener.onRecoveryDone(recoveryState);
                        }
                    } catch (Exception e) {
                        recoveryListener.onRecoveryFailure(recoveryState,
                                new RecoveryFailedException(recoveryState, null, e), true);
                    }
                });
            } else {
                final RuntimeException e;
                if (numShards == -1) {
                    e = new IndexNotFoundException(resizeSourceIndex);
                } else {
                    e = new IllegalStateException("not all required shards of index " + resizeSourceIndex
                            + " are started yet, expected " + numShards + " found " + startedShards.size() + " can't recover shard "
                            + shardId());
                }
                throw e;
            }
            break;
        default:
            throw new IllegalArgumentException("Unknown recovery source " + recoveryState.getRecoverySource());
    }
}

markAsRecovering("from store", recpveryState);

public IndexShardState markAsRecovering(String reason, RecoveryState recoveryState) throws IndexShardStartedException,
        IndexShardRelocatedException, IndexShardRecoveringException, IndexShardClosedException {
    synchronized (mutex) {
        if (state == IndexShardState.CLOSED) {
            throw new IndexShardClosedException(shardId);
        }
        if (state == IndexShardState.STARTED) {
            throw new IndexShardStartedException(shardId);
        }
        if (state == IndexShardState.RECOVERING) {
            throw new IndexShardRecoveringException(shardId);
        }
        if (state == IndexShardState.POST_RECOVERY) {
            throw new IndexShardRecoveringException(shardId);
        }
        this.recoveryState = recoveryState;
        return changeState(IndexShardState.RECOVERING, reason);
    }
}

threadPool.generic().execute(() -> {    
    try {
        //主分片从本地恢复
        if (recoverFromStore()) {
            //恢复成功
            //向Master发送action为 internal:cluster/shard/started的RPC请求
            recoveryListener.onRecoveryDone(recoveryState);
        }
    } catch (Exception e) {
        //恢复失败
        //关闭Engin，向Master发送 internal:cluster/shard/failure的RPC请求
        recoveryListener.onRecoveryFailure(recoveryState,
                            new RecoveryFailedException(recoveryState, null, e), true);
    }
});

public void prepareForIndexRecovery() {
   
    if (state != IndexShardState.RECOVERING) {
        throw new IndexShardNotRecoveringException(shardId, state);
    }
    recoveryState.setStage(RecoveryState.Stage.INDEX);
    assert currentEngineReference.get() == null;
}

private void internalRecoverFromStore(IndexShard indexShard) throws IndexShardRecoveryException {
    final RecoveryState recoveryState = indexShard.recoveryState();
    final boolean indexShouldExists = recoveryState.getRecoverySource().getType() != RecoverySource.Type.EMPTY_STORE;
    //进入INDEX阶段
    indexShard.prepareForIndexRecovery();
    SegmentInfos si = null;
    final Store store = indexShard.store();
    store.incRef();
    try {
        try {
            store.failIfCorrupted();
            try {
                //获取Lucene最后一次提交的分段信息，得到其中的版本号
                si = store.readLastCommittedSegmentsInfo();
            } catch (Exception e) {
                String files = "_unknown_";
                try {
                    files = Arrays.toString(store.directory().listAll());
                } catch (Exception inner) {
                    inner.addSuppressed(e);
                    files += " (failure=" + ExceptionsHelper.detailedMessage(inner) + ")";
                }
                if (indexShouldExists) {
                    throw new IndexShardRecoveryException(shardId,
                            "shard allocated for local recovery (post api), should exist, but doesn't, current files: " + files, e);
                }
            }
            if (si != null && indexShouldExists == false) {
                // it exists on the directory, but shouldn't exist on the FS, its a leftover (possibly dangling)
                // its a "new index create" API, we have to do something, so better to clean it than use same data
                logger.trace("cleaning existing shard, shouldn't exists");
                Lucene.cleanLuceneIndex(store.directory());
                si = null;
            }
        } catch (Exception e) {
            throw new IndexShardRecoveryException(shardId, "failed to fetch index version after copying it over", e);
        }
        if (recoveryState.getRecoverySource().getType() == RecoverySource.Type.LOCAL_SHARDS) {
            assert indexShouldExists;
            bootstrap(indexShard, store);
            writeEmptyRetentionLeasesFile(indexShard);
        } else if (indexShouldExists) {
            if (recoveryState.getRecoverySource().shouldBootstrapNewHistoryUUID()) {
                store.bootstrapNewHistory();
                writeEmptyRetentionLeasesFile(indexShard);
            }
            // since we recover from local, just fill the files and size
            try {
                final RecoveryState.Index index = recoveryState.getIndex();
                if (si != null) {
                    addRecoveredFileDetails(si, store, index);
                }
            } catch (IOException e) {
                logger.debug("failed to list file details", e);
            }
        } else {
              store.createEmpty(indexShard.indexSettings().getIndexVersionCreated().luceneVersion);
            final String translogUUID = Translog.createEmptyTranslog(
                    indexShard.shardPath().resolveTranslog(), SequenceNumbers.NO_OPS_PERFORMED, shardId,
                    indexShard.getPendingPrimaryTerm());
            store.associateIndexWithNewTranslog(translogUUID);
                writeEmptyRetentionLeasesFile(indexShard);
        }
        //从tanslog恢复数据
        indexShard.openEngineAndRecoverFromTranslog();
        indexShard.getEngine().fillSeqNoGaps(indexShard.getPendingPrimaryTerm());
        //进入FINALIZE状态
        indexShard.finalizeRecovery();
        //进入DONE阶段
        indexShard.postRecovery("post recovery from shard_store");
    } catch (EngineException | IOException e) {
        throw new IndexShardRecoveryException(shardId, "failed to recover from gateway", e);
    } finally {
        store.decRef();
    }
}

index.shard.check_on_startup=true

/**
 * 重放translog日志
 * @param translogRecoveryRunner
 * @param recoverUpToSeqNo
 * @throws IOException
 */
private void recoverFromTranslogInternal(TranslogRecoveryRunner translogRecoveryRunner, long recoverUpToSeqNo) throws IOException {
    Translog.TranslogGeneration translogGeneration = translog.getGeneration();
    final int opsRecovered;
    //根据最后一次提交的信息生成translog快照
    final long translogFileGen = Long.parseLong(lastCommittedSegmentInfos.getUserData().get(Translog.TRANSLOG_GENERATION_KEY));
    try (Translog.Snapshot snapshot = translog.newSnapshotFromGen(
            new Translog.TranslogGeneration(translog.getTranslogUUID(), translogFileGen), recoverUpToSeqNo)) {
        //重放这些日志
        opsRecovered = translogRecoveryRunner.run(this, snapshot);
    } catch (Exception e) {
        throw new EngineException(shardId, "failed to recover from translog", e);
    }
    // flush if we recovered something or if we have references to older translogs
    // note: if opsRecovered == 0 and we have older translogs it means they are corrupted or 0 length.
    assert pendingTranslogRecovery.get() : "translogRecovery is not pending but should be";
    pendingTranslogRecovery.set(false); // we are good - now we can commit
    //将重放后新生成的数据刷入硬盘
    if (opsRecovered > 0) {
        logger.trace("flushing post recovery from translog. ops recovered [{}]. committed translog id [{}]. current id [{}]",
                     opsRecovered, translogGeneration == null ? null :
                    translogGeneration.translogFileGeneration, translog.currentFileGeneration());
        commitIndexWriter(indexWriter, translog, null);
        refreshLastCommittedSegmentInfos();
        refresh("translog_recovery");
    }
    translog.trimUnreferencedReaders();
}

@Override
public int run(Engine engine, Translog.Snapshot snapshot) throws IOException {
    int opsRecovered = 0;
    Translog.Operation operation;
    while ((operation = snapshot.next()) != null) {
        //执行具体的写操作
        applyOperation(engine, convertToEngineOp(operation, Engine.Operation.Origin.LOCAL_TRANSLOG_RECOVERY));
        opsRecovered++;
        appliedOperations.incrementAndGet();
    }
    engine.syncTranslog();
    return opsRecovered;
}

public void finalizeRecovery() {
    recoveryState().setStage(RecoveryState.Stage.FINALIZE);
    Engine engine = getEngine();
    //执行refresh操作
    engine.refresh("recovery_finalization");
    engine.config().setEnableGcDeletes(true);
}

public void postRecovery(String reason) throws IndexShardStartedException, IndexShardRelocatedException, IndexShardClosedException {
   
    synchronized (postRecoveryMutex) {
        //再次执行刷新
        getEngine().refresh("post_recovery");
        synchronized (mutex) {
            if (state == IndexShardState.CLOSED) {
                throw new IndexShardClosedException(shardId);
            }
            if (state == IndexShardState.STARTED) {
                throw new IndexShardStartedException(shardId);
            }
            recoveryState.setStage(RecoveryState.Stage.DONE);
            //更新分片状态
            changeState(IndexShardState.POST_RECOVERY, reason);
        }
    }
}

sendShardAction(SHARD_STARTED_ACTION_NAME, currentState, entry, listener);

private void failAndRemoveShard(ShardRouting shardRouting, boolean sendShardFailure, String message, @Nullable Exception failure,
                                    ClusterState state) {
    try {
        AllocatedIndex<? extends Shard> indexService = indicesService.indexService(shardRouting.shardId().getIndex());
        if (indexService != null) {
            //关闭shard
            indexService.removeShard(shardRouting.shardId().id(), message);
        }
    } catch (ShardNotFoundException e) {
        // the node got closed on us, ignore it
    } catch (Exception inner) {
        inner.addSuppressed(failure);
        logger.warn(() -> new ParameterizedMessage(
                    "[{}][{}] failed to remove shard after failure ([{}])",
                    shardRouting.getIndexName(),
                    shardRouting.getId(),
                    message),
                inner);
    }
    if (sendShardFailure) {
        //向Master 节点发生 SHARD_FAILED_ACTION_NAME 请求
        sendFailShard(shardRouting, message, failure, state);
    }
}

public void startRecovery(final IndexShard indexShard, final DiscoveryNode sourceNode, final RecoveryListener listener) {
    final long recoveryId = onGoingRecoveries.startRecovery(indexShard, sourceNode, listener, recoverySettings.activityTimeout());
    threadPool.generic().execute(new RecoveryRunner(recoveryId));
}

public void prepareForIndexRecovery() {
    if (state != IndexShardState.RECOVERING) {
        throw new IndexShardNotRecoveringException(shardId, state);
    }
    //设置状态为index
    recoveryState.setStage(RecoveryState.Stage.INDEX);
    assert currentEngineReference.get() == null;
}

private void doRecovery(final long recoveryId) {
    cancellableThreads.executeIO(() ->
                transportService.submitRequest(request.sourceNode(), PeerRecoverySourceService.Actions.START_RECOVERY, request,
                    new TransportResponseHandler<RecoveryResponse>() {
                        @Override
                        public void handleResponse(RecoveryResponse recoveryResponse) {
                            final TimeValue recoveryTime = new TimeValue(timer.time());
                            // do this through ongoing recoveries to remove it from the collection
                            //响应成功，设置为DONE阶段
                            onGoingRecoveries.markRecoveryAsDone(recoveryId);
                        }
                        @Override
                        public void handleException(TransportException e) {
                            handleException.accept(e);
                        }

                        @Override
                        public String executor() {
                            return ThreadPool.Names.GENERIC;
                        }

                        @Override
                        public RecoveryResponse read(StreamInput in) throws IOException {
                            return new RecoveryResponse(in);
                        }
                    })
            );
}

public long recoverLocallyUpToGlobalCheckpoint() {
        assert Thread.holdsLock(mutex) == false : "recover locally under mutex";
        if (state != IndexShardState.RECOVERING) {
            throw new IndexShardNotRecoveringException(shardId, state);
        }
        assert recoveryState.getStage() == RecoveryState.Stage.INDEX : "unexpected recovery stage [" + recoveryState.getStage() + "]";
        assert routingEntry().recoverySource().getType() == RecoverySource.Type.PEER : "not a peer recovery [" + routingEntry() + "]";
        final Optional<SequenceNumbers.CommitInfo> safeCommit;
        final long globalCheckpoint;
        try {
            final String translogUUID = store.readLastCommittedSegmentsInfo().getUserData().get(Translog.TRANSLOG_UUID_KEY);
            globalCheckpoint = Translog.readGlobalCheckpoint(translogConfig.getTranslogPath(), translogUUID);
            safeCommit = store.findSafeIndexCommit(globalCheckpoint);
        } catch (org.apache.lucene.index.IndexNotFoundException e) {
            logger.trace("skip local recovery as no index commit found");
            return UNASSIGNED_SEQ_NO;
        } catch (Exception e) {
            logger.debug("skip local recovery as failed to find the safe commit", e);
            return UNASSIGNED_SEQ_NO;
        }
        if (safeCommit.isPresent() == false) {
            logger.trace("skip local recovery as no safe commit found");
            return UNASSIGNED_SEQ_NO;
        }
        assert safeCommit.get().localCheckpoint <= globalCheckpoint : safeCommit.get().localCheckpoint + " > " + globalCheckpoint;
        try {
            //验证索引
            maybeCheckIndex(); // check index here and won't do it again if ops-based recovery occurs
            //进入translog阶段
            recoveryState.setStage(RecoveryState.Stage.TRANSLOG);
            if (safeCommit.get().localCheckpoint == globalCheckpoint) {
                logger.trace("skip local recovery as the safe commit is up to date; safe commit {} global checkpoint {}",
                    safeCommit.get(), globalCheckpoint);
                recoveryState.getTranslog().totalLocal(0);
                return globalCheckpoint + 1;
            }
            if (indexSettings.getIndexMetaData().getState() == IndexMetaData.State.CLOSE ||
                IndexMetaData.INDEX_BLOCKS_WRITE_SETTING.get(indexSettings.getSettings())) {
                logger.trace("skip local recovery as the index was closed or not allowed to write; safe commit {} global checkpoint {}",
                    safeCommit.get(), globalCheckpoint);
                recoveryState.getTranslog().totalLocal(0);
                return safeCommit.get().localCheckpoint + 1;
            }
            try {
                final Engine.TranslogRecoveryRunner translogRecoveryRunner = (engine, snapshot) -> {
                    recoveryState.getTranslog().totalLocal(snapshot.totalOperations());
                    final int recoveredOps = runTranslogRecovery(engine, snapshot, Engine.Operation.Origin.LOCAL_TRANSLOG_RECOVERY,
                        recoveryState.getTranslog()::incrementRecoveredOperations);
                    recoveryState.getTranslog().totalLocal(recoveredOps); // adjust the total local to reflect the actual count
                    return recoveredOps;
                };
                innerOpenEngineAndTranslog(() -> globalCheckpoint);
                //从translog恢复
                getEngine().recoverFromTranslog(translogRecoveryRunner, globalCheckpoint);
                logger.trace("shard locally recovered up to {}", getEngine().getSeqNoStats(globalCheckpoint));
            } finally {
                synchronized (engineMutex) {
                    IOUtils.close(currentEngineReference.getAndSet(null));
                }
            }
        } catch (Exception e) {
            logger.debug(new ParameterizedMessage("failed to recover shard locally up to global checkpoint {}", globalCheckpoint), e);
            return UNASSIGNED_SEQ_NO;
        }
        try {
            // we need to find the safe commit again as we should have created a new one during the local recovery
            final Optional<SequenceNumbers.CommitInfo> newSafeCommit = store.findSafeIndexCommit(globalCheckpoint);
            assert newSafeCommit.isPresent() : "no safe commit found after local recovery";
            return newSafeCommit.get().localCheckpoint + 1;
        } catch (Exception e) {
            logger.debug(new ParameterizedMessage(
                "failed to find the safe commit after recovering shard locally up to global checkpoint {}", globalCheckpoint), e);
            return UNASSIGNED_SEQ_NO;
        }
}

public void finalizeRecovery(final long globalCheckpoint, final long trimAboveSeqNo, ActionListener<Void> listener) {
    ActionListener.completeWith(listener, () -> {
        //更新全局检查点，还没有进入FINALIZE阶段
        indexShard.updateGlobalCheckpointOnReplica(globalCheckpoint, "finalizing recovery");
        // Persist the global checkpoint.
        indexShard.sync();
        indexShard.persistRetentionLeases();
        if (trimAboveSeqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) {
            indexShard.rollTranslogGeneration();
            indexShard.afterWriteOperation();
            indexShard.trimOperationOfPreviousPrimaryTerms(trimAboveSeqNo);
        }
        if (hasUncommittedOperations()) {
            indexShard.flush(new FlushRequest().force(true).waitIfOngoing(true));
        }
        //进入FINALIZE阶段
        indexShard.finalizeRecovery();
        return null;
    });
}

/**
* 处理副分片的索引恢复响应
*/
class StartRecoveryTransportRequestHandler implements TransportRequestHandler<StartRecoveryRequest> {
    @Override
    public void messageReceived(final StartRecoveryRequest request, final TransportChannel channel, Task task) throws Exception {
        recover(request, new ChannelActionListener<>(channel, Actions.START_RECOVERY, request));
    }
}

//获取一个保留锁，使得translog不被清理
final Closeable retentionLock = shard.acquireRetentionLock();

final boolean isSequenceNumberBasedRecovery
                = request.startingSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO
                && isTargetSameHistory()
                && shard.hasCompleteHistoryOperations("peer-recovery", request.startingSeqNo())
                && (useRetentionLeases == false
                || (retentionLeaseRef.get() != null && retentionLeaseRef.get().retainingSequenceNumber() <= request.startingSeqNo()));

#单个节点执行副分片recovery时的最大并发数，默认2
cluster.routing.allocation.node_concurrent_recoveries: 2

#节点间复制数据是的限速，适当提高或取消限速，默认40M，0不限速
indices.recovery.max_bytes_per_sec: 40M

#单个节点主分片recovery时的最大并发数，默认4，多个数据磁盘的情况下，可以适当提高
cluster.routing.allocation.node_initial_primaries_recoveries: 4

#重启集群前，先停止写入端，执行sync flush，让恢复过程有机会跳过phase1

#适当多保留一些translog
index.translog.retention.size: 512MB
index.translog.retention.age: 12h

#合并Lucene分段，对于冷索引甚至不再更新的索引执行_forcemerge
#较少的Lucene分段可以提升恢复效率，例如，减少对比，降低文件传输请求数量

expectedVersion = 写副分片请求中的version = 写主分片成功后的version
通过下面的方法判断当前操作的版本号是否低于Lucene中的版本号:
EXTERNAL( (byte) 1) {
    @Override
    public boolean isVersionConflictForWrites (long currentVersion, long expectedVersion, boolean deleted) {
        if (currentVersion == Versions .NOT_ FOUND) {
            return false;
        }
        if (expectedVersion == Versions . MATCH ANY) {
            return true;
        }
        if (currentVersion >= expectedVersion) {
            return true ;
        }
        return false;
    }
}
如果translog重放的操作在写一条“老”数据，则compareOpToLuceneDocBasedOnVersions会返回OpVsLuceneDocStatus.OP_STALE_OR_EQUAL。

private OpVsLuceneDocStatus compa reOpToLucene DocBasedOnVersions (final Operation оp) throws IOException {
    final VersionValue versionValue = resolveDocVersion(op) ;
    if (versionValue == null) {
        return OpVs LuceneDocSta tus . LUCENE_ DOC_ NOT_ FOUND;
    } else if {
        return op.versionType().isVersionConflictForWrites (versionValue.version, op.version(), versionValue.isDelete()) ? OpVsLuceneDocStatus.OP_STALE_OR_EQUAL : OpVsLuceneDocStatus.OP_NEWER;
    }
}

// non-primary mode (i.e.n replica or recovery)
plan = planIndexingAsNonPrimary (index) ;

curl localhost: 9200/_cat/recovery
website  O  30ms  peer done  127.0.0.1 fc6s0S0 127 .0.0.1 iV6xmvW n/a n/a O O 0.0% 0000.0% 0 0 0 100.0%
website  0  30ms  existing_ store done n/a n/a 127.0.0.1 fc6s0S0 n/a n/a0OO100.0%14OO100.0%13825OO100.0%

使用?v参数可以显示列名: GET_ cat/ recovery?v

curl "localhost:9200/website/_recovery?pretty" {
    "id" : 4
    "type" :
    "PEER",
    "stage" : "DONE",
    "primary" : false,
    "source" : {
        ......
    }
    "target" : {
        ......
    }
    "index" : {
        "size" : {
            "total_ in_ bytes" : O,
            "reused_ in_ bytes" : O,
            "recovered_ in_ bytes" : 0,
            "percent" : "0.0%"
        }
        "total_ time_ in_ millis" : 6,
        "source_ throttle_ time_ in_ millis" : 0,
        "target_ throttle_ time_ in_ millis" : 0
    }
    "translog" : {
        " recovered" : 0 ,
        "total" : 0,
        "percent" : "100.0%",
        "total on start" : - 1 ,
        "total time in millis" : 13
    },
    "verify_ index" : {
        "check index time in millis" : O,
        "total time in millis" : 0
    }
}       

curl -s "127.0.0.1:9200/_all/_stats?level=shards&pretty" | grep sync_id | wC -1

查看 recovery 状态的命令:

curl localhost:9200/{index}/_stats?level=shards&pretty
curl localhost:9200/{index}/_recovery?pretty&human&detailed=true
curl localhost:9200/_cat/recovery