摘要：摘要目前是最流行的開(kāi)源分布式搜索引擎系統(tǒng)，其使用作為單機(jī)存儲(chǔ)引擎并提供強(qiáng)大的搜索查詢能力。前言分布式一致性原理剖析系列將會(huì)對(duì)的分布式一致性原理進(jìn)行詳細(xì)的剖析，介紹其實(shí)現(xiàn)方式原理以及其存在的問(wèn)題等基于版本。相當(dāng)于一次正常情況的新節(jié)點(diǎn)加入。

摘要： ES目前是最流行的開(kāi)源分布式搜索引擎系統(tǒng)，其使用Lucene作為單機(jī)存儲(chǔ)引擎并提供強(qiáng)大的搜索查詢能力。學(xué)習(xí)其搜索原理，則必須了解Lucene，而學(xué)習(xí)ES的架構(gòu)，就必須了解其分布式如何實(shí)現(xiàn)，而一致性是分布式系統(tǒng)的核心之一。

前言
“Elasticsearch分布式一致性原理剖析”系列將會(huì)對(duì)Elasticsearch的分布式一致性原理進(jìn)行詳細(xì)的剖析，介紹其實(shí)現(xiàn)方式、原理以及其存在的問(wèn)題等(基于6.2版本)。

ES目前是最流行的開(kāi)源分布式搜索引擎系統(tǒng)，其使用Lucene作為單機(jī)存儲(chǔ)引擎并提供強(qiáng)大的搜索查詢能力。學(xué)習(xí)其搜索原理，則必須了解Lucene，而學(xué)習(xí)ES的架構(gòu)，就必須了解其分布式如何實(shí)現(xiàn)，而一致性是分布式系統(tǒng)的核心之一。
本篇將介紹ES的集群組成、節(jié)點(diǎn)發(fā)現(xiàn)與Master選舉，錯(cuò)誤檢測(cè)與擴(kuò)縮容相關(guān)的內(nèi)容。ES在處理節(jié)點(diǎn)發(fā)現(xiàn)與Master選舉等方面沒(méi)有選擇Zookeeper等外部組件，而是自己實(shí)現(xiàn)的一套，本文會(huì)介紹ES的這套機(jī)制是如何工作的，存在什么問(wèn)題。本文的主要內(nèi)容如下：

ES集群構(gòu)成

節(jié)點(diǎn)發(fā)現(xiàn)

Master選舉

錯(cuò)誤檢測(cè)

集群擴(kuò)縮容

與Zookeeper、raft等實(shí)現(xiàn)方式的比較

小結(jié)

首先，一個(gè)Elasticsearch集群(下面簡(jiǎn)稱ES集群)是由許多節(jié)點(diǎn)(Node)構(gòu)成的，Node可以有不同的類型，通過(guò)以下配置，可以產(chǎn)生四種不同類型的Node：

conf/elasticsearch.yml:
    node.master: true/false
    node.data: true/false

四種不同類型的Node是一個(gè)node.master和node.data的true/false的兩兩組合。當(dāng)然還有其他類型的Node，比如IngestNode(用于數(shù)據(jù)預(yù)處理等)，不在本文討論范圍內(nèi)。

當(dāng)node.master為true時(shí)，其表示這個(gè)node是一個(gè)master的候選節(jié)點(diǎn)，可以參與選舉，在ES的文檔中常被稱作master-eligible node，類似于MasterCandidate。ES正常運(yùn)行時(shí)只能有一個(gè)master(即leader)，多于1個(gè)時(shí)會(huì)發(fā)生腦裂。

當(dāng)node.data為true時(shí)，這個(gè)節(jié)點(diǎn)作為一個(gè)數(shù)據(jù)節(jié)點(diǎn)，會(huì)存儲(chǔ)分配在該node上的shard的數(shù)據(jù)并負(fù)責(zé)這些shard的寫入、查詢等。

此外，任何一個(gè)集群內(nèi)的node都可以執(zhí)行任何請(qǐng)求，其會(huì)負(fù)責(zé)將請(qǐng)求轉(zhuǎn)發(fā)給對(duì)應(yīng)的node進(jìn)行處理，所以當(dāng)node.master和node.data都為false時(shí)，這個(gè)節(jié)點(diǎn)可以作為一個(gè)類似proxy的節(jié)點(diǎn)，接受請(qǐng)求并進(jìn)行轉(zhuǎn)發(fā)、結(jié)果聚合等。

上圖是一個(gè)ES集群的示意圖，其中Node_A是當(dāng)前集群的Master，Node_B和Node_C是Master的候選節(jié)點(diǎn)，其中Node_A和Node_B同時(shí)也是數(shù)據(jù)節(jié)點(diǎn)(DataNode)，此外，Node_D是一個(gè)單純的數(shù)據(jù)節(jié)點(diǎn)，Node_E是一個(gè)proxy節(jié)點(diǎn)。

到這里，我們提一個(gè)問(wèn)題，供讀者思考：一個(gè)ES集群應(yīng)當(dāng)配置多少個(gè)master-eligible node，當(dāng)集群的存儲(chǔ)或者計(jì)算資源不足，需要擴(kuò)容時(shí)，新擴(kuò)上去的節(jié)點(diǎn)應(yīng)該設(shè)置為何種類型？

Node啟動(dòng)后，首先要通過(guò)節(jié)點(diǎn)發(fā)現(xiàn)功能加入集群。ZenDiscovery是ES自己實(shí)現(xiàn)的一套用于節(jié)點(diǎn)發(fā)現(xiàn)和選主等功能的模塊，沒(méi)有依賴Zookeeper等工具，官方文檔：

https://www.elastic.co/guide/...

簡(jiǎn)單來(lái)說(shuō)，節(jié)點(diǎn)發(fā)現(xiàn)依賴以下配置：

conf/elasticsearch.yml:

discovery.zen.ping.unicast.hosts: [1.1.1.1, 1.1.1.2, 1.1.1.3]

這個(gè)配置可以看作是，在本節(jié)點(diǎn)到每個(gè)hosts中的節(jié)點(diǎn)建立一條邊，當(dāng)整個(gè)集群所有的node形成一個(gè)聯(lián)通圖時(shí)，所有節(jié)點(diǎn)都可以知道集群中有哪些節(jié)點(diǎn)，不會(huì)形成孤島。

官方推薦這里設(shè)置為所有的master-eligible node，讀者可以想想這樣有何好處：

It is recommended that the unicast hosts list be maintained as the list of master-eligible nodes in the cluster.

Master選舉

上圖是一個(gè)ES集群的示意圖，其中Node_A是當(dāng)前集群的Master，Node_B和Node_C是Master的候選節(jié)點(diǎn)，其中Node_A和Node_B同時(shí)也是數(shù)據(jù)節(jié)點(diǎn)(DataNode)，此外，Node_D是一個(gè)單純的數(shù)據(jù)節(jié)點(diǎn)，Node_E是一個(gè)proxy節(jié)點(diǎn)。

到這里，我們提一個(gè)問(wèn)題，供讀者思考：一個(gè)ES集群應(yīng)當(dāng)配置多少個(gè)master-eligible node，當(dāng)集群的存儲(chǔ)或者計(jì)算資源不足，需要擴(kuò)容時(shí)，新擴(kuò)上去的節(jié)點(diǎn)應(yīng)該設(shè)置為何種類型？

Node啟動(dòng)后，首先要通過(guò)節(jié)點(diǎn)發(fā)現(xiàn)功能加入集群。ZenDiscovery是ES自己實(shí)現(xiàn)的一套用于節(jié)點(diǎn)發(fā)現(xiàn)和選主等功能的模塊，沒(méi)有依賴Zookeeper等工具，官方文檔：

https://www.elastic.co/guide/...

簡(jiǎn)單來(lái)說(shuō)，節(jié)點(diǎn)發(fā)現(xiàn)依賴以下配置：

conf/elasticsearch.yml:

discovery.zen.ping.unicast.hosts: [1.1.1.1, 1.1.1.2, 1.1.1.3]

這個(gè)配置可以看作是，在本節(jié)點(diǎn)到每個(gè)hosts中的節(jié)點(diǎn)建立一條邊，當(dāng)整個(gè)集群所有的node形成一個(gè)聯(lián)通圖時(shí)，所有節(jié)點(diǎn)都可以知道集群中有哪些節(jié)點(diǎn)，不會(huì)形成孤島。
官方推薦這里設(shè)置為所有的master-eligible node，讀者可以想想這樣有何好處：

It is recommended that the unicast hosts list be maintained as the list of master-eligible nodes in the cluster.

上面提到，集群中可能會(huì)有多個(gè)master-eligible node，此時(shí)就要進(jìn)行master選舉，保證只有一個(gè)當(dāng)選master。如果有多個(gè)node當(dāng)選為master，則集群會(huì)出現(xiàn)腦裂，腦裂會(huì)破壞數(shù)據(jù)的一致性，導(dǎo)致集群行為不可控，產(chǎn)生各種非預(yù)期的影響。

為了避免產(chǎn)生腦裂，ES采用了常見(jiàn)的分布式系統(tǒng)思路，保證選舉出的master被多數(shù)派(quorum)的master-eligible node認(rèn)可，以此來(lái)保證只有一個(gè)master。這個(gè)quorum通過(guò)以下配置進(jìn)行配置：

conf/elasticsearch.yml:
    discovery.zen.minimum_master_nodes: 2

這個(gè)配置對(duì)于整個(gè)集群非常重要。

加粗文字1 master選舉誰(shuí)發(fā)起，什么時(shí)候發(fā)起？
master選舉是由master-eligible節(jié)點(diǎn)發(fā)起，當(dāng)一個(gè)master-eligible節(jié)點(diǎn)發(fā)現(xiàn)滿足以下條件時(shí)發(fā)起選舉：

該master-eligible節(jié)點(diǎn)的當(dāng)前狀態(tài)不是master。
該master-eligible節(jié)點(diǎn)通過(guò)ZenDiscovery模塊的ping操作詢問(wèn)其已知的集群其他節(jié)點(diǎn)，沒(méi)有任何節(jié)點(diǎn)連接到master。
包括本節(jié)點(diǎn)在內(nèi)，當(dāng)前已有超過(guò)minimum_master_nodes個(gè)節(jié)點(diǎn)沒(méi)有連接到master。
總結(jié)一句話，即當(dāng)一個(gè)節(jié)點(diǎn)發(fā)現(xiàn)包括自己在內(nèi)的多數(shù)派的master-eligible節(jié)點(diǎn)認(rèn)為集群沒(méi)有master時(shí)，就可以發(fā)起master選舉。

2 當(dāng)需要選舉master時(shí)，選舉誰(shuí)？
首先是選舉誰(shuí)的問(wèn)題，如下面源碼所示，選舉的是排序后的第一個(gè)MasterCandidate(即master-eligible node)。

public MasterCandidate electMaster(Collection candidates) {
    assert hasEnoughCandidates(candidates);
    List sortedCandidates = new ArrayList<>(candidates);
    sortedCandidates.sort(MasterCandidate::compare);
    return sortedCandidates.get(0);
}

那么是按照什么排序的？

public static int compare(MasterCandidate c1, MasterCandidate c2) {

// we explicitly swap c1 and c2 here. the code expects "better" is lower in a sorted
// list, so if c2 has a higher cluster state version, it needs to come first.
int ret = Long.compare(c2.clusterStateVersion, c1.clusterStateVersion);
if (ret == 0) {
    ret = compareNodes(c1.getNode(), c2.getNode());
}
return ret;

}
如上面源碼所示，先根據(jù)節(jié)點(diǎn)的clusterStateVersion比較，clusterStateVersion越大，優(yōu)先級(jí)越高。clusterStateVersion相同時(shí)，進(jìn)入compareNodes，其內(nèi)部按照節(jié)點(diǎn)的Id比較(Id為節(jié)點(diǎn)第一次啟動(dòng)時(shí)隨機(jī)生成)。

總結(jié)一下：

當(dāng)clusterStateVersion越大，優(yōu)先級(jí)越高。這是為了保證新Master擁有最新的clusterState(即集群的meta)，避免已經(jīng)commit的meta變更丟失。因?yàn)镸aster當(dāng)選后，就會(huì)以這個(gè)版本的clusterState為基礎(chǔ)進(jìn)行更新。(一個(gè)例外是集群全部重啟，所有節(jié)點(diǎn)都沒(méi)有meta，需要先選出一個(gè)master，然后master再通過(guò)持久化的數(shù)據(jù)進(jìn)行meta恢復(fù)，再進(jìn)行meta同步)。
當(dāng)clusterStateVersion相同時(shí)，節(jié)點(diǎn)的Id越小，優(yōu)先級(jí)越高。即總是傾向于選擇Id小的Node，這個(gè)Id是節(jié)點(diǎn)第一次啟動(dòng)時(shí)生成的一個(gè)隨機(jī)字符串。之所以這么設(shè)計(jì)，應(yīng)該是為了讓選舉結(jié)果盡可能穩(wěn)定，不要出現(xiàn)都想當(dāng)master而選不出來(lái)的情況。
3 怎么算選舉成功？
當(dāng)一個(gè)master-eligible node(我們假設(shè)為Node_A)發(fā)起一次選舉時(shí)，它會(huì)按照上述排序策略選出一個(gè)它認(rèn)為的master。

假設(shè)Node_A選Node_B當(dāng)Master：
Node_A會(huì)向Node_B發(fā)送join請(qǐng)求，那么此時(shí)：

(1) 如果Node_B已經(jīng)成為Master，Node_B就會(huì)把Node_A加入到集群中，然后發(fā)布最新的cluster_state, 最新的cluster_state就會(huì)包含Node_A的信息。相當(dāng)于一次正常情況的新節(jié)點(diǎn)加入。對(duì)于Node_A，等新的cluster_state發(fā)布到Node_A的時(shí)候，Node_A也就完成join了。

(2) 如果Node_B在競(jìng)選Master，那么Node_B會(huì)把這次join當(dāng)作一張選票。對(duì)于這種情況，Node_A會(huì)等待一段時(shí)間，看Node_B是否能成為真正的Master，直到超時(shí)或者有別的Master選成功。

(3) 如果Node_B認(rèn)為自己不是Master(現(xiàn)在不是，將來(lái)也選不上)，那么Node_B會(huì)拒絕這次join。對(duì)于這種情況，Node_A會(huì)開(kāi)啟下一輪選舉。

假設(shè)Node_A選自己當(dāng)Master：
此時(shí)NodeA會(huì)等別的node來(lái)join，即等待別的node的選票，當(dāng)收集到超過(guò)半數(shù)的選票時(shí)，認(rèn)為自己成為master，然后變更c(diǎn)luster_state中的master node為自己，并向集群發(fā)布這一消息。

有興趣的同學(xué)可以看看下面這段源碼：

    if (transportService.getLocalNode().equals(masterNode)) {
            final int requiredJoins = Math.max(0, electMaster.minimumMasterNodes() - 1); // we count as one
            logger.debug("elected as master, waiting for incoming joins ([{}] needed)", requiredJoins);
            nodeJoinController.waitToBeElectedAsMaster(requiredJoins, masterElectionWaitForJoinsTimeout,
                    new NodeJoinController.ElectionCallback() {
                        @Override
                        public void onElectedAsMaster(ClusterState state) {
                            synchronized (stateMutex) {
                                joinThreadControl.markThreadAsDone(currentThread);
                            }
                        }

                        @Override
                        public void onFailure(Throwable t) {
                            logger.trace("failed while waiting for nodes to join, rejoining", t);
                            synchronized (stateMutex) {
                                joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                            }
                        }
                    }

            );
        } else {
            // process any incoming joins (they will fail because we are not the master)
            nodeJoinController.stopElectionContext(masterNode + " elected");

            // send join request
            final boolean success = joinElectedMaster(masterNode);

            synchronized (stateMutex) {
                if (success) {
                    DiscoveryNode currentMasterNode = this.clusterState().getNodes().getMasterNode();
                    if (currentMasterNode == null) {
                        // Post 1.3.0, the master should publish a new cluster state before acking our join request. we now should have
                        // a valid master.
                        logger.debug("no master node is set, despite of join request completing. retrying pings.");
                        joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                    } else if (currentMasterNode.equals(masterNode) == false) {
                        // update cluster state
                        joinThreadControl.stopRunningThreadAndRejoin("master_switched_while_finalizing_join");
                    }

                    joinThreadControl.markThreadAsDone(currentThread);
                } else {
                    // failed to join. Try again...
                    joinThreadControl.markThreadAsDoneAndStartNew(currentThread);
                }

按照上述流程，我們描述一個(gè)簡(jiǎn)單的場(chǎng)景來(lái)幫助大家理解：

假如集群中有3個(gè)master-eligible node，分別為Node_A、 Node_B、 Node_C, 選舉優(yōu)先級(jí)也分別為Node_A、Node_B、Node_C。三個(gè)node都認(rèn)為當(dāng)前沒(méi)有master，于是都各自發(fā)起選舉，選舉結(jié)果都為Node_A(因?yàn)檫x舉時(shí)按照優(yōu)先級(jí)排序，如上文所述)。于是Node_A開(kāi)始等join(選票)，Node_B、Node_C都向Node_A發(fā)送join，當(dāng)Node_A接收到一次join時(shí)，加上它自己的一票，就獲得了兩票了(超過(guò)半數(shù))，于是Node_A成為Master。此時(shí)cluster_state(集群狀態(tài))中包含兩個(gè)節(jié)點(diǎn)，當(dāng)Node_A再收到另一個(gè)節(jié)點(diǎn)的join時(shí)，cluster_state包含全部三個(gè)節(jié)點(diǎn)。

4 選舉怎么保證不腦裂？
基本原則還是多數(shù)派的策略，如果必須得到多數(shù)派的認(rèn)可才能成為Master，那么顯然不可能有兩個(gè)Master都得到多數(shù)派的認(rèn)可。

上述流程中，master候選人需要等待多數(shù)派節(jié)點(diǎn)進(jìn)行join后才能真正成為master，就是為了保證這個(gè)master得到了多數(shù)派的認(rèn)可。但是我這里想說(shuō)的是，上述流程在絕大部份場(chǎng)景下沒(méi)問(wèn)題，聽(tīng)上去也非常合理，但是卻是有bug的。

因?yàn)樯鲜隽鞒滩](méi)有限制在選舉過(guò)程中，一個(gè)Node只能投一票，那么什么場(chǎng)景下會(huì)投兩票呢？比如Node_B投Node_A一票，但是Node_A遲遲不成為Master，Node_B等不及了發(fā)起了下一輪選主，這時(shí)候發(fā)現(xiàn)集群里多了個(gè)Node_0，Node_0優(yōu)先級(jí)比Node_A還高，那Node_B肯定就改投Node_0了。假設(shè)Node_0和Node_A都處在等選票的環(huán)節(jié)，那顯然這時(shí)候Node_B其實(shí)發(fā)揮了兩票的作用，而且投給了不同的人。

那么這種問(wèn)題應(yīng)該怎么解決呢，比如raft算法中就引入了選舉周期(term)的概念，保證了每個(gè)選舉周期中每個(gè)成員只能投一票，如果需要再投就會(huì)進(jìn)入下一個(gè)選舉周期，term+1。假如最后出現(xiàn)兩個(gè)節(jié)點(diǎn)都認(rèn)為自己是master，那么肯定有一個(gè)term要大于另一個(gè)的term，而且因?yàn)閮蓚€(gè)term都收集到了多數(shù)派的選票，所以多數(shù)節(jié)點(diǎn)的term是較大的那個(gè)，保證了term小的master不可能commit任何狀態(tài)變更(commit需要多數(shù)派節(jié)點(diǎn)先持久化日志成功，由于有term檢測(cè)，不可能達(dá)到多數(shù)派持久化條件)。這就保證了集群的狀態(tài)變更總是一致的。

而ES目前(6.2版本)并沒(méi)有解決這個(gè)問(wèn)題，構(gòu)造類似場(chǎng)景的測(cè)試case可以看到會(huì)選出兩個(gè)master，兩個(gè)node都認(rèn)為自己是master，向全集群發(fā)布狀態(tài)變更，這個(gè)發(fā)布也是兩階段的，先保證多數(shù)派節(jié)點(diǎn)“接受”這次變更，然后再要求全部節(jié)點(diǎn)commit這次變更。很不幸，目前兩個(gè)master可能都完成第一個(gè)階段，進(jìn)入commit階段，導(dǎo)致節(jié)點(diǎn)間狀態(tài)出現(xiàn)不一致，而在raft中這是不可能的。那么為什么都能完成第一個(gè)階段呢，因?yàn)榈谝粋€(gè)階段ES只是將新的cluster_state做簡(jiǎn)單的檢查后放入內(nèi)存隊(duì)列，如果當(dāng)前cluster_state的master為空，不會(huì)對(duì)新的cluster_state中的master做檢查，即在接受了Node_A成為master的cluster_state后(還未commit)，還可以繼續(xù)接受Node_B成為cluster_state。這就使Node_A和Node_B都能達(dá)到commit條件，發(fā)起commit命令，從而將集群狀態(tài)引向不一致。當(dāng)然，這種腦裂很快會(huì)自動(dòng)恢復(fù)，因?yàn)椴灰恢掳l(fā)生后某個(gè)master再次發(fā)布cluster_state時(shí)就會(huì)發(fā)現(xiàn)無(wú)法達(dá)到多數(shù)派條件，或者是發(fā)現(xiàn)它的follower并不構(gòu)成多數(shù)派而自動(dòng)降級(jí)為candidate等。

這里要表達(dá)的是，ES的ZenDiscovery模塊與成熟的一致性方案相比，在某些特殊場(chǎng)景下存在缺陷，下面講ES的meta變更流程時(shí)也會(huì)分析其他的ES無(wú)法滿足一致性的場(chǎng)景。

錯(cuò)誤檢測(cè)

MasterFaultDetection與NodesFaultDetection

這里的錯(cuò)誤檢測(cè)可以理解為類似心跳的機(jī)制，有兩類錯(cuò)誤檢測(cè)，一類是Master定期檢測(cè)集群內(nèi)其他的Node，另一類是集群內(nèi)其他的Node定期檢測(cè)當(dāng)前集群的Master。檢查的方法就是定期執(zhí)行ping請(qǐng)求。ES文檔：

There are two fault detection processes running. The first is by the master, to ping all the other nodes in the cluster and verify that they are alive. And on the other end, each node pings to master to verify if its still alive or an election process needs to be initiated.

如果Master檢測(cè)到某個(gè)Node連不上了，會(huì)執(zhí)行removeNode的操作，將節(jié)點(diǎn)從cluster_state中移除，并發(fā)布新的cluster_state。當(dāng)各個(gè)模塊apply新的cluster_state時(shí)，就會(huì)執(zhí)行一些恢復(fù)操作，比如選擇新的primaryShard或者replica，執(zhí)行數(shù)據(jù)復(fù)制等。

如果某個(gè)Node發(fā)現(xiàn)Master連不上了，會(huì)清空pending在內(nèi)存中還未commit的new cluster_state，然后發(fā)起rejoin，重新加入集群(如果達(dá)到選舉條件則觸發(fā)新master選舉)。

2. rejoin
除了上述兩種情況，還有一種情況是Master發(fā)現(xiàn)自己已經(jīng)不滿足多數(shù)派條件(>=minimumMasterNodes)了，需要主動(dòng)退出master狀態(tài)(退出master狀態(tài)并執(zhí)行rejoin)以避免腦裂的發(fā)生，那么master如何發(fā)現(xiàn)自己需要rejoin呢？

上面提到，當(dāng)有節(jié)點(diǎn)連不上時(shí)，會(huì)執(zhí)行removeNode。在執(zhí)行removeNode時(shí)判斷剩余的Node是否滿足多數(shù)派條件，如果不滿足，則執(zhí)行rejoin。

     if (electMasterService.hasEnoughMasterNodes(remainingNodesClusterState.nodes()) == false) {
                final int masterNodes = electMasterService.countMasterNodes(remainingNodesClusterState.nodes());
                rejoin.accept(LoggerMessageFormat.format("not enough master nodes (has [{}], but needed [{}])",
                                                         masterNodes, electMasterService.minimumMasterNodes()));
                return resultBuilder.build(currentState);
            } else {
                return resultBuilder.build(allocationService.deassociateDeadNodes(remainingNodesClusterState, true, describeTasks(tasks)));
            }

在publish新的cluster_state時(shí)，分為send階段和commit階段，send階段要求多數(shù)派必須成功，然后再進(jìn)行commit。如果在send階段沒(méi)有實(shí)現(xiàn)多數(shù)派返回成功，那么可能是有了新的master或者是無(wú)法連接到多數(shù)派個(gè)節(jié)點(diǎn)等，則master需要執(zhí)行rejoin。

   try {
            publishClusterState.publish(clusterChangedEvent, electMaster.minimumMasterNodes(), ackListener);
        } catch (FailedToCommitClusterStateException t) {
            // cluster service logs a WARN message
            logger.debug("failed to publish cluster state version [{}](not enough nodes acknowledged, min master nodes [{}])",
                newState.version(), electMaster.minimumMasterNodes());

            synchronized (stateMutex) {
                pendingStatesQueue.failAllStatesAndClear(
                    new ElasticsearchException("failed to publish cluster state"));

                rejoin("zen-disco-failed-to-publish");
            }
            throw t;
        }

在對(duì)其他節(jié)點(diǎn)進(jìn)行定期的ping時(shí)，發(fā)現(xiàn)有其他節(jié)點(diǎn)也是master，此時(shí)會(huì)比較本節(jié)點(diǎn)與另一個(gè)master節(jié)點(diǎn)的cluster_state的version，誰(shuí)的version大誰(shuí)成為master，version小的執(zhí)行rejoin。

   if (otherClusterStateVersion > localClusterState.version()) {
            rejoin("zen-disco-discovered another master with a new cluster_state [" + otherMaster + "][" + reason + "]");
        } else {
            // TODO: do this outside mutex
            logger.warn("discovered [{}] which is also master but with an older cluster_state, telling [{}] to rejoin the cluster ([{}])", otherMaster, otherMaster, reason);
            try {
                // make sure we"re connected to this node (connect to node does nothing if we"re already connected)
                // since the network connections are asymmetric, it may be that we received a state but have disconnected from the node
                // in the past (after a master failure, for example)
                transportService.connectToNode(otherMaster);
                transportService.sendRequest(otherMaster, DISCOVERY_REJOIN_ACTION_NAME, new RejoinClusterRequest(localClusterState.nodes().getLocalNodeId()), new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {

                    @Override
                    public void handleException(TransportException exp) {
                        logger.warn((Supplier) () -> new ParameterizedMessage("failed to send rejoin request to [{}]", otherMaster), exp);
                    }
                });
            } catch (Exception e) {
                logger.warn((Supplier) () -> new ParameterizedMessage("failed to send rejoin request to [{}]", otherMaster), e);
            }
        }

集群擴(kuò)縮容
上面講了節(jié)點(diǎn)發(fā)現(xiàn)、Master選舉、錯(cuò)誤檢測(cè)等機(jī)制，那么現(xiàn)在我們可以來(lái)看一下如何對(duì)集群進(jìn)行擴(kuò)縮容。

1 擴(kuò)容DataNode
假設(shè)一個(gè)ES集群存儲(chǔ)或者計(jì)算資源不夠了，我們需要進(jìn)行擴(kuò)容，這里我們只針對(duì)DataNode，即配置為：

conf/elasticsearch.yml:

node.master: false
node.data: true

然后需要配置集群名、節(jié)點(diǎn)名等其他配置，為了讓該節(jié)點(diǎn)能夠加入集群，我們把discovery.zen.ping.unicast.hosts配置為集群中的master-eligible node。

conf/elasticsearch.yml:

cluster.name: es-cluster
node.name: node_Z
discovery.zen.ping.unicast.hosts: ["x.x.x.x", "x.x.x.y", "x.x.x.z"]

然后啟動(dòng)節(jié)點(diǎn)，節(jié)點(diǎn)會(huì)自動(dòng)加入到集群中，集群會(huì)自動(dòng)進(jìn)行rebalance，或者通過(guò)reroute api進(jìn)行手動(dòng)操作。

https://www.elastic.co/guide/...

https://www.elastic.co/guide/...

2 縮容DataNode
假設(shè)一個(gè)ES集群使用的機(jī)器數(shù)太多了，需要縮容，我們?cè)趺窗踩牟僮鱽?lái)保證數(shù)據(jù)安全，并且不影響可用性呢？

首先，我們選擇需要縮容的節(jié)點(diǎn)，注意本節(jié)只針對(duì)DataNode的縮容，MasterNode縮容涉及到更復(fù)雜的問(wèn)題，下面再講。

然后，我們需要把這個(gè)Node上的Shards遷移到其他節(jié)點(diǎn)上，方法是先設(shè)置allocation規(guī)則，禁止分配Shard到要縮容的機(jī)器上，然后讓集群進(jìn)行rebalance。

PUT _cluster/settings
{
"transient" : {

"cluster.routing.allocation.exclude._ip" : "10.0.0.1"

}
}
等這個(gè)節(jié)點(diǎn)上的數(shù)據(jù)全部遷移完成后，節(jié)點(diǎn)可以安全下線。

更詳細(xì)的操作方式可以參考官方文檔：

https://www.elastic.co/guide/...

3 擴(kuò)容MasterNode
假如我們想擴(kuò)容一個(gè)MasterNode(master-eligible node)， 那么有個(gè)需要考慮的問(wèn)題是，上面提到為了避免腦裂，ES是采用多數(shù)派的策略，需要配置一個(gè)quorum數(shù)：

conf/elasticsearch.yml:

discovery.zen.minimum_master_nodes: 2

假設(shè)之前3個(gè)master-eligible node，我們可以配置quorum為2，如果擴(kuò)容到4個(gè)master-eligible node，那么quorum就要提高到3。

所以我們應(yīng)該先把discovery.zen.minimum_master_nodes這個(gè)配置改成3，再擴(kuò)容master，更改這個(gè)配置可以通過(guò)API的方式：

curl -XPUT localhost:9200/_cluster/settings -d "{

"persistent" : {
    "discovery.zen.minimum_master_nodes" : 3
}

}"
這個(gè)API發(fā)送給當(dāng)前集群的master，然后新的值立即生效，然后master會(huì)把這個(gè)配置持久化到cluster meta中，之后所有節(jié)點(diǎn)都會(huì)以這個(gè)配置為準(zhǔn)。

但是這種方式有個(gè)問(wèn)題在于，配置文件中配置的值和cluster meta中的值很可能出現(xiàn)不一致，不一致很容易導(dǎo)致一些奇怪的問(wèn)題，比如說(shuō)集群重啟后，在恢復(fù)cluster meta前就需要進(jìn)行master選舉，此時(shí)只可能拿配置中的值，拿不到cluster meta中的值，但是cluster meta恢復(fù)后，又需要以cluster meta中的值為準(zhǔn)，這中間肯定存在一些正確性相關(guān)的邊界case。

總之，動(dòng)master節(jié)點(diǎn)以及相關(guān)的配置一定要謹(jǐn)慎，master配置錯(cuò)誤很有可能導(dǎo)致腦裂甚至數(shù)據(jù)寫壞、數(shù)據(jù)丟失等場(chǎng)景。

4 縮容MasterNode
縮容MasterNode與擴(kuò)容跟擴(kuò)容是相反的流程，我們需要先把節(jié)點(diǎn)縮下來(lái)，再把quorum數(shù)調(diào)下來(lái)，不再詳細(xì)描述。

與Zookeeper、raft等實(shí)現(xiàn)方式的比較

與使用Zookeeper相比

本篇講了ES集群中節(jié)點(diǎn)相關(guān)的幾大功能的實(shí)現(xiàn)方式：

節(jié)點(diǎn)發(fā)現(xiàn)
Master選舉
錯(cuò)誤檢測(cè)
集群擴(kuò)縮容
試想下，如果我們使用Zookeeper來(lái)實(shí)現(xiàn)這幾個(gè)功能，會(huì)帶來(lái)哪些變化？

Zookeeper介紹
我們首先介紹一下Zookeeper，熟悉的同學(xué)可以略過(guò)。

Zookeeper分布式服務(wù)框架是Apache Hadoop 的一個(gè)子項(xiàng)目，它主要是用來(lái)解決分布式應(yīng)用中經(jīng)常遇到的一些數(shù)據(jù)管理問(wèn)題，如：統(tǒng)一命名服務(wù)、狀態(tài)同步服務(wù)、集群管理、分布式應(yīng)用配置項(xiàng)的管理等。

簡(jiǎn)單來(lái)說(shuō)，Zookeeper就是用于管理分布式系統(tǒng)中的節(jié)點(diǎn)、配置、狀態(tài)，并完成各個(gè)節(jié)點(diǎn)間進(jìn)行配置和狀態(tài)的同步等。大量的分布式系統(tǒng)依賴于Zookeeper或者是類似的組件。

Zookeeper通過(guò)目錄樹(shù)的形式來(lái)管理數(shù)據(jù)，每個(gè)節(jié)點(diǎn)稱為一個(gè)znode，每個(gè)znode由3部分組成:

此為狀態(tài)信息, 描述該znode的版本, 權(quán)限等信息.
與該znode關(guān)聯(lián)的數(shù)據(jù).
該znode下的子節(jié)點(diǎn).
stat中有一項(xiàng)是ephemeralOwner，如果有值，代表是一個(gè)臨時(shí)節(jié)點(diǎn)，臨時(shí)節(jié)點(diǎn)會(huì)在session結(jié)束后刪除，可以用來(lái)輔助應(yīng)用進(jìn)行master選舉和錯(cuò)誤檢測(cè)。

Zookeeper提供watch功能，可以用于監(jiān)聽(tīng)相應(yīng)的事件，比如某個(gè)znode下的子節(jié)點(diǎn)的增減，某個(gè)znode本身的增減，某個(gè)znode的更新等。

怎么使用Zookeeper實(shí)現(xiàn)ES的上述功能

節(jié)點(diǎn)發(fā)現(xiàn)：每個(gè)節(jié)點(diǎn)的配置文件中配置一下Zookeeper服務(wù)器的地址，節(jié)點(diǎn)啟動(dòng)后到Zookeeper中某個(gè)目錄中注冊(cè)一個(gè)臨時(shí)的znode。當(dāng)前集群的master監(jiān)聽(tīng)這個(gè)目錄的子節(jié)點(diǎn)增減的事件，當(dāng)發(fā)現(xiàn)有新節(jié)點(diǎn)時(shí)，將新節(jié)點(diǎn)加入集群。

master選舉：當(dāng)一個(gè)master-eligible
node啟動(dòng)時(shí)，都嘗試到固定位置注冊(cè)一個(gè)名為master的臨時(shí)znode，如果注冊(cè)成功，即成為master，如果注冊(cè)失敗則監(jiān)聽(tīng)這個(gè)znode的變化。當(dāng)master出現(xiàn)故障時(shí)，由于是臨時(shí)znode，會(huì)自動(dòng)刪除，這時(shí)集群中其他的master-eligible
node就會(huì)嘗試再次注冊(cè)。使用Zookeeper后其實(shí)是把選master變成了搶master。

錯(cuò)誤檢測(cè)：由于節(jié)點(diǎn)的znode和master的znode都是臨時(shí)znode，如果節(jié)點(diǎn)故障，會(huì)與Zookeeper斷開(kāi)session，znode自動(dòng)刪除。集群的master只需要監(jiān)聽(tīng)znode變更事件即可，如果master故障，其他的候選master則會(huì)監(jiān)聽(tīng)到master
znode被刪除的事件，嘗試成為新的master。

集群擴(kuò)縮容：擴(kuò)縮容將不再需要考慮minimum_master_nodes配置的問(wèn)題，會(huì)變得更容易。

使用Zookeeper的好處是，把一些復(fù)雜的分布式一致性問(wèn)題交給Zookeeper來(lái)做，ES本身的邏輯就可以簡(jiǎn)化很多，正確性也有保證，這也是大部分分布式系統(tǒng)實(shí)踐過(guò)的路子。而ES的這套ZenDiscovery機(jī)制經(jīng)歷過(guò)很多次bug fix，到目前仍有一些邊角的場(chǎng)景存在bug，而且運(yùn)維也不簡(jiǎn)單。

那為什么ES不使用Zookeeper呢，大概是官方開(kāi)發(fā)覺(jué)得增加Zookeeper依賴后會(huì)多依賴一個(gè)組件，使集群部署變得更復(fù)雜，用戶在運(yùn)維時(shí)需要多運(yùn)維一個(gè)Zookeeper。

那么在自主實(shí)現(xiàn)這條路上，還有什么別的算法選擇嗎？當(dāng)然有的，比如raft。

2. 與使用raft相比
raft算法是近幾年很火的一個(gè)分布式一致性算法，其實(shí)現(xiàn)相比paxos簡(jiǎn)單，在各種分布式系統(tǒng)中也得到了應(yīng)用。這里不再描述其算法的細(xì)節(jié)，我們單從master選舉算法角度，比較一下raft與ES目前選舉算法的異同點(diǎn)：

相同點(diǎn)
多數(shù)派原則：必須得到超過(guò)半數(shù)的選票才能成為master。
選出的leader一定擁有最新已提交數(shù)據(jù)：在raft中，數(shù)據(jù)更新的節(jié)點(diǎn)不會(huì)給數(shù)據(jù)舊的節(jié)點(diǎn)投選票，而當(dāng)選需要多數(shù)派的選票，則當(dāng)選人一定有最新已提交數(shù)據(jù)。在es中，version大的節(jié)點(diǎn)排序優(yōu)先級(jí)高，同樣用于保證這一點(diǎn)。
不同點(diǎn)
正確性論證：raft是一個(gè)被論證過(guò)正確性的算法，而ES的算法是一個(gè)沒(méi)有經(jīng)過(guò)論證的算法，只能在實(shí)踐中發(fā)現(xiàn)問(wèn)題，做bug fix，這是我認(rèn)為最大的不同。
是否有選舉周期term：raft引入了選舉周期的概念，每輪選舉term加1，保證了在同一個(gè)term下每個(gè)參與人只能投1票。ES在選舉時(shí)沒(méi)有term的概念，不能保證每輪每個(gè)節(jié)點(diǎn)只投一票。
選舉的傾向性：raft中只要一個(gè)節(jié)點(diǎn)擁有最新的已提交的數(shù)據(jù)，則有機(jī)會(huì)選舉成為master。在ES中，version相同時(shí)會(huì)按照NodeId排序，總是NodeId小的人優(yōu)先級(jí)高。

raft從正確性上看肯定是更好的選擇，而ES的選舉算法經(jīng)過(guò)幾次bug fix也越來(lái)越像raft。當(dāng)然，在ES最早開(kāi)發(fā)時(shí)還沒(méi)有raft，而未來(lái)ES如果繼續(xù)沿著這個(gè)方向走很可能最終就變成一個(gè)raft實(shí)現(xiàn)。

raft不僅僅是選舉，下一篇介紹meta數(shù)據(jù)一致性時(shí)也會(huì)繼續(xù)比較ES目前的實(shí)現(xiàn)與raft的異同。

本篇介紹了Elasticsearch集群的組成、節(jié)點(diǎn)發(fā)現(xiàn)、master選舉、故障檢測(cè)和擴(kuò)縮容等方面的實(shí)現(xiàn)，與一般的文章不同，本文對(duì)其原理、存在的問(wèn)題也進(jìn)行了一些分析，并與其他實(shí)現(xiàn)方式進(jìn)行了比較。

作為Elasticsearch分布式一致性原理剖析系列的第一篇，本文先從節(jié)點(diǎn)入手，下一篇會(huì)介紹meta數(shù)據(jù)變更的一致性問(wèn)題，會(huì)在本文的基礎(chǔ)上對(duì)ES的分布式原理做進(jìn)一步分析。
詳情請(qǐng)閱讀原文