From de96142109fa47d44fd87d67638b0dd9f02d6343 Mon Sep 17 00:00:00 2001 From: "haidong.pang" Date: Tue, 13 Jan 2026 06:50:19 +0000 Subject: [PATCH 1/2] [zbs]: Try to reconnect after zbs ping failure Resolves: ZSTAC-79447 Change-Id: I757275676874656b727a6172646961687a726870 --- .../storage/zbs/ZbsPrimaryStorageMdsBase.java | 6 + .../storage/zbs/ZbsStorageController.java | 88 +++++++++++-- .../addon/zbs/ZbsPrimaryStorageCase.groovy | 123 +++++++++++++++--- 3 files changed, 189 insertions(+), 28 deletions(-) diff --git a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java index 81309d43642..292beca133a 100644 --- a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java +++ b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java @@ -206,6 +206,12 @@ public void run(FlowTrigger trigger, Map data) { restf.asyncJsonPost(ZbsAgentUrl.primaryStorageUrl(getSelf().getAddr(), SYNC_METADATA_PATH), cmd, new JsonAsyncRESTCallback(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(); } diff --git a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java index 512712a3cf3..49e5924b5ab 100644 --- a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java +++ b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java @@ -57,6 +57,7 @@ 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; @@ -503,6 +504,46 @@ 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 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(); + } + }); + } + @Override public void ping(ReturnValueCompletion completion) { reloadDbInfo(); @@ -528,17 +569,46 @@ public void fail(ErrorCode errorCode) { })).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 disconnectedMds = mds.stream() + .filter(m -> MdsStatus.Disconnected.equals(m.getSelf().getStatus())) + .collect(Collectors.toList()); + + if (disconnectedMds.isEmpty()) { + completion.success(new PingResult(addonInfo)); return; } - completion.success(new PingResult(addonInfo)); + + final boolean allDisconnected = disconnectedMds.size() == mds.size(); + String notConnectedIps = disconnectedMds.stream() + .map(ZbsPrimaryStorageMdsBase::getSelf) + .map(MdsInfo::getAddr) + .collect(Collectors.joining(", ")); + + // it should be covered in primary storage reconnect phase + if (allDisconnected) { + completion.success(new PingResult(addonInfo, String.format("All MDS are not connected, disconnected MDS addresses: %s", notConnectedIps))); + return; + } + + // Try to reconnect disconnected MDS, the reconnection results are recorded in addonInfo + // and will not cause the ping operation to fail. + logger.warn(String.format("Some MDS are not connected for ZBS primary storage[uuid:%s], disconnected MDS addresses: %s, try to reconnect them", + self.getUuid(), notConnectedIps)); + + reconnectMdss(disconnectedMds, new Completion(completion) { + @Override + public void success() { + completion.success(new PingResult(addonInfo)); + } + + @Override + public void fail(ErrorCode errorCode) { + logger.warn(String.format("Some MDS are still not connected after reconnection for ZBS primary storage[uuid:%s], error: %s", + self.getUuid(), errorCode)); + completion.success(new PingResult(addonInfo)); + } + }); + } }); } diff --git a/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy b/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy index f2394a852ba..f6954bf32c2 100644 --- a/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy +++ b/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy @@ -1,6 +1,7 @@ package org.zstack.test.integration.storage.primary.addon.zbs import org.springframework.http.HttpEntity +import org.zstack.cbd.AddonInfo import org.zstack.core.cloudbus.EventCallback import org.zstack.core.cloudbus.EventFacade import org.zstack.core.db.Q @@ -13,6 +14,7 @@ import org.zstack.header.storage.primary.PrimaryStorageStatus import org.zstack.header.storage.primary.PrimaryStorageVO import org.zstack.cbd.MdsUri import org.zstack.kvm.KVMAgentCommands +import org.zstack.cbd.MdsStatus import org.zstack.sdk.* import org.zstack.storage.addon.primary.ExternalPrimaryStorageCanonicalEvent import org.zstack.storage.primary.PrimaryStorageGlobalConfig @@ -28,6 +30,8 @@ import org.zstack.testlib.SubCase import org.zstack.utils.data.SizeUnit import org.zstack.utils.gson.JSONObjectUtil +import java.util.concurrent.atomic.AtomicInteger + /** * @author Xingwei Yu * @date 2024/4/19 10:09 @@ -159,7 +163,6 @@ class ZbsPrimaryStorageCase extends SubCase { diskOffering = env.inventoryByName("diskOffering") as DiskOfferingInventory kvm = env.inventoryByName("kvm-1") as KVMHostInventory evtf = bean(EventFacade.class) - testDefaultConfig() testUpdateExternalPrimaryStorage() testLifecycle() @@ -170,6 +173,7 @@ class ZbsPrimaryStorageCase extends SubCase { testAddExternalPrimaryStorageWithMalformedJsonRejectedByInterceptor() testDataVolumeNegativeScenario() testDecodeMdsUriWithSpecialPassword() + testMdsReconnectAfterMaximumPingFailures() } } @@ -345,55 +349,85 @@ class ZbsPrimaryStorageCase extends SubCase { env.afterSimulator(ZbsPrimaryStorageMdsBase.PING_PATH) { rsp, HttpEntity e -> def cmd = JSONObjectUtil.toObject(e.body, ZbsPrimaryStorageMdsBase.PingCmd.class) - ZbsPrimaryStorageMdsBase.PingRsp pingRsp = new ZbsPrimaryStorageMdsBase.PingRsp() if (cmd.addr.equals("127.0.1.1")) { - pingRsp.success = false - pingRsp.error = "on purpose" + rsp.success = false + rsp.error = "on purpose" + } + + return rsp + } + + env.afterSimulator(ZbsPrimaryStorageMdsBase.SYNC_METADATA_PATH) { ZbsPrimaryStorageMdsBase.SyncMetadataRsp rsp, HttpEntity e -> + ZbsPrimaryStorageMdsBase.PingCmd cmd = JSONObjectUtil.toObject(e.body, ZbsPrimaryStorageMdsBase.PingCmd.class) + if (cmd.getAddr().equals("127.0.1.1")) { + rsp.setSuccess(false) + rsp.setError("on purpose") } - return pingRsp + return rsp } - sleep(2000) + retryInSecs { + assert count == 1 + } addonInfo = Q.New(ExternalPrimaryStorageVO.class).select(ExternalPrimaryStorageVO_.addonInfo).eq(ExternalPrimaryStorageVO_.uuid, ps.uuid).findValue() assert addonInfo == "{\"clusterInfo\":{\"uuid\":\"123456789\",\"version\":\"1.6.1-for-test\"},\"mdsInfos\":[{\"username\":\"root\",\"password\":\"password\",\"port\":22,\"addr\":\"127.0.1.1\",\"externalAddr\":\"127.0.0.1\",\"status\":\"Disconnected\"},{\"username\":\"root\",\"password\":\"password\",\"port\":22,\"addr\":\"127.0.1.2\",\"externalAddr\":\"127.0.0.1\",\"status\":\"Connected\"},{\"username\":\"root\",\"password\":\"password\",\"port\":22,\"addr\":\"127.0.1.3\",\"externalAddr\":\"127.0.0.1\",\"status\":\"Connected\"}],\"logicalPoolInfos\":[{\"physicalPoolID\":1,\"redundanceAndPlaceMentPolicy\":{\"copysetNum\":300,\"replicaNum\":3,\"zoneNum\":3},\"logicalPoolID\":1,\"usedSize\":322961408,\"quota\":0,\"createTime\":1735875794,\"type\":0,\"rawWalUsedSize\":0,\"allocateStatus\":0,\"rawUsedSize\":968884224,\"physicalPoolName\":\"pool1\",\"capacity\":579933831168,\"logicalPoolName\":\"lpool1\",\"userPolicy\":\"eyJwb2xpY3kiIDogMX0=\",\"allocatedSize\":3221225472}]}" assert Q.New(ExternalPrimaryStorageVO.class).select(ExternalPrimaryStorageVO_.status).eq(ExternalPrimaryStorageVO_.uuid, ps.uuid).findValue() == PrimaryStorageStatus.Connected - assert count == 1 assert data.uuid == ps.uuid env.afterSimulator(ZbsPrimaryStorageMdsBase.PING_PATH) { rsp, HttpEntity e -> def cmd = JSONObjectUtil.toObject(e.body, ZbsPrimaryStorageMdsBase.PingCmd.class) - ZbsPrimaryStorageMdsBase.PingRsp pingRsp = new ZbsPrimaryStorageMdsBase.PingRsp() if (cmd.addr.equals("127.0.1.1")) { - pingRsp.success = false - pingRsp.error = "on purpose" + rsp.success = false + rsp.error = "on purpose" } else if (cmd.addr.equals("127.0.1.2")) { - pingRsp.success = false - pingRsp.error = "on purpose" + rsp.success = false + rsp.error = "on purpose" } else if (cmd.addr.equals("127.0.1.3")) { - pingRsp.success = false - pingRsp.error = "on purpose" + rsp.success = false + rsp.error = "on purpose" } - return pingRsp + return rsp } - sleep(2500) + env.afterSimulator(ZbsPrimaryStorageMdsBase.SYNC_METADATA_PATH) { ZbsPrimaryStorageMdsBase.SyncMetadataRsp rsp, HttpEntity e -> + ZbsPrimaryStorageMdsBase.PingCmd cmd = JSONObjectUtil.toObject(e.body, ZbsPrimaryStorageMdsBase.PingCmd.class) + if (cmd.getAddr().equals("127.0.1.1")) { + rsp.setSuccess(false) + rsp.setError("on purpose") + } else if (cmd.getAddr().equals("127.0.1.2")) { + rsp.setSuccess(false) + rsp.setError("on purpose") + } else if (cmd.getAddr().equals("127.0.1.3")) { + rsp.setSuccess(false) + rsp.setError("on purpose") + } + + return rsp + } + + retryInSecs { + assert count == 2 + } addonInfo = Q.New(ExternalPrimaryStorageVO.class).select(ExternalPrimaryStorageVO_.addonInfo).eq(ExternalPrimaryStorageVO_.uuid, ps.uuid).findValue() assert addonInfo == "{\"clusterInfo\":{\"uuid\":\"123456789\",\"version\":\"1.6.1-for-test\"},\"mdsInfos\":[{\"username\":\"root\",\"password\":\"password\",\"port\":22,\"addr\":\"127.0.1.1\",\"externalAddr\":\"127.0.0.1\",\"status\":\"Disconnected\"},{\"username\":\"root\",\"password\":\"password\",\"port\":22,\"addr\":\"127.0.1.2\",\"externalAddr\":\"127.0.0.1\",\"status\":\"Disconnected\"},{\"username\":\"root\",\"password\":\"password\",\"port\":22,\"addr\":\"127.0.1.3\",\"externalAddr\":\"127.0.0.1\",\"status\":\"Disconnected\"}],\"logicalPoolInfos\":[{\"physicalPoolID\":1,\"redundanceAndPlaceMentPolicy\":{\"copysetNum\":300,\"replicaNum\":3,\"zoneNum\":3},\"logicalPoolID\":1,\"usedSize\":322961408,\"quota\":0,\"createTime\":1735875794,\"type\":0,\"rawWalUsedSize\":0,\"allocateStatus\":0,\"rawUsedSize\":968884224,\"physicalPoolName\":\"pool1\",\"capacity\":579933831168,\"logicalPoolName\":\"lpool1\",\"userPolicy\":\"eyJwb2xpY3kiIDogMX0=\",\"allocatedSize\":3221225472}]}" - assert Q.New(ExternalPrimaryStorageVO.class).select(ExternalPrimaryStorageVO_.status).eq(ExternalPrimaryStorageVO_.uuid, ps.uuid).findValue() == PrimaryStorageStatus.Disconnected - assert count == 2 + retryInSecs { + assert Q.New(ExternalPrimaryStorageVO.class).select(ExternalPrimaryStorageVO_.status).eq(ExternalPrimaryStorageVO_.uuid, ps.uuid).findValue() == PrimaryStorageStatus.Disconnected + } assert data.uuid == ps.uuid env.cleanAfterSimulatorHandlers() - sleep(2000) + retryInSecs { + assert count >= 3 + } Q.New(ExternalPrimaryStorageVO.class).select(ExternalPrimaryStorageVO_.status).eq(ExternalPrimaryStorageVO_.uuid, ps.uuid).findValue() == PrimaryStorageStatus.Connected @@ -563,6 +597,57 @@ class ZbsPrimaryStorageCase extends SubCase { assert uri.password == specialPassword } + void testMdsReconnectAfterMaximumPingFailures() { + env.cleanSimulatorAndMessageHandlers() + Integer originalPingInterval = PrimaryStorageGlobalConfig.PING_INTERVAL.value().toInteger() + PrimaryStorageGlobalConfig.PING_INTERVAL.updateValue(1) + final Integer MAX_PING_CNT = ZbsConstants.PRIMARY_STORAGE_MDS_MAXIMUM_PING_FAILURE + AtomicInteger pingFailureCount = new AtomicInteger(0) + Boolean reconnectTriggered = false + + env.simulator(ZbsPrimaryStorageMdsBase.PING_PATH) { HttpEntity e, EnvSpec spec -> + ZbsPrimaryStorageMdsBase.PingCmd cmd = JSONObjectUtil.toObject(e.body, ZbsPrimaryStorageMdsBase.PingCmd.class) + ZbsPrimaryStorageMdsBase.PingRsp rsp = new ZbsPrimaryStorageMdsBase.PingRsp() + if (cmd.getAddr().equals("127.0.1.1") && !reconnectTriggered) { + rsp.success = false + rsp.error = "Mds ping failed on purpose" + pingFailureCount.incrementAndGet() + return rsp + } + return rsp + + } + + env.simulator(ZbsPrimaryStorageMdsBase.SYNC_METADATA_PATH) { HttpEntity e, EnvSpec spec -> + ZbsPrimaryStorageMdsBase.SyncMetadataCmd cmd = JSONObjectUtil.toObject(e.body, ZbsPrimaryStorageMdsBase.SyncMetadataCmd.class) + ZbsPrimaryStorageMdsBase.SyncMetadataRsp rsp = new ZbsPrimaryStorageMdsBase.SyncMetadataRsp() + rsp.setExternalAddr(cmd.getAddr()) + if (cmd.getAddr().equals("127.0.1.1")) { + assert pingFailureCount.intValue().equals(MAX_PING_CNT) + reconnectTriggered = true + } + return rsp + } + + sleep((MAX_PING_CNT + 1) * 1000 + 500) + retryInSecs { + assert reconnectTriggered + } + retryInSecs { + AddonInfo addonInfo = JSONObjectUtil.toObject( + Q.New(ExternalPrimaryStorageVO.class) + .select(ExternalPrimaryStorageVO_.addonInfo) + .eq(ExternalPrimaryStorageVO_.uuid, ps.uuid) + .findValue() + .toString(), + AddonInfo.class + ) + assert MdsStatus.Connected.equals(addonInfo.getMdsInfos().find { it.addr.equals("127.0.1.1") }.getStatus()) + } + + PrimaryStorageGlobalConfig.PING_INTERVAL.updateValue(originalPingInterval) + env.cleanAfterSimulatorHandlers() + } void deleteVolume(String volUuid) { deleteDataVolume { From ca7439e482bc08826a0ed6a47940d5df067832bd Mon Sep 17 00:00:00 2001 From: "haidong.pang" Date: Wed, 28 Jan 2026 05:33:30 +0000 Subject: [PATCH 2/2] [zbs]: Try to reconnect after all mds disconnected - After the ZBS primary storage agent on the MDS restarts, it may fail to start up independently due to lost firewall rules or metadata. - Redeploy by the MN is required. Resolves: ZSTAC-79447 Change-Id: I726b7467627269627779696d6e72756f75706e76 --- .../org/zstack/storage/zbs/ZbsConstants.java | 3 +- .../storage/zbs/ZbsPrimaryStorageMdsBase.java | 2 +- .../storage/zbs/ZbsStorageController.java | 109 ++++++++++++++---- .../addon/zbs/ZbsPrimaryStorageCase.groovy | 8 +- 4 files changed, 94 insertions(+), 28 deletions(-) diff --git a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsConstants.java b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsConstants.java index 8f547cfeb29..987138c3360 100644 --- a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsConstants.java +++ b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsConstants.java @@ -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"; diff --git a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java index 292beca133a..986ad438b42 100644 --- a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java +++ b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsPrimaryStorageMdsBase.java @@ -313,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 stepCount = new ArrayList<>(); for (int i = 1; i <= MAX_PING_CNT; i++) { stepCount.add(i); diff --git a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java index 49e5924b5ab..4db9baf0d1d 100644 --- a/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java +++ b/plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java @@ -62,6 +62,7 @@ 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.*; /** @@ -86,6 +87,7 @@ public class ZbsStorageController implements PrimaryStorageControllerSvc, Primar private AddonInfo addonInfo; private Config config; + private final Map 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"; @@ -544,6 +546,61 @@ public void done(ErrorCodeList errorCodeList) { }); } + 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 getDisconnectedMds(List mdsBases) { + return mdsBases.stream() + .filter(m -> MdsStatus.Disconnected.equals(m.getSelf().getStatus())) + .collect(Collectors.toList()); + } + + private boolean allMdssDisconnected(List mdsBases) { + List disconnectedMds = getDisconnectedMds(mdsBases); + return disconnectedMds.size() == mdsBases.size(); + } + + private List getMdssNeedToReconnect(List mdsBases) { + List 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 completion) { reloadDbInfo(); @@ -558,44 +615,36 @@ public void ping(ReturnValueCompletion 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) { - List disconnectedMds = mds.stream() - .filter(m -> MdsStatus.Disconnected.equals(m.getSelf().getStatus())) - .collect(Collectors.toList()); + List mdsNeedToReconnect = getMdssNeedToReconnect(mds); - if (disconnectedMds.isEmpty()) { + if (mdsNeedToReconnect.isEmpty()) { completion.success(new PingResult(addonInfo)); return; } - final boolean allDisconnected = disconnectedMds.size() == mds.size(); - String notConnectedIps = disconnectedMds.stream() - .map(ZbsPrimaryStorageMdsBase::getSelf) - .map(MdsInfo::getAddr) + String mdsNeedToReconnectInfo = mdsNeedToReconnect.stream() + .map(m -> String.format("%s (Consecutive Ping Failure: %d)", + m.getSelf().getAddr(), + mdsConsecutivePingFailureCount.get(m.getSelf()))) .collect(Collectors.joining(", ")); - // it should be covered in primary storage reconnect phase - if (allDisconnected) { - completion.success(new PingResult(addonInfo, String.format("All MDS are not connected, disconnected MDS addresses: %s", notConnectedIps))); - return; - } - - // Try to reconnect disconnected MDS, the reconnection results are recorded in addonInfo - // and will not cause the ping operation to fail. - logger.warn(String.format("Some MDS are not connected for ZBS primary storage[uuid:%s], disconnected MDS addresses: %s, try to reconnect them", - self.getUuid(), notConnectedIps)); + logger.warn(String.format("Some MDS need to reconnect for ZBS primary storage[uuid:%s], MDS : %s", + self.getUuid(), mdsNeedToReconnectInfo)); - reconnectMdss(disconnectedMds, new Completion(completion) { + reconnectMdss(mdsNeedToReconnect, new Completion(completion) { @Override public void success() { completion.success(new PingResult(addonInfo)); @@ -603,8 +652,25 @@ public void success() { @Override public void fail(ErrorCode errorCode) { - logger.warn(String.format("Some MDS are still not connected after reconnection for ZBS primary storage[uuid:%s], error: %s", - self.getUuid(), errorCode)); + List 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)); } }); @@ -1188,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); diff --git a/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy b/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy index f6954bf32c2..e337588caf5 100644 --- a/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy +++ b/test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy @@ -601,7 +601,8 @@ class ZbsPrimaryStorageCase extends SubCase { env.cleanSimulatorAndMessageHandlers() Integer originalPingInterval = PrimaryStorageGlobalConfig.PING_INTERVAL.value().toInteger() PrimaryStorageGlobalConfig.PING_INTERVAL.updateValue(1) - final Integer MAX_PING_CNT = ZbsConstants.PRIMARY_STORAGE_MDS_MAXIMUM_PING_FAILURE + final Integer MDS_PING_RETRY_PER_CYCLE = ZbsConstants.MDS_PING_RETRY_PER_CYCLE + final Integer MDS_PING_FAIL_CYCLE_THRESHOLD = ZbsConstants.MDS_PING_FAIL_CYCLE_THRESHOLD AtomicInteger pingFailureCount = new AtomicInteger(0) Boolean reconnectTriggered = false @@ -622,14 +623,13 @@ class ZbsPrimaryStorageCase extends SubCase { ZbsPrimaryStorageMdsBase.SyncMetadataCmd cmd = JSONObjectUtil.toObject(e.body, ZbsPrimaryStorageMdsBase.SyncMetadataCmd.class) ZbsPrimaryStorageMdsBase.SyncMetadataRsp rsp = new ZbsPrimaryStorageMdsBase.SyncMetadataRsp() rsp.setExternalAddr(cmd.getAddr()) - if (cmd.getAddr().equals("127.0.1.1")) { - assert pingFailureCount.intValue().equals(MAX_PING_CNT) + if (cmd.getAddr().equals("127.0.1.1") && pingFailureCount.intValue().equals(MDS_PING_RETRY_PER_CYCLE * MDS_PING_FAIL_CYCLE_THRESHOLD)) { reconnectTriggered = true } return rsp } - sleep((MAX_PING_CNT + 1) * 1000 + 500) + sleep((MDS_PING_RETRY_PER_CYCLE * MDS_PING_FAIL_CYCLE_THRESHOLD + 1) * 1000 + 500) retryInSecs { assert reconnectTriggered }