mirror/dble
1
0
Fork 0
mirror of https://github.com/actiontech/dble.git synced 2026-01-05 04:10:32 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: narrow down the scope of syncing dbGroup state

Browse Source
This commit is contained in:
guoaomen
2023-06-26 17:00:37 +08:00
parent c878bf6546
commit 9bca8e313c
3 changed files with 69 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/main/java/com/actiontech/dble/cluster/ClusterHelper.java
View File

@@ -1,6 +1,7 @@
package com.actiontech.dble.cluster;
import com.actiontech.dble.cluster.general.bean.KvBean;
import com.actiontech.dble.services.manager.response.ReloadConfig;
import javax.annotation.Nullable;
import java.util.Collections;
@@ -57,12 +58,12 @@ public final class ClusterHelper {
return ClusterGeneralConfig.getInstance().getClusterSender().getOnlineMap();
}
public static void writeConfToCluster() throws Exception {
public static void writeConfToCluster(ReloadConfig.ReloadResult reloadResult) throws Exception {
ClusterLogic.syncSequenceJsonToCluster();
ClusterLogic.syncDbJsonToCluster();
ClusterLogic.syncShardingJsonToCluster();
ClusterLogic.syncUseJsonToCluster();
ClusterLogic.syncDbGroupStatusToCluster();
ClusterLogic.syncDbGroupStatusToCluster(reloadResult);
}
public static String getPathValue(String path) throws Exception {

28
src/main/java/com/actiontech/dble/cluster/ClusterLogic.java
View File

@@ -288,8 +288,8 @@ public final class ClusterLogic {
return;
}
try {
boolean result = ReloadConfig.reloadByConfig(Integer.parseInt(params), false);
if (!checkLocalResult(result)) {
ReloadConfig.ReloadResult result = ReloadConfig.reloadByConfig(Integer.parseInt(params), false);
if (!checkLocalResult(result.isSuccess())) {
return;
}
} catch (Exception e) {
@@ -957,6 +957,30 @@ public final class ClusterLogic {
LOGGER.info("syncDbGroupStatusToCluster success");
}
public static void syncDbGroupStatusToCluster(ReloadConfig.ReloadResult reloadResult) throws Exception {
LOGGER.info("syncDbGroupStatusToCluster start");
HaConfigManager.getInstance().init(true);
Map<String, String> dbGroupStatusMap = HaConfigManager.getInstance().getSourceJsonList();
Map<String, PhysicalDbGroup> recycleHostMap = reloadResult.getRecycleHostMap();
if (recycleHostMap != null) {
for (Map.Entry<String, PhysicalDbGroup> groupEntry : recycleHostMap.entrySet()) {
String dbGroupName = groupEntry.getKey();
LOGGER.debug("delete dbGroup_status:{}", dbGroupName);
ClusterHelper.cleanKV(ClusterPathUtil.getHaStatusPath(dbGroupName));
}
}
Map<String, PhysicalDbGroup> addOrChangeHostMap = reloadResult.getAddOrChangeHostMap();
if (addOrChangeHostMap != null) {
for (Map.Entry<String, PhysicalDbGroup> groupEntry : addOrChangeHostMap.entrySet()) {
String dbGroupStatusJson = dbGroupStatusMap.get(groupEntry.getKey());
LOGGER.debug("add dbGroup_status:{}---{}", groupEntry.getKey(), dbGroupStatusJson);
ClusterHelper.setKV(ClusterPathUtil.getHaStatusPath(groupEntry.getKey()), dbGroupStatusJson);
}
}
LOGGER.info("syncDbGroupStatusToCluster success");
}
public static void checkBinlogStatusRelease(String crashNode) {
try {
//check the latest bing_log status

55
src/main/java/com/actiontech/dble/services/manager/response/ReloadConfig.java
View File

@@ -130,14 +130,14 @@ public final class ReloadConfig {
lock.writeLock().lock();
try {
//step 2 reload the local config file
boolean reloadResult;
ReloadResult reloadResult;
if (confStatus.getStatus().equals(ConfStatus.Status.MANAGER_INSERT) || confStatus.getStatus().equals(ConfStatus.Status.MANAGER_UPDATE) ||
confStatus.getStatus().equals(ConfStatus.Status.MANAGER_DELETE)) {
reloadResult = reloadByConfig(loadAllMode, true);
} else {
reloadResult = reloadByLocalXml(loadAllMode);
}
if (!reloadResult) {
if (!reloadResult.isSuccess()) {
packetResult.setSuccess(false);
packetResult.setErrorMsg("Reload config failure.The reason is reload interruputed by others,config should be reload");
packetResult.setErrorCode(ErrorCode.ER_RELOAD_INTERRUPUTED);
@@ -147,7 +147,7 @@ public final class ReloadConfig {
ClusterDelayProvider.delayAfterMasterLoad();
//step 3 if the reload with no error ,than write the config file into cluster center remote
ClusterHelper.writeConfToCluster();
ClusterHelper.writeConfToCluster(reloadResult);
ReloadLogHelper.info("reload config: sent config file to cluster center", LOGGER);
//step 4 write the reload flag and self reload result into cluster center,notify the other dble to reload
@@ -199,17 +199,17 @@ public final class ReloadConfig {
packetResult.setErrorCode(ErrorCode.ER_YES);
return;
}
boolean reloadResult;
ReloadResult reloadResult;
if (confStatus.getStatus().equals(ConfStatus.Status.MANAGER_INSERT) || confStatus.getStatus().equals(ConfStatus.Status.MANAGER_UPDATE) ||
confStatus.getStatus().equals(ConfStatus.Status.MANAGER_DELETE)) {
reloadResult = reloadByConfig(loadAllMode, true);
} else {
reloadResult = reloadByLocalXml(loadAllMode);
}
if (reloadResult && returnFlag) {
if (reloadResult.isSuccess() && returnFlag) {
// ok package
return;
} else if (!reloadResult) {
} else if (!reloadResult.isSuccess()) {
packetResult.setSuccess(false);
packetResult.setErrorMsg("Reload config failure.The reason is reload interruputed by others,metadata should be reload");
packetResult.setErrorCode(ErrorCode.ER_RELOAD_INTERRUPUTED);
@@ -228,11 +228,11 @@ public final class ReloadConfig {
}
@Deprecated
public static boolean reloadByLocalXml(final int loadAllMode) throws Exception {
public static ReloadResult reloadByLocalXml(final int loadAllMode) throws Exception {
return reload(loadAllMode, null, null, null, null);
}
public static boolean reloadByConfig(final int loadAllMode, boolean isWriteToLocal) throws Exception {
public static ReloadResult reloadByConfig(final int loadAllMode, boolean isWriteToLocal) throws Exception {
String userConfig = DbleTempConfig.getInstance().getUserConfig();
userConfig = StringUtil.isBlank(userConfig) ? DbleServer.getInstance().getConfig().getUserConfig() : userConfig;
String dbConfig = DbleTempConfig.getInstance().getDbConfig();
@@ -241,14 +241,14 @@ public final class ReloadConfig {
shardingConfig = StringUtil.isBlank(shardingConfig) ? DbleServer.getInstance().getConfig().getShardingConfig() : shardingConfig;
String sequenceConfig = DbleTempConfig.getInstance().getSequenceConfig();
sequenceConfig = StringUtil.isBlank(sequenceConfig) ? DbleServer.getInstance().getConfig().getSequenceConfig() : sequenceConfig;
boolean reloadResult = reload(loadAllMode, userConfig, dbConfig, shardingConfig, sequenceConfig);
ReloadResult reloadResult = reload(loadAllMode, userConfig, dbConfig, shardingConfig, sequenceConfig);
DbleTempConfig.getInstance().clean();
//sync json to local
DbleServer.getInstance().getConfig().syncJsonToLocal(isWriteToLocal);
return reloadResult;
}
private static boolean reload(final int loadAllMode, String userConfig, String dbConfig, String shardingConfig, String sequenceConfig) throws Exception {
private static ReloadResult reload(final int loadAllMode, String userConfig, String dbConfig, String shardingConfig, String sequenceConfig) throws Exception {
TraceManager.TraceObject traceObject = TraceManager.threadTrace("self-reload");
try {
/*
@@ -295,7 +295,7 @@ public final class ReloadConfig {
}
}
private static boolean intelligentReloadAll(int loadAllMode, ConfigInitializer loader) throws Exception {
private static ReloadResult intelligentReloadAll(int loadAllMode, ConfigInitializer loader) throws Exception {
TraceManager.TraceObject traceObject = TraceManager.threadTrace("self-intelligent-reload");
try {
/* 2.1.1 get diff of dbGroups */
@@ -367,7 +367,7 @@ public final class ReloadConfig {
if (!loader.isFullyConfigured()) {
recycleServerConnections();
}
return result;
return new ReloadResult(result, addOrChangeHosts, recycleHosts);
} catch (Exception e) {
initFailed(newDbGroups);
throw e;
@@ -402,7 +402,7 @@ public final class ReloadConfig {
}
}
private static boolean forceReloadAll(final int loadAllMode, ConfigInitializer loader) throws Exception {
private static ReloadResult forceReloadAll(final int loadAllMode, ConfigInitializer loader) throws Exception {
TraceManager.TraceObject traceObject = TraceManager.threadTrace("self-force-reload");
try {
ServerConfig config = DbleServer.getInstance().getConfig();
@@ -451,7 +451,8 @@ public final class ReloadConfig {
ReloadLogHelper.info("reload config: apply new config start", LOGGER);
boolean result;
try {
result = config.reload(newUsers, newSchemas, newShardingNodes, newDbGroups, config.getDbGroups(), newErRelations,
Map<String, PhysicalDbGroup> oldDbGroupMap = config.getDbGroups();
result = config.reload(newUsers, newSchemas, newShardingNodes, newDbGroups, oldDbGroupMap, newErRelations,
newSystemVariables, loader.isFullyConfigured(), loadAllMode, newBlacklistConfig, newFunctions,
loader.getUserConfig(), loader.getSequenceConfig(), loader.getShardingConfig(), loader.getDbConfig());
CronScheduler.getInstance().init(config.getSchemas());
@@ -463,7 +464,7 @@ public final class ReloadConfig {
if (!loader.isFullyConfigured()) {
recycleServerConnections();
}
return result;
return new ReloadResult(result, newDbGroups, oldDbGroupMap);
} catch (Exception e) {
initFailed(newDbGroups);
throw e;
@@ -607,4 +608,28 @@ public final class ReloadConfig {
service.writeErrMessage(errorCode, errorMsg);
}
}
public static class ReloadResult {
private final boolean success;
private final Map<String, PhysicalDbGroup> addOrChangeHostMap;
private final Map<String, PhysicalDbGroup> recycleHostMap;
public ReloadResult(boolean success, Map<String, PhysicalDbGroup> addOrChangeHostMap, Map<String, PhysicalDbGroup> recycleHostMap) {
this.success = success;
this.addOrChangeHostMap = addOrChangeHostMap;
this.recycleHostMap = recycleHostMap;
}
public boolean isSuccess() {
return success;
}
public Map<String, PhysicalDbGroup> getAddOrChangeHostMap() {
return addOrChangeHostMap;
}
public Map<String, PhysicalDbGroup> getRecycleHostMap() {
return recycleHostMap;
}
}
}