1.概述

转载：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, 都是如此。

【Elasticsearch】Elasticsearch gateway 流程分析相关推荐

Elasticsearch 5.x segments merge 流程分析
2019独角兽企业重金招聘Python工程师标准>>> Elasticsearch 5.x segments merge 流程分析这两周主要看了下 Elasticsearch(其 ...
Elasticsearch CCR源码分析（补充）
接上篇TODO Elasticsearch CCR源码分析上篇TODO: http请求(ccr/follow)接收到后,follow集群节点开始全量同步,是以snapshot的模式去拉leader集 ...
用于Elasticsearch数据可视化和分析的强大工具
The goal is to turn data into information, and information into insight. 目标是将数据转化为信息,并将信息转化为洞察力. ―Ca ...
springcloud gateway 请求执行流程分析
一.示例 pom.xml <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http:// ...
java基础流程分析，及原理解析，因为bu满，而qian行
基本功 =>同样的流程别人的解释*=>基础二面向对象的特征四个基本特征:抽象,继承, 封装, 多态抽象: 就好比用程序描述一个人,肯定得抽象的通过(身高,体重,年龄 , 胖瘦)这些 ...
浅显易懂的Django架构流程分析
Django的运行方式运行Django项目的方法很多,一种是在开发和调试中经常用到的runserver方法,使用Django自己的Web Server.另外一种就是使用fastcgi, uWSGI等 ...
RequestRateLimiter 的配置装配流程分析
背景这几天分析 RequestRateLimterGatewayFilter 的源码,一直对 RedisRateLimter 的限流参数是怎么绑定的这个问题困扰. 今天开机后看到周末跟踪代码打开的窗 ...
VLC架构及流程分析
0x00 前置信息 VLC是一个非常庞大的工程,我从它的架构及流程入手进行分析,涉及到一些很细的概念先搁置一边,日后详细分析. 0x01 源码结构(Android Java相关的暂未分析) # bui ...
动态执行流程分析和性能瓶颈分析的利器——gperftools的Cpu Profiler
在<动态执行流程分析和性能瓶颈分析的利器--valgrind的callgrind>中,我们领略了valgrind对流程和性能瓶颈分析的强大能力.本文将介绍拥有相似能力的gperftools ...

【Elasticsearch】Elasticsearch gateway 流程分析