【Elasticsearch】Elasticsearch gateway 流程分析

淡淡的烟草味﹌ 2022-10-13 14:54 135阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]

# 1.概述 #

转载：[elasticsearch gateway 流程分析][elasticsearch gateway]

es 存储的数据有以下几种形式:

state
    index
    translog

index 为 lucene 生成的索引文件  
translog 为es 产生的事务日志  
state 是元数据信息,有以下几种:

nodes/0/_state/global-82.st 集群层面元信息
    nodes/0/indices/website/_state/state-1.st 索引层面元信息
    nodes/0/indices/website/0/_state/state-0.st 分片层面元信息

分别对应 es 中的数据结构:

MetaData: 主要是 settings, templates
    IndexMetaData: 主要是numberOfShards,mappings等
    ShardStateMetaData: 主要是version,indexUUID,primary

上述信息被持久化到磁盘,要注意的是:持久化的 state 不包括某个 shard 存在于哪个 node 这种内容路由信息,依靠 gateway 的 recovery 过程重建 RoutingTable

在集群 full restart时,达到 recovery 条件时,进入 gateway流程, recovery 条件由以下三个配置控制:

gateway.expected_nodes
    gateway.recover_after_time
    gateway.recover_after_nodes

假设取值为10,5m,8,则集群启动时节点达到10个则立即进入 recovery; 如果一直没有达到10个,5分钟超时后如果节点达到8个也进入 recovery

gateway 模块负责当`集群 full restart` 时的元信息(state)数据恢复.恢复以后的结果包括集群级,索引级,但不包括 shard 级,当集群级,索引级元数据选举完毕后,执行 `submitStateUpdateTask` 提交一个source 为`local-gateway-elected-state` 的任务,触发获取 shard 级元数据的操作,这个fetch过程是异步的,可能比较长,然后submit的这个任务就结束, gateway 流程结束

因此, `gateway 流程由 gateway 模块和 allocation 模块共同完成的`,在 gateway 将集群级,索引级元数据选举完毕后,`submitStateUpdateTask`提交的任务中会执行 allocation 模块的 reroute 继续后面的流程.

主要实现在 GatewayService 类,他继承自ClusterStateListener, 由clusterChanged时触发.

# 2.一致性 #

由于`元数据信息是根据版本号选举出来的`,而元数据写入成功的条件是多数,因此,保证进入 recovery 的条件为节点数量为多数,可以保证集群级和索引级的一致性,而 shard 级需要从另一方面考虑.

默认情况下,`doc写入成功条件为多数shard 写入成功`,例外的,当副本为1(数据存2份)时,成功条件为1个 shard 写入成功.就是说,只有主分片为 active时,可以正常执行写操作.在这种条件下,进入 recovery 流程时,被选出的主分片不一定是拥有最新数据的分片.当主分片被选举成功,他就作为 shard 的一致性参考,即使后来原本拥有最新数据shard 的节点加入集群,也会被同步成为当前主分片一致.因此,当集群状态为非 green, 写操作是有数据丢失的风险.

# 3.流程分析 #

Master 选举成功之后,达到 `gateway.recovery` 条件时,进入恢复流程,此时所有 shard 都标记为 `Unassigned` 状态。`Master 从各个节点主动获取元数据信息,选举版本号最大的作为最新元数据,`包括集群级,索引级.两者确定之后,执行 reroute, 挨个获取各个 shard 级别元数据,默认超时13s,`取版本号最大的为Primary.`

因此,`元数据信息每次都是选举出来的`,取版本号`最新`来用,而Primary shard 位于哪个 node 却是 reroute `动态计算`出来的,以前Primary shard是哪个节点,下次不一定是他.

gateway流程的后半部分在 clusterService 的 updateTask 线程中,伴随着这个updateTask结束,整个流程结束. 最终 gateway 流程完毕时,向各节点索取 shard 信息的流程一般还没结束,因此大部分主分片尚未确定, “某个分片存在于那个节点” 这种信息也还没有.这部分由cluster\_reroute(async\_shard\_fetch)触发下一个流程,那便是 allocation .执行数据分配.

选举集群级和索引级别的元数据  
执行线程为:generic main  
实现位于:

gateway.Gateway#performStateRecovery

首先向有 Master 资格的节点发起请求,获取元数据

TransportNodesListGatewayMetaState.NodesGatewayMetaState nodesState = listGatewayMetaState.list(nodesIds, null).actionGet();

获取的响应数量必须达到 requiredAllocation:

int requiredAllocation = Math.max(1, minimumMasterNodesProvider.get());

然后接下来就是通过版本号选取集群级和索引级元数据的实现.这里不贴代码

## 3.2 动态构建路由表 ##

执行线程为:clusterService\#updateTask  
选举完成后,调用

listener.onSuccess(builder.build());

执行恢复后的逻辑,主要实现位于:

gateway.GatewayService.GatewayRecoveryListener#onSuccess

通过clusterService.submitStateUpdateTask提交一个更新任务,submitStateUpdateTask函数执行Executor对应的 Task,Executor就是submitStateUpdateTask时传参进去的

执行恢复成功后的主要流程:

// update the state to reflect the new metadata and routing
    ClusterState updatedState = ClusterState.builder(currentState)
            .blocks(blocks)
            .metaData(metaDataBuilder)
            .build();
    
    //先初始化,路由表中的信息基本除了 nodeid 之外都根据设置信息准备好了
    // initialize all index routing tables as empty 
    RoutingTable.Builder routingTableBuilder = RoutingTable.builder(updatedState.routingTable());
    for (ObjectCursor<IndexMetaData> cursor : updatedState.metaData().indices().values()) { 
        routingTableBuilder.addAsRecovery(cursor.value);
    }
    // start with 0 based versions for routing table
    routingTableBuilder.version(0);
    
    //根据当前集群活跃节点重新生成内容路由信息
    // now, reroute reroute, 
    RoutingAllocation.Result routingResult = allocationService.reroute(
            ClusterState.builder(updatedState).routingTable(routingTableBuilder.build()).build(),
            "state recovered");
    
    return ClusterState.builder(updatedState).routingResult(routingResult).build();

注意这里会阻塞集群元信息更新,直到 gateway 流程结束, 此阶段如果有新机器想加入集群是加入不了的

## 3.3 reroute ##

在上面的 allocationService.reroute 函数对集群所有的 shard 统一执行一次 reroute 调用,收集 shard 级元数据.

详细的过程参考 allocation 流程分析

# 4.结束语 #

reroute 会更新集群信息,集群信息的应用进而触发各个模块对其进行处理,很多流程就这样触发的.从选主完到 gateway, 再到 allocation,recovery, 都是如此。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]: /images/20221005/f63dc6afe0514e4081be3c6f9965c064.png
[elasticsearch gateway]: https://www.easyice.cn/archives/226