《E往無前》系列將著重展現騰訊云ES在持續(xù)深入優(yōu)化客戶所關心的「省！快！穩(wěn)！」訴求，能夠在低成本的同時兼顧高可用、高性能、高穩(wěn)定等特性，可以滿足微盟、小紅書、微信支付等內外部大客戶的核心場景需求。

E往無前?|?海量數據ES擴展難？騰訊云ES 擴展百萬級分片也“So Easy~”

導語：ES 以易用性與實效性著稱，易用性得益于 ES 有穩(wěn)健的分布式架構。但在 ES 走向海量規(guī)模的背景下，對日常運維、研發(fā)的專業(yè)度要求比較高，往往需要深入理解集群架構原理，本文結合最新版本源碼剖析 ES 的分布式底層架構原理，為廣大 ES 愛好者揭開底層架構的面紗，幫助運維、研發(fā)同學進一步熟悉內核。于此同時，海量場景 ES 的擴展性也面臨諸多挑戰(zhàn)，如分片數、節(jié)點數等，本文也為大家分享 TencentES Oteam 內核是如何優(yōu)化擴展性瓶頸的。

1、背景

? ?Elasticsearch 是一個實時的分布式搜索分析引擎，簡稱 ES。一個集群由多個節(jié)點組成，節(jié)點的角色可以根據用戶的使用場景自由配置，集群可以以節(jié)點為單位自由擴縮容，數據以索引、分片的形式散列在各個節(jié)點上。本文介紹 ES 分布式架構基礎原理，剖析分布式元數據管理模型，并介紹騰訊云 ES 在分布式擴展性層面相關的優(yōu)化，解析代碼基于 8.5 版本。

2、分布式架構思路

? ? ? ? 我們首先來看看 ES 的分布式架構。一個 ES 集群包含多個節(jié)點，節(jié)點的類型有多種，最新版本主要的角色類型包括：

• master 節(jié)點：可設置專屬 master 節(jié)點，也可以和其它節(jié)點角色共享。

• 數據節(jié)點：?data、data_hot、data_warm、data_cold、data_frozen 等。

• 功能節(jié)點：?ingest、ml、remote_cluster_client、transform 等。

????節(jié)點角色分為三類：主節(jié)點（master）用于管理集群；數據節(jié)點（data）用于存儲管理數據，數據節(jié)點按照數據特點分為多種類型，冷溫熱等等；功能節(jié)點，包括 ETL、機器學習、遠程管理等等。下面我們拿一種典型的大規(guī)模集群場景架構來舉例說明。

ES 分布式架構

? ??上圖所示的分布式架構中，上面部分是專屬 master 節(jié)點，負責管理集群的元數據，他們之間通過基于類 Raft 的分布式一致性協(xié)議進行選主、元數據同步。下面部分是專屬 data 節(jié)點，data 節(jié)點上包含索引分片。索引對應傳統(tǒng)關系數據庫的表，是邏輯概念，一個索引包含多個分片，分片是節(jié)點上的數據存儲單元，它按照主鍵 hash 或用戶自定義數據路由（routing）均勻分布到各個節(jié)點。分片一般包含一主（primary）、多從（replica）分片，一個索引主從分片之間的數據復制模型基于微軟提出的 PacificA 協(xié)議。

3、 讀寫模型

3.1 寫入模型

ES 的任意節(jié)點可作為寫入請求的協(xié)調節(jié)點，接收用戶請求，協(xié)調節(jié)點先將寫入請求 hash 至分片粒度并先轉發(fā)對應主分片寫入，主分片寫入成功再轉發(fā)至從分片，主從分片均寫入完畢經協(xié)調節(jié)點返回客戶端成功。

? ??騰訊云 ES 內核團隊實現了寫入分組定向路由、主從分片物理復制能力，可以減少從分片的寫入棧計算開銷，寫入性能提升50%+。

分布式寫入模型

? ??本文不詳細解析寫入過程，有興趣同學可參考寫入流程解析：https://cloud.tencent.com/developer/article/1370501

3.2 查詢模型

? ??和寫入一樣，ES 的任意節(jié)點可以作為查詢請求的協(xié)調節(jié)點，請求轉發(fā)至對應一個或多個（取決于路由規(guī)則，不指定路由默認索引所有分片均查詢）數據分片的主或者從分片進行查詢，查詢根據復雜度分不同類型，QUERY_THEN_FETCH（兩階段），QUERY_AND_FETCH（一階段），各個分片查詢結果最后在協(xié)調節(jié)點匯聚，返回最終結果給客戶端。

分布式查詢模型

本文不詳細解析查詢過程，有興趣同學可參考查詢流程解析：https://cloud.tencent.com/developer/article/1154813

4、分布式架構元數據模型分類

? ? ? ? 在大規(guī)模分布式存儲架構中，元數據模型主要分為中心化架構和對等架構兩類。中心化架構的典型代表包括 HDFS、ES、BigTable 等，對等架構典型的系統(tǒng)包括 Cassandra、Dynamo 等。

4.1 中心化架構

? ? ? ??中心化架構的特點是專有的主節(jié)點管理集群元數據，HDFS 中 Namenode 統(tǒng)一管理元數據，data node 不單獨維護，客戶端找 Namenode 獲取路由信息。中心化架構的優(yōu)勢是在大規(guī)模集群場景下，元數據同步范圍收斂，效率更高。由于主節(jié)點提供路由查詢信息，因此其主要缺點是主節(jié)點易成為瓶頸，一般通過 Federation 機制優(yōu)化，或者將元數據存儲到數據表中分布在部分節(jié)點上管理。ES 也采用了中心化架構，稍后我們展開介紹。

中心化架構

4.2 對等架構

? ? ? ? 對等架構的特點是去中心化，沒有獨立的主節(jié)點。節(jié)點之間通過 Gossip 協(xié)議傳播同步元數據，所有節(jié)點保存全量的元數據。這種方式架構簡單清晰，沒有中心化瓶頸。缺點是通過 Gossip 協(xié)議收斂元數據效率偏低，受節(jié)點數量限制，擴展性弱。

對等架構

5、元數據模型

????????前面我們介紹了 ES 的分布式架構基礎原理及讀寫基本流程，也了解了業(yè)內常用的分布式架構元數據管理模型。下面我們來看看 ES 是如何管理集群的，其核心元數據模型是如何運作的。

??????? ES 的元數據由 master 節(jié)點維護管理，同時其它節(jié)點也維護著全量的元數據，目的是為了確保每個節(jié)點都能承擔數據路由的能力。元數據包括節(jié)點信息、索引信息、分片路由信息、配置信息等等，下面我們先揭開元數據在內存中的神秘面紗，然后再看看元數據是如何持久化的。

5.1 內存結構

??????? ES 元數據的內存數據結構對象是 ClusterState，代碼層面對應的是 ClusterState.java。其主要的成員包含：

? ? 上面的內存結構中，對象占比最大的是 nodes、metaData、routingTable 和 routingNodes，這四塊是元數據的核心組成部分：

元數據核心結構

??????接下來對這四個部分的內部結構進行展開分析。

5.1.1 DiscoveryNodes

? ? ? ? DiscoveryNodes 對象描述的是集群的節(jié)點信息。里面包含如下主要的成員信息：

// 當前 master
private final String masterNodeId;
// 本地節(jié)點
private final String localNodeId;
// 完整的節(jié)點列表
private final ImmutableOpenMap<String, DiscoveryNode> nodes;


// 其它分類型節(jié)點列表
private final ImmutableOpenMap<String, DiscoveryNode> dataNodes;
private final ImmutableOpenMap<String, DiscoveryNode> masterNodes;
private final ImmutableOpenMap<String, DiscoveryNode> ingestNodes;

? ? ? ??從變量名稱很容易理解，其包含了本地節(jié)點、當前 master 節(jié)點的 id 信息，以及集群全量節(jié)點列表（nodes），并將這個列表分類為不同類型的節(jié)點列表。

5.1.2 MetaData

? ? ? ? MetaData 包含集群全局的配置信息。主要的成員包括：

// 集群的動態(tài)設定，transient 全量重啟后就沒有了，persistent 會持久化
private final Settings transientSettings;
private final Settings persistentSettings;
// 索引、模板信息
private final ImmutableOpenMap<String, IndexMetaData> indices;
private final ImmutableOpenMap<String, IndexTemplateMetaData> templates;
// 集群選舉管理信息
private final CoordinationMetaData coordinationMetaData;
// raft 選舉使用的 term
private final long term;
// 最近提交的選舉節(jié)點列表，內部是一個 Set<String> nodeIds
private final VotingConfiguration lastCommittedConfiguration;
// 最近接收的選舉節(jié)點列表
private final VotingConfiguration lastAcceptedConfiguration;
// 用戶通過 _cluster/voting_config_exclusions 接口設定的選舉排除節(jié)點列表
private final Set<VotingConfigExclusion> votingConfigExclusions;

5.1.3 RoutingTable

??????? RoutingTable 包含主要的數據路由信息，在查詢、寫入請求的時候提供分片到節(jié)點的路由，動態(tài)構建不會持久化。我們先來通過一張圖了解這個數據路由結構的包含關系：

分片到節(jié)點的映射（路由表）

????????一個 RoutingTable 包含集群的所有索引，一個索引包含多個分片，一個分片包含一個主、多個從本分片，最底層的 ShardRouting 描述具體的某個副本分片所在的節(jié)點以及正在搬遷的目標節(jié)點信息。用戶請求時只指定索引信息，請求到達協(xié)調節(jié)點，由協(xié)調節(jié)點根據該路由表來獲取底層分片所在節(jié)點并轉發(fā)請求。

????????接下來看看對應的數據結構，RoutingTable 頂層主要的成員只有一個：

// key 為索引名，IndexRoutingTable 為該索引包含的分片信息
private final ImmutableOpenMap<String, IndexRoutingTable> indicesRouting;

??????? IndexRoutingTable 對象：

// 索引信息，包含索引名稱、uuid 等。
private final Index index;
// 索引的分片列表，key 為分片的編號，例如 2 號分片，3 號分片
private final ImmutableOpenIntMap<IndexShardRoutingTable> shards;

????????一套分片信息包含一個主、多個從分片，其對象結構 IndexShardRoutingTable：

// 當前分片信息，主要包含分片的編號、分片對應的索引信息
final ShardId shardId;
// 當前分片的主分片信息
final ShardRouting primary;
// 當前分片的多個從分片列表
final List<ShardRouting> replicas;
// 當前分片全量的分片列表，包含主、副本
final List<ShardRouting> shards;
// 當前分片已經 started 的活躍分片列表
final List<ShardRouting> activeShards;
// 當前分片已經分配了節(jié)點的分片列表
final List<ShardRouting> assignedShards;

? ? ? ??分片最底層的數據結構是 ShardRouting ，它描述這個分片的狀態(tài)、節(jié)點歸屬等信息：

private final ShardId shardId;
// 分片所在的當前節(jié)點 id
private final String currentNodeId;
// 如果分片正在搬遷，這里為目標節(jié)點 id
private final String relocatingNodeId;
// 是否是主分片
private final boolean primary;
// 分片的狀態(tài)，UNASSIGNED/INITIALIZING/STARTED/RELOCATING
private final ShardRoutingState state;
// 每一個分片分配后都有一個唯一標識
private final AllocationId allocationId;

5.1.4 RoutingNodes

????????該對象為節(jié)點到分片的映射關系。主要用于統(tǒng)計每個節(jié)點的分片分配情況，在分片分配、均衡的時候使用，需要時根據 RoutingTable 動態(tài)構建，不會持久化。

節(jié)點到分片的映射

?? ? ? RoutingNodes 的主要成員包括：

// 節(jié)點到分片的映射，key 為 nodeId，value 為一個包含分片列表的節(jié)點信息
private final Map<String, RoutingNode> nodesToShards = new HashMap<>();
// 未分配的分片列表，每次 reroute 從這里獲取分片分配
private final UnassignedShards unassignedShards = new UnassignedShards(this);
// 已經分配的分片列表
private final Map<ShardId, List<ShardRouting>> assignedShards = new HashMap<>();
// 當前分片遠程恢復的統(tǒng)計數據 incoming/outcoming
private final Map<String, Recoveries> recoveriesPerNode = new HashMap<>();

? ? ? ??上面 RoutingNode 的主要結構包括：

private final String nodeId;
// 該節(jié)點上的分片列表
private final LinkedHashMap<ShardId, ShardRouting> shards; 
// 該節(jié)點上的初始化中的分片列表
private final LinkedHashSet<ShardRouting> initializingShards;
// 該節(jié)點上的正在搬遷（搬走）的分片列表
private final LinkedHashSet<ShardRouting> relocatingShards;

????????前面就是所有元數據在內存中的結構，最后我們借助這張 UML 圖來看元數據的整體結構和關系：

元數據類結構

5.2 元數據持久化

???????前面介紹的元數據內存結構中，只有部分信息會被持久化存儲到磁盤上，其它結構都是在節(jié)點啟動后動態(tài)構建或者直接在內存中動態(tài)維護的。持久化的內容包含兩部分，索引元數據和集群元數據也叫全局元數據，下面我們分別來介紹這兩部分持久化的內容。

5.2.1 索引元數據（index metadata）

? ? ? ?索引元數據維護的是各個索引獨有的配置信息，持久化的內容主要包括：

• in_sync_allocations：每個分片分配之后都有一個唯一的 allocationId，該列表是主分片用來維護被認為跟自己保持數據一致的副本列表。從這個列表中踢出的分片不會接收新的寫入，需要走分片恢復流程將數據追齊之后才能再被放入該隊列。

• mappings：索引的 mapping 信息，定義各個字段的類型、屬性。

• settings：索引自己的配置信息。

• state：索引狀態(tài)，OPEN/CLOSED。

• aliases：索引別名信息。

• routing_num_shards：索引分片數量。

• primary_terms：每一輪分片切主都會產生新的 primary term，用于保持主從分片之間的一致性。

5.2.2 全局元數據（global metadata）

????????全局元數據持久化的內容主要是前面描述的 MetaData 對象中去除 indices 索引部分。包括動態(tài)配置、模板信息、選舉信息等。主要包含三部分：

• manifest-數字.st：該文件是一個磁盤元數據管理入口，主要管理元數據的版本信息。

• global-數字.st：MetaData 中的全局元數據的主要信息就是持久化到這個文件，包括動態(tài)配置、模板信息等。

• node-數字.st：當前節(jié)點的元數據信息，包括節(jié)點的 nodeId 和版本信息。

5.2.3 元數據文件分布

????????在 7.6.0 版本之前，元數據直接存放在每個節(jié)點磁盤上，且在專有 master 節(jié)點上每個索引一個元數據目錄。在 7.6.0 版本之后，ES 將元數據放到節(jié)點本地獨立的 lucene 索引中保存。7.6.0 之前持久化的數據目錄包含的內容：

├── indices
│   ├── 00IUbRWZSzGsN71k0TlPmA
│   │   └── _state
│   │   └── state-8.st
│   ├── 01RTGqCbRe-9PYO55j8zAQ
│   │   └── _state
│   │   └── state-10.st
│   ├── 02DSrGNzRzizuI42Yf6aJg
│   │   └── _state
│   │   └── state-17.st


├── node.lock
└── _state
├── global-4928.st
└── node-0.st

????????上圖中上半部分是每個索引的元數據，它們位于 indices 目錄下，一個索引對應一個 uuid 目錄，下面有一個 _state 文件夾，索引自己的元數據就放在這個 _state 下面。下半部分就是我們上面描述的全局元數據持久化的文件。

??????? 7.6.0 之后 data/nodes/0 里面保存的是 segment 文件，元數據以本地 Lucene index 方式持久化，收斂文件數量。其中 node-0.st 是舊版升級后的兼容文件：

├── node.lock
└── _state
├── _3q.cfe
├── _3q.cfs
├── _3q.si
├── _3r.cfe
├── _3r.cfs
├── _3r.si
├── node-0.st
├── segments_6i
└── write.lock

? ? ? ?元數據索引同時保存全局、索引元數據，用 type 字段區(qū)分，索引包含三個字段：

• type：cluster 級別元數據 “global”；索引級別元數據 “index”。

• index_uuid：索引元數據對應索引的 UUID。

• data：metadata 元數據內容。

? ? ? ?同時，在每次提交保存的時候，會存一份 commit user data，包括選舉的 term，最新的版本、node id、node version 等信息：

* +------------------------------+-----------------------------+----------------------------------------------+
* | "type" (string field)        | "index_uuid" (string field) | "data" (stored binary field in SMILE format) |
* +------------------------------+-----------------------------+----------------------------------------------+
* | GLOBAL_TYPE_NAME == "global" | (omitted)                   | Global metadata                              |
* | INDEX_TYPE_NAME  == "index"  | Index UUID                  | Index metadata                               |
* +------------------------------+-----------------------------+----------------------------------------------+
*
* Additionally each commit has the following user data:
*
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+
* |        Key symbol         |       Key literal       |                                     Value                                     |
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+
* | CURRENT_TERM_KEY          | "current_term"          | Node's "current" term (≥ last-accepted term and the terms of all sent joins)  |
* | LAST_ACCEPTED_VERSION_KEY | "last_accepted_version" | The cluster state version corresponding with the persisted metadata           |
* | NODE_ID_KEY               | "node_id"               | The (persistent) ID of the node that wrote this metadata                      |
* | NODE_VERSION_KEY          | "node_version"          | The (ID of the) version of the node that wrote this metadata                  |
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+

6、元數據管理

??????? ES 的索引創(chuàng)建、刪除、mapping 更新、template 更新、節(jié)點加入脫離等 DDL 操作都會涉及到元數據變更。前面我們從內存、持久化層面介紹了 ES 元數據的組成部分，接下來我們看看 ES 是如何對元數據進行管理的。

6.1 元數據初始化

??????? ES 節(jié)點的啟動入口在 Node.java 的 start 函數中，里面會初始化各種 service。元數據管理相關的 service 也是在這個環(huán)節(jié)進行初始化。GatewayMetaState 對象負責元數據的存取，在節(jié)點啟動過程中會根據節(jié)點的類型，確定元數據的存取方式。分為以下幾種場景：

• master 節(jié)點。基于 LucenePersistedState 對象同步落盤，每當節(jié)點收到新的元數據的時候會馬上保存到磁盤。并更新緩存，讀取的時候優(yōu)先讀取緩存對象，節(jié)點啟動的時候緩存會從磁盤加載。

• data 節(jié)點。包括前述 data 前綴的 role，基于 AsyncPersistedState 對象異步落盤，對應異步線程名稱是 AsyncLucenePersistedState#updateTask。

• 其它功能節(jié)點。非 master 屬性且不保存數據的節(jié)點，例如 ingest、ml、transform 等?；?InMemoryPersistedState 對象直接保存到內存不做持久化。無論是在內存中的還是持久化的元數據對象，它們都對外暴露一個 PersistedState 接口。提供保存最后接收的元數據（ setLastAcceptedState），以及獲取最后接收的元數據（getLastAcceptedState）。

GatewayMetaState.java 代碼片段：

if (DiscoveryNode.isMasterNode(settings) || DiscoveryNode.canContainData(settings)) {
    ......
    if (DiscoveryNode.isMasterNode(settings)) {
        persistedState = new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState);
    } else {
        persistedState = new AsyncPersistedState(
            settings,
            transportService.getThreadPool(),
            new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState)
        );
    }        
    ......
    this.persistedState.set(persistedState);
} else {
    ......
    persistedState.set(new InMemoryPersistedState(currentTerm, clusterState));
}

6.2 發(fā)布流程

????????接下來我們以最典型的 DDL 創(chuàng)建索引操作為例，介紹整個元數據發(fā)布流程。ES 的 DDL 操作都是通過元數據變更推導模式實現的，例如創(chuàng)建索引，首先 master 會產生一版帶新增索引的元數據，并將該新版元數據發(fā)布至各個節(jié)點，各個節(jié)點和自己上一個持久化的元數據版本進行比對，產生差異化的索引執(zhí)行創(chuàng)建索引、分片的任務。下圖是一張宏觀的索引創(chuàng)建元數據變更流程，方便大家從整體架構上有個初步的了解。

索引創(chuàng)建元數據變更流程

????????接下來分別從 master、data 節(jié)點維度分別看著兩階段的處理流程。下面是兩個階段的各個狀態(tài)：

enum PublicationTargetState {
    NOT_STARTED,
    FAILED,
    SENT_PUBLISH_REQUEST, // 已發(fā)送 publish 請求
    WAITING_FOR_QUORUM,   // 等待大多數節(jié)點響應
    SENT_APPLY_COMMIT,    // 已發(fā)送 commit 請求
    APPLIED_COMMIT,       // 元數據應用完畢
}

6.2.1 Master publish 階段

????????用戶發(fā)起 create index 請求，會先到達 rest 層，由 RestCreateIndexAction 解析請求，并產生 transport 層的請求轉發(fā)給 master 處理。調用鏈：

RestCreateIndexAction -> TransportCreateIndexAction -> MetadataCreateIndexService

所有參數解析完畢之后，就進入提交集群元數據變更任務流程：

private void onlyCreateIndex(final CreateIndexClusterStateUpdateRequest request, final ActionListener<AcknowledgedResponse> listener) {
    normalizeRequestSetting(request);
    //提交元數據變更任務，索引創(chuàng)建優(yōu)先級為 URGENT
    submitUnbatchedTask(
        "create-index [" + request.index() + "], cause [" + request.cause() + "]",
        new AckedClusterStateUpdateTask(Priority.URGENT, request, listener) {


            @Override
            public ClusterState execute(ClusterState currentState) throws Exception {
                // 分配分片產生新版本的元數據
                return applyCreateIndexRequest(currentState, request, false);
            }


            @Override
            public void onFailure(Exception e) {
                ......
            }
        }
    );
}

????????代碼中 applyCreateIndexRequest 主要負責索引創(chuàng)建過程中，分配分片產生新版本的元數據，里面涉及復雜的分片分配、均衡策略流程，騰訊云 ES 內核結合單個索引分片數、節(jié)點主分片數、節(jié)點總分片數、節(jié)點存儲空間多個維度深度定制優(yōu)化了均衡策略，徹底解決社區(qū)版各個維度負載不均的問題，這里不展開，以后再單獨介紹。

????????上述任務經過 MasterService.submitStateUpdateTask 包裝之后提交給線程池執(zhí)行：

public void submitTask(BatchedTask task, @Nullable TimeValue timeout) throws EsRejectedExecutionException {
    tasksPerBatchingKey.compute(task.batchingKey, (k, existingTasks) -> {
        if (existingTasks == null) {
            existingTasks = Collections.synchronizedSet(new LinkedHashSet<>());
        } else {
            assert assertNoDuplicateTasks(task, existingTasks);
        }
        // 按 executor 匯總 task
        existingTasks.add(task);
        return existingTasks;
    });


    if (timeout != null) {
        // 指定超時時間，不指定默認元數據變更 30s 超時
        threadExecutor.execute(task, timeout, () -> onTimeoutInternal(task, timeout));
    } else {
        threadExecutor.execute(task);
    }
}

????????元數據任務支持 batch 方式執(zhí)行，相同類別的任務例如 template 變更任務可以批量變更，參考 MetadataIndexTemplateService.TEMPLATE_TASK_EXECUTOR。創(chuàng)建索引任務不能批量執(zhí)行。元數據任務執(zhí)行采用的是單線程多任務按優(yōu)先級串行執(zhí)行的模式，線程名稱為 _masterService#updateTask_，上述 threadExecutor 的初始化過程：

protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
    return EsExecutors.newSinglePrioritizing(
        // masterService#updateTask，我們經常在 jstack 堆棧中看到的 master 節(jié)點元數據變更線程
        nodeName + "/" + MASTER_UPDATE_THREAD_NAME,
        daemonThreadFactory(nodeName, MASTER_UPDATE_THREAD_NAME),
        threadPool.getThreadContext(),
        threadPool.scheduler(),
        ......
    );
}

元數據變更的優(yōu)先級：

public enum Priority {
    IMMEDIATE((byte) 0), // 節(jié)點加入、脫離等
    URGENT((byte) 1), // 索引、template 創(chuàng)建等
    HIGH((byte) 2),
    NORMAL((byte) 3),
    LOW((byte) 4),
    LANGUID((byte) 5);
}

元數據變更任務產生了新的元數據之后，就會進入元數據 publish 階段，調用棧：

MasterService.publishClusterStateUpdate() -> MasterService.publish() -> Coordinator.publish()

核心邏輯在 Coordinator 中處理，該類負責管理元數據的 publish、commit 流程。publish 函數的代碼片段：

@Override
public void publish(
    ClusterStatePublicationEvent clusterStatePublicationEvent,
    ActionListener<Void> publishListener,
    AckListener ackListener
) {
    try {
        synchronized (mutex) {
           ......
            try {
                ......
                // 獲取目標發(fā)布節(jié)點
                final DiscoveryNodes publishNodes = publishRequest.getAcceptedState().nodes();
                // leader checker 在數據節(jié)點上用于追蹤 master，和 master 保持心跳
                leaderChecker.setCurrentNodes(publishNodes);
                // 設置當前 leader 需要追蹤的 follower 節(jié)點，主要是用于心跳保持
                followersChecker.setCurrentNodes(publishNodes);
                // lag 探測器，如果某個節(jié)點超過指定時間默認 90s 沒有 publish 成功則踢出
                lagDetector.setTrackedNodes(publishNodes);
                // 啟動 publish 任務
                publication.start(followersChecker.getFaultyNodes());
            } finally {
                publicationContext.decRef();
            }
        }
    } catch (Exception e) {
       ......
    }
}

依次發(fā)送給各節(jié)點：

public void start(Set<DiscoveryNode> faultyNodes) {
    logger.trace("publishing {} to {}", publishRequest, publicationTargets);
    publicationTargets.forEach(PublicationTarget::sendPublishRequest);
}
void sendPublishRequest() {
    Publication.this.sendPublishRequest(discoveryNode, publishRequest, new PublishResponseHandler());
}

??????? sendPublishRequest 內部會分全量和增量兩種發(fā)送方式，在節(jié)點脫離之后重新加入或者有版本落后的情況下元數據會全量發(fā)送。至此 master 節(jié)點的發(fā)送流程就結束了，接下來就會在上面的 PublishResponseHandler 中等待多數節(jié)點響應后發(fā)起 commit 請求。

6.2.2 數據節(jié)點接收 publish

????????實際處理 publish 的節(jié)點不止數據節(jié)點，還包括不帶數據屬性的所有節(jié)點，也包括當前 master 節(jié)點本身。只是我們?yōu)榱朔奖忝枋?，以數據?jié)點處理 publish 邏輯為主。數據節(jié)點接收 publish 請求的入口在 PublicationTransportHandler.handleIncomingPublishRequest()，接收的 ClusterState 分為全量和增量兩種，用一個 boolean 區(qū)分，代碼片段：

private PublishWithJoinResponse handleIncomingPublishRequest(BytesTransportRequest request) throws IOException {
    StreamInput in = request.bytes().streamInput();
    try {
        ......
        // 全量或增量標記
        if (in.readBoolean()) {
            final ClusterState incomingState;
            ......
            final PublishWithJoinResponse response = acceptState(incomingState);
            // 這里 lastSeenClusterState 的目的就是為了下次接收增量之后應用 diff
            lastSeenClusterState.set(incomingState);
            return response;
        } else {
            final ClusterState lastSeen = lastSeenClusterState.get();
            if (lastSeen == null) {
                throw new IncompatibleClusterStateVersionException("have no local cluster state");
            } else {
                // 基于 lastSeen 應用增量 diff 產生新的完整的 incomingState
                ClusterState incomingState;
                try {
                    final Diff<ClusterState> diff;
                    // Close stream early to release resources used by the de-compression as early as possible
                    try (StreamInput input = in) {
                        diff = ClusterState.readDiffFrom(input, lastSeen.nodes().getLocalNode());
                    }
                    incomingState = diff.apply(lastSeen); // might throw IncompatibleClusterStateVersionException
                } catch (Exception e) {
                    throw e;
                }
                ......
                final PublishWithJoinResponse response = acceptState(incomingState);
                // 本地變量 CAS 替換成全局
                lastSeenClusterState.compareAndSet(lastSeen, incomingState);
                return response;
            }
        }
    } finally {
        IOUtils.close(in);
    }
}

????????序列化完畢后進入本地 accept 階段。上述片段中 acceptState 函數內部先會做元數據的各種校驗，比如 term、clusterUUID 等最終更新 lastAcceptedState 并持久化，在元數據未 commit 之前元數據的 metaData 部分的 clusterUUIDCommitted 屬性為 false。元數據的標識是通過 clusterUUID 來區(qū)分的，判斷一個集群的多個節(jié)點具備相同的元數據版本可根據此屬性，并結合 clusterUUIDCommitted 屬性確定元數據是已經 commit 的版本。處理接收請求代碼片段：

public PublishResponse handlePublishRequest(PublishRequest publishRequest) {
    final ClusterState clusterState = publishRequest.getAcceptedState();
    if (clusterState.term() != getCurrentTerm()) {
       ......
    }
    if (clusterState.term() == getLastAcceptedTerm() && clusterState.version() <= getLastAcceptedVersion()) {
       ......
    }
    // 更新 lastAcceptedState
    persistedState.setLastAcceptedState(clusterState);
    return new PublishResponse(clusterState.term(), clusterState.version());
}

6.2.3 Master 節(jié)點發(fā)起 commit

????????在 master 節(jié)點發(fā)送 publish 請求給各個節(jié)點后，會在等待半數以上節(jié)點響應才會進入 commit 流程。代碼片段：

public Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse) {
    ......
    publishVotes.addVote(sourceNode);
    if (isPublishQuorum(publishVotes)) {
        return Optional.of(new ApplyCommitRequest(localNode, publishResponse.getTerm(), publishResponse.getVersion()));
    }
    return Optional.empty();
}

一旦達到多數節(jié)點響應后，master 發(fā)起 apply commit 請求：

@Override
protected void sendApplyCommit(
    DiscoveryNode destination,
    ApplyCommitRequest applyCommit,
    ActionListener<Empty> responseActionListener
) {
    transportService.sendRequest(
        destination,
        COMMIT_STATE_ACTION_NAME,
        applyCommit,
        COMMIT_STATE_REQUEST_OPTIONS,
        new ActionListenerResponseHandler<>(wrapWithMutex(responseActionListener), in -> Empty.INSTANCE, Names.CLUSTER_COORDINATION)
    );
}

ApplyCommitRequest 請求比較簡單，僅包含 sourceNode、term、version 三個字段。

6.2.4 數據節(jié)點處理 commit

????????數據節(jié)點處理 commit 請求的入口在 Coordinator.java。一方面標記接收的元數據為 commited 狀態(tài)，另外進行元數據應用。代碼片段：

private void handleApplyCommit(ApplyCommitRequest applyCommitRequest, ActionListener<Void> applyListener) {
    synchronized (mutex) {
        // 將我們之前接收到的 acceptedState 的元數據改為 commit 狀態(tài)
        coordinationState.get().handleCommit(applyCommitRequest);
        ......
        if (applyCommitRequest.getSourceNode().equals(getLocalNode())) {
            // master node applies the committed state at the end of the publication process, not here.
            applyListener.onResponse(null);
        } else {
            // 元數據接收完畢，進入應用環(huán)節(jié)
            clusterApplier.onNewClusterState(applyCommitRequest.toString(), () -> applierState, applyListener.map(r -> {
                onClusterStateApplied();
                return r;
            }));
        }
    }
}

? ? ? ??數據節(jié)點 commit 元數據之后，元數據的發(fā)布流程就完畢了，之后數據節(jié)點進入異步應用環(huán)節(jié)。異步單線程處理，線程名稱是 clusterApplierService#updateTask 我們在很多 jstack 中看到該線程名稱即為數據節(jié)點的元數據應用線程。

????????元數據應用流程核心邏輯在 ClusterApplierService.java 的 applyChanges 函數中，主要是處理一些列元數據變更附屬的任務，例如創(chuàng)建、刪除索引、template 維護等，另外 master 節(jié)點上還會回調一些元數據變更完成后關聯的 listener。

private void applyChanges(ClusterState previousClusterState, ClusterState newClusterState, String source, Recorder stopWatch) {
    ......
    logger.debug("apply cluster state with version {}", newClusterState.version());
    // 進行元數據應用，會按優(yōu)先級分為 low/normal/high 來應用，例如創(chuàng)建索引屬于 high
    callClusterStateAppliers(clusterChangedEvent, stopWatch);


    logger.debug("set locally applied cluster state to version {}", newClusterState.version());
    // 更新本地最新 commited 的 clusterState
    state.set(newClusterState);


    // 這里一般是 master 節(jié)點發(fā)起元數據 commit 結束后，再回調相應的 listener
    callClusterStateListeners(clusterChangedEvent, stopWatch);
}

????????我們主要以創(chuàng)建索引來介紹元數據變更主流程，上述 callClusterStateAppliers apply 元數據的環(huán)節(jié)會進到 IndicesClusterStateService.java applyClusterState 函數，該函數有一連串應用最新 state 的流程：

@Override
public synchronized void applyClusterState(final ClusterChangedEvent event) {
    // 最新 commit 的元數據
    final ClusterState state = event.state();
    ......
    // 刪除索引、分片，清理磁盤分片目錄
    deleteIndices(event); // also deletes shards of deleted indices


    // 刪除索引、分片，只是清理內存對象，主要是針對 Close/Open 操作
    removeIndicesAndShards(event); // also removes shards of removed indices


    // 更新索引 settings、mapping 等
    updateIndices(event); // can also fail shards, but these are then guaranteed to be in failedShardsCache


    // 創(chuàng)建索引、分片
    createIndicesAndUpdateShards(state);
}

??????? createIndicesAndUpdateShards 主要處理索引創(chuàng)建任務，根據接收到的元數據，對比內存最新的索引列表，找出需要創(chuàng)建的索引列表進行創(chuàng)建，同時創(chuàng)建缺失的分片信息：

private?void?createIndicesAndUpdateShards(final?ClusterState?state)?{
    DiscoveryNodes nodes = state.nodes();
    // 通過節(jié)點到分片的映射，獲取屬于該節(jié)點的索引分片信息
    RoutingNode localRoutingNode = state.getRoutingNodes().node(nodes.getLocalNodeId());
  
    // 對比接收到的元數據和本節(jié)點內存的索引列表，找出新增的需要創(chuàng)建的索引
    final Map<Index, List<ShardRouting>> indicesToCreate = new HashMap<>();
    for (ShardRouting shardRouting : localRoutingNode) {
        ShardId shardId = shardRouting.shardId();
        if (failedShardsCache.containsKey(shardId) == false) {
            final Index index = shardRouting.index();
            final var indexService = indicesService.indexService(index);
            if (indexService == null) {
                // 不存在的索引就是需要創(chuàng)建的
                indicesToCreate.computeIfAbsent(index, k -> new ArrayList<>()).add(shardRouting);
            } else {
                // 索引存在看看有沒需要創(chuàng)建或更新的分片
                createOrUpdateShard(state, nodes, routingTable, shardRouting, indexService);
            }
        }
    }


    // 過濾出來的待創(chuàng)建索引列表，遍歷依次創(chuàng)建
    for (Map.Entry<Index, List<ShardRouting>> entry : indicesToCreate.entrySet()) {
        final Index index = entry.getKey();
        final IndexMetadata indexMetadata = state.metadata().index(index);
        logger.debug("[{}] creating index", index);


        try {
            // 創(chuàng)建索引
            indicesService.createIndex(indexMetadata, buildInIndexListener, true);
        } catch (Exception e) {
            ......
        }
        // 創(chuàng)建索引對應的本地分片
        for (ShardRouting shardRouting : entry.getValue()) {
            createOrUpdateShard(state, nodes, routingTable, shardRouting, indexService);
        }
    }
}

????????至此，一個完整的元數據變更流程就介紹完了?？傮w來看，master 經過兩階段提交元數據后，進入元數據應用流程，各個節(jié)點對比自己本地的信息和接收的元數據，根據差異處理相關流程。

7、擴展性優(yōu)化

????????前面我們介紹了 ES 元數據管理模型，接下來我們結合元數據管理模型看看 ES 分布式架構存在的擴展性瓶頸及優(yōu)化措施。

7.1 擴展性瓶頸

????????社區(qū)版本建議控制整個集群分片數在 3 萬以下，節(jié)點數不超過100，超過之后在創(chuàng)建、刪除索引、維護 mapping、template 等元數據變更操作時可能出現較嚴重的卡頓。例如 ES 在寫入觸發(fā)創(chuàng)建的場景，大批量 bulk 請求在索引創(chuàng)建完畢之前會堆積內存，節(jié)點有被打垮的風險。其次，因為單集群支持的分片數、節(jié)點數有限，導致用戶大規(guī)模數據場景下，需要建大量的小規(guī)格集群滿足業(yè)務需求，從而導致集群碎片化資源嚴重，整體 TCO 偏高。從前面元數據整體的變更流程中我們總結出如下主要的瓶頸：

• master 構建新元數據，例如節(jié)點分片分配、均衡策略時，會基于 RoutingTable 全量構建 RoutingNodes。

• master 節(jié)點元數據序列化，全量比對新舊元數據，構建 diff 并序列化。

• data 節(jié)點元數據 diff 推導，基于 RoutingTable 全量構建 RoutingNodes，多次全量遍歷本節(jié)點分片和 diff 比對。

• 單個任務多次元數據變更，例如創(chuàng)建一個索引，會先創(chuàng)建主分片再創(chuàng)建從分片，導致單任務多次元數據變更。

• 路由全節(jié)點分發(fā)，單次元數據變更全節(jié)點分發(fā)，GC、網絡抖動等長尾節(jié)點影響變更。

• 部分場景元數據同步落盤，HDD 場景性能影響嚴重。
  
  ???總結來說，分片到節(jié)點的映射（RoutingTable）和節(jié)點到分片的映射（RoutingNodes）兩者之間的全量構建，以及新舊元數據全量對比、全節(jié)點分發(fā)、同步落盤等是影響擴展性的主要瓶頸。

7.2 優(yōu)化措施

????????結合前面分析的瓶頸點，優(yōu)化措施主要包括映射增量維護、收斂元數據變更范圍、重啟性能優(yōu)化等方向。

7.2.1 映射增量維護

??????? RoutingTable 和 RoutingNodes 全量構建的問題，在分片總數達到 5 萬以上時，多處這種相互構建的性能瓶頸明顯，我們可以采用兩者增量維護的方式，避免全量相互構建：

映射增量維護

???????元數據發(fā)送前序列化階段也不需要全量對比，因為我們上面已經維護了一個 diff 的增量，可以省去一些全量對比的環(huán)節(jié)，直接將產生好的 diff 發(fā)送給數據節(jié)點即可。數據節(jié)點上的元數據推導也不需要再進行全量對比，直接拿增量的 diff 進行應用。對于部分持久化存在同步刷盤的情況改為異步刷盤，緩解 HDD 盤場景的性能瓶頸。整體的優(yōu)化方向是流程增量、異步化：

元數據變更增量、異步化改造

7.2.2 收斂變更范圍

????????對于元數據發(fā)布全節(jié)點的瓶頸，在節(jié)點數多了例如數百個之后，容易因個別節(jié)點的抖動受影響。我們可以控制元數據發(fā)布的范圍，只發(fā)生在 master 節(jié)點之間，索引、分片的創(chuàng)建采用定向節(jié)點下發(fā)的方式避免請求扇出嚴重，而數據節(jié)點可以通過學習的方式動態(tài)構建元數據。

收斂元數據變更范圍

7.2.3 重啟性能優(yōu)化

????????在集群的總分片數和節(jié)點數達到一定規(guī)模后，重啟恢復的性能會成為瓶頸，時間會比較長，也是我們需要主要優(yōu)化的方向。因為重啟會涉及到大量的分片分配、恢復，主要的優(yōu)化思路包括：

• 排序粗化：分片恢復有優(yōu)先級，按優(yōu)先級排序時，我們可以從分片維度優(yōu)化到索引維度。上面就是整體的擴展性優(yōu)化的主要方向，還有其它方面的優(yōu)化，例如統(tǒng)計接口層面的優(yōu)化，主要思路是構建緩存、低頻的方式提升接口性能。

• 批量 fetch：在恢復之前，master 會先獲取所有數據節(jié)點上分片最新的狀態(tài)，可以從單分片請求優(yōu)化為按節(jié)點批量獲取。

• 遍歷裁剪：在分片恢復過程中，每分配一批（并發(fā)恢復分片數控制）分片，都需要全量遍歷所有分片，遍歷過程中可以根據并發(fā)控制的分片數對遍歷進行高效裁剪，去掉不需要分配的無用分片的遍歷過程。

7.2.4 優(yōu)化效果

????????在經過一系列全方位擴展性瓶頸優(yōu)化之后，我們將集群的分片數擴展至百萬級，節(jié)點數擴展至數百上千級。于此同時，我們也將部分熱點瓶頸優(yōu)化反饋給了社區(qū)。附部分已合并 PR：

https://github.com/elastic/elasticsearch/pull/87723

https://github.com/elastic/elasticsearch/pull/64753

https://github.com/elastic/elasticsearch/pull/46520

https://github.com/elastic/elasticsearch/pull/65045

https://github.com/elastic/elasticsearch/pull/56870

https://github.com/elastic/elasticsearch/pull/65172

https://github.com/elastic/elasticsearch/pull/60564

8、總結

????????本文通過介紹 ES 分布式架構，對比業(yè)界主流的分布式元數據管理模式，之后剖析了 ES 核心元數據管理模型。最后結合集群擴展性的瓶頸介紹了騰訊在 ES 擴展性方面所做的相關優(yōu)化，現階段重點解決了元數據的擴展性，節(jié)點的擴展性后續(xù)還需持續(xù)優(yōu)化以支持更大規(guī)模，目前騰訊云 ES 內核已能滿足絕大部分用戶的擴展性需求。文章來源地址http://www.zghlxwxcb.cn/news/detail-461310.html