batch query CRedis instances status && concurrent get instances statu…

…s from XPipe and CRedis
ctripcorp · Sep 23, 2024 · a8eb019 · a8eb019
1 parent 232b062
commit a8eb019
Show file tree

Hide file tree

Showing 12 changed files with 225 additions and 207 deletions.
diff --git a/core/src/main/java/com/ctrip/xpipe/api/migration/OuterClientService.java b/core/src/main/java/com/ctrip/xpipe/api/migration/OuterClientService.java
@@ -764,6 +764,11 @@ public String getActiveDc() {
 		public void setActiveDc(String activeDc) {
 			this.activeDc = activeDc;
 		}
+
+		@Override
+		public String toString() {
+			return String.format("[%s:%s]%s", clusterName, activeDc, hostPortDcStatuses);
+		}
 	}
 
 	@JsonIgnoreProperties(ignoreUnknown = true)
@@ -815,6 +820,11 @@ public void setDc(String dc) {
 		public void setCanRead(boolean canRead) {
 			this.canRead = canRead;
 		}
+
+		@Override
+		public String toString() {
+			return String.format("{%s:%d-%s|%s}", host, port, dc, canRead);
+		}
 	}
 
 	enum ClusterType {

diff --git a/core/src/main/java/com/ctrip/xpipe/concurrent/AggregatorStateSetterManager.java b/core/src/main/java/com/ctrip/xpipe/concurrent/AggregatorStateSetterManager.java
diff --git a/redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/CheckerConfig.java b/redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/CheckerConfig.java
@@ -108,8 +108,8 @@ public interface CheckerConfig {
 
     int getKeeperCheckerIntervalMilli();
 
-    int getInstancePullIntervalSeconds();
+    int getMarkInstanceBaseDelayMilli();
 
-    int getInstancePullRandomSeconds();
+    int getMarkInstanceMaxDelayMilli();
 
 }
diff --git a/...edis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/impl/CheckConfigBean.java b/...edis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/impl/CheckConfigBean.java
@@ -113,6 +113,10 @@ public class CheckConfigBean extends AbstractConfigBean {
 
     public static final String KEY_KEEPER_CHECKER_INTERVAL = "keeper.checker.interval";
 
+    public static final String KEY_CHECKER_MARK_DELAY_BASE = "checker.health.mark.delay.base.milli";
+
+    public static final String KEY_CHECKER_MARK_DELAY_MAX = "checker.health.mark.delay.max.milli";
+
     public static final String KEY_CHECKER_INSTANCE_PULL_INTERVAL = "checker.instance.pull.interval";
 
     public static final String KEY_CHECKER_INSTANCE_PULL_RANDOM = "checker.instance.pull.random";
@@ -346,6 +350,14 @@ public int getKeeperCheckerIntervalMilli() {
         return getIntProperty(KEY_KEEPER_CHECKER_INTERVAL, 60 * 1000);
     }
 
+    public int getMarkInstanceBaseDelayMilli() {
+        return getIntProperty(KEY_CHECKER_MARK_DELAY_BASE, 3000);
+    }
+
+    public int getMarkInstanceMaxDelayMilli() {
+        return getIntProperty(KEY_CHECKER_MARK_DELAY_MAX, 10000);
+    }
+
     public int getInstancePullIntervalSeconds() {
         return getIntProperty(KEY_CHECKER_INSTANCE_PULL_INTERVAL, 5);
     }

diff --git a/.../com/ctrip/xpipe/redis/checker/healthcheck/actions/interaction/AggregatorPullService.java b/.../com/ctrip/xpipe/redis/checker/healthcheck/actions/interaction/AggregatorPullService.java
@@ -8,7 +8,7 @@
 
 public interface AggregatorPullService {
 
-    Set<HostPortDcStatus> getNeedAdjustInstances(Set<HostPort> instances) throws Exception;
+    Set<HostPortDcStatus> getNeedAdjustInstances(String cluster, Set<HostPort> instances) throws Exception;
 
     void doMarkInstances(String clusterName, Set<HostPortDcStatus> instances) throws OuterClientException;
 

diff --git a/...rip/xpipe/redis/checker/healthcheck/actions/interaction/DefaultAggregatorPullService.java b/...rip/xpipe/redis/checker/healthcheck/actions/interaction/DefaultAggregatorPullService.java
@@ -1,9 +1,10 @@
 package com.ctrip.xpipe.redis.checker.healthcheck.actions.interaction;
 
+import com.ctrip.xpipe.api.command.CommandFuture;
 import com.ctrip.xpipe.api.migration.OuterClientException;
 import com.ctrip.xpipe.api.migration.OuterClientService;
 import com.ctrip.xpipe.api.migration.OuterClientService.*;
-import com.ctrip.xpipe.endpoint.ClusterShardHostPort;
+import com.ctrip.xpipe.command.AbstractCommand;
 import com.ctrip.xpipe.endpoint.HostPort;
 import com.ctrip.xpipe.redis.checker.CheckerService;
 import com.ctrip.xpipe.redis.checker.RemoteCheckerManager;
@@ -12,12 +13,15 @@
 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.meta.MetaCache;
+import com.ctrip.xpipe.utils.XpipeThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.PostConstruct;
 import java.util.*;
+import java.util.concurrent.*;
 
 
 @Component
@@ -34,15 +38,31 @@ public class DefaultAggregatorPullService implements AggregatorPullService{
     private OuterClientService outerClientService = OuterClientService.DEFAULT;
     private static final Logger logger =  LoggerFactory.getLogger(DefaultAggregatorPullService.class);
 
+    private ExecutorService executors;
+
+    @PostConstruct
+    public void postConstruct() {
+        this.executors = new ThreadPoolExecutor(100, 100, 0L, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue(256), XpipeThreadFactory.create("DefaultAggregatorPullService"),
+                new ThreadPoolExecutor.CallerRunsPolicy());
+    }
+
     @Override
-    public Set<HostPortDcStatus> getNeedAdjustInstances(Set<HostPort> instances) throws Exception{
+    public Set<HostPortDcStatus> getNeedAdjustInstances(String cluster, Set<HostPort> instances) throws Exception {
         Set<HostPortDcStatus> instanceNeedAdjust = new HashSet<>();
-        Map<HostPort, Boolean> xpipeAllHealthStatus = getXpipeAllHealthStatus(instances);
-        logger.debug("[DefaultAggregatorPullService][getNeedAdjustInstances]xpipeAllHealthStatus:{}", xpipeAllHealthStatus);
-        Map<HostPort, Boolean> outerClientAllHealthStatus = getOuterClientAllHealthStatus(instances);
-        logger.debug("[DefaultAggregatorPullService][getNeedAdjustInstances]outerClientAllHealthStatus:{}", outerClientAllHealthStatus);
+
+        QueryXPipeInstanceStatusCmd queryXPipeInstanceStatusCmd = new QueryXPipeInstanceStatusCmd(cluster, instances);
+        QueryOuterClintInstanceStatusCmd queryOuterClintInstanceStatusCmd = new QueryOuterClintInstanceStatusCmd(cluster, instances);
+
+        CommandFuture<Map<HostPort, Boolean>> xpipeQueryFuture = queryXPipeInstanceStatusCmd.execute(executors);
+        CommandFuture<Map<HostPort, Boolean>> outerClientQueryFuture = queryOuterClintInstanceStatusCmd.execute(executors);
+
+        Map<HostPort, Boolean> xpipeAllHealthStatus = xpipeQueryFuture.get();
+        Map<HostPort, Boolean> outerClientAllHealthStatus = outerClientQueryFuture.get();
+
         for (Map.Entry<HostPort, Boolean> entry : xpipeAllHealthStatus.entrySet()) {
-            if (!outerClientAllHealthStatus.containsKey(entry.getKey()) || !entry.getValue().equals(outerClientAllHealthStatus.get(entry.getKey()))) {
+            if (!outerClientAllHealthStatus.containsKey(entry.getKey())
+                    || !entry.getValue().equals(outerClientAllHealthStatus.get(entry.getKey()))) {
                 instanceNeedAdjust.add(
                         new HostPortDcStatus(entry.getKey().getHost(), entry.getKey().getPort(), metaCache.getDc(entry.getKey()), entry.getValue()));
             }
@@ -57,29 +77,10 @@ public void doMarkInstances(String clusterName, Set<HostPortDcStatus> instances)
         outerClientService.batchMarkInstance(markInstanceRequest);
     }
 
-    public Map<HostPort, Boolean> getXpipeAllHealthStatus(Set<HostPort> instances) {
-        XPipeInstanceHealthHolder xPipeInstanceHealthHolder = new XPipeInstanceHealthHolder();
-        for (CheckerService checkerService : remoteCheckerManager.getAllCheckerServices()) {
-            xPipeInstanceHealthHolder.add(checkerService.getAllClusterInstanceHealthStatus(instances));
-        }
-        return xPipeInstanceHealthHolder.getAllHealthStatus(checkerConfig.getQuorum());
-    }
-
-    public Map<HostPort, Boolean> getOuterClientAllHealthStatus(Set<HostPort> hostPorts) throws Exception {
-        Map<ClusterShardHostPort, Boolean> instancesUp = new HashMap<>();
-        for (HostPort hostPort : hostPorts) {
-            ClusterShardHostPort clusterShardHostPort = new ClusterShardHostPort(null, null, hostPort);
-            instancesUp.put(clusterShardHostPort, outerClientService.isInstanceUp(clusterShardHostPort));
-        }
-        Map<HostPort, Boolean> result = new HashMap<>();
-        for (Map.Entry<ClusterShardHostPort, Boolean> entry : instancesUp.entrySet()) {
-            result.put(entry.getKey().getHostPort(), entry.getValue());
-        }
-        return result;
-    }
+    private void alertMarkInstance(String clusterName, Set<HostPortDcStatus> instances) {
+        if (instances.isEmpty()) return;
 
-    public void alertMarkInstance(String clusterName, Set<HostPortDcStatus> instances) {
-        if (!instances.isEmpty()) {
+        try {
             for (HostPortDcStatus instance : instances) {
                 if (instance.isCanRead()) {
                     alertManager.alert(clusterName, null,
@@ -89,6 +90,65 @@ public void alertMarkInstance(String clusterName, Set<HostPortDcStatus> instance
                             new HostPort(instance.getHost(), instance.getPort()), ALERT_TYPE.MARK_INSTANCE_DOWN, "Mark Instance Down");
                 }
             }
+        } catch (Throwable th) {
+            logger.info("[alertMarkInstance][{}] fail", clusterName, th);
         }
     }
+
+    private class QueryXPipeInstanceStatusCmd extends AbstractCommand<Map<HostPort, Boolean>> {
+
+        private String cluster;
+
+        private Set<HostPort> instances;
+
+        public QueryXPipeInstanceStatusCmd(String cluster, Set<HostPort> instances) {
+            this.cluster = cluster;
+            this.instances = instances;
+        }
+
+        @Override
+        protected void doExecute() throws Throwable {
+            XPipeInstanceHealthHolder xpipeInstanceHealthHolder = new XPipeInstanceHealthHolder();
+            for (CheckerService checkerService : remoteCheckerManager.getAllCheckerServices()) {
+                xpipeInstanceHealthHolder.add(checkerService.getAllClusterInstanceHealthStatus(instances));
+            }
+            future().setSuccess(xpipeInstanceHealthHolder.getAllHealthStatus(checkerConfig.getQuorum()));
+        }
+
+        @Override
+        protected void doReset() {
+        }
+
+        @Override
+        public String getName() {
+            return "QueryXPipeInstanceStatusCmd";
+        }
+    }
+
+    private class QueryOuterClintInstanceStatusCmd extends AbstractCommand<Map<HostPort, Boolean>> {
+
+        private String cluster;
+
+        private Set<HostPort> instances;
+
+        public QueryOuterClintInstanceStatusCmd(String cluster, Set<HostPort> instances) {
+            this.cluster = cluster;
+            this.instances = instances;
+        }
+
+        @Override
+        protected void doExecute() throws Throwable {
+            future().setSuccess(outerClientService.batchQueryInstanceStatus(cluster, instances));
+        }
+
+        @Override
+        protected void doReset() {
+        }
+
+        @Override
+        public String getName() {
+            return "[QueryOuterClintInstanceStatusCmd]" + cluster;
+        }
+    }
+
 }