fix bugs for console and keeper

core/src/main/java/com/ctrip/xpipe/cluster/AbstractLeaderElector.java

@@ -33,7 +33,8 @@
     @Autowired
     private ZkClient zkClient;
 
-    private ExecutorService executors = Executors.newCachedThreadPool(XpipeThreadFactory.create(getClass().getSimpleName()));
+    // single thread make zk events handled sequentially
+    private ExecutorService executors = Executors.newSingleThreadExecutor(XpipeThreadFactory.create(getClass().getSimpleName()));
 
     private ApplicationContext applicationContext;
 
@@ -63,8 +64,8 @@
                     try{
                         logger.info("[isLeader][notify]{}", entry.getKey());
                         entry.getValue().isleader();
-                    }catch (Exception e){
-                        logger.error("[isLeader]" + entry, e);
+                    }catch (Throwable th){
+                        logger.error("[isLeader]" + entry, th);
                     }
                 }
             }
@@ -79,8 +80,8 @@
                     try{
                         logger.info("[notLeader][notify]{}", entry.getKey());
                         entry.getValue().notLeader();
-                    }catch (Exception e){
-                        logger.error("[notLeader]" + entry, e);
+                    }catch (Throwable th){
+                        logger.error("[notLeader]" + entry, th);
                     }
                 }
             }

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/proxy/impl/DefaultProxyChainCollector.java

@@ -25,6 +25,7 @@
 import org.springframework.web.client.RestOperations;
 
 import javax.annotation.Resource;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
@@ -114,16 +115,16 @@
 
     @Override
     protected void doStart() throws Exception {
-        scheduled.scheduleWithFixedDelay(() -> {
+        future = scheduled.scheduleWithFixedDelay(() -> {
             if (!taskTrigger.get()) {
                 return;
             }
             logger.debug("proxy chain collector started");
-            getAllDcProxyChains();
+            fetchAllDcProxyChains();
         }, getStartTime(), getPeriodic(), TimeUnit.MILLISECONDS);
     }
 
-    private void getAllDcProxyChains() {
+    protected void fetchAllDcProxyChains() {
         ParallelCommandChain commandChain = new ParallelCommandChain(MoreExecutors.directExecutor(), false);
         consoleConfig.getConsoleDomains().forEach((dc, domain)->{
             logger.debug("begin to get proxy chain from dc {} {}", dc, domain);
@@ -152,18 +153,21 @@
         dcProxyChainMap.forEach((dc, proxyChainMap) -> {
             proxyChainMap.forEach((clusterShard, proxyChain) -> {
                 DefaultTunnelInfo tunnel = proxyChain.getTunnelInfos().get(0);
-                if (tempShardProxyChain.containsKey(clusterShard)) {
-                    tempShardProxyChain.get(clusterShard).getTunnelInfos().add(tunnel);
-                } else {
-                    tempShardProxyChain.put(clusterShard, proxyChain);
+                if (!tempShardProxyChain.containsKey(clusterShard)) {
+                    tempShardProxyChain.put(clusterShard, new DefaultProxyChain(proxyChain.getBackupDcId(),
+                            proxyChain.getClusterId(), proxyChain.getShardId(), proxyChain.getPeerDcId(), new ArrayList<>()));
                 }
-
+                tempShardProxyChain.get(clusterShard).getTunnelInfos().add(tunnel);
                 tempTunnelClusterShardMap.put(tunnel.getTunnelId(), clusterShard);
             });
         });
         synchronized (DefaultProxyChainCollector.this) {
-            tunnelClusterShardMap = tempTunnelClusterShardMap;
-            shardProxyChainMap = tempShardProxyChain;
+            if (taskTrigger.get()) {
+                tunnelClusterShardMap = tempTunnelClusterShardMap;
+                shardProxyChainMap = tempShardProxyChain;
+            } else {
+                clear();
+            }
         }
     }
 
@@ -173,6 +177,12 @@
         return this;
     }
 
+    @VisibleForTesting
+    DefaultProxyChainCollector setHttpService(DefaultHttpService httpService) {
+        this.httpService = httpService;
+        return this;
+    }
+
     @Override
     public Map<DcClusterShardPeer, ProxyChain> getShardProxyChainMap() {
         return shardProxyChainMap;
@@ -186,15 +196,19 @@
         return 1000;
     }
 
+    protected void clear() {
+        shardProxyChainMap.clear();
+        tunnelClusterShardMap.clear();
+        dcProxyChainMap.clear();
+    }
+
     @Override
     protected void doStop() throws Exception {
         if(future != null) {
             future.cancel(true);
             future = null;
         }
-        shardProxyChainMap.clear();
-        tunnelClusterShardMap.clear();
-        dcProxyChainMap.clear();
+        clear();
     }
 
     @Override

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

@@ -85,6 +85,7 @@ public void notLeader() {
     }
 
     private synchronized void stopLoadMeta(){
+        logger.info("[loadMeta][stop]{}", this);
         if (future != null)
             future.cancel(true);
         future = null;
@@ -98,7 +99,7 @@ private synchronized void stopLoadMeta(){
     }
 
     public void startLoadMeta() {
-        logger.info("[loadMeta]{}", this);
+        logger.info("[loadMeta][start]{}", this);
 
         refreshIntervalMilli = consoleConfig.getCacheRefreshInterval();
 

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/service/impl/DefaultDelayService.java

@@ -2,6 +2,8 @@
 
 import com.ctrip.xpipe.api.foundation.FoundationService;
 import com.ctrip.xpipe.cluster.ClusterType;
+import com.ctrip.xpipe.command.AbstractCommand;
+import com.ctrip.xpipe.command.ParallelCommandChain;
 import com.ctrip.xpipe.endpoint.HostPort;
 import com.ctrip.xpipe.redis.checker.healthcheck.BiDirectionSupport;
 import com.ctrip.xpipe.redis.checker.healthcheck.OneWaySupport;
@@ -25,11 +27,15 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.GLOBAL_EXECUTOR;
+
 /**
  * @author chen.zhu
  * <p>
@@ -58,6 +64,9 @@
     @Autowired
     private FoundationService foundationService;
 
+    @Resource(name = GLOBAL_EXECUTOR)
+    protected ExecutorService executors;
+
     @Override
     public void updateRedisDelays(Map<HostPort, Long> redisDelays) {
         hostPort2Delay.putAll(redisDelays);
@@ -221,12 +230,24 @@
         }
 
         UnhealthyInfoModel infoAggregation = new UnhealthyInfoModel();
+        ParallelCommandChain commandChain = new ParallelCommandChain(executors);
         for (String dcId : xpipeMeta.getDcs().keySet()) {
-            UnhealthyInfoModel unhealthyInfo = getDcActiveClusterUnhealthyInstance(dcId);
-            if (null == unhealthyInfo) infoAggregation.getAttachFailDc().add(dcId);
-            else infoAggregation.merge(unhealthyInfo);
+            FetchDcUnhealthyInstanceCmd cmd = new FetchDcUnhealthyInstanceCmd(dcId);
+            commandChain.add(cmd);
+            cmd.future().addListener(commandFuture -> {
+                if (commandFuture.isSuccess() && null != commandFuture.get()) {
+                    infoAggregation.merge(commandFuture.get());
+                } else {
+                    infoAggregation.getAttachFailDc().add(dcId);
+                }
+            });
         }
 
+        try {
+            commandChain.execute().get();
+        } catch (Throwable th) {
+            logger.info("[getAllUnhealthyInstance][fail] {}", th.getMessage());
+        }
         return infoAggregation;
     }
 
@@ -239,4 +260,29 @@
     public void setFoundationService(FoundationService foundationService) {
         this.foundationService = foundationService;
     }
+
+    class FetchDcUnhealthyInstanceCmd extends AbstractCommand<UnhealthyInfoModel> {
+
+        private String dc;
+
+        public FetchDcUnhealthyInstanceCmd(String dc) {
+            this.dc = dc;
+        }
+
+        @Override
+        protected void doExecute() throws Throwable {
+            future().setSuccess(getDcActiveClusterUnhealthyInstance(dc));
+        }
+
+        @Override
+        protected void doReset() {
+            // do nothing
+        }
+
+        @Override
+        public String getName() {
+            return getClass().getSimpleName();
+        }
+    }
+
 }

redis/redis-console/src/test/java/com/ctrip/xpipe/redis/console/proxy/impl/DefaultProxyChainCollectorTest.java

@@ -1,18 +1,116 @@
 package com.ctrip.xpipe.redis.console.proxy.impl;
 
+import com.ctrip.xpipe.api.foundation.FoundationService;
+import com.ctrip.xpipe.redis.checker.model.DcClusterShardPeer;
 import com.ctrip.xpipe.redis.console.AbstractConsoleTest;
+import com.ctrip.xpipe.redis.console.config.ConsoleConfig;
+import com.ctrip.xpipe.redis.console.model.ProxyModel;
+import com.ctrip.xpipe.redis.console.proxy.ProxyChain;
+import com.ctrip.xpipe.redis.console.proxy.ProxyChainAnalyzer;
 import com.ctrip.xpipe.redis.console.proxy.ProxyChainCollector;
+import com.ctrip.xpipe.redis.console.reporter.DefaultHttpService;
+import org.apache.http.HttpException;
+import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestOperations;
 
+import javax.annotation.Resource;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.IntStream;
+
+import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.SCHEDULED_EXECUTOR;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.when;
+
+
+@RunWith(MockitoJUnitRunner.class)
 public class DefaultProxyChainCollectorTest extends AbstractConsoleTest {
 
-    @Autowired
-    ProxyChainCollector collector;
+    @Mock
+    private ProxyChainAnalyzer proxyChainAnalyzer;
+
+    @Mock
+    private ConsoleConfig consoleConfig;
+
+    @Mock
+    private DefaultHttpService httpService;
+
+    @Mock
+    private RestOperations restTemplate;
+
+    @InjectMocks
+    DefaultProxyChainCollector collector;
+
+    private Map<String,String> consoles = new HashMap<String, String>() {{
+        put("dc1","http://dc1");
+        put("dc2","http://dc2");
+    }};
+
+    @Before
+    public void setupDefaultProxyChainCollectorTest() {
+        when(httpService.getRestTemplate()).thenReturn(restTemplate);
+        this.collector.setHttpService(httpService);
+        when(consoleConfig.getConsoleDomains()).thenReturn(consoles);
+    }
+
+    private Map<DcClusterShardPeer, DefaultProxyChain> generateProxyChains(int cnt) {
+        Map<DcClusterShardPeer, DefaultProxyChain> result = new HashMap<>();
+        IntStream.range(0, cnt).forEach(i -> {
+            DcClusterShardPeer dcClusterShardPeer = new DcClusterShardPeer("dc1", "cluster" + i, "shard" + i, "dc2");
+            List<DefaultTunnelInfo> tunnels = new ArrayList<>();
+            tunnels.add(new DefaultTunnelInfo(new ProxyModel(), "tunnel" + i));
+            DefaultProxyChain proxyChain = new DefaultProxyChain("dc1", "cluster" + i, "shard" + i, "dc2", tunnels);
+            result.put(dcClusterShardPeer, proxyChain);
+        });
+
+        return result;
+    }
 
     @Test
-    public void testUpdateProxyChains() {
-//        collector.updateShardProxyChainMap();
+    public void remoteDcDown_noMemLeak() {
+        ResponseEntity<Map<DcClusterShardPeer, DefaultProxyChain>> resp = new ResponseEntity(generateProxyChains(100), HttpStatus.OK);
+        when(restTemplate.exchange(anyString(), any(), any(), any(ParameterizedTypeReference.class), anyString()))
+                .thenReturn(resp);
+        IntStream.range(0, 10).forEach(i -> collector.fetchAllDcProxyChains());
+
+        Map<String, Map<DcClusterShardPeer, ProxyChain>> dcProxyChainMap = collector.getDcProxyChainMap();
+        for (Map<DcClusterShardPeer, ProxyChain> proxyChainMap: dcProxyChainMap.values()) {
+            for (ProxyChain proxyChain: proxyChainMap.values()) {
+                Assert.assertEquals(2, proxyChain.getTunnelInfos().size());
+            }
+        }
+
+        doAnswer(inov -> {
+            String uri = inov.getArgument(0);
+            String host = consoles.values().iterator().next();
+            if (uri.startsWith(host)) throw new HttpException("mock");
+            else return resp;
+        }).when(restTemplate).exchange(anyString(), any(), any(), any(ParameterizedTypeReference.class));
+
+        IntStream.range(0, 10).forEach(i -> collector.fetchAllDcProxyChains());
+        dcProxyChainMap = collector.getDcProxyChainMap();
+        for (Map<DcClusterShardPeer, ProxyChain> proxyChainMap: dcProxyChainMap.values()) {
+            for (ProxyChain proxyChain: proxyChainMap.values()) {
+                Assert.assertEquals(2, proxyChain.getTunnelInfos().size());
+            }
+        }
     }
 
+
 }

redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/protocal/cmd/RdbOnlyPsync.java

@@ -30,14 +30,12 @@
 
 	protected void failPsync(Throwable throwable) {
 		super.failPsync(throwable);
-		if (psyncState == PSYNC_STATE.PSYNC_COMMAND_WAITING_REPONSE) {
-			try {
-				getLogger().debug("[failPsync] psync fail before beginReadRdb");
-				currentReplicationStore.close();
-				currentReplicationStore.destroy();
-			} catch (Throwable th) {
-				getLogger().warn("[failPsync] release rdb file fail", th);
-			}
+		try {
+			getLogger().info("[failPsync][release rdb]");
+			currentReplicationStore.close();
+			currentReplicationStore.destroy();
+		} catch (Throwable th) {
+			getLogger().warn("[failPsync][release rdb] fail", th);
 		}
 	}