Skip to content

<fix>[mn]: synchronize hash ring operations #4432

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.

Sign up for GitHub

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

Open
zstack-robot-2 wants to merge 1 commit into
base: 5.4.10
Choose a base branch
from
Open

<fix>[mn]: synchronize hash ring operations #4432

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
28 changes: 14 additions & 14 deletions core/src/main/java/org/zstack/core/cloudbus/ResourceDestinationMakerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,27 +27,27 @@ public class ResourceDestinationMakerImpl implements ManagementNodeChangeListene
private DatabaseFacade dbf;

@Override
public void nodeJoin(ManagementNodeInventory inv) {
public synchronized void nodeJoin(ManagementNodeInventory inv) {
nodeHash.add(inv.getUuid());
nodes.put(inv.getUuid(), new NodeInfo(inv));
}

@Override
public void nodeLeft(ManagementNodeInventory inv) {
public synchronized void nodeLeft(ManagementNodeInventory inv) {
String nodeId = inv.getUuid();
nodeHash.remove(nodeId);
nodes.remove(nodeId);
}

@Override
public void iAmDead(ManagementNodeInventory inv) {
public synchronized void iAmDead(ManagementNodeInventory inv) {
String nodeId = inv.getUuid();
nodeHash.remove(nodeId);
nodes.remove(nodeId);
}

@Override
public void iJoin(ManagementNodeInventory inv) {
public synchronized void iJoin(ManagementNodeInventory inv) {
List<ManagementNodeVO> lst = Q.New(ManagementNodeVO.class).list();
lst.forEach((ManagementNodeVO node) -> {
nodeHash.add(node.getUuid());
Expand All @@ -56,7 +56,7 @@ public void iJoin(ManagementNodeInventory inv) {
}

@Override
public String makeDestination(String resourceUuid) {
public synchronized String makeDestination(String resourceUuid) {
String nodeUuid = nodeHash.get(resourceUuid);
if (nodeUuid == null) {
throw new CloudRuntimeException("Cannot find any available management node to send message");
Expand All @@ -66,18 +66,18 @@ public String makeDestination(String resourceUuid) {
}

@Override
public boolean isManagedByUs(String resourceUuid) {
public synchronized boolean isManagedByUs(String resourceUuid) {
String nodeUuid = makeDestination(resourceUuid);
return nodeUuid.equals(Platform.getManagementServerId());
}

@Override
public Collection<String> getManagementNodesInHashRing() {
return nodeHash.getNodes();
public synchronized Collection<String> getManagementNodesInHashRing() {
return new ArrayList<>(nodeHash.getNodes());
}

@Override
public NodeInfo getNodeInfo(String nodeUuid) {
public synchronized NodeInfo getNodeInfo(String nodeUuid) {
NodeInfo info = nodes.get(nodeUuid);
if (info == null) {
ManagementNodeVO vo = dbf.findByUuid(nodeUuid, ManagementNodeVO.class);
Expand All @@ -93,17 +93,17 @@ public NodeInfo getNodeInfo(String nodeUuid) {
}

@Override
public Collection<NodeInfo> getAllNodeInfo() {
return nodes.values();
public synchronized Collection<NodeInfo> getAllNodeInfo() {
return new ArrayList<>(nodes.values());
}

@Override
public int getManagementNodeCount() {
return nodes.values().size();
public synchronized int getManagementNodeCount() {
return nodes.size();
}


public boolean isNodeInCircle(String nodeId) {
public synchronized boolean isNodeInCircle(String nodeId) {
return nodeHash.hasNode(nodeId);
}
}
101 changes: 68 additions & 33 deletions portal/src/main/java/org/zstack/portal/managementnode/ManagementNodeManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -106,6 +107,15 @@ public class ManagementNodeManagerImpl extends AbstractService implements Manage
// A dictionary (nodeId -> ManagementNodeInventory) of joined management Node
final private Map<String, ManagementNodeInventory> joinedManagementNodes = new ConcurrentHashMap<>();

// Lock to serialize lifecycle events from heartbeat reconciliation and canonical event callbacks,
// preventing race conditions where a nodeJoin event is immediately followed by a stale nodeLeft
// from the heartbeat thread, or vice versa. See ZSTAC-77711.
private final Object lifecycleLock = new Object();

// Track nodes found in hash ring but missing from DB. Only call nodeLeft after a node
// is missing for two consecutive heartbeat cycles, to avoid removing nodes that just joined.
private final Set<String> suspectedMissingFromDb = new HashSet<>();

private static int NODE_STARTING = 0;
private static int NODE_RUNNING = 1;
private static int NODE_FAILED = -1;
Expand Down Expand Up @@ -367,12 +377,16 @@ protected void run(Map tokens, Object data) {

ManagementNodeLifeCycleData d = (ManagementNodeLifeCycleData) data;

if (LifeCycle.NodeJoin.toString().equals(d.getLifeCycle())) {
nodeLifeCycle.nodeJoin(d.getInventory());
} else if (LifeCycle.NodeLeft.toString().equals(d.getLifeCycle())) {
nodeLifeCycle.nodeLeft(d.getInventory());
} else {
throw new CloudRuntimeException(String.format("unknown lifecycle[%s]", d.getLifeCycle()));
synchronized (lifecycleLock) {
if (LifeCycle.NodeJoin.toString().equals(d.getLifeCycle())) {
// Clear from suspected set since the node is confirmed alive
suspectedMissingFromDb.remove(d.getInventory().getUuid());
nodeLifeCycle.nodeJoin(d.getInventory());
} else if (LifeCycle.NodeLeft.toString().equals(d.getLifeCycle())) {
nodeLifeCycle.nodeLeft(d.getInventory());
} else {
throw new CloudRuntimeException(String.format("unknown lifecycle[%s]", d.getLifeCycle()));
}
}
}
};
Expand Down Expand Up @@ -853,34 +867,55 @@ private void checkAllNodesHealth() {

Set<String> nodeUuidsInDb = nodesInDb.stream().map(ManagementNodeVO::getUuid).collect(Collectors.toSet());

// When a node is dying, we may not receive the the dead notification because the message bus may be also dead
// at that moment. By checking if the node UUID is still in our hash ring, we know what nodes should be kicked out
destinationMaker.getManagementNodesInHashRing().forEach(nodeUuid -> {
if (!nodeUuidsInDb.contains(nodeUuid)) {
logger.warn(String.format("found that a management node[uuid:%s] had no heartbeat in database but still in our hash ring," +
"notify that it's dead", nodeUuid));
ManagementNodeInventory inv = new ManagementNodeInventory();
inv.setUuid(nodeUuid);
inv.setHostName(destinationMaker.getNodeInfo(nodeUuid).getNodeIP());

nodeLifeCycle.nodeLeft(inv);
}
});

// check if any node missing in our hash ring
nodesInDb.forEach(n -> {
if (n.getUuid().equals(node().getUuid()) || suspects.contains(n)) {
return;
}

new Runnable() {
@Override
@AsyncThread
public void run() {
nodeLifeCycle.nodeJoin(ManagementNodeInventory.valueOf(n));
// Reconcile hash ring with DB under lifecycleLock to prevent race with
// canonical event callbacks (nodeJoin/nodeLeft). See ZSTAC-77711.
synchronized (lifecycleLock) {
// When a node is dying, we may not receive the dead notification because the message bus may be also dead
// at that moment. By checking if the node UUID is still in our hash ring, we know what nodes should be kicked out.
// Use two-round confirmation: first round marks as suspected, second round actually removes.
Set<String> currentSuspected = new HashSet<>();
destinationMaker.getManagementNodesInHashRing().forEach(nodeUuid -> {
if (!nodeUuidsInDb.contains(nodeUuid)) {
if (suspectedMissingFromDb.contains(nodeUuid)) {
// Second consecutive detection — confirmed missing, remove from hash ring
logger.warn(String.format("management node[uuid:%s] confirmed missing from database for two consecutive" +
" heartbeat cycles, removing from hash ring", nodeUuid));
ManagementNodeInventory inv = new ManagementNodeInventory();
inv.setUuid(nodeUuid);
try {
inv.setHostName(destinationMaker.getNodeInfo(nodeUuid).getNodeIP());
} catch (Exception e) {
logger.warn(String.format("cannot get node info for node[uuid:%s], use empty hostname", nodeUuid));
}

nodeLifeCycle.nodeLeft(inv);
} else {
// First detection — mark as suspected, defer removal to next cycle
logger.warn(String.format("management node[uuid:%s] not found in database but still in hash ring," +
" marking as suspected (will remove on next heartbeat if still missing)", nodeUuid));
currentSuspected.add(nodeUuid);
}
}
}.run();
});
});
// Update suspected set: only keep nodes that are newly suspected this round
suspectedMissingFromDb.clear();
suspectedMissingFromDb.addAll(currentSuspected);

// check if any node missing in our hash ring
nodesInDb.forEach(n -> {
if (n.getUuid().equals(node().getUuid()) || suspects.contains(n)) {
return;
}

new Runnable() {
@Override
@AsyncThread
public void run() {
nodeLifeCycle.nodeJoin(ManagementNodeInventory.valueOf(n));
}
}.run();
});
}
}

@Override
Expand Down