add redis client name

ctripcorp · Nov 5, 2024 · a4cd17e · a4cd17e
1 parent 97e9741
commit a4cd17e
Show file tree

Hide file tree

Showing 13 changed files with 218 additions and 26 deletions.
diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/AbstractNettyRequestResponseCommand.java b/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 RedisAsyncNettyClient && !((RedisAsyncNettyClient) nettyClient).getDoAfterConnectedOver()) {
+				commandTimeoutMilli += ((RedisAsyncNettyClient) 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>() {
 

diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/AsyncNettyClient.java b/core/src/main/java/com/ctrip/xpipe/netty/commands/AsyncNettyClient.java
@@ -3,12 +3,15 @@
 import com.ctrip.xpipe.api.endpoint.Endpoint;
 import com.ctrip.xpipe.api.proxy.ProxyEnabled;
 import com.ctrip.xpipe.utils.ChannelUtil;
+import com.sun.org.apache.xpath.internal.operations.Bool;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 
 /**
  * @author chen.zhu
@@ -21,6 +24,8 @@ public class AsyncNettyClient extends DefaultNettyClient {
 
     private ChannelFuture future;
 
+    protected final AtomicReference<Boolean> doAfterConnectedOver = new AtomicReference<>(false);
+
     public AsyncNettyClient(ChannelFuture future, Endpoint endpoint) {
         super(future.channel());
         this.future = future;
@@ -35,6 +40,8 @@ public void operationComplete(ChannelFuture future) {
                     } else {
                         desc.set(ChannelUtil.getDesc(future.channel()));
                     }
+                    doAfterConnected();
+                    doAfterConnectedOver.set(true);
                 } else {
                     logger.info("[async][connect-fail] endpint: {}", endpoint, future.cause());
                 }
@@ -93,4 +100,16 @@ public String toString() {
         return super.toString();
     }
 
+    protected void doAfterConnected() {
+
+    }
+
+    protected boolean getDoAfterConnectedOver() {
+        return doAfterConnectedOver.get();
+    }
+
+    protected int getAfterConnectCommandTimeoutMill() {
+        return 0;
+    }
+
 }
diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/NettyRedisPoolClientFactory.java b/core/src/main/java/com/ctrip/xpipe/netty/commands/NettyRedisPoolClientFactory.java
@@ -0,0 +1,25 @@
+package com.ctrip.xpipe.netty.commands;
+
+import com.ctrip.xpipe.api.endpoint.Endpoint;
+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;
+
+    public NettyRedisPoolClientFactory(int eventLoopThreads, String clientName) {
+        super(eventLoopThreads);
+        this.clientName = clientName;
+    }
+
+    @Override
+    public PooledObject<NettyClient> makeObject(Endpoint key) throws Exception {
+        ChannelFuture f = b.connect(key.getHost(), key.getPort());
+        NettyClient nettyClient = new RedisAsyncNettyClient(f, key, clientName);
+        f.channel().attr(NettyClientHandler.KEY_CLIENT).set(nettyClient);
+        return new DefaultPooledObject<NettyClient>(nettyClient);
+    }
+
+}
diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/RedisAsyncNettyClient.java b/core/src/main/java/com/ctrip/xpipe/netty/commands/RedisAsyncNettyClient.java
@@ -0,0 +1,117 @@
+package com.ctrip.xpipe.netty.commands;
+
+import com.ctrip.xpipe.api.codec.Codec;
+import com.ctrip.xpipe.api.endpoint.Endpoint;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.util.ReferenceCountUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+
+public class RedisAsyncNettyClient extends AsyncNettyClient{
+
+    protected Logger logger = LoggerFactory.getLogger(getClass());
+
+    private String clientName;
+
+    private static final String CLIENT_SET_NAME = "CLIENT SETNAME ";
+
+    private static final String EXPECT_RESP = "OK";
+
+    private static final int DEFAULT_REDIS_COMMAND_TIME_OUT_MILLI = 660;
+
+    public RedisAsyncNettyClient(ChannelFuture future, Endpoint endpoint, String clientName) {
+        super(future, endpoint);
+        this.clientName = clientName;
+    }
+
+    @Override
+    protected void doAfterConnected() {
+        String command = CLIENT_SET_NAME + clientName + "\r\n";
+        ByteBuf byteBuf = Unpooled.wrappedBuffer(command.getBytes(Codec.defaultCharset));
+        try {
+            sendRequest(byteBuf, new ByteBufReceiver() {
+                @Override
+                public RECEIVER_RESULT receive(Channel channel, ByteBuf byteBuf) {
+                    String result = doReceiveResponse(byteBuf);
+                    if (result == null) {
+                        return RECEIVER_RESULT.CONTINUE;
+                    }
+                    if (EXPECT_RESP.equalsIgnoreCase(result)) {
+                        return RECEIVER_RESULT.SUCCESS;
+                    }
+                    return RECEIVER_RESULT.FAIL;
+                }
+
+                @Override
+                public void clientClosed(NettyClient nettyClient) {
+                    logger.warn("[redisAsync][wont-send][{}] {}", desc, nettyClient.channel());
+                }
+
+                @Override
+                public void clientClosed(NettyClient nettyClient, Throwable th) {
+                    logger.warn("[redisAsync][wont-send][{}] {}", desc, nettyClient.channel(), th);
+                }
+            });
+        } finally {
+            ReferenceCountUtil.release(byteBuf);
+        }
+    }
+
+    @Override
+    protected int getAfterConnectCommandTimeoutMill() {
+        return DEFAULT_REDIS_COMMAND_TIME_OUT_MILLI;
+    }
+
+    public static String doReceiveResponse(ByteBuf byteBuf) {
+        ByteArrayOutputStream baous = new ByteArrayOutputStream();
+        CRLF_STATE crlfState = CRLF_STATE.CONTENT;
+        int readable = byteBuf.readableBytes();
+        for(int i=0; i < readable ;i++){
+
+            byte data = byteBuf.readByte();
+            baous.write(data);
+            switch(data){
+                case '\r':
+                    crlfState = CRLF_STATE.CR;
+                    break;
+                case '\n':
+                    if(crlfState == CRLF_STATE.CR){
+                        crlfState = CRLF_STATE.CRLF;
+                        break;
+                    }
+                default:
+                    crlfState = CRLF_STATE.CONTENT;
+                    break;
+            }
+
+            if(crlfState == CRLF_STATE.CRLF){
+                break;
+            }
+        }
+        if (baous.toByteArray().length != 0 && crlfState == CRLF_STATE.CRLF) {
+            String data = new String(baous.toByteArray(), Codec.defaultCharset);
+            int beginIndex = 0;
+            int endIndex = data.length();
+            if(data.charAt(0) == '+'){
+                beginIndex = 1;
+            }
+            if(data.charAt(endIndex - 2) == '\r' && data.charAt(endIndex - 1) == '\n'){
+                endIndex -= 2;
+            }
+            return data.substring(beginIndex, endIndex);
+        }
+        return null;
+    }
+
+    public enum CRLF_STATE{
+        CR,
+        CRLF,
+        CONTENT
+    }
+
+}
diff --git a/...a/com/ctrip/xpipe/redis/checker/healthcheck/actions/redismaster/MasterOverOneMonitor.java b/...a/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)

diff --git a/...e/redis/checker/healthcheck/actions/sentinel/collector/DefaultSentinelHelloCollector.java b/...e/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;

diff --git a/...r/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/impl/ProxyConnectedChecker.java b/...r/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)

diff --git a/redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/resource/Resource.java b/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";
 

diff --git a/...s-checker/src/test/java/com/ctrip/xpipe/redis/checker/AbstractCheckerIntegrationTest.java b/...s-checker/src/test/java/com/ctrip/xpipe/redis/checker/AbstractCheckerIntegrationTest.java
@@ -148,7 +148,7 @@ 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));
             LifecycleHelper.initializeIfPossible(keyedObjectPool);

diff --git a/...in/java/com/ctrip/xpipe/redis/console/proxy/impl/DefaultProxyMonitorCollectorManager.java b/...in/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

diff --git a/...dis-console/src/main/java/com/ctrip/xpipe/redis/console/redis/DefaultSentinelManager.java b/...dis-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)

diff --git a/...src/main/java/com/ctrip/xpipe/redis/console/service/model/impl/ShardModelServiceImpl.java b/...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;