Merge branch 'bugfix/migrate-and-delaycheck-and-crossdcelector' into 'master'

vzc祝辰 · vzc祝辰 · commit b5f4010cde11 · 2020-08-20T14:07:34.000+08:00
bugfix for migration pre-check, cross dc leader elector and redis delay get api

See merge request framework/x-pipe!23
diff --git a/core/src/main/java/com/ctrip/xpipe/migration/DefaultOuterClientService.java b/core/src/main/java/com/ctrip/xpipe/migration/DefaultOuterClientService.java
@@ -75,6 +75,7 @@ public MigrationPublishResult doMigrationPublish(String clusterName, String shar
 	@Override
 	public ClusterInfo getClusterInfo(String clusterName) {
 		ClusterInfo clusterInfo = new ClusterInfo();
+		clusterInfo.setName(clusterName);
 		clusterInfo.setGroups(Lists.newArrayList(new GroupInfo()));
 		return clusterInfo;
 	}
diff --git a/redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/console/impl/ConsoleServiceManager.java b/redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/console/impl/ConsoleServiceManager.java
@@ -1,6 +1,7 @@
 package com.ctrip.xpipe.redis.console.console.impl;
 
 import com.ctrip.xpipe.endpoint.HostPort;
+import com.ctrip.xpipe.exception.XpipeRuntimeException;
 import com.ctrip.xpipe.redis.console.config.ConsoleConfig;
 import com.ctrip.xpipe.redis.console.console.ConsoleService;
 import com.ctrip.xpipe.redis.console.healthcheck.actions.interaction.HEALTH_STATE;
@@ -93,15 +94,20 @@ public UnhealthyInfoModel getAllUnhealthyInstanceFromParallelService() {
         return parallelService.getAllUnhealthyInstance();
     }
 
-    private ConsoleService getServiceByDc(String activeIdc) {
-        String dcId = activeIdc.toUpperCase();
-        ConsoleService service = services.get(dcId);
+    private ConsoleService getServiceByDc(String dcId) {
+        String upperCaseDcId = dcId.toUpperCase();
+        ConsoleService service = services.get(upperCaseDcId);
         if (service == null) {
             synchronized (this) {
-                service = services.get(dcId);
+                service = services.get(upperCaseDcId);
                 if (service == null) {
-                    service = new DefaultConsoleService(consoleConfig.getConsoleDomains().get(dcId));
-                    services.put(activeIdc, service);
+                    Optional<String> optionalKey = consoleConfig.getConsoleDomains().keySet().stream().filter(dcId::equalsIgnoreCase).findFirst();
+                    if (!optionalKey.isPresent()) {
+                        throw new XpipeRuntimeException("unknown dc id " + dcId);
+                    }
+
+                    service = new DefaultConsoleService(consoleConfig.getConsoleDomains().get(optionalKey.get()));
+                    services.put(upperCaseDcId, service);
                 }
             }
         }
diff --git a/redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/election/CrossDcLeaderElectionAction.java b/redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/election/CrossDcLeaderElectionAction.java
@@ -1,6 +1,7 @@
 package com.ctrip.xpipe.redis.console.election;
 
 import com.ctrip.xpipe.api.foundation.FoundationService;
+import com.ctrip.xpipe.cluster.ClusterType;
 import com.ctrip.xpipe.redis.console.config.ConsoleConfig;
 import com.ctrip.xpipe.redis.console.dao.ConfigDao;
 import com.ctrip.xpipe.redis.console.model.ConfigModel;
@@ -146,14 +147,20 @@ private long calculateElectDelay() {
 
     private float calculateActiveClusterRatio() {
         XpipeMeta xpipeMeta = metaCache.getXpipeMeta();
+        if (null == xpipeMeta) return 1L;
+
         long count;
         long totalCluster = 0;
         long activeClusterCount = 0;
 
         for (DcMeta dcMeta : xpipeMeta.getDcs().values()) {
             count = 0;
             for (ClusterMeta clusterMeta : dcMeta.getClusters().values()) {
-                if (!clusterMeta.getActiveDc().equals(dcMeta.getId())) {
+                ClusterType clusterType = ClusterType.lookup(clusterMeta.getType());
+                if (clusterType.supportSingleActiveDC() && !clusterMeta.getActiveDc().equals(dcMeta.getId())) {
+                    continue;
+                }
+                if (clusterType.supportMultiActiveDC() && !dcMeta.getId().equalsIgnoreCase(dataCenter)) {
                     continue;
                 }
 
diff --git a/redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/healthcheck/actions/delay/CrossMasterDelayService.java b/redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/healthcheck/actions/delay/CrossMasterDelayService.java
@@ -19,6 +19,7 @@
 import org.springframework.stereotype.Component;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
@@ -67,13 +68,18 @@ public void stopWatch(HealthCheckAction action) {
 
     public Map<String, Pair<HostPort, Long>> getPeerMasterDelayFromCurrentDc(String clusterId, String shardId) {
         Map<String, Pair<HostPort, Long>> peerMasterDelays = crossMasterDelays.get(new DcClusterShard(currentDcId, clusterId, shardId));
-        if (null != peerMasterDelays) {
-            peerMasterDelays.forEach((key, delay) -> {
-                if (delay.getValue() > 0) delay.setValue(TimeUnit.NANOSECONDS.toMillis(delay.getValue()));
-            });
-        }
+        if (null == peerMasterDelays) return null;
+
+        Map<String, Pair<HostPort, Long>> result = new HashMap<>(peerMasterDelays.size());
+        peerMasterDelays.forEach((targetDc, delay) -> {
+            if (delay.getValue() > 0) {
+                result.put(targetDc, Pair.of(delay.getKey(), TimeUnit.NANOSECONDS.toMillis(delay.getValue())));
+            } else {
+                result.put(targetDc, Pair.of(delay.getKey(), delay.getValue()));
+            }
+        });
 
-        return peerMasterDelays;
+        return result;
     }
 
     public Map<String, Pair<HostPort, Long>> getPeerMasterDelayFromSourceDc(String sourceDcId, String clusterId, String shardId) {
diff --git a/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/AllTests.java b/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/AllTests.java
@@ -6,6 +6,7 @@
 import com.ctrip.xpipe.redis.console.cluster.ConsoleCrossDcServerTest;
 import com.ctrip.xpipe.redis.console.config.impl.DefaultConsoleConfigTest;
 import com.ctrip.xpipe.redis.console.config.impl.DefaultConsoleDbConfigTest;
+import com.ctrip.xpipe.redis.console.console.impl.ConsoleServiceManagerTest;
 import com.ctrip.xpipe.redis.console.controller.api.HealthControllerTest;
 import com.ctrip.xpipe.redis.console.controller.api.data.*;
 import com.ctrip.xpipe.redis.console.controller.api.data.meta.CheckPrepareRequestTest;
@@ -225,7 +226,8 @@
         TombstoneSizeCheckActionTest.class,
         TombstoneSizeMetricListenerTest.class,
         DefaultRedisMasterActionListenerTest.class,
-        DefaultMetaCacheTest.class
+        DefaultMetaCacheTest.class,
+        ConsoleServiceManagerTest.class
 })
 public class AllTests {
 
diff --git a/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/console/impl/ConsoleServiceManagerTest.java b/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/console/impl/ConsoleServiceManagerTest.java
@@ -1,13 +1,19 @@
 package com.ctrip.xpipe.redis.console.console.impl;
 
+import com.ctrip.xpipe.exception.ExceptionUtils;
+import com.ctrip.xpipe.exception.XpipeRuntimeException;
 import com.ctrip.xpipe.redis.console.AbstractConsoleTest;
 import com.ctrip.xpipe.redis.console.config.ConsoleConfig;
+import com.ctrip.xpipe.redis.console.console.ConsoleService;
 import com.ctrip.xpipe.redis.console.healthcheck.actions.interaction.HEALTH_STATE;
+import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 
+import java.lang.reflect.Method;
+import java.util.HashMap;
 import java.util.List;
 
 import static org.mockito.Mockito.when;
@@ -50,4 +56,29 @@ public void testMulti(){
         logger.info("{}", health_states);
 
     }
+
+    @Test
+    public void testGetServiceByDcIgnoreCase() throws Exception {
+        when(consoleConfig.getConsoleDomains()).thenReturn(new HashMap<String, String>(){{
+            put("jq", "http://127.0.0.1:8080");
+            put("OY", "http://127.0.0.1:8081");
+        }});
+
+        ConsoleServiceManager manager = new ConsoleServiceManager(consoleConfig);
+        Method method = ConsoleServiceManager.class.getDeclaredMethod("getServiceByDc", String.class);
+        method.setAccessible(true);
+
+        ConsoleService consoleService = (ConsoleService) method.invoke(manager, "jq");
+        Assert.assertNotNull(consoleService);
+
+        consoleService = (ConsoleService) method.invoke(manager, "oy");
+        Assert.assertNotNull(consoleService);
+
+        try {
+            method.invoke(manager, "rb");
+            Assert.fail();
+        } catch (Exception e) {
+            logger.info("[testGetServiceByDc] get unknown dc", e);
+        }
+    }
 }
diff --git a/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/election/CrossDcLeaderElectionActionTest.java b/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/election/CrossDcLeaderElectionActionTest.java
@@ -4,6 +4,7 @@
 import com.ctrip.xpipe.AbstractTest;
 import com.ctrip.xpipe.api.observer.Observable;
 import com.ctrip.xpipe.api.observer.Observer;
+import com.ctrip.xpipe.cluster.ClusterType;
 import com.ctrip.xpipe.redis.console.config.ConsoleConfig;
 import com.ctrip.xpipe.redis.console.dao.ConfigDao;
 import com.ctrip.xpipe.redis.console.exception.DalInsertException;
@@ -153,18 +154,21 @@ private XpipeMeta mockXpipeMeta() {
             xpipeMeta.addDc(dcMeta);
         });
 
-        ClusterMeta oyClusterMeta = new ClusterMeta();
-        ClusterMeta oyClusterMeta2 = new ClusterMeta();
-        ClusterMeta rbClusterMeta = new ClusterMeta();
+        ClusterMeta oyClusterMeta = new ClusterMeta("oyCluster");
+        ClusterMeta oyClusterMeta2 = new ClusterMeta("oyCluster2");
+        ClusterMeta rbClusterMeta = new ClusterMeta("rbCluster");
+        ClusterMeta biClusterMeta = new ClusterMeta("bi-cluster");
         oyClusterMeta.setActiveDc("oy");
         oyClusterMeta2.setActiveDc("oy");
         rbClusterMeta.setActiveDc("rb");
-        oyClusterMeta.setId("oyCluster");
-        oyClusterMeta2.setId("oyCluster2");
-        rbClusterMeta.setId("rbCluster");
-
-        xpipeMeta.getDcs().get("oy").addCluster(oyClusterMeta).addCluster(rbClusterMeta).addCluster(oyClusterMeta2);
-        xpipeMeta.getDcs().get("rb").addCluster(oyClusterMeta).addCluster(rbClusterMeta).addCluster(oyClusterMeta2);
+        oyClusterMeta.setType(ClusterType.ONE_WAY.toString());
+        oyClusterMeta2.setType(ClusterType.ONE_WAY.toString());
+        rbClusterMeta.setType(ClusterType.ONE_WAY.toString());
+        biClusterMeta.setType(ClusterType.BI_DIRECTION.toString());
+
+        xpipeMeta.getDcs().get("fq").addCluster(biClusterMeta);
+        xpipeMeta.getDcs().get("oy").addCluster(oyClusterMeta).addCluster(rbClusterMeta).addCluster(oyClusterMeta2).addCluster(biClusterMeta);
+        xpipeMeta.getDcs().get("rb").addCluster(oyClusterMeta).addCluster(rbClusterMeta).addCluster(oyClusterMeta2).addCluster(biClusterMeta);
         return xpipeMeta;
     }
 
diff --git a/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/healthcheck/actions/delay/CrossMasterDelayServiceTest.java b/redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/healthcheck/actions/delay/CrossMasterDelayServiceTest.java
@@ -54,7 +54,7 @@ public void setupCrossMasterDelayServiceTest() {
 
     @Test
     public void testOnAction() {
-        service.onAction(new DelayActionContext(instance, 10L));
+        service.onAction(new DelayActionContext(instance, 10 * 1000 * 1000L));
         Assert.assertEquals(Collections.singletonMap(remoteDcId, Pair.of(new HostPort(), 10L)), service.getPeerMasterDelayFromCurrentDc(clusterId, shardId));
         Assert.assertEquals(Collections.singletonMap(remoteDcId, Pair.of(new HostPort(), 10L)), service.getPeerMasterDelayFromSourceDc(dcId, clusterId, shardId));
     }

Original file line number	Diff line number	Diff line change
`@@ -75,6 +75,7 @@ public MigrationPublishResult doMigrationPublish(String clusterName, String shar`
`75`	`75`	`@Override`
`76`	`76`	`public ClusterInfo getClusterInfo(String clusterName) {`
`77`	`77`	`ClusterInfo clusterInfo = new ClusterInfo();`
	`78`	`+ clusterInfo.setName(clusterName);`
`78`	`79`	`clusterInfo.setGroups(Lists.newArrayList(new GroupInfo()));`
`79`	`80`	`return clusterInfo;`
`80`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ public void setupCrossMasterDelayServiceTest() {`
`54`	`54`
`55`	`55`	`@Test`
`56`	`56`	`public void testOnAction() {`
`57`		`- service.onAction(new DelayActionContext(instance, 10L));`
	`57`	`+ service.onAction(new DelayActionContext(instance, 10 * 1000 * 1000L));`
`58`	`58`	`Assert.assertEquals(Collections.singletonMap(remoteDcId, Pair.of(new HostPort(), 10L)), service.getPeerMasterDelayFromCurrentDc(clusterId, shardId));`
`59`	`59`	`Assert.assertEquals(Collections.singletonMap(remoteDcId, Pair.of(new HostPort(), 10L)), service.getPeerMasterDelayFromSourceDc(dcId, clusterId, shardId));`
`60`	`60`	`}`