Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Refactor&Enhancement] Enhance naming server status check. #12573

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,13 @@ public interface ConsistencyProtocol<T extends Config, P extends RequestProcesso
*/
void memberChange(Set<String> addresses);

/**
* Whether protocol is ready to work, such as contain leader, finish load snapshot and so on.
*
* @return {@code true} when protocol ready to work, otherwise {@code false}
*/
boolean isReady();

/**
* Consistency agreement service shut down .
* 一致性协议服务关闭
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,14 @@ public APProtocol getApProtocol() {
return apProtocol;
}

public boolean isCpInit() {
return cpInit;
}

public boolean isApInit() {
return apInit;
}

@PreDestroy
@Override
public void destroy() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -224,4 +224,9 @@ public boolean isLeader(String group) {
}
return node.isLeader();
}

@Override
public boolean isReady() {
return raftServer.isReady();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -520,6 +520,17 @@ public Node findNodeByGroup(final String group) {
return null;
}

public boolean isReady() {
if (raftConfig.isStrictMode()) {
for (RequestProcessor4CP each : processors) {
if (null == getLeader(each.group())) {
return false;
}
}
}
return isStarted;
}

Map<String, RaftGroupTuple> getMultiRaftGroup() {
return multiRaftGroup;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ public class RaftConfig implements Config<RequestProcessor4CP> {

private Set<String> members = Collections.synchronizedSet(new HashSet<>());

private boolean strictMode;

@Override
public void setMembers(String self, Set<String> members) {
this.selfAddress = self;
Expand Down Expand Up @@ -95,6 +97,14 @@ public String getValOfDefault(String key, String defaultVal) {
return data.getOrDefault(key, defaultVal);
}

public void setStrictMode(boolean strictMode) {
this.strictMode = strictMode;
}

public boolean isStrictMode() {
return strictMode;
}

@Override
public String toString() {
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -118,4 +118,10 @@ void testIsLeader() {
verify(serverMock).findNodeByGroup(groupId);
verify(nodeMock).isLeader();
}

@Test
void testIsReady() {
raftProtocol.isReady();
verify(serverMock).isReady();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import org.springframework.mock.env.MockEnvironment;
import org.springframework.test.util.ReflectionTestUtils;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
Expand Down Expand Up @@ -96,6 +97,8 @@ class JRaftServerTest {

private JRaftServer server;

private RaftConfig config;

@Mock
private RequestProcessor4CP mockProcessor4CP;

Expand Down Expand Up @@ -133,7 +136,7 @@ static void beforeClass() {
@BeforeEach
void before() throws NoSuchFieldException, IllegalAccessException {
initPeersAndConfiguration();
RaftConfig config = new RaftConfig();
config = new RaftConfig();
Collection<Member> initEvent = Collections.singletonList(Member.builder().ip("1.1.1.1").port(7848).build());
config.setMembers("1.1.1.1:7848", ProtocolManager.toCPMembersInfo(initEvent));

Expand All @@ -148,7 +151,8 @@ boolean peerChange(JRaftMaintainService maintainService, Set<String> newPeers) {
server.init(config);

Map<String, JRaftServer.RaftGroupTuple> map = new HashMap<>();
map.put("test_nacos", new JRaftServer.RaftGroupTuple(node, requestProcessor, raftGroupService, nacosStateMachine));
map.put("test_nacos",
new JRaftServer.RaftGroupTuple(node, requestProcessor, raftGroupService, nacosStateMachine));
server.mockMultiRaftGroup(map);

mockcliClientService();
Expand Down Expand Up @@ -221,19 +225,21 @@ void testInvokeToLeader()
when(cliClientServiceMock.getRpcClient()).thenReturn(rpcClient);
setLeaderAs(peerId1);
int timeout = 3000;
Method invokeToLeaderMethod = JRaftServer.class.getDeclaredMethod("invokeToLeader", String.class, Message.class, int.class,
FailoverClosure.class);
Method invokeToLeaderMethod = JRaftServer.class.getDeclaredMethod("invokeToLeader", String.class, Message.class,
int.class, FailoverClosure.class);
invokeToLeaderMethod.setAccessible(true);
invokeToLeaderMethod.invoke(server, groupId, this.readRequest, timeout, null);
verify(cliClientServiceMock).getRpcClient();
verify(rpcClient).invokeAsync(eq(peerId1.getEndpoint()), eq(readRequest), any(InvokeCallback.class), any(long.class));
verify(rpcClient).invokeAsync(eq(peerId1.getEndpoint()), eq(readRequest), any(InvokeCallback.class),
any(long.class));
}

@Test
void testRefreshRouteTable() {
server.refreshRouteTable(groupId);
verify(cliClientServiceMock, times(1)).connect(peerId1.getEndpoint());
verify(cliClientServiceMock).getLeader(eq(peerId1.getEndpoint()), any(CliRequests.GetLeaderRequest.class), eq(null));
verify(cliClientServiceMock).getLeader(eq(peerId1.getEndpoint()), any(CliRequests.GetLeaderRequest.class),
eq(null));
}

@Test
Expand Down Expand Up @@ -324,6 +330,16 @@ public RestResult<String> execute(Map<String, String> args) {
changed.set(false);
}

@Test
void testIsReady() {
assertTrue(server.isReady());
config.setStrictMode(true);
((Collection<RequestProcessor4CP>) ReflectionTestUtils.getField(server, "processors")).add(mockProcessor4CP);
assertTrue(server.isReady());
setLeaderAs(null);
assertFalse(server.isReady());
}

@AfterEach
void shutdown() {
server.shutdown();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@

import com.alibaba.nacos.common.utils.StringUtils;
import com.alibaba.nacos.core.distributed.ProtocolManager;
import com.alibaba.nacos.naming.constants.Constants;
import com.alibaba.nacos.core.distributed.distro.DistroProtocol;
import com.alibaba.nacos.naming.misc.GlobalConfig;
import com.alibaba.nacos.naming.misc.GlobalExecutor;
import com.alibaba.nacos.naming.misc.SwitchDomain;
import com.alipay.sofa.jraft.RouteTable;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
Expand All @@ -36,13 +36,20 @@
@Service
public class ServerStatusManager {

private final GlobalConfig globalConfig;

private final DistroProtocol distroProtocol;

private final ProtocolManager protocolManager;

private final SwitchDomain switchDomain;

private ServerStatus serverStatus = ServerStatus.STARTING;

public ServerStatusManager(ProtocolManager protocolManager, SwitchDomain switchDomain) {
public ServerStatusManager(GlobalConfig globalConfig, DistroProtocol distroProtocol,
ProtocolManager protocolManager, SwitchDomain switchDomain) {
this.globalConfig = globalConfig;
this.distroProtocol = distroProtocol;
this.protocolManager = protocolManager;
this.switchDomain = switchDomain;
}
Expand All @@ -59,30 +66,37 @@ private void refreshServerStatus() {
return;
}

if (hasLeader()) {
if (isReady()) {
serverStatus = ServerStatus.UP;
} else {
serverStatus = ServerStatus.DOWN;
}
}

private boolean hasLeader() {
if (protocolManager.getCpProtocol() == null) {
private boolean isReady() {
if (!globalConfig.isDataWarmup()) {
return true;
}
if (!protocolManager.isCpInit() || protocolManager.getCpProtocol() == null) {
return false;
}
return null != RouteTable.getInstance().selectLeader(Constants.NAMING_PERSISTENT_SERVICE_GROUP);
return protocolManager.getCpProtocol().isReady() && distroProtocol.isInitialized();
}

public ServerStatus getServerStatus() {
return serverStatus;
}

public Optional<String> getErrorMsg() {
if (hasLeader()) {
if (isReady()) {
return Optional.empty();
}
return Optional.of("No leader for raft group " + Constants.NAMING_PERSISTENT_SERVICE_GROUP
+ ", please see logs `alipay-jraft.log` or `naming-raft.log` to see details.");
if (!distroProtocol.isInitialized()) {
return Optional.of(
"Distro snapshot load failed, please see logs `protocol-distro.log` or `naming-distro.log` to see details.");
}
return Optional.of(
"No leader for raft, please see logs `alipay-jraft.log` or `naming-raft.log` to see details.");
}

public class ServerStatusUpdater implements Runnable {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Map;
import java.util.Optional;

/**
* Filter incoming traffic to refuse or revise unexpected requests.
Expand Down Expand Up @@ -98,8 +99,9 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha
}

final String statusMsg = "server is " + serverStatusManager.getServerStatus().name() + "now";
if (serverStatusManager.getErrorMsg().isPresent()) {
resp.getWriter().write(statusMsg + ", detailed error message: " + serverStatusManager.getErrorMsg());
Optional<String> errorMsg = serverStatusManager.getErrorMsg();
if (errorMsg.isPresent()) {
resp.getWriter().write(statusMsg + ", detailed error message: " + errorMsg.get());
} else {
resp.getWriter().write(statusMsg + ", please try again later!");
}
Expand Down
Loading
Loading