标签：Term JOIN 请求 Leader public Elasticsearch 主流程 节点 ES
Raft协议
Raft是分布式系统中的一种共识算法，用于在集群中选举Leader管理集群。Raft协议中有以下角色：
Leader（领导者）：集群中的领导者，负责管理集群。
Candidate（候选者）：具有竞选Leader资格的角色，如果集群需要选举Leader，节点需要先转为候选者角色才可以发起竞选。
Follower（跟随者 ）：Leader的跟随者，接收和处理来自Leader的消息，与Leader之间保持通信，如果通信超时或者其他原因导致节点与Leader之间通信失败，节点会认为集群中没有Leader，就会转为候选者发起竞选，推荐自己成为Leader。
Raft协议中还有一个Term（任期）的概念，任期是随着选举的举行而变化，一般是单调进行递增，比如说集群中当前的任期为1，此时某个节点发现集群中没有Leader，开始发起竞选，此时任期编号就会增加为2，表示进行了新一轮的选举。一般会为Term较大的那个节点进行投票，当某个节点收到的投票数达到了Quorum，一般是集群中的节点数/2 + 1，将会被选举为Leader。
Elasticsearch选主
Elasticsearch在7.0版本以前采用Bully算法进行选主，7.0以后使用了Raft协议，但没有完全按照Raft协议来实现，而是做了一些调整，ES选主流程如下：
-
节点的初始化状态为Candidate;
-
启动选举任务，向探测到的集群中其他节点发送PRE_VOTE投票请求，请求中会携带节点的Term信息;
-
其他节点收到PRE_VOTE投票请求后，对请求进行处理：
（1）更新自己收到过的最大的Term
如果请求中的Term比自己的Term大并且当前节点是Leader节点，意味着当前的Leader可能已经过期，其他节点已经开始竞选Leader，所以此时当前节点需要放弃Leader的身份，重新发起选举。
（2）根据当前节点记录的Leader信息决定是否投票给发起者，然后向发起者返回投票响应信息：
-
如果当前节点记录的集群Leader为空，同意投票给发起者。
-
如果当前节点记录的集群Leader不为空，但是与本次发起的节点一致，同样同意投票。
-
如果当前节点记录的集群Leader为空，但是与本次发起的节点不同，拒绝投票给发起者。
-
-
发起者收到其他节点对PRE_VOTE投票请求的响应，判断是否得到了大多数投票，如果是进入下一步；
-
发起者向集群中的节点发送StartJoin请求，邀请节点加入集群，发送StartJoin请求的时候会将Term增加1，但是发起者的Term暂不更新，这与Raft协议在发起选举的时候就对Term增加的操作不一样；
-
其他节点收到StartJoin请求，更新自己的Term信息，处理完毕后向发起者发送JOIN请求，JOIN请求中携带了节点的Term信息；
收到StartJoin请求时，只要请求中的Term比当前节点的Term大，当前节点都会同意为发起者进行投票，这里也与Raft协议规定的每个任期内只能为一个节点进行投票不一致。
既然节点可以多次进行投票，那么就有可能产生多个Leader，对于这种情况，Elasticsearch会选择最后那个选举成功的节点成为Leader。
-
发起者收到其他节点发送的JOIN请求后，会统计收到的JOIN请求个数，如果达到了大多数投票，即可成为Leader；
发起者收到JOIN请求时也会校验自己的Term是否比JOIN请求中的Term大，在第5步中发起者并未更新自己的Term，所以首次收到JOIN请求后，Term信息会小于JOIN请求中的Term，这里发起者会模拟一个JOIN请求给自己，也就是自己为自己投一票。
-
发起者成为Leader；
ES选主存在的问题
由于每个节点可以多次进行投票，有可能出现节点竞争激烈导致一直未选出leader的问题。关于问题的解决方案可以参考以下两篇文章：
【张超】留意Elasticsearch 7.x 可能无法选主的问题
【Guohang Huang】腾讯 Elasticsearch 7.x 大集群选主优化
Elasticsearch选举流程分析
在ES启动节点的时候，会调用Coordinator的startInitialJoin方法开启选举：
// Node
public class Node implements Closeable {
public Node start() throws NodeValidationException {
// ...
// 启动集群选举
coordinator.startInitialJoin();
// ...
}
}
// Coordinator
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
public void startInitialJoin() {
synchronized (mutex) {
// 先转为候选者
becomeCandidate("startInitialJoin");
}
// 启动选举任务
clusterBootstrapService.scheduleUnconfiguredBootstrap();
}
}
成为候选节点
becomeCandidate方法主要做一些Leader选举的前置工作：
- 判断节点的角色是否是候选者，因为Raft协议中候选者才可以发起leader选举，所以第一步需要把当前节点转为候选者节点；
- 初始化
PreVoteCollector里面状态信息，它是一个二元组Tuple<DiscoveryNode, PreVoteResponse>，DiscoveryNode记录了集群的leader节点，PreVoteResponse里面记录节点的Term信息，包括当前Term、上一次接受的Term（集群Term）和上一次接受的版本（集群版本），在投票选举的时候会用到；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
void becomeCandidate(String method) {
// 判断是否持有锁
assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
logger.debug("{}: coordinator becoming CANDIDATE in term {} (was {}, lastKnownLeader was [{}])", method,
getCurrentTerm(), mode, lastKnownLeader);
// 如果不是CANDIDATE
if (mode != Mode.CANDIDATE) {
final Mode prevMode = mode;
// 设置为CANDIDATE
mode = Mode.CANDIDATE;
cancelActivePublication("become candidate: " + method);
//...
// 如果之前是Leader
if (prevMode == Mode.LEADER) {
// 清除Master相关信息
cleanMasterService();
}
// ...
}
// 更新PreVoteCollector里面记录的leader节点和Term信息，这里还没有选举出leader，所以传入的是null
preVoteCollector.update(getPreVoteResponse(), null);
}
private PreVoteResponse getPreVoteResponse() {
// 创建PreVoteResponse，记录当前Term、上一次接受的Term和上一次接受的版本
return new PreVoteResponse(
getCurrentTerm(),
coordinationState.get().getLastAcceptedTerm(),
coordinationState.get().getLastAcceptedState().version()
);
}
}
PreVoteCollector的二元组如下，DiscoveryNode为leader节点，PreVoteResponse记录了Term相关信息，其他节点发起选举时，返回给发起者的投票结果就是PreVoteResponse：
public class PreVoteCollector {
// 二元组
private volatile Tuple<DiscoveryNode, PreVoteResponse> state;
public void update(final PreVoteResponse preVoteResponse, @Nullable final DiscoveryNode leader) {
logger.trace("updating with preVoteResponse={}, leader={}", preVoteResponse, leader);
// 初始化状态信息
state = new Tuple<>(leader, preVoteResponse);
}
}
Leader选举
在scheduleUnconfiguredBootstrap方法中，对节点是否有Master角色权限进行了判断，如果没有Master角色权限，直接返回终止选举，否则启动选举任务，获取集群中发现的节点，调用startBootstrap开始启动：
public class ClusterBootstrapService {
scheduleUnconfiguredBootstrap() {
if (unconfiguredBootstrapTimeout == null) {
return;
}
// Master角色权限校验
if (transportService.getLocalNode().isMasterNode() == false) {
return;
}
logger.info(
"no discovery configuration found, will perform best-effort cluster bootstrapping after [{}] "
+ "unless existing master is discovered",
unconfiguredBootstrapTimeout
);
// 执行启动任务
transportService.getThreadPool().scheduleUnlessShuttingDown(unconfiguredBootstrapTimeout, Names.GENERIC, new Runnable() {
@Override
public void run() {
// 获取集群中发现的节点
final Set<DiscoveryNode> discoveredNodes = getDiscoveredNodes();
logger.debug("performing best-effort cluster bootstrapping with {}", discoveredNodes);
// 启动
startBootstrap(discoveredNodes, emptyList());
}
// ...
});
}
}
启动选举
在startBootstrap方法中，首先判断探测到的集群节点discoveryNodes是否有Master角色权限，然后调用doBootstrap进行启动。
在doBootstrap方法中，创建了VotingConfiguration，然后调用votingConfigurationConsumer触发选举，并进行了异常捕捉，如果出现异常进行重试：
public class ClusterBootstrapService {
private void startBootstrap(Set<DiscoveryNode> discoveryNodes, List<String> unsatisfiedRequirements) {
// 判断发现的节点是否有Master角色权限
assert discoveryNodes.stream().allMatch(DiscoveryNode::isMasterNode) : discoveryNodes;
assert unsatisfiedRequirements.size() < discoveryNodes.size() : discoveryNodes + " smaller than " + unsatisfiedRequirements;
if (bootstrappingPermitted.compareAndSet(true, false)) {
// 启动
doBootstrap(
// 创建VotingConfiguration
new VotingConfiguration(
Stream.concat(
discoveryNodes.stream().map(DiscoveryNode::getId),
unsatisfiedRequirements.stream().map(s -> BOOTSTRAP_PLACEHOLDER_PREFIX + s)
).collect(Collectors.toSet())
)
);
}
}
private void doBootstrap(VotingConfiguration votingConfiguration) {
assert transportService.getLocalNode().isMasterNode();
try {
// 触发投票
votingConfigurationConsumer.accept(votingConfiguration);
} catch (Exception e) {
logger.warn(() -> "exception when bootstrapping with " + votingConfiguration + ", rescheduling", e);
// 如果出现异常，进行重试
transportService.getThreadPool().scheduleUnlessShuttingDown(TimeValue.timeValueSeconds(10), Names.GENERIC, new Runnable() {
@Override
public void run() {
doBootstrap(votingConfiguration);
}
// ...
});
}
}
}
votingConfigurationConsumer是一个函数式编程接口，它接收一个表达式，在Coordinator的构造函数中可以看到对ClusterBootstrapService进行实例化时，传入的是setInitialConfiguration方法，所以votingConfigurationConsumer.accept(votingConfiguration)会执行Coordinator的setInitialConfiguration方法：
public class ClusterBootstrapService {
// votingConfigurationConsumer
private final Consumer<VotingConfiguration> votingConfigurationConsumer;
public ClusterBootstrapService(
Settings settings,
TransportService transportService,
Supplier<Iterable<DiscoveryNode>> discoveredNodesSupplier,
BooleanSupplier isBootstrappedSupplier,
Consumer<VotingConfiguration> votingConfigurationConsumer
) {
//...
// 设置votingConfigurationConsumer
this.votingConfigurationConsumer = votingConfigurationConsumer;
}
}
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
public Coordinator(
// ...
) {
// ...
// 初始化ClusterBootstrapService
this.clusterBootstrapService = new ClusterBootstrapService(
settings,
transportService,
this::getFoundPeers,
this::isInitialConfigurationSet,
this::setInitialConfiguration // 传入setInitialConfiguration方法
);
// ...
}
}
setInitialConfiguration方法的处理逻辑如下：
- 首先进行一系列的校验，如果校验不通过不能进行选举：
- 是否已经初始化过；
- 当前节点是有Master角色权限；
- 集群中的节点是否包含当前节点；
- 集群中的节点个数是否达到了Quorum个；
- 调用
preVoteCollector的update方法，更新当前节点记录的Leader节点和Term信息； - 调用
startElectionScheduler方法启动选举；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
public boolean setInitialConfiguration(final VotingConfiguration votingConfiguration) {
synchronized (mutex) {
// 获取集群状态
final ClusterState currentState = getStateForMasterService();
// 判断是否初始化过
if (isInitialConfigurationSet()) {
logger.debug("initial configuration already set, ignoring {}", votingConfiguration);
return false;
}
// 校验Master角色权限
if (getLocalNode().isMasterNode() == false) {
logger.debug("skip setting initial configuration as local node is not a master-eligible node");
throw new CoordinationStateRejectedException(
"this node is not master-eligible, but cluster bootstrapping can only happen on a master-eligible node"
);
}
// 如果节点ID中不包含当前节点的ID
if (votingConfiguration.getNodeIds().contains(getLocalNode().getId()) == false) {
logger.debug("skip setting initial configuration as local node is not part of initial configuration");
throw new CoordinationStateRejectedException("local node is not part of initial configuration");
}
// ...
// 判断节点个数是否达到Quorum
if (votingConfiguration.hasQuorum(knownNodes.stream().map(DiscoveryNode::getId).toList()) == false) {
// ...
throw new CoordinationStateRejectedException(
"not enough nodes discovered to form a quorum in the initial configuration "
+ "[knownNodes="
+ knownNodes
+ ", "
+ votingConfiguration
+ "]"
);
}
// ...
// 更新
preVoteCollector.update(getPreVoteResponse(), null);
// 开始选举
startElectionScheduler();
return true;
}
}
}
发起选举
startElectionScheduler方法用于启动选举任务，任务是异步执行的：
- 校验节点是否是CANDIDATE节点，如果是继续往下进行；
- 如果当前节点集群健康状态处于UNHEALTHY，直接返回；
- 调用
PreVoteCollector的start方法发起投票；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private void startElectionScheduler() {
assert electionScheduler == null : electionScheduler;
// 校验Master角色权限
if (getLocalNode().isMasterNode() == false) {
return;
}
final TimeValue gracePeriod = TimeValue.ZERO;
// 启动选举任务
electionScheduler = electionSchedulerFactory.startElectionScheduler(gracePeriod, new Runnable() {
@Override
public void run() {
synchronized (mutex) {
// 如果是CANDIDATE节点
if (mode == Mode.CANDIDATE) {
// 获取之前的集群状态
final ClusterState lastAcceptedState = coordinationState.get().getLastAcceptedState();
if (localNodeMayWinElection(lastAcceptedState) == false) {
logger.trace("skip prevoting as local node may not win election: {}", lastAcceptedState.coordinationMetadata());
return;
}
// 获取集群状态信息
final StatusInfo statusInfo = nodeHealthService.getHealth();
// 如果处于UNHEALTHY状态
if (statusInfo.getStatus() == UNHEALTHY) {
logger.debug("skip prevoting as local node is unhealthy: [{}]", statusInfo.getInfo());
return;
}
if (prevotingRound != null) {
prevotingRound.close();
}
// 发起投票
prevotingRound = preVoteCollector.start(lastAcceptedState, getDiscoveredNodes());
}
}
}
// ...
});
}
}
PreVoteCollector的start方法中，创建了PreVotingRound，然后调用PreVotingRound的start的方法发起投票：
public class PreVoteCollector {
public Releasable start(final ClusterState clusterState, final Iterable<DiscoveryNode> broadcastNodes) {
// 创建PreVotingRound
PreVotingRound preVotingRound = new PreVotingRound(clusterState, state.v2().getCurrentTerm());
// 发起投票
preVotingRound.start(broadcastNodes);
return preVotingRound;
}
}
发送PRE_VOTE投票请求
PreVotingRound是PreVoteCollector的内部类，在start方法中，会遍历探测到的集群节点，然后进行遍历，向每一个节点发送PRE_VOTE投票请求，投票请求响应信息处理是在handlePreVoteResponse方法中处理的：
public class PreVoteCollector {
private class PreVotingRound implements Releasable {
PreVotingRound(final ClusterState clusterState, final long currentTerm) {
// 集群状态
this.clusterState = clusterState;
// 构建投票请求
preVoteRequest = new PreVoteRequest(transportService.getLocalNode(), currentTerm);
}
void start(final Iterable<DiscoveryNode> broadcastNodes) {
logger.debug("{} requesting pre-votes from {}", this, broadcastNodes);
// 遍历发现的节点，当前节点向每一个节点发送投票请求
broadcastNodes.forEach(
// 发送PRE_VOTE请求
n -> transportService.sendRequest(
n,
REQUEST_PRE_VOTE_ACTION_NAME,
preVoteRequest,
new TransportResponseHandler<PreVoteResponse>() {
// ...
@Override
public void handleResponse(PreVoteResponse response) {
// 处理返回的响应
handlePreVoteResponse(response, n);
}
// ...
}
)
);
}
}
}
节点对PRE_VOTE投票请求的处理
在PreVoteCollector的构造函数中可以看到，注册了REQUEST_PRE_VOTE_ACTION_NAME请求处理器，对PRE_VOTE请求的处理是调用handlePreVoteRequest方法进行的，处理完毕后调用sendResponse返回响应信息：
public class PreVoteCollector {
// 选举任务
private final Runnable startElection;
// 更新最大Term
private final LongConsumer updateMaxTermSeen;
PreVoteCollector(
final TransportService transportService,
final Runnable startElection,
final LongConsumer updateMaxTermSeen,
final ElectionStrategy electionStrategy,
NodeHealthService nodeHealthService
) {
this.transportService = transportService;
this.startElection = startElection;
this.updateMaxTermSeen = updateMaxTermSeen;
this.electionStrategy = electionStrategy;
this.nodeHealthService = nodeHealthService;
// 注册PRE_VOTE请求处理器
transportService.registerRequestHandler(
REQUEST_PRE_VOTE_ACTION_NAME,
Names.CLUSTER_COORDINATION,
false,
false,
PreVoteRequest::new,
(request, channel, task) -> channel.sendResponse(handlePreVoteRequest(request)) // 调用handlePreVoteRequest处理请求
);
}
}
在handlePreVoteRequest之前，首先看Coordinator的构造函数对PreVoteCollector实例化时传入的参数，主要关注startElection和updateMaxTermSeen，它们都是函数式编程接口，从实例化的代码中可以看到分别对应Coordinator的传入的startElection和updateMaxTermSeen方法，在后面会用到这两个方法：
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
public Coordinator(
// ...
){
// ...
this.preVoteCollector = new PreVoteCollector(
transportService,
this::startElection, // 传入startElection方法，启动选举
this::updateMaxTermSeen, // 传入updateMaxTermSeen，更新收到的最大Term
electionStrategy,
nodeHealthService
);
// ...
}
}
handlePreVoteRequest方法处理逻辑如下：
- 对term进行比较，调用
updateMaxTermSeen.accept()更新收到的最大Term； - 获取当前节点记录的集群Leader节点和Term信息；
- 如果Leader节点为空，表示还没有Leader节点，返回响应同意发起投票的节点成为leader；
- 如果leader不为空，但是与发起请求的节点是同一个节点，同样支持发起请求的节点成为leader；
- 其他情况，表示已经存在leader，拒绝投票请求；
public class PreVoteCollector {
private PreVoteResponse handlePreVoteRequest(final PreVoteRequest request) {
// 比较Term，更新maxTermSeen
updateMaxTermSeen.accept(request.getCurrentTerm());
Tuple<DiscoveryNode, PreVoteResponse> state = this.state;
assert state != null : "received pre-vote request before fully initialised";
// 获取当前节点记录的集群Leader节点
final DiscoveryNode leader = state.v1();
// 获取当前节点的Term信息
final PreVoteResponse response = state.v2();
// 获取健康状态
final StatusInfo statusInfo = nodeHealthService.getHealth();
// 如果当前节点的状态处于UNHEALTHY
if (statusInfo.getStatus() == UNHEALTHY) {
String message = "rejecting " + request + " on unhealthy node: [" + statusInfo.getInfo() + "]";
logger.debug(message);
throw new NodeHealthCheckFailureException(message);
}
// 如果leader为空，表示还没有Leader节点，返回响应同意发起投票的节点成为leader
if (leader == null) {
return response;
}
// 如果leader不为空，但是与发起请求的节点是同一个节点，同样支持发起请求的节点成为leader
if (leader.equals(request.getSourceNode())) {
return response;
}
// 其他情况，表示已经存在leader，拒绝投票请求
throw new CoordinationStateRejectedException("rejecting " + request + " as there is already a leader");
}
}
updateMaxTermSeen
上面说过updateMaxTermSeen指向Coordinator的updateMaxTermSeen方法，处理逻辑如下：
- 比较当前节点收到过的最大的Term与请求中的Term，选择较大的那个作为
maxTermSeen的值进行更新； - 如果当前节点是Leader并且请求中的Term大于当前节点的Term，表示当前节点的信息可能已经过期，需要放弃当前的Leader角色，重新发起选举：
- 调用ensureTermAtLeast检查Term，确保是最新的Term，在ensureTermAtLeast方法中会判断，如果当前节点Term小于请求中的Term将当前节点转为候选者；
- 调用startElection方法重新进行选举；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private void updateMaxTermSeen(final long term) {
synchronized (mutex) {
// 当前节点收到过的最大的Term与请求中的term，如果请求中的Term较大，maxTermSeen的值将被更新为请求中的Term的值
maxTermSeen = Math.max(maxTermSeen, term);
// 获取当前节点的term
final long currentTerm = getCurrentTerm();
// 如果当前节点是Leader并且maxTermSeen大于当前节点的Term，请求中的Term较大，这里maxTermSeen的值就是请求中的Term，所以也是在比较请求中的Term是否大于当前节点的Term
if (mode == Mode.LEADER && maxTermSeen > currentTerm) {
if (publicationInProgress()) {
logger.debug("updateMaxTermSeen: maxTermSeen = {} > currentTerm = {}, enqueueing term bump", maxTermSeen, currentTerm);
} else {
try {
logger.debug("updateMaxTermSeen: maxTermSeen = {} > currentTerm = {}, bumping term", maxTermSeen, currentTerm);
// 确保Term是最新
ensureTermAtLeast(getLocalNode(), maxTermSeen);
// 发起选举
startElection();
} catch (Exception e) {
logger.warn(new ParameterizedMessage("failed to bump term to {}", maxTermSeen), e);
becomeCandidate("updateMaxTermSeen");
}
}
}
}
}
}
ensureTermAtLeast
在ensureTermAtLeast方法中，判断当前节点的Term是否小于请求中的Term：
-
如果是则创建StartJoinRequest然后调用
joinLeaderInTerm方法，joinLeaderInTerm方法会返回一个JOIN信息；在集群选举Leader的时候，某个节点成为Leader之前，会向其他节点发送StartJoin请求，这里进行模拟发送，当前节点向自己发送一个StartJoinRequest进行处理，更新当前节点的Term信息，后面会详细讲解StartJoin请求的处理。
-
如果不是，返回一个空的JOIN信息；
在joinLeaderInTerm方法中，会调用handleStartJoin处理StartJoin请求，它会更新当前节点Term信息为最新，之后判断当前节点是否是CANDIDATE，如果不是需要将节点转为CANDIDATE:
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private Optional<Join> ensureTermAtLeast(DiscoveryNode sourceNode, long targetTerm) {
assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
// 判断当前节点Term是否小于请求中的Term
if (getCurrentTerm() < targetTerm) {
// 调用joinLeaderInTerm
return Optional.of(joinLeaderInTerm(new StartJoinRequest(sourceNode, targetTerm)));
}
return Optional.empty();
}
private Join joinLeaderInTerm(StartJoinRequest startJoinRequest) {
synchronized (mutex) {
logger.debug("joinLeaderInTerm: for [{}] with term {}", startJoinRequest.getSourceNode(), startJoinRequest.getTerm());
final Join join = coordinationState.get().handleStartJoin(startJoinRequest);
lastJoin = Optional.of(join);
peerFinder.setCurrentTerm(getCurrentTerm());
// 如果不是CANDIDATE转为CANDIDATE
if (mode != Mode.CANDIDATE) {
becomeCandidate("joinLeaderInTerm");
followersChecker.updateFastResponseState(getCurrentTerm(), mode);
preVoteCollector.update(getPreVoteResponse(), null);
}
return join;
}
}
}
PRE_VOTE响应处理
发起者收到集群节点返回的PRE_VOTE请求响应时，在handlePreVoteResponse方法中进行处理：
- 同样调用
updateMaxTermSeen更新当前节点收到的最大Term； - 如果响应中的Term大于当前节点的Term， 或者Term相等但是版本号大于当前节点的版本号，直接返回不进行处理，否则进入下一步；
- 走到这里表示认同当前节点成为Leader节点，将得到的投票信息放入
preVotesReceived； - 判断是否得到了大多数投票，也就是收到的投票数是否超过了Quorum，如果未超过直接返回，如果超过表示当前节点可以成为Leader；
- 通过
startElection开始处理成为Leader前的操作；
public class PreVoteCollector {
private class PreVotingRound implements Releasable {
private void handlePreVoteResponse(final PreVoteResponse response, final DiscoveryNode sender) {
if (isClosed.get()) {
logger.debug("{} is closed, ignoring {} from {}", this, response, sender);
return;
}
// 处理最大Term
updateMaxTermSeen.accept(response.getCurrentTerm());
// 如果响应中的Term大于当前节点的Term， 或者Term相等但是版本号大于当前节点的版本号
if (response.getLastAcceptedTerm() > clusterState.term()
|| (response.getLastAcceptedTerm() == clusterState.term() && response.getLastAcceptedVersion() > clusterState.version())) {
logger.debug("{} ignoring {} from {} as it is fresher", this, response, sender);
return;
}
// 记录得到的投票
preVotesReceived.put(sender, response);
// ...
// 判断是否得到了大多数投票
if (electionStrategy.isElectionQuorum(
clusterState.nodes().getLocalNode(),
localPreVoteResponse.getCurrentTerm(),
localPreVoteResponse.getLastAcceptedTerm(),
localPreVoteResponse.getLastAcceptedVersion(),
clusterState.getLastCommittedConfiguration(),
clusterState.getLastAcceptedConfiguration(),
voteCollection
) == false) {
logger.debug("{} added {} from {}, no quorum yet", this, response, sender);
return;
}
// ...
// 开始选举
startElection.run();
}
}
}
成为Leader
邀请节点加入集群
在成为Leader前，需要向集群中的节点发送StartJoin请求，邀请节点加入集群：
- 创建
StartJoin请求，请求中设置了Term信息，取当前节点的Term和收到过最大的Term中较大的那个值并加1； - 调用
sendStartJoinRequest发送StartJoin请求；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private void startElection() {
synchronized (mutex) {
// 是否是CANDIDATE
if (mode == Mode.CANDIDATE) {
if (localNodeMayWinElection(getLastAcceptedState()) == false) {
logger.trace("skip election as local node may not win it: {}", getLastAcceptedState().coordinationMetadata());
return;
}
// 创建StartJoin请求，这里可以看到在请求中的Term，设置为最大Term + 1
final StartJoinRequest startJoinRequest = new StartJoinRequest(getLocalNode(), Math.max(getCurrentTerm(), maxTermSeen) + 1);
logger.debug("starting election with {}", startJoinRequest);
// 调用sendStartJoinRequest发送StartJoin请求
getDiscoveredNodes().forEach(node -> joinHelper.sendStartJoinRequest(startJoinRequest, node));
}
}
}
}
StartJoin请求发送
StartJoin请求表示邀请节点加入集群信息，接收者收到请求后会向发起者发送JOIN请求表示进行加入，所以发起者对StartJoin的响应不需要做什么处理，等待接收者发送JOIN请求即可：
public class JoinHelper {
void sendStartJoinRequest(final StartJoinRequest startJoinRequest, final DiscoveryNode destination) {
assert startJoinRequest.getSourceNode().isMasterNode()
: "sending start-join request for master-ineligible " + startJoinRequest.getSourceNode();
// 发送START_JOIN请求
transportService.sendRequest(destination, START_JOIN_ACTION_NAME, startJoinRequest, new TransportResponseHandler.Empty() {
@Override
public void handleResponse(TransportResponse.Empty response) {
// 什么也不处理
logger.debug("successful response to {} from {}", startJoinRequest, destination);
}
@Override
public void handleException(TransportException exp) {
logger.debug(new ParameterizedMessage("failure in response to {} from {}", startJoinRequest, destination), exp);
}
});
}
}
StartJoin请求处理
JoinHelper的构造函数中，注册了START_JOIN请求处理器，在收到START_JOIN请求时，会调用joinLeaderInTerm处理，然后调用sendJoinRequest向发送者发送JOIN请求：
public class JoinHelper {
JoinHelper(
// ...
) {
// 注册START_JOIN_ACTION_NAME请求处理
transportService.registerRequestHandler(
START_JOIN_ACTION_NAME,
Names.CLUSTER_COORDINATION,
false,
false,
StartJoinRequest::new,
(request, channel, task) -> {
final DiscoveryNode destination = request.getSourceNode();
// 发送join请求
sendJoinRequest(destination, currentTermSupplier.getAsLong(), Optional.of(joinLeaderInTerm.apply(request))); // 调用joinLeaderInTerm处理
channel.sendResponse(Empty.INSTANCE);
}
);
}
}
joinLeaderInTerm
joinLeaderInTerm方法用于处理StartJoin请求，返回一个Join对象并发送给发起者，发起者会根据返回的Join信息计算得到的票数，以此决定是否成为Leader，joinLeaderInTerm处理逻辑如下：
- 调用
handleStartJoin处理StartJoin请求，它会从请求中获取Term信息并更新到当前节点的CurrentTerm中，并返回Join对象，用于向发起者回复投票结果； - 如果节点不是
CANDIDATE，将节点转为CANDIDATE；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private Join joinLeaderInTerm(StartJoinRequest startJoinRequest) {
synchronized (mutex) {
logger.debug("joinLeaderInTerm: for [{}] with term {}", startJoinRequest.getSourceNode(), startJoinRequest.getTerm());
// 处理StartJoin请求
final Join join = coordinationState.get().handleStartJoin(startJoinRequest);
lastJoin = Optional.of(join);
peerFinder.setCurrentTerm(getCurrentTerm());
// 如果节点不是CANDIDATE，转为CANDIDATE
if (mode != Mode.CANDIDATE) {
becomeCandidate("joinLeaderInTerm");
} else {
followersChecker.updateFastResponseState(getCurrentTerm(), mode);
preVoteCollector.update(getPreVoteResponse(), null);
}
return join;
}
}
}
更新CurrentTerm
在handleStartJoin方法中从请求中获取Term信息并更新到当前节点的CurrentTerm中：
- 如果
StartJoin请求中的Term小于或者等于当前节点的Term，抛出异常； - 更新当前节点的CurrentTerm为StartJoin请求中的Term；
- 返回一个Join对象，里面记录当前节点加入集群的信息，包括当前节点信息、发送startJoin请求的节点（选举为Leader的节点），当前节点的Term，当前节点上一次接受的Term、当前节点上一次接受的版本；
handleStartJoin方法中只要请求中的Term大于当前节点的Term，都会继续往下进行，最后返回一个Join对象，这意味着当前节点同意为发起者进行投票，也就是说Elasticsearch允许一个节点多次进行投票，并没有按照Raft协议中的规定每个任期内只能给一个节点投票。
public class CoordinationState {
public Join handleStartJoin(StartJoinRequest startJoinRequest) {
// 如果StartJoin请求中的Term小于或者等于当前节点的Term，抛出异常
if (startJoinRequest.getTerm() <= getCurrentTerm()) {
logger.debug(
"handleStartJoin: ignoring [{}] as term provided is not greater than current term [{}]",
startJoinRequest,
getCurrentTerm()
);
throw new CoordinationStateRejectedException(
"incoming term " + startJoinRequest.getTerm() + " not greater than current term " + getCurrentTerm()
);
}
logger.debug("handleStartJoin: leaving term [{}] due to {}", getCurrentTerm(), startJoinRequest);
// ...
// 更新当前节点的CurrentTerm
persistedState.setCurrentTerm(startJoinRequest.getTerm());
// 判断当前节点的Term是否与startJoin请求的一致
assert getCurrentTerm() == startJoinRequest.getTerm();
lastPublishedVersion = 0;
lastPublishedConfiguration = getLastAcceptedConfiguration();
startedJoinSinceLastReboot = true;
electionWon = false;
joinVotes = new VoteCollection();
publishVotes = new VoteCollection();
// 返回JOIN信息，包括当前节点、发送startJoin请求的节点、当前节点的Term、当前节点上一次接受的Term、当前节点上一次接受的版本
return new Join(localNode, startJoinRequest.getSourceNode(), getCurrentTerm(), getLastAcceptedTerm(), getLastAcceptedVersion());
}
}
节点加入集群
向Leader发送JOIN请求
StartJoin请求处理完毕后调用sendJoinRequest向发起者发送JOIN请求，表示加入集群：
public class JoinHelper {
public void sendJoinRequest(DiscoveryNode destination, long term, Optional<Join> optionalJoin) {
assert destination.isMasterNode() : "trying to join master-ineligible " + destination;
final StatusInfo statusInfo = nodeHealthService.getHealth();
// 如果处于UNHEALTHY状态不进行发送
if (statusInfo.getStatus() == UNHEALTHY) {
logger.debug("dropping join request to [{}]: [{}]", destination, statusInfo.getInfo());
return;
}
// 构建JOIN请求体
final JoinRequest joinRequest = new JoinRequest(transportService.getLocalNode(), term, optionalJoin);
// ...
if (pendingOutgoingJoins.putIfAbsent(dedupKey, pendingJoinInfo) == null) {
logger.debug("attempting to join {} with {}", destination, joinRequest);
pendingJoinInfo.message = PENDING_JOIN_CONNECTING;
// 连接节点
transportService.connectToNode(destination, new ActionListener<>() {
@Override
public void onResponse(Releasable connectionReference) {
// ...
clusterApplier.onNewClusterState(
"joining " + destination.descriptionWithoutAttributes(),
() -> null,
new ActionListener<>() {
@Override
public void onResponse(Void unused) {
// ....
pendingJoinInfo.message = PENDING_JOIN_WAITING_RESPONSE;
// 发送JOIN请求
transportService.sendRequest(
destination,
JOIN_ACTION_NAME,
joinRequest,
TransportRequestOptions.of(null, TransportRequestOptions.Type.PING),
new TransportResponseHandler.Empty() {
@Override
public void handleResponse(TransportResponse.Empty response) {
pendingJoinInfo.message = PENDING_JOIN_WAITING_STATE;
pendingOutgoingJoins.remove(dedupKey);
logger.debug("successfully joined {} with {}", destination, joinRequest);
lastFailedJoinAttempt.set(null);
}
// ...
}
);
}
// ...
}
);
}
// ...
});
} else {
logger.debug("already attempting to join {} with request {}, not sending request", destination, joinRequest);
}
}
}
Leader处理Join请求
JoinHelper的构造函数中，注册了JOIN请求处理器，是通过joinHandler来处理请求的，它同样是函数式编程接口，在Coordinator对JoinHelper进行实例化的时候，可以看到传入的是handleJoinRequest方法：
public class JoinHelper {
JoinHelper(
// ...
BiConsumer<JoinRequest, ActionListener<Void>> joinHandler,
// ...
) {
// ...
transportService.registerRequestHandler(
JOIN_ACTION_NAME,
Names.CLUSTER_COORDINATION,
false,
false,
JoinRequest::new,
(request, channel, task) -> joinHandler.accept(
request,
new ChannelActionListener<Empty, JoinRequest>(channel, JOIN_ACTION_NAME, request).map(ignored -> Empty.INSTANCE)
)
);
// ...
}
}
// Coordinator
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
public Coordinator(
// ...
) {
// ...
this.joinHelper = new JoinHelper(
allocationService,
masterService,
clusterApplier,
transportService,
this::getCurrentTerm,
this::handleJoinRequest, // handleJoinRequest方法
// ...
);
// ...
}
}
Coordinator的handleJoinRequest方法中，会对发送JOIN的节点进行连接，进行JOIN请求验证：
- 先调用
processJoinRequest处理收到的JOIN请求； - 调用
validateJoinRequest方法对JOIN请求进行验证；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private void handleJoinRequest(JoinRequest joinRequest, ActionListener<Void> joinListener) {
// ...
// 连接节点
transportService.connectToNode(joinRequest.getSourceNode(), new ActionListener<>() {
@Override
public void onResponse(Releasable response) {
boolean retainConnection = false;
try {
// 对JOIN请求进行验证
validateJoinRequest(
joinRequest,
ActionListener.runBefore(joinListener, () -> Releasables.close(response))
.delegateFailure((l, ignored) -> processJoinRequest(joinRequest, l)) // 处理请求
);
retainConnection = true;
} catch (Exception e) {
joinListener.onFailure(e);
} finally {
if (retainConnection == false) {
Releasables.close(response);
}
}
}
// ...
});
}
}
JOIN请求处理
processJoinRequest处理逻辑如下：
- 调用
updateMaxTermSeen更新收到最大的Term； - 判断是否已经成功竞选为Leader，因为发起者会收到多个节点发送的JOIN请求，每次处理JOIN请求会判断是否获取了大多数投票，并将结果更新到
CoordinationState的electionWon变量中，为了不重复调用becomeLeader，这里先获取最近一次更新的值，记为prevElectionWon，用于判断后面是否需要调用becomeLeader成为Leader； - 调用
handleJoin进行处理，处理的时候会判断是否获取了大多数的投票，并更新CoordinationState中electionWon的值； - 再次从CoordinationState中获取
electionWon值进行判断，如果prevElectionWon为false但是当前的electionWon为true，也就是之前未收到大多数投票的，但是处理当前的JOIN请求时达到了大多数投票，成功竞选为Leader，则调用becomeLeader成为Leader；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private void processJoinRequest(JoinRequest joinRequest, ActionListener<Void> joinListener) {
assert Transports.assertNotTransportThread("blocking on coordinator mutex and maybe doing IO to increase term");
// 获取JOIN信息
final Optional<Join> optionalJoin = joinRequest.getOptionalJoin();
try {
synchronized (mutex) {
// 更新最大Term
updateMaxTermSeen(joinRequest.getTerm());
// 获取集群协调状态
final CoordinationState coordState = coordinationState.get();
// 获取上一次的状态，是否成功选举为Leader
final boolean prevElectionWon = coordState.electionWon();
// 处理JOIN
optionalJoin.ifPresent(this::handleJoin);
joinAccumulator.handleJoinRequest(joinRequest.getSourceNode(), joinListener);
// 如果之前未成为Leader并且当前选举Leader成功
if (prevElectionWon == false && coordState.electionWon()) {
// 成为Leader
becomeLeader();
}
}
} catch (Exception e) {
joinListener.onFailure(e);
}
}
}
接下来看下handleJoin的处理过程：
-
首先调用ensureTermAtLeast方法确保当前节点是最新的Term，ensureTermAtLeast前面已经讲过，会确保当前的节点Term是最新，如果已经是最新什么也不做，如果不是将创建StartJoinRequest然后调用
joinLeaderInTerm方法，joinLeaderInTerm方法会返回一个JOIN信息，表示当前节点要加入一个集群的信息；在节点发送StartJoin请求时可知，对请求中的Term进行了加1但是节点自己的Term并未更新，所以首次收到发回的JOIN请求进入
handleJoin时，JOIN请求中的Term会比当前节点的Term大1，那么ensureTermAtLeast就会返回一个JOIN信息，然后再次调用handleJoin处理JOIN请求，这里可以理解为节点向自己发了一个JOIN请求（通过创建JOIN对象的方式），给自己投一票； -
上面说过
CoordinationState中electionWon记录了是否已经选举为Leader，所以这里进行判断，如果已经被选举成为了Leader，调用handleJoinIgnoringExceptions处理JOIN请求，这个方法底层还是调用CoordinationState的handleJoin进行处理，只不过在外层进行了异常捕捉，会忽略抛出的异常，因为节点之前已经成功选举了Leader，所以本次JION请求处理无关紧要，为了不让异常影响后续的流程，所以对异常进行一个捕捉； -
如果还未成功选举为Leader，调用
CoordinationState的handleJoin处理请求，与第一步不一样的是这个不会对异常进行捕捉，因为此时还没成为Leader，如果有异常信息需要抛出；
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
// 获取CoordinationState
private final SetOnce<CoordinationState> coordinationState = new SetOnce<>();
private void handleJoin(Join join) {
synchronized (mutex) {
// 确保Term最新，如果不是最新，会返回一个JOIN对象，调用handleJoin进行处理，这里可以理解为节点给自己投了一票
ensureTermAtLeast(getLocalNode(), join.getTerm()).ifPresent(this::handleJoin);
// 如果已经被选举为Leader
if (coordinationState.get().electionWon()) {
// 调用对异常进行捕捉的handleJoin方法
final boolean isNewJoinFromMasterEligibleNode = handleJoinIgnoringExceptions(join);
final boolean establishedAsMaster = mode == Mode.LEADER && getLastAcceptedState().term() == getCurrentTerm();
if (isNewJoinFromMasterEligibleNode && establishedAsMaster && publicationInProgress() == false) {
scheduleReconfigurationIfNeeded();
}
} else { // 如果还未为成为Leader
// CoordinationState的handleJoin处理请求
coordinationState.get().handleJoin(join);
}
}
}
private boolean handleJoinIgnoringExceptions(Join join) {
try {
// CoordinationState的handleJoin处理请求
return coordinationState.get().handleJoin(join);
} catch (CoordinationStateRejectedException e) {
logger.debug(() -> "failed to add " + join + " - ignoring", e);
return false;
}
}
}
在CoordinationState的handleJoin中，首先会对Term和版本信息进行一系列的校验，如果校验通过，记录收到的JOIN请求个数，表示当前已经成功收到的投票数，然后调用isElectionQuorum判断是否获得了大多数的投票，也就是获得的投票数达到了Quorum，并将值更新到electionWon中：
public class CoordinationState {
public boolean handleJoin(Join join) {
assert join.targetMatches(localNode) : "handling join " + join + " for the wrong node " + localNode;
// 如果收到的JOIN请求Term与当前节点的Term不一致抛出异常
if (join.getTerm() != getCurrentTerm()) {
logger.debug("handleJoin: ignored join due to term mismatch (expected: [{}], actual: [{}])", getCurrentTerm(), join.getTerm());
throw new CoordinationStateRejectedException(
"incoming term " + join.getTerm() + " does not match current term " + getCurrentTerm()
);
}
// ...
// 获取上一次的Term
final long lastAcceptedTerm = getLastAcceptedTerm();
// 如果请求中的上一次接受的Term大于当前节点的lastAcceptedTerm，抛出异常
if (join.getLastAcceptedTerm() > lastAcceptedTerm) {
logger.debug( "handleJoin: ignored join as joiner has a better last accepted term (expected: <=[{}], actual: [{}])", lastAcceptedTerm, join.getLastAcceptedTerm());
throw new CoordinationStateRejectedException( "incoming last accepted term "
+ join.getLastAcceptedTerm() + " of join higher than current last accepted term "
+ lastAcceptedTerm
);
}
// 对比版本
if (join.getLastAcceptedTerm() == lastAcceptedTerm && join.getLastAcceptedVersion() > getLastAcceptedVersion()) {
logger.debug("handleJoin: ignored join as joiner has a better last accepted version (expected: <=[{}], actual: [{}]) in term {}", getLastAcceptedVersion(), join.getLastAcceptedVersion(), lastAcceptedTerm);
throw new CoordinationStateRejectedException("incoming last accepted version "
+ join.getLastAcceptedVersion() + " of join higher than current last accepted version "
+ getLastAcceptedVersion() + " in term " + lastAcceptedTerm);
}
// ...
// 记录JOIN投票
boolean added = joinVotes.addJoinVote(join);
boolean prevElectionWon = electionWon;
// 判断是否得到了大多数投票，这里会更新electionWon的值
electionWon = isElectionQuorum(joinVotes);
assert prevElectionWon == false || electionWon : // we cannot go from won to not won
"locaNode= " + localNode + ", join=" + join + ", joinVotes=" + joinVotes;
logger.debug(
"handleJoin: added join {} from [{}] for election, electionWon={} lastAcceptedTerm={} lastAcceptedVersion={}", join, join.getSourceNode(), electionWon, lastAcceptedTerm, getLastAcceptedVersion()
);
// 如果得到了大多数投票并且上一次没有选举为Leader
if (electionWon && prevElectionWon == false) {
logger.debug("handleJoin: election won in term [{}] with {}", getCurrentTerm(), joinVotes;
lastPublishedVersion = getLastAcceptedVersion();
}
return added;
}
}
转为Leader
当节点收到了大多数投票后，就会调用becomeLeader转为Leader，这里会将节点由CANDIDATE转为LEADER角色，然后调用preVoteCollector的update更新Term和Leader节点信息：
public class Coordinator extends AbstractLifecycleComponent implements ClusterStatePublisher {
private void becomeLeader() {
assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
// 是否是CANDIDATE
assert mode == Mode.CANDIDATE : "expected candidate but was " + mode;
// 是否有Master角色权限
assert getLocalNode().isMasterNode() : getLocalNode() + " became a leader but is not master-eligible";
logger.debug("handleJoinRequest: coordinator becoming LEADER in term {} (was {}, lastKnownLeader was [{}])", getCurrentTerm(), mode,lastKnownLeader);
// 转为Leader
mode = Mode.LEADER;
joinAccumulator.close(mode);
// 设置为LeaderJoinAccumulator
joinAccumulator = joinHelper.new LeaderJoinAccumulator();
lastKnownLeader = Optional.of(getLocalNode());
peerFinder.deactivate(getLocalNode());
clusterFormationFailureHelper.stop();
closePrevotingAndElectionScheduler();
// 更新Leader信息和Term信息
preVoteCollector.update(getPreVoteResponse(), getLocalNode());
assert leaderChecker.leader() == null : leaderChecker.leader();
followersChecker.updateFastResponseState(getCurrentTerm(), mode);
}
}
参考
【张超】深入理解 Elasticsearch 7.x 新的集群协调层
【政采云技术团队】Elasticsearch系列之二选主7.x之后
Elasticsearch版本：8.3
标签：Term,JOIN,请求,Leader,public,Elasticsearch,主流程,节点,ES 来源： https://www.cnblogs.com/shanml/p/16684887.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。