call batchMarkInstance with dc instance cnt

Author: qifanwang

core/src/main/java/com/ctrip/xpipe/api/migration/OuterClientService.java

@@ -732,13 +732,17 @@ class MarkInstanceRequest{
 		private String clusterName;
 		private String activeDc;
 
+		private Map<String, Integer> instanceCnt;
+
 		public MarkInstanceRequest() {
 		}
 
-		public MarkInstanceRequest(Set<HostPortDcStatus> hostPortDcStatuses, String clusterName, String activeDc) {
+		public MarkInstanceRequest(Set<HostPortDcStatus> hostPortDcStatuses, String clusterName,
+								   String activeDc, Map<String, Integer> instanceCnt) {
 			this.hostPortDcStatuses = hostPortDcStatuses;
 			this.clusterName = clusterName;
 			this.activeDc = activeDc;
+			this.instanceCnt = instanceCnt;
 		}
 
 		public Set<HostPortDcStatus> getHostPortDcStatuses() {
@@ -765,6 +769,14 @@ public void setActiveDc(String activeDc) {
 			this.activeDc = activeDc;
 		}
 
+		public Map<String, Integer> getInstanceCnt() {
+			return instanceCnt;
+		}
+
+		public void setInstanceCnt(Map<String, Integer> instanceCnt) {
+			this.instanceCnt = instanceCnt;
+		}
+
 		@Override
 		public String toString() {
 			return String.format("[%s:%s]%s", clusterName, activeDc, hostPortDcStatuses);

core/src/main/java/com/ctrip/xpipe/migration/DefaultOuterClientService.java

@@ -4,6 +4,7 @@
 import com.ctrip.xpipe.endpoint.ClusterShardHostPort;
 import com.ctrip.xpipe.endpoint.HostPort;
 import com.ctrip.xpipe.utils.DateTimeUtils;
+import com.ctrip.xpipe.utils.VisibleForTesting;
 import com.google.common.collect.Lists;
 
 import java.net.InetSocketAddress;
@@ -19,6 +20,8 @@ public class DefaultOuterClientService extends AbstractOuterClientService {
 
 	private Map<HostPort, Boolean> instanceStatus = new ConcurrentHashMap<>();
 
+	private Map<String, Integer> cntMap = new ConcurrentHashMap<>();
+
 	@Override
 	public void markInstanceUp(ClusterShardHostPort clusterShardHostPort) throws OuterClientException {
 		logger.info("[markInstanceUp]{}", clusterShardHostPort);
@@ -131,6 +134,7 @@ public void batchMarkInstance(MarkInstanceRequest markInstanceRequest) throws Ou
 		for (HostPortDcStatus hostPortDcStatus : markInstanceRequest.getHostPortDcStatuses()) {
 			instanceStatus.put(new HostPort(hostPortDcStatus.getHost(), hostPortDcStatus.getPort()), hostPortDcStatus.isCanRead());
 		}
+		this.cntMap = markInstanceRequest.getInstanceCnt();
 	}
 
 	@Override
@@ -140,4 +144,8 @@ public OuterClientDataResp<List<ClusterExcludedIdcInfo>> getAllExcludedIdcs() th
 		resp.setResult(Collections.emptyList());
 		return resp;
 	}
+
+	public Map<String, Integer> getCntMap() {
+		return cntMap;
+	}
 }

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/interaction/DefaultAggregatorPullService.java

@@ -12,13 +12,17 @@
 import com.ctrip.xpipe.redis.checker.alert.AlertManager;
 import com.ctrip.xpipe.redis.checker.config.CheckerConfig;
 import com.ctrip.xpipe.redis.checker.healthcheck.actions.interaction.compensator.data.XPipeInstanceHealthHolder;
+import com.ctrip.xpipe.redis.core.entity.ClusterMeta;
+import com.ctrip.xpipe.redis.core.entity.ShardMeta;
 import com.ctrip.xpipe.redis.core.meta.MetaCache;
+import com.ctrip.xpipe.redis.core.meta.XpipeMetaManager;
 import com.ctrip.xpipe.utils.XpipeThreadFactory;
 import com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import com.ctrip.xpipe.redis.core.entity.DcMeta;
 
 import javax.annotation.PostConstruct;
 import java.util.*;
@@ -87,7 +91,8 @@ public Set<HostPortDcStatus> getNeedAdjustInstances(String cluster, Set<HostPort
     @Override
     public void doMarkInstances(String clusterName, String activeDc, Set<HostPortDcStatus> instances) throws OuterClientException {
         alertMarkInstance(clusterName, instances);
-        MarkInstanceRequest markInstanceRequest = new MarkInstanceRequest(instances, clusterName, activeDc);
+        Map<String, Integer> dcInstancesCnt = metaCache.getClusterCntMap(clusterName);
+        MarkInstanceRequest markInstanceRequest = new MarkInstanceRequest(instances, clusterName, activeDc, dcInstancesCnt);
         outerClientService.batchMarkInstance(markInstanceRequest);
     }
 
@@ -170,4 +175,9 @@ protected void setExecutors(Executor executors) {
         this.executors = executors;
     }
 
+    @VisibleForTesting
+    protected void setMetaCache(MetaCache metaCache) {
+        this.metaCache = metaCache;
+    }
+
 }

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/impl/TestMetaCache.java

@@ -227,4 +227,9 @@ public Set<String> getAllShardNamesByClusterName(String clusterName) {
         return Collections.emptySet();
     }
 
+    @Override
+    public Map<String, Integer> getClusterCntMap(String clusterName) {
+        return Collections.emptyMap();
+    }
+
 }

redis/redis-checker/src/test/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/interaction/DefaultAggregatorPullServiceTest.java

@@ -141,5 +141,4 @@ public void testQuorumWithLastMark() throws Exception {
         Assert.assertEquals("jq", status.getDc());
         Assert.assertTrue(status.isCanRead());
     }
-
 }

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/resources/AbstractMetaCache.java

@@ -703,6 +703,23 @@ public Set<String> getAllShardNamesByClusterName(String clusterName) {
         return shards;
     }
 
+    @Override
+    public Map<String, Integer> getClusterCntMap(String clusterName) {
+        Map<String, Integer> clusterCntMap = new HashMap<>();
+        for (DcMeta dcMeta : meta.getKey().getDcs().values()) {
+            ClusterMeta clusterMeta = dcMeta.getClusters().get(clusterName);
+            if(clusterMeta == null) {
+                continue;
+            }
+            int cnt = 0;
+            for(ShardMeta shardMeta : clusterMeta.getShards().values()) {
+                cnt += shardMeta.getRedises().size();
+            }
+            clusterCntMap.put(dcMeta.getId(), cnt);
+        }
+        return clusterCntMap;
+    }
+
     @VisibleForTesting
     public AbstractMetaCache setMeta(Pair<XpipeMeta, XpipeMetaManager> meta) {
         this.meta = meta;
@@ -714,4 +731,5 @@ public AbstractMetaCache setMonitor2ClusterShard(Map<String, Triple<String, Stri
         this.monitor2ClusterShard = monitorMap;
         return this;
     }
+
 }

redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/resources/DefaultMetaCacheTest.java

@@ -356,6 +356,14 @@ public void checkClustersCntTest() {
         }
     }
 
+    @Test
+    public void testClusterDnCnt() {
+        Map<String, Integer> cntMap = metaCache.getClusterCntMap("cluster1");
+        Assert.assertEquals(cntMap.size(), 2);
+        Assert.assertEquals(cntMap.get("jq").intValue(), 3);
+        Assert.assertEquals(cntMap.get("oy").intValue(), 2);
+    }
+
     protected String getXpipeMetaConfigFile() {
         return "dc-meta-test.xml";
     }

redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/meta/MetaCache.java

@@ -98,4 +98,6 @@ public interface MetaCache {
     boolean isAsymmetricCluster(String clusterName);
 
     Set<String> getAllShardNamesByClusterName(String clusterName);
+
+    Map<String, Integer> getClusterCntMap(String clusterName);
 }

redis/redis-keeper/src/test/java/com/ctrip/xpipe/redis/keeper/impl/RordbReplicationSupportTest.java

@@ -198,7 +198,7 @@ public void endWriteRdb() {
             }
         });
         psync.execute();
-        future.get(5, TimeUnit.SECONDS);
+        future.get(10, TimeUnit.SECONDS);
 
         return psync;
     }

redis/redis-keeper/src/test/java/com/ctrip/xpipe/redis/keeper/impl/fakeredis/FakeRedisHalfRdbServerFail.java

@@ -67,7 +67,7 @@ public void redisFailKeeperRestartDumpNewRdb() throws Exception {
 		SimplePsyncObserver simplePsyncObserver = new SimplePsyncObserver();
 		InMemoryPsync inMemoryPsync = sendInmemoryPsync("localhost", redisKeeperServer.getListeningPort(), simplePsyncObserver);
 		//wait
-		simplePsyncObserver.getOnline().get(6000, TimeUnit.MILLISECONDS);
+		simplePsyncObserver.getOnline().get(8000, TimeUnit.MILLISECONDS);
 		//wait for commands
 		sleep(1000);