Add redis client name

core/src/main/java/com/ctrip/xpipe/netty/commands/AbstractNettyRequestResponseCommand.java

@@ -56,16 +56,21 @@ protected void doSendRequest(final NettyClient nettyClient, ByteBuf byteBuf) {
 		}
 
 		if(getCommandTimeoutMilli() > 0 && scheduled != null){
+			int commandTimeoutMilli = getCommandTimeoutMilli();
+			if (nettyClient instanceof RedisNettyClient && !((RedisNettyClient) nettyClient).getDoAfterConnectedOver()) {
+				commandTimeoutMilli += ((RedisNettyClient) nettyClient).getAfterConnectCommandTimeoutMill();
+			}
 
-			getLogger().debug("[doSendRequest][schedule timeout]{}, {}", this, getCommandTimeoutMilli());
+			getLogger().debug("[doSendRequest][schedule timeout]{}, {}", this, commandTimeoutMilli);
+			int finalCommandTimeoutMilli = commandTimeoutMilli;
 			timeoutFuture = scheduled.schedule(new AbstractExceptionLogTask() {
 
 				@Override
 				public void doRun() {
 					getLogger().info("[{}][run][timeout]{}", AbstractNettyRequestResponseCommand.this, nettyClient);
-					future().setFailure(new CommandTimeoutException("timeout " +  + getCommandTimeoutMilli()));
+					future().setFailure(new CommandTimeoutException("timeout " + finalCommandTimeoutMilli));
 				}
-			}, getCommandTimeoutMilli(), TimeUnit.MILLISECONDS);
+			}, commandTimeoutMilli, TimeUnit.MILLISECONDS);
 
 			future().addListener(new CommandFutureListener<V>() {
 

core/src/main/java/com/ctrip/xpipe/netty/commands/RedisNettyClient.java

@@ -0,0 +1,9 @@
+package com.ctrip.xpipe.netty.commands;
+
+public interface RedisNettyClient {
+
+    boolean getDoAfterConnectedOver();
+
+    int getAfterConnectCommandTimeoutMill();
+
+}

core/src/main/java/com/ctrip/xpipe/pool/XpipeNettyClientPool.java

@@ -61,7 +61,7 @@ public NettyClient borrowObject() throws BorrowObjectException {
 		try {
 			return objectPool.borrowObject();
 		} catch (Exception e) {
-			logger.error("[borrowObject]" + factory,  e);
+			logger.error("[borrowObject] NumIdle:{}, NumActive:{}" + factory, objectPool.getNumIdle(), objectPool.getNumActive(), e);
 			throw new BorrowObjectException("borrow " + factory, e);
 		}
 	}

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/CheckerConfig.java

@@ -112,4 +112,6 @@ public interface CheckerConfig {
 
     int getMarkInstanceMaxDelayMilli();
 
+    boolean getShouldDoAfterNettyClientConnected();
+
 }

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

@@ -37,6 +37,8 @@ public class DataCenterConfigBean extends AbstractConfigBean {
 
     public static final String KEY_HTTP_ACCEPT_ENCODING = "http.accept.encoding";
 
+    private static final String KEY_NETTY_CLIENT_DO_AFTER_CONNECTED = "netty.client.do.after.connected";
+
     private AtomicReference<String> zkConnection = new AtomicReference<>();
     private AtomicReference<String> zkNameSpace = new AtomicReference<>();
 
@@ -95,4 +97,8 @@ public String getHttpAcceptEncoding() {
         return getProperty(KEY_HTTP_ACCEPT_ENCODING, null);
     }
 
+    public boolean getDoAfterNettyClientConnected() {
+        return getBooleanProperty(KEY_NETTY_CLIENT_DO_AFTER_CONNECTED, true);
+    }
+
 }

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/redismaster/MasterOverOneMonitor.java

@@ -33,7 +33,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.REDIS_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.SCHEDULED_EXECUTOR;
 
 /**
@@ -49,7 +49,7 @@ public class MasterOverOneMonitor implements RedisMasterActionListener, OneWaySu
     @Autowired
     private AlertManager alertManager;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = REDIS_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     @Resource(name = SCHEDULED_EXECUTOR)

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/actions/sentinel/collector/DefaultSentinelHelloCollector.java

@@ -46,7 +46,7 @@
 import java.util.stream.Collectors;
 
 import static com.ctrip.xpipe.redis.checker.healthcheck.actions.sentinel.SentinelHelloCheckAction.LOG_TITLE;
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.SENTINEL_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.THREAD_POOL_TIME_OUT;
 
 /**
@@ -77,7 +77,7 @@ public class DefaultSentinelHelloCollector implements SentinelHelloCollector {
     @Autowired
     private PersistenceCache persistenceCache;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = SENTINEL_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     private SentinelLeakyBucket leakyBucket;

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

@@ -19,7 +19,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledExecutorService;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.PROXY_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.SCHEDULED_EXECUTOR;
 
 @Component
@@ -28,7 +28,7 @@ public class ProxyConnectedChecker implements ProxyChecker {
     @Autowired
     private CheckerConfig checkerConfig;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = PROXY_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     @Resource(name = SCHEDULED_EXECUTOR)

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/resource/Resource.java

@@ -8,11 +8,15 @@ public class Resource {
 
     public static final String REDIS_COMMAND_EXECUTOR = "redisCommandExecutor";
 
-    public static final String KEYED_NETTY_CLIENT_POOL = "keyedClientPool";
+    public static final String REDIS_KEYED_NETTY_CLIENT_POOL = "redisKeyedClientPool";
 
-    public static final String REDIS_SESSION_NETTY_CLIENT_POOL = "redisSessionClientPool";
+    public static final String SENTINEL_KEYED_NETTY_CLIENT_POOL = "sentinelKeyedClientPool";
+
+    public static final String KEEPER_KEYED_NETTY_CLIENT_POOL = "keeperKeyedClientPool";
 
-    public static final String MIGRATE_KEEPER_CLIENT_POOL = "migrateKeeperClientPool";
+    public static final String PROXY_KEYED_NETTY_CLIENT_POOL = "proxyKeyedClientPool";
+
+    public static final String REDIS_SESSION_NETTY_CLIENT_POOL = "redisSessionClientPool";
 
     public static final String PING_DELAY_INFO_EXECUTORS = "pingDelayInfoExecutors";
 

redis/redis-checker/src/test/java/com/ctrip/xpipe/redis/checker/AbstractCheckerIntegrationTest.java

@@ -5,6 +5,7 @@
 import com.ctrip.xpipe.concurrent.DefaultExecutorFactory;
 import com.ctrip.xpipe.lifecycle.LifecycleHelper;
 import com.ctrip.xpipe.netty.commands.NettyKeyedPoolClientFactory;
+import com.ctrip.xpipe.redis.core.client.NettyRedisPoolClientFactory;
 import com.ctrip.xpipe.pool.XpipeNettyClientKeyedObjectPool;
 import com.ctrip.xpipe.redis.checker.cluster.GroupCheckerLeaderElector;
 import com.ctrip.xpipe.redis.checker.config.CheckerConfig;
@@ -148,17 +149,33 @@ public ScheduledExecutorService getRedisCommandExecutor() {
             );
         }
 
-        @Bean(name = KEYED_NETTY_CLIENT_POOL)
+        @Bean(name = REDIS_KEYED_NETTY_CLIENT_POOL)
         public XpipeNettyClientKeyedObjectPool getReqResNettyClientPool() throws Exception {
-            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(8));
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getRedisPoolClientFactory(8));
             LifecycleHelper.initializeIfPossible(keyedObjectPool);
             LifecycleHelper.startIfPossible(keyedObjectPool);
             return keyedObjectPool;
         }
 
         @Bean(name = REDIS_SESSION_NETTY_CLIENT_POOL)
         public XpipeNettyClientKeyedObjectPool getRedisSessionNettyClientPool() throws Exception {
-            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(12));
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getRedisPoolClientFactory(12));
+            LifecycleHelper.initializeIfPossible(keyedObjectPool);
+            LifecycleHelper.startIfPossible(keyedObjectPool);
+            return keyedObjectPool;
+        }
+
+        @Bean(name = SENTINEL_KEYED_NETTY_CLIENT_POOL)
+        public XpipeNettyClientKeyedObjectPool getSentinelReqResNettyClientPool() throws Exception {
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(8));
+            LifecycleHelper.initializeIfPossible(keyedObjectPool);
+            LifecycleHelper.startIfPossible(keyedObjectPool);
+            return keyedObjectPool;
+        }
+
+        @Bean(name = PROXY_KEYED_NETTY_CLIENT_POOL)
+        public XpipeNettyClientKeyedObjectPool getProxyReqResNettyClientPool() throws Exception {
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(8));
             LifecycleHelper.initializeIfPossible(keyedObjectPool);
             LifecycleHelper.startIfPossible(keyedObjectPool);
             return keyedObjectPool;
@@ -195,6 +212,10 @@ public ScheduledExecutorService getHelloCheckScheduled() {
         private NettyKeyedPoolClientFactory getKeyedPoolClientFactory(int eventLoopThreads) {
             return new NettyKeyedPoolClientFactory(eventLoopThreads);
         }
+
+        private NettyKeyedPoolClientFactory getRedisPoolClientFactory(int eventLoopThreads) {
+            return new NettyRedisPoolClientFactory(eventLoopThreads, "xpipe", () -> true);
+        }
 
         @Bean
         public FoundationService foundationService() {

redis/redis-checker/src/test/java/com/ctrip/xpipe/redis/checker/TestConfig.java

@@ -315,4 +315,9 @@ public int getMarkInstanceMaxDelayMilli() {
         return 2000;
     }
 
+    @Override
+    public boolean getShouldDoAfterNettyClientConnected() {
+        return true;
+    }
+
 }

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/config/impl/DefaultConsoleConfig.java

@@ -587,6 +587,11 @@ public String getHttpAcceptEncoding() {
         return dataCenterConfigBean.getHttpAcceptEncoding();
     }
 
+    @Override
+    public boolean getShouldDoAfterNettyClientConnected() {
+        return dataCenterConfigBean.getDoAfterNettyClientConnected();
+    }
+
     @Override
     public void addListener(ConfigKeyListener listener) {
         this.listenersSet.add(listener);

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

@@ -35,7 +35,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.PROXY_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.SCHEDULED_EXECUTOR;
 
 @Component
@@ -49,7 +49,7 @@ public class DefaultProxyMonitorCollectorManager extends AbstractStartStoppable
     @Resource(name = SCHEDULED_EXECUTOR)
     private ScheduledExecutorService scheduled;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = PROXY_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     @Autowired

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/redis/DefaultSentinelManager.java

@@ -36,8 +36,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
-import static com.ctrip.xpipe.redis.checker.resource.Resource.REDIS_COMMAND_EXECUTOR;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.*;
 
 /**
  * @author chen.zhu
@@ -53,7 +52,7 @@ public class DefaultSentinelManager implements SentinelManager, ShardEventHandle
 
     private static final int LONG_SENTINEL_COMMAND_TIMEOUT = 2000;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = SENTINEL_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedClientPool;
 
     @Resource(name = REDIS_COMMAND_EXECUTOR)

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

@@ -69,7 +69,7 @@ public class ShardModelServiceImpl implements ShardModelService{
     @Resource(name = REDIS_COMMAND_EXECUTOR)
     private ScheduledExecutorService scheduled;
 
-    @Resource(name = MIGRATE_KEEPER_CLIENT_POOL)
+    @Resource(name = KEEPER_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     private RetryCommandFactory<Object> retryCommandFactory;

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/spring/ResourceConfig.java

@@ -3,11 +3,14 @@
 import com.ctrip.xpipe.concurrent.DefaultExecutorFactory;
 import com.ctrip.xpipe.lifecycle.LifecycleHelper;
 import com.ctrip.xpipe.netty.commands.NettyKeyedPoolClientFactory;
+import com.ctrip.xpipe.redis.core.client.NettyRedisPoolClientFactory;
 import com.ctrip.xpipe.pool.XpipeNettyClientKeyedObjectPool;
+import com.ctrip.xpipe.redis.checker.config.CheckerConfig;
 import com.ctrip.xpipe.redis.core.spring.AbstractRedisConfigContext;
 import com.ctrip.xpipe.utils.OsUtils;
 import com.ctrip.xpipe.utils.XpipeThreadFactory;
 import com.google.common.util.concurrent.MoreExecutors;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -22,11 +25,14 @@
 @Configuration
 public class ResourceConfig extends AbstractRedisConfigContext {
 
+    @Autowired
+    private CheckerConfig consoleConfig;
+
     private final static int REDIS_SESSION_CLIENT_POOL_SIZE = Integer.parseInt(System.getProperty("REDIS_SESSION_CLIENT_POOL_SIZE", "12"));
 
     private final static int KEYED_CLIENT_POOL_SIZE = Integer.parseInt(System.getProperty("KEYED_CLIENT_POOL_SIZE", "8"));
 
-    private final static int MIGRATE_KEEPER_CLIENT_POOL_SIZE = Integer.parseInt(System.getProperty("MIGRATE_KEEPER_CLIENT_POOL_SIZE", "1"));
+    private final static String DEFAULT_CLIENT_NAME = "xpipe";
 
     @Bean(name = REDIS_COMMAND_EXECUTOR)
     public ScheduledExecutorService getRedisCommandExecutor() {
@@ -40,25 +46,41 @@ public ScheduledExecutorService getRedisCommandExecutor() {
         );
     }
 
-    @Bean(name = KEYED_NETTY_CLIENT_POOL)
+    @Bean(name = REDIS_KEYED_NETTY_CLIENT_POOL)
     public XpipeNettyClientKeyedObjectPool getReqResNettyClientPool() throws Exception {
-        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(KEYED_CLIENT_POOL_SIZE));
+        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getRedisPoolClientFactory(KEYED_CLIENT_POOL_SIZE));
         LifecycleHelper.initializeIfPossible(keyedObjectPool);
         LifecycleHelper.startIfPossible(keyedObjectPool);
         return keyedObjectPool;
     }
 
     @Bean(name = REDIS_SESSION_NETTY_CLIENT_POOL)
     public XpipeNettyClientKeyedObjectPool getRedisSessionNettyClientPool() throws Exception {
-        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(REDIS_SESSION_CLIENT_POOL_SIZE));
+        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getRedisPoolClientFactory(REDIS_SESSION_CLIENT_POOL_SIZE));
         LifecycleHelper.initializeIfPossible(keyedObjectPool);
         LifecycleHelper.startIfPossible(keyedObjectPool);
         return keyedObjectPool;
     }
 
-    @Bean(name = MIGRATE_KEEPER_CLIENT_POOL)
-    public XpipeNettyClientKeyedObjectPool getMigrateKeeperClientPool() throws Exception {
-        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(MIGRATE_KEEPER_CLIENT_POOL_SIZE));
+    @Bean(name = KEEPER_KEYED_NETTY_CLIENT_POOL)
+    public XpipeNettyClientKeyedObjectPool getKeeperReqResNettyClientPool() throws Exception {
+        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(KEYED_CLIENT_POOL_SIZE));
+        LifecycleHelper.initializeIfPossible(keyedObjectPool);
+        LifecycleHelper.startIfPossible(keyedObjectPool);
+        return keyedObjectPool;
+    }
+
+    @Bean(name = SENTINEL_KEYED_NETTY_CLIENT_POOL)
+    public XpipeNettyClientKeyedObjectPool getSentinelReqResNettyClientPool() throws Exception {
+        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(KEYED_CLIENT_POOL_SIZE));
+        LifecycleHelper.initializeIfPossible(keyedObjectPool);
+        LifecycleHelper.startIfPossible(keyedObjectPool);
+        return keyedObjectPool;
+    }
+
+    @Bean(name = PROXY_KEYED_NETTY_CLIENT_POOL)
+    public XpipeNettyClientKeyedObjectPool getProxyReqResNettyClientPool() throws Exception {
+        XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(KEYED_CLIENT_POOL_SIZE));
         LifecycleHelper.initializeIfPossible(keyedObjectPool);
         LifecycleHelper.startIfPossible(keyedObjectPool);
         return keyedObjectPool;
@@ -96,4 +118,8 @@ private NettyKeyedPoolClientFactory getKeyedPoolClientFactory(int eventLoopThrea
         return new NettyKeyedPoolClientFactory(eventLoopThreads);
     }
 
+    private NettyKeyedPoolClientFactory getRedisPoolClientFactory(int eventLoopThreads) {
+        return new NettyRedisPoolClientFactory(eventLoopThreads, DEFAULT_CLIENT_NAME, () -> consoleConfig.getShouldDoAfterNettyClientConnected());
+    }
+
 }

redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/client/AsyncConnectionCondition.java

@@ -0,0 +1,7 @@
+package com.ctrip.xpipe.redis.core.client;
+
+public interface AsyncConnectionCondition {
+
+    boolean shouldDo();
+
+}

redis/redis-core/src/main/java/com/ctrip/xpipe/redis/core/client/NettyRedisPoolClientFactory.java

@@ -0,0 +1,31 @@
+package com.ctrip.xpipe.redis.core.client;
+
+import com.ctrip.xpipe.api.endpoint.Endpoint;
+import com.ctrip.xpipe.netty.commands.NettyClient;
+import com.ctrip.xpipe.netty.commands.NettyClientHandler;
+import com.ctrip.xpipe.netty.commands.NettyKeyedPoolClientFactory;
+import io.netty.channel.ChannelFuture;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+
+public class NettyRedisPoolClientFactory extends NettyKeyedPoolClientFactory {
+
+    private String clientName;
+
+    private AsyncConnectionCondition asyncConnectionCondition;
+
+    public NettyRedisPoolClientFactory(int eventLoopThreads, String clientName, AsyncConnectionCondition asyncConnectionCondition) {
+        super(eventLoopThreads);
+        this.clientName = clientName;
+        this.asyncConnectionCondition = asyncConnectionCondition;
+    }
+
+    @Override
+    public PooledObject<NettyClient> makeObject(Endpoint key) throws Exception {
+        ChannelFuture f = b.connect(key.getHost(), key.getPort());
+        NettyClient nettyClient = new RedisAsyncNettyClient(f, key, clientName, asyncConnectionCondition);
+        f.channel().attr(NettyClientHandler.KEY_CLIENT).set(nettyClient);
+        return new DefaultPooledObject<NettyClient>(nettyClient);
+    }
+
+}