ES7.9.1 publish原理详解

ES集群中，master负责实施维护集群元数据的更新，然后再分发给data节点，分发的过程就是publish，也就是本文的重点。本文将以索引创建的流程来讲解这个过程。

master首先创建新的集群元数据

当master接收到创建索引的请求后，首先进入如下MetadataCreateIndexService.onlyCreateIndex()函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
private void onlyCreateIndex(final CreateIndexClusterStateUpdateRequest request,
                                final ActionListener<ClusterStateUpdateResponse> listener) {
       clusterService.submitStateUpdateTask(
           "create-index [" + request.index() + "], cause [" + request.cause() + "]",
           // 每个create使用这惟一的key
           new AckedClusterStateUpdateTask<ClusterStateUpdateResponse>(Priority.URGENT, request, listener) { 
               protected ClusterStateUpdateResponse newResponse(boolean acknowledged) {
                   return new ClusterStateUpdateResponse(acknowledged);
               }
               @Override
               public ClusterState execute(ClusterState currentState) throws Exception {
                   // 产生新的ClusterState，还没成为本地的ClusterState
                   return applyCreateIndexRequest(currentState, request, false); 
               }
           });
   }

master进行任何更新State的操作时，都会调用submitStateUpdateTask() -> ClusterService.submitStateUpdateTasks() -> MasterService.submitStateUpdateTasks() -> TaskBatcher.submitTasks()，在submitTasks()中会对tasks分类合并，有些task可以合并执行以加快速度:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public void submitTasks(List<? extends BatchedTask> tasks, @Nullable TimeValue timeout) throws EsRejectedExecutionException {
    final BatchedTask firstTask = tasks.get(0);
    final Map<Object, BatchedTask> tasksIdentity = tasks.stream().collect(Collectors.toMap( 
        BatchedTask::getTask,
        Function.identity(),
        (a, b) -> { throw new IllegalStateException("cannot add duplicate task: " + a); },
        IdentityHashMap::new));
    //比如多个shartedShard过后来，都会先锁着。
    synchronized (tasksPerBatchingKey) { 
        LinkedHashSet<BatchedTask> existingTasks = tasksPerBatchingKey.computeIfAbsent(firstTask.batchingKey,
            k -> new LinkedHashSet<>(tasks.size()));
        // 对应存量的
        existingTasks.addAll(tasks);
    }
    if (timeout != null) {
        threadExecutor.execute(firstTask, timeout, () -> onTimeoutInternal(tasks, timeout)); // 这个类就是[node][masterService#updateTask][T#1]
    } else {
        threadExecutor.execute(firstTask);
    }
}

可以看到，该函数会去检查task的batchingKey是否一致，若一致的话，放在相同的batchingKey下，最常见的startShard/failShard元数据更新就是可以合并执行，batchingKey相同的前提是调用ClusterService.submitStateUpdateTask时，使用了相同的ClusterStateTaskExecutor，而startShard使用了全局唯一的ShardStartedClusterStateTaskExecutor作为key。在create中，我们明显可以看到每个index创建都会产生新的AckedClusterStateUpdateTask作为batchingKey，索引创建流程只能逐个全局同步。这里会从线程池中产生我们熟悉的[node][masterService#updateTask][T#1]线程进行构造新的ClusterStae。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
private void runTasks(TaskInputs taskInputs) {
    final ClusterState previousClusterState = state();
    final long computationStartTime = threadPool.relativeTimeInMillis();
     // 去执行每个task如何产生新的ClusterState
    final TaskOutputs taskOutputs = calculateTaskOutputs(taskInputs, previousClusterState);
    taskOutputs.notifyFailedTasks();
    final TimeValue computationTime = getTimeSince(computationStartTime);
    logExecutionTime(computationTime, "compute cluster state update", summary);
    if (taskOutputs.clusterStateUnchanged()) {
        ......
    } else {
        final ClusterState newClusterState = taskOutputs.newClusterState;
        final long publicationStartTime = threadPool.relativeTimeInMillis();
        try {
            ClusterChangedEvent clusterChangedEvent = new ClusterChangedEvent(summary, newClusterState, previousClusterState);
            // 这里回去进行真正广播
            publish(clusterChangedEvent, taskOutputs, publicationStartTime);  
        } catch (Exception e) {
            handleException(summary, publicationStartTime, newClusterState, e);
        }
    }
}

这个函数主要做了如下事情：
1.调用calculateTaskOutputs以产生新的ClusterState。产生过程可以参考前面自定义类的AckedClusterStateUpdateTask.execute调用applyCreateIndexRequest产生新的集群状态，这里将不是本文重点。
2.调用publish()进行全局广播，全局广播包括主master本身。

master全局广播

预处理

进行广播前，master还会做如下预处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void publish(ClusterChangedEvent clusterChangedEvent, ActionListener<Void> publishListener, AckListener ackListener) {
    final PublishRequest publishRequest = coordinationState.get().handleClientValue(clusterState);
    final CoordinatorPublication publication = new CoordinatorPublication(publishRequest, publicationContext,
           new ListenableFuture<>(), ackListener, publishListener);
    currentPublication = Optional.of(publication);
    // 从最新的集群状态中的所有的节点
    final DiscoveryNodes publishNodes = publishRequest.getAcceptedState().nodes();
    // master仅仅更新下数据节点列表。当master接收到data的心跳时会校验
    leaderChecker.setCurrentNodes(publishNodes); 
    // master会更新本地维持的、对数据节点的心跳连接
    followersChecker.setCurrentNodes(publishNodes);
    // master更新本地需要维护的、数据节点响应的ClusterVersion版本号，便于剔除对版本超时未同步的节点。 
    lagDetector.setTrackedNodes(publishNodes);
    // 真正开始publish，需要注意的是`followersChecker.getFaultyNodes()`记录的是心跳超时重试未成功的节点，此时同步时直接置位这些节点PublicationTargetState状态为failed
    publication.start(followersChecker.getFaultyNodes()); 
}

在CoordinatorPublication初始化时，会针对整个广播设置cancel超时时间(cluster.publish.timeout publish，默认30s)+info超时(cluster.publish.info_timeout, 默认10s)，info超时日志如下：

1
[INFO ] after [10s] publication of cluster state version [407258] is still waiting for {node2}[SENT_APPLY_COMMIT], {node1}[SENT_APPLY_COMMIT]

info超时会打印master还没接收到的commit响应的所有节点。cancel超时后，会将整个publish过程置为 cancelled+isCompleted。对于还未完成第二次响应的节点，直接置为失败，master会直接进入整个publish收尾阶段。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void cancel(String reason) {
    if (isCompleted) {
        return;
    }

    assert cancelled == false;
    cancelled = true; // 那么会先去取消
    // 若第一轮响应的master节点还没超过一半(那么直接置位失败)
    if (applyCommitRequest.isPresent() == false) { 
        final Exception e = new ElasticsearchException("publication cancelled before committing: " + reason);
        // 标记还未成的PublicationTarget为失败
        publicationTargets.stream().filter(PublicationTarget::isActive).forEach(pt -> pt.setFailed(e));
    }
    onPossibleCompletion(); 
}

onPossibleCompletion()会在后面介绍。

向每个节点广播请求

开始对每个数据节点发送元数据请求：

1
2
3
4
5
6
7
void sendPublishRequest() {
    if (isFailed()) {
        return;
    }
    state = PublicationTargetState.SENT_PUBLISH_REQUEST;
    Publication.this.sendPublishRequest(discoveryNode, publishRequest, new PublishResponseHandler());
}

master在对每个目标节点(包含master节点本身)发送publish前，会分别对每个目标节点构建PublicationTarget对象，来跟踪publish state，表明当前节点的publish进行到了哪个阶段：

1
2
3
4
5
6
7
8
enum PublicationTargetState {
    NOT_STARTED,  // 对象初始化时的状态
    FAILED,  //该目标节点的publish失败了，比如超时，或者节点异常等
    SENT_PUBLISH_REQUEST,  // master已经向数据节点发送了第一次的publish_request
    WAITING_FOR_QUORUM,  // 数据节点已经响应了master,但是masater在等待第二次commit的条件
    SENT_APPLY_COMMIT, // master已经向数据节点发送了commit请求
    APPLIED_COMMIT,  //  master收到了数据节点对commit请求的响应
}

当对目标节点置位SENT_PUBLISH_REQUEST后，进入PublicationContext.sendPublishRequest()进行clusterState的发送。同时定义了PublishResponseHandler作为master响应目前节点第一次response的处理类。在sendPublishRequest时，ES会判断是否向目标节点发送全量ClusterState还是仅仅发送diff的ClusterState。最常见的全量发布ClusterState的情况就是有新的节点加入到集群。

目标节点接收到maser发送的publish请求

目标节点的PublicationTransportHandler.handleIncomingPublishRequest首先接收到master发送的请求，做了以下三件事情：
1.然后第一步就是解析出最新的ClusterState。
2.其次进入acceptState()->CoordinationState.handlePublishRequest()构建响应master的response。
3.再次调用becomeFollower()变身Follow（非master本身）。
我们看下第二步构建Response时做了哪些事情：

1
2
3
4
5
public PublishResponse handlePublishRequest(PublishRequest publishRequest) {
    final ClusterState clusterState = publishRequest.getAcceptedState();
    persistedState.setLastAcceptedState(clusterState);
    return new PublishResponse(clusterState.term(), clusterState.version());
}

该函数主要做了如下事情：将接受到的集群元数据进行落盘。针对不同角色的节点，persistedState也不同：
1.目标节点为masters属性的角色时，persistedState=LucenePersistedState。
2.目标节点为仅仅为data属性的角色时，persistedState=AsyncLucenePersistedState。
这里还有lastSeenClusterState元数据，它的作用仅仅值是为了第二次接受到commit请求时做版本等校验用的。并不会作为接收到的临时元数据使用。
这里字面上可以知道：对master节点，对于ClusterState落盘时同步操作，若IO压力大的话，对落盘相当耗时，会拖累整个集群publish耗时；对数据节点，采用异步落盘的方式，避免阻塞整个落盘导致的响应超时。我们看下如何落盘，以及落盘落了哪些信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
void writeIncrementalStateAndCommit(long currentTerm, ClusterState previousClusterState,
                                    ClusterState clusterState) throws IOException {
    try {
        final long startTimeMillis = relativeTimeMillisSupplier.getAsLong();
         // 进行lucene.flush()构建segment
        final WriterStats stats = updateMetadata(previousClusterState.metadata(), clusterState.metadata());
        // 进行lucene.flush()刷盘
        commit(currentTerm, clusterState.version()); 
        final long durationMillis = relativeTimeMillisSupplier.getAsLong() - startTimeMillis;
        final TimeValue finalSlowWriteLoggingThreshold = slowWriteLoggingThresholdSupplier.get();
        // check是否超时
        if (durationMillis >= finalSlowWriteLoggingThreshold.getMillis()) {
            logger.warn("writing cluster state took [{}ms] which is above the warn threshold of [{}]; " +
                    "wrote global metadata [{}] and metadata for [{}] indices and skipped [{}] unchanged indices",
                durationMillis, finalSlowWriteLoggingThreshold, stats.globalMetaUpdated, stats.numIndicesUpdated,
                stats.numIndicesUnchanged);
        } else {
        ......
        }
    } finally {
        closeIfAnyIndexWriterHasTragedyOrIsClosed();
    }
}
private WriterStats updateMetadata(Metadata previouslyWrittenMetadata, Metadata metadata) throws IOException {
    // globalMeta是否有发生变化
    final boolean updateGlobalMeta = Metadata.isGlobalStateEquals(previouslyWrittenMetadata, metadata) == false;
    // 若变化了，那么全部覆盖掉
    if (updateGlobalMeta) { 
         // 重新产生gloabal Metadata
        try (ReleasableDocument globalMetadataDocument = makeGlobalMetadataDocument(metadata)) {
             // 配置的数据盘，每个盘都会写一份
            for (MetadataIndexWriter metadataIndexWriter : metadataIndexWriters) {
                metadataIndexWriter.updateGlobalMetadata(globalMetadataDocument.getDocument());
            }
        }
    }
    // 获取旧的索引version
    final Map<String, Long> indexMetadataVersionByUUID = new HashMap<>(previouslyWrittenMetadata.indices().size());
    for (ObjectCursor<IndexMetadata> cursor : previouslyWrittenMetadata.indices().values()) {
        final IndexMetadata indexMetadata = cursor.value;
        final Long previousValue = indexMetadataVersionByUUID.putIfAbsent(indexMetadata.getIndexUUID(), indexMetadata.getVersion());
    }
    int numIndicesUpdated = 0;
    int numIndicesUnchanged = 0;
     // 遍历新的IndexMetadata
    for (ObjectCursor<IndexMetadata> cursor : metadata.indices().values()) {
        final IndexMetadata indexMetadata = cursor.value;
        final Long previousVersion = indexMetadataVersionByUUID.get(indexMetadata.getIndexUUID());
        // IndexMetadata新创建的，或者发生了改变
        if (previousVersion == null || indexMetadata.getVersion() != previousVersion) { 
            logger.trace("updating metadata for [{}], changing version from [{}] to [{}]",
                indexMetadata.getIndex(), previousVersion, indexMetadata.getVersion());
            numIndicesUpdated++;
            try (ReleasableDocument indexMetadataDocument = makeIndexMetadataDocument(indexMetadata)) {
                for (MetadataIndexWriter metadataIndexWriter : metadataIndexWriters) {
                    metadataIndexWriter.updateIndexMetadataDocument(indexMetadataDocument.getDocument(), indexMetadata.getIndex());
                }
            }
        } else { 
            numIndicesUnchanged++;
        }
        indexMetadataVersionByUUID.remove(indexMetadata.getIndexUUID());
    }
    // 存在旧的，但是没有在新的中存在了，那么就是被删除了。
    for (String removedIndexUUID : indexMetadataVersionByUUID.keySet()) {
        for (MetadataIndexWriter metadataIndexWriter : metadataIndexWriters) {
            metadataIndexWriter.deleteIndexMetadata(removedIndexUUID);
        }
    }
    // 去flush一次
    // Flush, to try and expose a failure (e.g. out of disk space) before committing, because we can handle a failure here more
    // gracefully than one that occurs during the commit process.
    for (MetadataIndexWriter metadataIndexWriter : metadataIndexWriters) {
        metadataIndexWriter.flush();
    }
    return new WriterStats(updateGlobalMeta, numIndicesUpdated, numIndicesUnchanged);
}

需要知道的是：
1.在global Metadate中，并没有存储所有indexMetadata，而是作为单独一项存储的。
2.存储的时候回遍历每个data.path分别都存储一份。所以在线上环境，我们需要严格将data和master节点区分开，以免data角色将磁盘IO占用过多，而影响元数据的同步落盘操作。
注意：数据节点仅仅是将集群元数据保存在了本地，并更新了，但是还没有真正合并到data节点当前使用的ClusterState中。真正将新的ClusterState当成本地元数据，是在接收到master发送的commit请求后。

master接收到目标节点发送的publish响应

master接收到data响应的响应是在Publication$PublicationTarget$PublishResponseHandler.onResponse()，首先将对该节点publish请求状态置为WAITING_FOR_QUORUM，然后进入PublicationTarget.handlePublishResponse()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
void handlePublishResponse(PublishResponse publishResponse) {
    // master属性节点响应过半后，才会赋值。接着直接第二次commit
    if (applyCommitRequest.isPresent()) { 
        sendApplyCommit();
    } else {
        // master检查是否有资格发送commit请求
        try {
             // 响应节点过半的，继续执行
            Publication.this.handlePublishResponse(discoveryNode, publishResponse).ifPresent(applyCommit -> { 
                assert applyCommitRequest.isPresent() == false;
                applyCommitRequest = Optional.of(applyCommit);
                ackListener.onCommit(TimeValue.timeValueMillis(currentTimeSupplier.getAsLong() - startTime)); 
                 // master对第一次响应的节点（状态为WAITING_FOR_QUORUM）开始进行第二次commit
                publicationTargets.stream().filter(PublicationTarget::isWaitingForQuorum)
                    .forEach(PublicationTarget::sendApplyCommit);
            });
        } catch (Exception e) {
            setFailed(e);
            onPossibleCommitFailure();
        }
    }
}

该函数主要做了如下事情：
1.首先检查是否已经对某些节点发送了applyCommitRequest请求。master可以对数据节点发送applyCommitRequest是有条件的：必须有一半的master属性的节点已经响应了。（raft协议的特性）。
2.若master还没有发送过applyCommitRequest请求，那么会检查是否有资格可以对data节点发送第二次commit请求了。若有资格发送了，那么对所有状态为WAITING_FOR_QUORUM的节发送commit请求。

master向目标节点发送二次commit请求

master收到过半master属性的第一次response请求后，开始对WAITING_FOR_QUORUM状态的节点发送commit请求：

1
2
3
4
5
void sendApplyCommit() {
    //对目标节点发送状态置为SENT_APPLY_COMMIT
    state = PublicationTargetState.SENT_APPLY_COMMIT; 
    Publication.this.sendApplyCommit(discoveryNode, applyCommitRequest.get(), new ApplyCommitResponseHandler());
}

目标节点接收到master发送的二次commit请求

目前目标节点收到master发送的commit请求后，首先进入了Coordinator.handleApplyCommit()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
private void handleApplyCommit(ApplyCommitRequest applyCommitRequest, ActionListener<Void> applyListener) {
    synchronized (mutex) {
         // master节点收到本节点的commit响应
        if (applyCommitRequest.getSourceNode().equals(getLocalNode())) { 
            // master合并元数据到全局将在收到所有数据节commit响应后(具体见CoordinatorPublication.onCompletion())，将跑到transportCommitCallback
            applyListener.onResponse(null); 
        } else { // 数据节点收到master发送的commit请求
            clusterApplier.onNewClusterState(applyCommitRequest.toString(), () -> applierState,
                new ClusterApplyListener() {

                    @Override
                    public void onFailure(String source, Exception e) {
                        applyListener.onFailure(e); // 将跑到PublicationTransportHandler.transportCommitCallback
                    }

                    @Override
                    public void onSuccess(String source) {
                        applyListener.onResponse(null);// 将跑到PublicationTransportHandler.transportCommitCallback
                    }
                });
        }
    }
}

针对不同角色，目标节点做了不同的反应：
1.若本节点就是主master节点，那么调用PublicationTransportHandler.transportCommitCallback()，作用仅仅是响应回去。 master只有在整个publish()完成后，才会将新元数据作为本地的全局元数据(后面会讲)。
2.若本目标节点是非主master节点，则调用ClusterApplierService.onNewClusterState将新ClusterState节点作为本节点维持的最新全局ClusterState。
我们看下本目标节点在替换元数据的时候哪些事情，实际进入的是ClusterApplierService.runTask()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
private void runTask(UpdateTask task) {
    final ClusterState previousClusterState = state.get();
    long startTimeMS = currentTimeInMillis();
    final StopWatch stopWatch = new StopWatch();
    final ClusterState newClusterState;
    try {
        try (Releasable ignored = stopWatch.timing("running task [" + task.source + ']')) {
            // 直接获取的是最新ClusterState
            newClusterState = task.apply(previousClusterState); 
        }
    } catch (Exception e) {
        ......
        return;
    }

    if (previousClusterState == newClusterState) {
        TimeValue executionTime = TimeValue.timeValueMillis(Math.max(0, currentTimeInMillis() - startTimeMS));
        warnAboutSlowTaskIfNeeded(executionTime, task.source, stopWatch);
        task.listener.onSuccess(task.source);
    } else {
        try {// 超级重要，当集群元数据修改后，会去做一系列检查，比如创建索引等，将分配给本节点的分片状态置位started等
            applyChanges(task, previousClusterState, newClusterState, stopWatch);
            // 会去调用PublicationTransportHandler.transportCommitCallback()，直接响应主master
            task.listener.onSuccess(task.source); 
        } catch (Exception e) {
            ......
        }
    }
}

非主master节点主要做了如下事情：
1.获取的是最新ClusterState。
2.检查新旧ClusterState是否一致，若一致，则不做任何操作。
3.若旧ClusterState有变化，则调用applyChanges()，根据最新ClusterState适配本地。

我们再看下applyChanges()如何适配本地的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
private void applyChanges(UpdateTask task, ClusterState previousClusterState, ClusterState newClusterState, StopWatch stopWatch) {
    ClusterChangedEvent clusterChangedEvent = new ClusterChangedEvent(task.source, newClusterState, previousClusterState);
    final DiscoveryNodes.Delta nodesDelta = clusterChangedEvent.nodesDelta();
     // 比如节点个数发生了变化，那么就跑到这里
    if (nodesDelta.hasChanges() && logger.isInfoEnabled()) {
        String summary = nodesDelta.shortSummary();
        if (summary.length() > 0) {
            logger.info("{}, term: {}, version: {}, reason: {}",
                summary, newClusterState.term(), newClusterState.version(), task.source);
        } // 会打印 removed {{、added {{日志
    } // 若added,那么已经认同加入集群了

    try (Releasable ignored = stopWatch.timing("connecting to new nodes")) {
        connectToNodesAndWait(newClusterState);
    }

    callClusterStateAppliers(clusterChangedEvent, stopWatch);
    
    nodeConnectionsService.disconnectFromNodesExcept(newClusterState.nodes());

    state.set(newClusterState);
    // 这里也比较重要，会去等待新的集群状态，然后触发某些操作（比如请求集群状态，但是此时没有maser,可见TransportMasterNodeAction$AsyncSingleAction.retry()）
    callClusterStateListeners(clusterChangedEvent, stopWatch);
}

非主master节点主要做了如下事情：
1.检查是否有节点掉线&新增，对于新增节点主动进行connect，同时打印如下日志：

1
[2020-08-10T12:12:22,781][INFO ][o.e.c.s.ClusterApplierService] [node1] added {{node2}}, term: 28, version: 483578, reason: ApplyCommitRequest{term=28, version=483578, sourceNode={master}}

2.本地调用callClusterStateAppliers()根据最新的ClusterState做一些操作，比如创建IndexService，删除索引数据，分配分片等操作。我们看下其中重要实现：

1
2
3
4
5
6
7
private void callClusterStateAppliers(ClusterChangedEvent clusterChangedEvent, StopWatch stopWatch) {
        clusterStateAppliers.forEach(applier -> {
            try (Releasable ignored = stopWatch.timing("running applier [" + applier + "]")) {
                applier.applyClusterState(clusterChangedEvent);
            }
        });
    }

clusterStateAppliers={highPriorityStateAppliers, normalPriorityStateAppliers, lowPriorityStateAppliers}，我们需要着重强调下highPriorityStateAppliers中的IndicesClusterStateService.applyClusterState()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public synchronized void applyClusterState(final ClusterChangedEvent event) {
    final ClusterState state = event.state();
    updateFailedShardsCache(state);
    deleteIndices(event); // also deletes shards of deleted indices
    removeIndices(event); // also removes shards of removed indices
    failMissingShards(state);
    // 删除被删掉的索引的shard
    removeShards(state);   // removes any local shards that doesn't match what the master expects
    // 本地更新本地索引元数据
    updateIndices(event); // can also fail shards, but these are then guaranteed to be in failedShardsCache
    //  在本地创建索引元数据
    createIndices(state);
    // 恢复或者创建分片
    createOrUpdateShards(state);
}

非master节点的本地维护元数据将在可以根据全局ClusterState进行及时调整。
3.关闭掉线节点的连接。
4.设置本地维护的最新全局ClusteState,存放在ClusterApplierService.state对象中。
5.调用callClusterStateListeners()来进行回调响应，比如数据节点需要请求master时，发现本地找不到master，那么就会创建一个listener，等待本地维护的ClusteState发生变化时，再去retry；又比如本节点是master,更新元数据后不是master，会做一些收尾处理等。

master接收到目标节点发送二次commit响应

master收到目标节点的二次响应后，最先进入ApplyCommitResponseHandler.onResponse()函数：

1
2
3
4
5
6
7
8
9
public void onResponse(TransportResponse.Empty ignored) {
    if (isFailed()) {
          return;
    }
    // 修改这个确定的二次确认为已完成,
    setAppliedCommit(); 
    // 同时检查是不是所有节点都二次响应
    onPossibleCompletion(); 
}

master主要做了如下操作:
1.确认针对目标节点的二次响应完成：修改目标节点的publish状态为APPLIED_COMMIT；进入CoordinatorPublication构造函数的lister中更新master维护的每个数据节点最新ClusterState version。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public void onNodeAck(DiscoveryNode node, Exception e) {
    // acking and cluster state application for local node is handled specially
    //本节点是master即为响应节点
    if (node.equals(getLocalNode())) { 
          synchronized (mutex) {
              if (e == null) {
                  // master本身第二次确认完成, 仅仅设置localNodeAckEvent为done
                  localNodeAckEvent.onResponse(null);
              } else {
                  localNodeAckEvent.onFailure(e);
               }
          }
    } else {// 响应节点为非master节点 
          // 会跑到 AckCountDownListener.onNodeAck()里面检查是否全部全部节点ack。数据节点不会finish
          ackListener.onNodeAck(node, e);
          // 这里比较重要,会去更新本节点维护的数据节点的version，若version落后超时，会有惩罚机制
          if (e == null) { 
               lagDetector.setAppliedVersion(node, publishRequest.getAcceptedState().version());
          }
    }
}

对每个目标节点二次commit响应做了如下操作：
1.1 若目标节点就是本主master节点，那么仅标记localNodeAckEvent状态为done(后面会用)
1.2 若目标节点是非主master节点，则更新本主master维护的其他节点的ClusterState version（若落后严重，会主动被master剔除集群，后面会介绍)。

2.调用onPossibleCompletion()检查整个publish是否完成了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
private void onPossibleCompletion() {
    // 若超时30s(cluster.publish.timeout),就cancelled=true,置为失败
    if (cancelled == false) { 
        for (final PublicationTarget target : publicationTargets) {  // 遍历每一个target
             // 只要还有一个没有第二次确认完成，就退出
            if (target.isActive()) { 
                return;
            }
        }
    } 
    // 此时1.要是cancelled=true；2.要么cancelled=false, 但是所有目标节点publish状态已经done->applyCommitRequest已经发送请求。
    if (applyCommitRequest.isPresent() == false) {
        //还没有任何节点进行第二次commit：超时导致的失败
        logger.debug("onPossibleCompletion: [{}] commit failed", this);
        assert isCompleted == false;
        isCompleted = true;
        onCompletion(false)
        return;
    }       
    isCompleted = true;
    //全部完成了才会去调用 这里还有大作用，会去调用CoordinatorPublication.applyClusterState()
    onCompletion(true); 
}

主要做了如下检查：
2.1 检查cancelled是否置为失败，若未失败，且还有至少一个目标节点未完成二次commit，那么就退出等待。
2.2 若applyCommitRequest为空，说明是超时导致的失败，代表整个publish已经失败的完成了，会进入onCompletion()。
2.3 此时所有节点已经完成二次commit响应，进入onCompletion()

再继续看下主master调用onCompletion()做了哪些事情：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
//master节点上，所有任务已经完成（isCompleted=true），可能任务全部失败了（超时30会设置），也可能任务全部成功了
protected void onCompletion(boolean committed) { 
    // master本身完成二次确认
    localNodeAckEvent.addListener(new ActionListener<Void>() { 
        @Override
        public void onResponse(Void ignore) {
            receivedJoinsProcessed = true;
            // 也是比较重要的, master合并元数据进本身的ClusterState。数据节点合并是在收到commit请求后就合并(详见Coordinator.handleApplyCommit()函数)
            clusterApplier.onNewClusterState(Coordinator.CoordinatorPublication.this.toString(), () -> applierState, // 进去会去调用
                    new ClusterApplier.ClusterApplyListener() {
                        @Override
                        public void onSuccess(String source) { // 本地master更新后
                            synchronized (mutex) {
                                currentPublication = Optional.empty();
                                // trigger term bump if new term was found during publication
                                updateMaxTermSeen(getCurrentTerm());

                                if (mode == Coordinator.Mode.LEADER) {
                                    .......
                                }
                                // 开始对滞后的节点进行处理
                                lagDetector.startLagDetector(publishRequest.getAcceptedState().version()); 
                                logIncompleteNodes(Level.WARN); // 超时30s的节点报警
                            }
                            cancelTimeoutHandlers(); // 取消超时
                            ackListener.onNodeAck(getLocalNode(), null); // 本节点也完成了
                            publishListener.onResponse(null);
                        }
                    });
        }
    }, EsExecutors.newDirectExecutorService(), transportService.getThreadPool().getThreadContext());
}

主要做了如下事情：若主master收到本节点的二次commti响应（设置localNodeAckEvent为done），那么
1.调用ClusterApplierService.onNewClusterState将新的ClusterState融合到本地节点中（参考data融合新的全局元数据）
2.开始针对本地维护的数据节点ClusterState version，若再超时时间外仍然低于当前同步的version，则将数据节点从集群中剔除，超时时间90s(由cluster.follower_lag.timeout参数决定)

1
2
3
4
5
6
7
8
9
10
11
12
13
void checkForLag(final long version) {
    if (appliedStateTrackersByNode.get(discoveryNode) != this) {
        logger.trace("{} no longer active when checking version {}", this, version);
        return;
    }

    long appliedVersion = this.appliedVersion.get();
    // 落后
    logger.warn(
            "node [{}] is lagging at cluster state version [{}], although publication of cluster state version [{}] completed [{}] ago",
            discoveryNode, appliedVersion, version, clusterStateApplicationTimeout);
    onLagDetected.accept(discoveryNode); // 在 Coordinator 构造函数中。惩罚将节点脱离集群，实际调用removeNode()函数
}

会打印如下日志:

1
[2020-08-10T14:12:24,781][WARN ][o.e.c.c.LagDetector] [master] node [node1] is lagging at cluster state version [483037], although publication of cluster state version [483038] completed [1.5m] ago

然后直接调用Coordinator.removeNode()再次广播全局元数据。
3.打印publish超时未完成日志。

1
[2020-08-10T12:12:24,781][WARN ][o.e.c.c.C.CoordinatorPublication] [master1] after [30.1s] publication of cluster state version [483038] is still waiting for {node1}[SENT_APPLY_COMMIT], {node2} [SENT_APPLY_COMMIT]

总结

master广播全过程分为第一次广播+第二次commit请求，只有过半master节点响应才能继续第二次广播。在30s超时时间后，主动设置publish状态为true, 在规定时间内元数据更新较慢的节点，master会主动将其剔除集群。

扫一扫，分享到微信

微信分享二维码
Error: Comments Not Initialized
Styling with Markdown is supported
Powered by Gitment