Copycat server之间的configure是如何，何时被同步的？
大家可以看到，只有leader可以同步配置
1. 显式的调用LeaderState.configure
LeaderState.configure
/**
* Commits the given configuration.
*/
protected CompletableFuture<Long> configure(Collection<Member> members) {
final long index;
try (ConfigurationEntry entry = context.getLog().create(ConfigurationEntry.class)) {
entry.setTerm(context.getTerm())
.setTimestamp(System.currentTimeMillis())
.setMembers(members);
index = context.getLog().append(entry); //先把configuration写入local log
LOGGER.debug("{} - Appended {}", context.getCluster().member().address(), entry); // Store the index of the configuration entry in order to prevent other configurations from
// being logged and committed concurrently. This is an important safety property of Raft.
configuring = index; //configuring用于互斥
context.getClusterState().configure(new Configuration(entry.getIndex(), entry.getTerm(), entry.getTimestamp(), entry.getMembers())); //更新ClusterState
} return appender.appendEntries(index).whenComplete((commitIndex, commitError) -> { //调用appendEntries同步configuration
context.checkThread();
if (isOpen()) {
// Reset the configuration index to allow new configuration changes to be committed.
configuring = 0; //configuring完成
}
});
}
何处显式调用？
private void appendInitialEntries() {
// Append a configuration entry to propagate the leader's cluster configuration.
configure(context.getCluster().members());
public CompletableFuture<JoinResponse> join(final JoinRequest request) {
configure(members).whenComplete((index, error) -> {
@Override
public CompletableFuture<ReconfigureResponse> reconfigure(final ReconfigureRequest request) {
configure(members).whenComplete((index, error) -> {
@Override
public CompletableFuture<LeaveResponse> leave(final LeaveRequest request) {
configure(members).whenComplete((index, error) -> {
2. 在发送hb，和appendEntries时，也会自动发生同步
因为上面调用configure也是会调用到
LeaderAppender.appendEntries(MemberState member)
和普通append不同的是，会走到这个逻辑，
// If the member term is less than the current term or the member's configuration index is less
// than the local configuration index, send a configuration update to the member.
// Ensure that only one configuration attempt per member is attempted at any given time by storing the
// member state in a set of configuring members.
// Once the configuration is complete sendAppendRequest will be called recursively.
else if (member.getConfigTerm() < context.getTerm() || member.getConfigIndex() < context.getClusterState().getConfiguration().index()) {
if (member.canConfigure()) {
sendConfigureRequest(member, buildConfigureRequest(member));
}
}
注意虽然是在appendEntries里面，
这里发出的是sendConfigureRequest，而不是appendRequest
因为leader的configuration发生变化，所以member.getConfigIndex一定是小的，所以要更新
AbstractAppender
/**
* Connects to the member and sends a configure request.
*/
protected void sendConfigureRequest(MemberState member, ConfigureRequest request) {
// Start the configure to the member.
member.startConfigure(); context.getConnections().getConnection(member.getMember().serverAddress()).whenComplete((connection, error) -> {
context.checkThread(); if (open) {
if (error == null) {
sendConfigureRequest(connection, member, request);
} else {
// Complete the configure to the member.
member.completeConfigure(); // 将configuring设置成false，表示configuring结束 // Trigger reactions to the request failure.
handleConfigureRequestFailure(member, request, error);
}
}
});
} /**
* Sends a configuration message.
*/
protected void sendConfigureRequest(Connection connection, MemberState member, ConfigureRequest request) {
LOGGER.debug("{} - Sent {} to {}", context.getCluster().member().address(), request, member.getMember().serverAddress());
connection.<ConfigureRequest, ConfigureResponse>send(request).whenComplete((response, error) -> {
context.checkThread(); // Complete the configure to the member.
member.completeConfigure(); if (open) {
if (error == null) {
LOGGER.debug("{} - Received {} from {}", context.getCluster().member().address(), response, member.getMember().serverAddress());
handleConfigureResponse(member, request, response);
} else {
LOGGER.warn("{} - Failed to configure {}", context.getCluster().member().address(), member.getMember().serverAddress());
handleConfigureResponseFailure(member, request, error);
}
}
});
}
LeaderAppender
@Override
protected void handleConfigureResponse(MemberState member, ConfigureRequest request, ConfigureResponse response) {
// Trigger commit futures if necessary.
updateHeartbeatTime(member, null); // 判断是否大部分member都已经完成configuration同步 super.handleConfigureResponse(member, request, response);
}
AbstractAppender
/**
* Handles a configuration response.
*/
protected void handleConfigureResponse(MemberState member, ConfigureRequest request, ConfigureResponse response) {
if (response.status() == Response.Status.OK) {
handleConfigureResponseOk(member, request, response);
} else {
handleConfigureResponseError(member, request, response);
}
}
/**
* Handles an OK configuration response.
*/
@SuppressWarnings("unused")
protected void handleConfigureResponseOk(MemberState member, ConfigureRequest request, ConfigureResponse response) {
// Reset the member failure count and update the member's status if necessary.
succeedAttempt(member); // Update the member's current configuration term and index according to the installed configuration.
member.setConfigTerm(request.term()).setConfigIndex(request.index()); // Recursively append entries to the member.
appendEntries(member);
}
在server上收到ConfigureRequest
ServerContext
public void connectServer(Connection connection) {
connection.handler(ConfigureRequest.class, request -> state.configure(request));
在connectServer中，说明configuration只能在server间调用，client是不能调用的
configure，只在两个state中有实现
InactiveState
@Override
public CompletableFuture<ConfigureResponse> configure(ConfigureRequest request) {
context.checkThread();
logRequest(request);
updateTermAndLeader(request.term(), request.leader()); // 更新leader term Configuration configuration = new Configuration(request.index(), request.term(), request.timestamp(), request.members()); // 根据request创建Configuration // Configure the cluster membership. This will cause this server to transition to the
// appropriate state if its type has changed.
context.getClusterState().configure(configuration); // 调用ClusterState.configure // If the configuration is already committed, commit it to disk.
// Check against the actual cluster Configuration rather than the received configuration in
// case the received configuration was an older configuration that was not applied.
if (context.getCommitIndex() >= context.getClusterState().getConfiguration().index()) {
context.getClusterState().commit(); // 调用commit
} return CompletableFuture.completedFuture(logResponse(ConfigureResponse.builder()
.withStatus(Response.Status.OK) // 返回
.build()));
}
FollowerState
public CompletableFuture<ConfigureResponse> configure(ConfigureRequest request) {
CompletableFuture<ConfigureResponse> future = super.configure(request);
resetHeartbeatTimeout();
return future;
}
可以看到Follower里面只是多了resetHB
继续，
可以看到首先是ClusterState.configure
ClusterState中保存了当前server，所知道cluster的所有信息，其中包含Configuration对象
final class ClusterState implements Cluster, AutoCloseable {
private final ServerContext context;
private final ServerMember member;
private volatile Configuration configuration;
configure
/**
* Configures the cluster state.
*
* @param configuration The cluster configuration.
* @return The cluster state.
*/
ClusterState configure(Configuration configuration) {
// If the configuration index is less than the currently configured index, ignore it.
// Configurations can be persisted and applying old configurations can revert newer configurations.
if (this.configuration != null && configuration.index() <= this.configuration.index()) //如果是老的configuration，丢弃
return this; Instant time = Instant.ofEpochMilli(configuration.time()); // Iterate through members in the new configuration, add any missing members, and update existing members.
boolean transition = false;
for (Member member : configuration.members()) {
if (member.equals(this.member)) { //如果有我的配置变更
transition = this.member.type().ordinal() < member.type().ordinal(); //看下type是否promote，只有promote才要transition，不让demote
this.member.update(member.type(), time).update(member.clientAddress(), time);
members.add(this.member);
} else { //如果是更新其他member的配置
// If the member state doesn't already exist, create it.
MemberState state = membersMap.get(member.id());
if (state == null) { //如果是新member，初始化
state = new MemberState(new ServerMember(member.type(), member.serverAddress(), member.clientAddress(), time), this);
state.resetState(context.getLog());
this.members.add(state.getMember());
this.remoteMembers.add(state);
membersMap.put(member.id(), state);
addressMap.put(member.address(), state);
joinListeners.accept(state.getMember());
} // If the member type has changed, update the member type and reset its state.
state.getMember().update(member.clientAddress(), time);
if (state.getMember().type() != member.type()) { //如果member的type发生了变化，更新数据
state.getMember().update(member.type(), time);
state.resetState(context.getLog());
} // If the member status has changed, update the local member status.
if (state.getMember().status() != member.status()) { //如果status发送变化，更新
state.getMember().update(member.status(), time);
} // Update the optimized member collections according to the member type.
for (List<MemberState> memberType : memberTypes.values()) {
memberType.remove(state);
} List<MemberState> memberType = memberTypes.get(member.type());
if (memberType == null) {
memberType = new CopyOnWriteArrayList<>();
memberTypes.put(member.type(), memberType);
}
memberType.add(state);
}
} // Transition the local member only if the member is being promoted and not demoted.
// Configuration changes that demote the local member are only applied to the local server
// upon commitment. This ensures that e.g. a leader that's removing itself from the quorum
// can commit the configuration change prior to shutting down.
if (transition) { //做state transition
context.transition(this.member.type());
} // Iterate through configured members and remove any that no longer exist in the configuration.
int i = 0;
while (i < this.remoteMembers.size()) {
MemberState member = this.remoteMembers.get(i);
if (!configuration.members().contains(member.getMember())) {
this.members.remove(member.getMember());
this.remoteMembers.remove(i);
for (List<MemberState> memberType : memberTypes.values()) {
memberType.remove(member);
}
membersMap.remove(member.getMember().id());
addressMap.remove(member.getMember().address());
leaveListeners.accept(member.getMember());
} else {
i++;
}
} // If the local member was removed from the cluster, remove it from the members list.
if (!configuration.members().contains(member)) {
members.remove(member);
} this.configuration = configuration; //更新configuration对象 // Store the configuration if it's already committed.
if (context.getCommitIndex() >= configuration.index()) { //如果这个configuration已经被commit，store到存储上
context.getMetaStore().storeConfiguration(configuration);
} // Reassign members based on availability.
reassign(); //更新passive member的assignment，因为member变了，所以follower所对应的passive可能需要重新分配 return this;
}
ClusterState.commit
/**
* Commit the current configuration to disk.
*
* @return The cluster state.
*/
ClusterState commit() {
// Apply the configuration to the local server state.
context.transition(member.type());
if (!configuration.members().contains(member) && leaveFuture != null) {
leaveFuture.complete(null);
} // If the local stored configuration is older than the committed configuration, overwrite it.
if (context.getMetaStore().loadConfiguration().index() < configuration.index()) {
context.getMetaStore().storeConfiguration(configuration);
}
return this;
}
逻辑，就是把configuration存入本地盘
reconfigure
在什么地方被调用，
ServerMember
@Override
public CompletableFuture<Void> promote() {
return configure(Type.values()[type.ordinal() + 1]);
}
可以用reconfigure来更新配置，比如promote，demote
/**
* Demotes the server to the given type.
*/
CompletableFuture<Void> configure(Member.Type type) {
CompletableFuture<Void> future = new CompletableFuture<>();
cluster.getContext().getThreadContext().executor().execute(() -> configure(type, future));
return future;
}
/**
* Recursively reconfigures the cluster.
*/
private void configure(Member.Type type, CompletableFuture<Void> future) {
// Set a timer to retry the attempt to leave the cluster.
configureTimeout = cluster.getContext().getThreadContext().schedule(cluster.getContext().getElectionTimeout(), () -> {
configure(type, future); //设定schedule，反复重试
}); // Attempt to leave the cluster by submitting a LeaveRequest directly to the server state.
// Non-leader states should forward the request to the leader if there is one. Leader states
// will log, replicate, and commit the reconfiguration.
cluster.getContext().getServerState().reconfigure(ReconfigureRequest.builder()
.withIndex(cluster.getConfiguration().index())
.withTerm(cluster.getConfiguration().term())
.withMember(new ServerMember(type, serverAddress(), clientAddress(), updated))
.build()).whenComplete((response, error) -> {
if (error == null) {
if (response.status() == Response.Status.OK) {
cancelConfigureTimer(); //如果成功就cancel掉schedule
cluster.configure(new Configuration(response.index(), response.term(), response.timestamp(), response.members())); //更新clusterState中的configuration配置
future.complete(null);
} else if (response.error() == null || response.error() == CopycatError.Type.NO_LEADER_ERROR) {
cancelConfigureTimer();
configureTimeout = cluster.getContext().getThreadContext().schedule(cluster.getContext().getElectionTimeout().multipliedBy(2), () -> configure(type, future));
} else {
cancelConfigureTimer();
future.completeExceptionally(response.error().createException());
}
}
});
}
处理ReconfigureRequest
public void connectServer(Connection connection) {
connection.handler(ReconfigureRequest.class, request -> state.reconfigure(request));
ReserveState
@Override
public CompletableFuture<ReconfigureResponse> reconfigure(ReconfigureRequest request) {
context.checkThread();
logRequest(request); if (context.getLeader() == null) {
return CompletableFuture.completedFuture(logResponse(ReconfigureResponse.builder()
.withStatus(Response.Status.ERROR)
.withError(CopycatError.Type.NO_LEADER_ERROR)
.build()));
} else {
return this.<ReconfigureRequest, ReconfigureResponse>forward(request)
.exceptionally(error -> ReconfigureResponse.builder()
.withStatus(Response.Status.ERROR)
.withError(CopycatError.Type.NO_LEADER_ERROR)
.build())
.thenApply(this::logResponse);
}
}
只是forward到leader
LeaderState
@Override
public CompletableFuture<ReconfigureResponse> reconfigure(final ReconfigureRequest request) {
// If another configuration change is already under way, reject the configuration.
// If the leader index is 0 or is greater than the commitIndex, reject the promote requests.
// Configuration changes should not be allowed until the leader has committed a no-op entry.
// See https://groups.google.com/forum/#!topic/raft-dev/t4xj6dJTP6E
if (configuring() || initializing()) { //如果正在configure或leader初始化，不能做配置变更
return CompletableFuture.completedFuture(logResponse(ReconfigureResponse.builder()
.withStatus(Response.Status.ERROR)
.build()));
} // If the configuration request index is less than the last known configuration index for
// the leader, fail the request to ensure servers can't reconfigure an old configuration.
if (request.index() > 0 && request.index() < context.getClusterState().getConfiguration().index() || request.term() != context.getClusterState().getConfiguration().term()
&& (existingMember.type() != request.member().type() || existingMember.status() != request.member().status())) {
return CompletableFuture.completedFuture(logResponse(ReconfigureResponse.builder()
.withStatus(Response.Status.ERROR)
.withError(CopycatError.Type.CONFIGURATION_ERROR)
.build()));
} Member member = request.member(); // If the client address is being set or has changed, update the configuration.
if (member.clientAddress() != null && (existingMember.clientAddress() == null || !existingMember.clientAddress().equals(member.clientAddress()))) {
existingMember.update(member.clientAddress(), Instant.now());
} // Update the member type.
existingMember.update(request.member().type(), Instant.now()); Collection<Member> members = context.getCluster().members(); CompletableFuture<ReconfigureResponse> future = new CompletableFuture<>();
configure(members).whenComplete((index, error) -> { //调用configure
context.checkThread();
if (isOpen()) {
if (error == null) { //如果成功，更新local配置
future.complete(logResponse(ReconfigureResponse.builder()
.withStatus(Response.Status.OK)
.withIndex(index)
.withTerm(context.getClusterState().getConfiguration().term())
.withTime(context.getClusterState().getConfiguration().time())
.withMembers(members)
.build()));
} else {
future.complete(logResponse(ReconfigureResponse.builder()
.withStatus(Response.Status.ERROR)
.withError(CopycatError.Type.INTERNAL_ERROR)
.build()));
}
}
});
return future;
}