Skip to content

<fix>[zbs]: backport MDS automatic reconnect #4532

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 wants to merge 2 commits into
base: 5.4.12
Choose a base branch
from
Open

<fix>[zbs]: backport MDS automatic reconnect #4532

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
de96142
<fix>[zbs]: Try to reconnect after zbs ping failure
Jan 13, 2026
ca7439e
<fix>[zbs]: Try to reconnect after all mds disconnected
Jan 28, 2026
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
3 changes: 2 additions & 1 deletion plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@ public interface ZbsConstants {
long ZBS_HEARTBEAT_VOLUME_SIZE_IN_GIGABYTE = 1;
String ZBS_CBD_LUN_PATH_FORMAT = "cbd:%s/%s/%s";
String ZBS_CBD_PREFIX_SCHEME = "cbd://";
Integer PRIMARY_STORAGE_MDS_MAXIMUM_PING_FAILURE = 3;
Integer MDS_PING_RETRY_PER_CYCLE = 3;
Integer MDS_PING_FAIL_CYCLE_THRESHOLD = 3;
String VOLUME_PHYSICAL_BLOCK_SIZE = "4096";
String MEGABYTE_SUPPORTED_VERSION = "1.6.1";
String MEGABYTE_UNIT = "M";
Expand Down
8 changes: 7 additions & 1 deletion plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,12 @@ public void run(FlowTrigger trigger, Map data) {
restf.asyncJsonPost(ZbsAgentUrl.primaryStorageUrl(getSelf().getAddr(), SYNC_METADATA_PATH), cmd, new JsonAsyncRESTCallback<SyncMetadataRsp>(trigger) {
@Override
public void success(SyncMetadataRsp ret) {
if (!ret.isSuccess()) {
trigger.fail(operr("unable to sync metadata from ZBS primary storage MDS[%s], because %s",
getSelf().getAddr(), ret.getError()));
return;
}

getSelf().setExternalAddr(ret.getExternalAddr());
trigger.next();
}
Expand Down Expand Up @@ -307,7 +313,7 @@ public String getName() {
}

private void pingMds(ClusterInfo clusterInfo, final Completion completion) {
final Integer MAX_PING_CNT = ZbsConstants.PRIMARY_STORAGE_MDS_MAXIMUM_PING_FAILURE;
final Integer MAX_PING_CNT = ZbsConstants.MDS_PING_RETRY_PER_CYCLE;
final List<Integer> stepCount = new ArrayList<>();
for (int i = 1; i <= MAX_PING_CNT; i++) {
stepCount.add(i);
Expand Down
155 changes: 145 additions & 10 deletions plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,10 +57,12 @@
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.zstack.core.Platform.operr;
import static org.zstack.storage.zbs.ZbsConstants.MDS_PING_FAIL_CYCLE_THRESHOLD;
import static org.zstack.storage.zbs.ZbsHelper.*;

/**
Expand All @@ -85,6 +87,7 @@ public class ZbsStorageController implements PrimaryStorageControllerSvc, Primar
private AddonInfo addonInfo;
private Config config;

private final Map<MdsInfo, Integer> mdsConsecutivePingFailureCount = new ConcurrentHashMap<>();
public static final String DEPLOY_CLIENT_PATH = "/zbs/primarystorage/client/deploy";
public static final String GET_CAPACITY_PATH = "/zbs/primarystorage/capacity";
public static final String GET_FACTS_PATH = "/zbs/primarystorage/facts";
Expand Down Expand Up @@ -503,6 +506,101 @@ public void handle(ErrorCode errCode, Map data) {
}).start();
}

private void reconnectSingleMds(ZbsPrimaryStorageMdsBase mdsBase, Completion completion) {
mdsBase.connect(new Completion(completion) {
@Override
public void success() {
completion.success();
}

@Override
public void fail(ErrorCode errorCode) {
completion.fail(errorCode);
}
});
}

private void reconnectMdss(List<ZbsPrimaryStorageMdsBase> mdsBases, Completion completion) {
new While<>(mdsBases).all((mdsBase, comp) -> {
reconnectSingleMds(mdsBase, new Completion(comp) {
@Override
public void success() {
comp.done();
}

@Override
public void fail(ErrorCode errorCode) {
comp.addError(errorCode);
comp.done();
}
});
}).run(new WhileDoneCompletion(completion) {
@Override
public void done(ErrorCodeList errorCodeList) {
if (!errorCodeList.getCauses().isEmpty()) {
completion.fail(errorCodeList.getCauses().get(0));
return;
}
completion.success();
}
});
}

private void incrementMdsPingFailureCount(ZbsPrimaryStorageMdsBase mds) {
// initialize the count if absent
mdsConsecutivePingFailureCount.putIfAbsent(mds.getSelf(), 0);
// increment the count only if it is below the threshold
// avoid unbounded growth of the count
if (mdsConsecutivePingFailureCount.get(mds.getSelf()) < MDS_PING_FAIL_CYCLE_THRESHOLD) {
mdsConsecutivePingFailureCount.merge(mds.getSelf(), 1, Integer::sum);
}
}

private void resetMdsPingFailureCount(ZbsPrimaryStorageMdsBase mds) {
mdsConsecutivePingFailureCount.remove(mds.getSelf());
}

private List<ZbsPrimaryStorageMdsBase> getDisconnectedMds(List<ZbsPrimaryStorageMdsBase> mdsBases) {
return mdsBases.stream()
.filter(m -> MdsStatus.Disconnected.equals(m.getSelf().getStatus()))
.collect(Collectors.toList());
}

private boolean allMdssDisconnected(List<ZbsPrimaryStorageMdsBase> mdsBases) {
List<ZbsPrimaryStorageMdsBase> disconnectedMds = getDisconnectedMds(mdsBases);
return disconnectedMds.size() == mdsBases.size();
}

private List<ZbsPrimaryStorageMdsBase> getMdssNeedToReconnect(List<ZbsPrimaryStorageMdsBase> mdsBases) {
List<ZbsPrimaryStorageMdsBase> disconnectedMds = getDisconnectedMds(mdsBases);
boolean allDisconnected = allMdssDisconnected(mdsBases);

if (disconnectedMds.isEmpty()) {
return Collections.emptyList();
}

// If all MDS are disconnected, try to reconnect all of them
if (allDisconnected) {
logger.info("All MDS are disconnected, ignore the ping failure count threshold and all of them need be reconnected");
return disconnectedMds;
}

String disconnectedMdsInfo = disconnectedMds.stream()
.map(m -> String.format("%s (Consecutive Ping Failure: %d)",
m.getSelf().getAddr(),
mdsConsecutivePingFailureCount.computeIfAbsent(m.getSelf(), k -> 0)))
.collect(Collectors.joining(", "));

logger.info(String.format("Some MDS are disconnected for ZBS primary storage[uuid:%s], disconnected MDS: [%s]",
self.getUuid(), disconnectedMdsInfo));

return disconnectedMds.stream()
.filter(m -> mdsConsecutivePingFailureCount
.computeIfAbsent(m.getSelf(), k -> 0) >= MDS_PING_FAIL_CYCLE_THRESHOLD)
.collect(Collectors.toList());

}

@Override
public void ping(ReturnValueCompletion<PingResult> completion) {
reloadDbInfo();
Expand All @@ -517,28 +615,66 @@ public void ping(ReturnValueCompletion<PingResult> completion) {
@Override
public void success() {
m.getSelf().setStatus(MdsStatus.Connected);
resetMdsPingFailureCount(m);
comp.done();
}

@Override
public void fail(ErrorCode errorCode) {
m.getSelf().setStatus(MdsStatus.Disconnected);
incrementMdsPingFailureCount(m);
comp.done();
}
})).run(new WhileDoneCompletion(completion) {
@Override
public void done(ErrorCodeList errorCodeList) {
boolean isConnected = addonInfo.getMdsInfos().stream().anyMatch(mdsInfo -> MdsStatus.Connected.equals(mdsInfo.getStatus()));
if (!isConnected) {
String notConnectedIps = addonInfo.getMdsInfos().stream()
.filter(mdsInfo -> !MdsStatus.Connected.equals(mdsInfo.getStatus()))
.map(MdsInfo::getAddr)
.collect(Collectors.joining(", "));

completion.success(new PingResult(addonInfo, String.format("all MDS are not connected, disconnected MDS addresses: %s", notConnectedIps)));
List<ZbsPrimaryStorageMdsBase> mdsNeedToReconnect = getMdssNeedToReconnect(mds);

if (mdsNeedToReconnect.isEmpty()) {
completion.success(new PingResult(addonInfo));
return;
}
completion.success(new PingResult(addonInfo));

String mdsNeedToReconnectInfo = mdsNeedToReconnect.stream()
.map(m -> String.format("%s (Consecutive Ping Failure: %d)",
m.getSelf().getAddr(),
mdsConsecutivePingFailureCount.get(m.getSelf())))
.collect(Collectors.joining(", "));

logger.warn(String.format("Some MDS need to reconnect for ZBS primary storage[uuid:%s], MDS : %s",
self.getUuid(), mdsNeedToReconnectInfo));

reconnectMdss(mdsNeedToReconnect, new Completion(completion) {
@Override
public void success() {
completion.success(new PingResult(addonInfo));
}

@Override
public void fail(ErrorCode errorCode) {
List<ZbsPrimaryStorageMdsBase> stillDisconnectedMds = getDisconnectedMds(mds);
boolean allDisconnected = allMdssDisconnected(mds);

if (allDisconnected) {
completion.success(new PingResult(addonInfo,
String.format("All MDS are still disconnected after reconnection for ZBS primary storage[uuid:%s]",
self.getUuid())));
return;
}

String stillDisconnectedMdsInfo = stillDisconnectedMds.stream()
.map(m -> String.format("%s (Consecutive Ping Failure: %d)",
m.getSelf().getAddr(),
mdsConsecutivePingFailureCount.get(m.getSelf())))
.collect(Collectors.joining(", "));

logger.warn(String.format("Some MDS are still disconnected after reconnection for ZBS primary storage[uuid:%s], disconnected MDS : %s, error: %s",
self.getUuid(), stillDisconnectedMdsInfo, errorCode));

completion.success(new PingResult(addonInfo));
}
});

}
});
}
Expand Down Expand Up @@ -1118,7 +1254,6 @@ public void fail(ErrorCode errorCode) {
});
}


@Override
public void syncAddonInfo(String addonInfo) {
this.addonInfo = StringUtils.isEmpty(addonInfo) ? new AddonInfo() : JSONObjectUtil.toObject(addonInfo, AddonInfo.class);
Expand Down
Loading