github/arthas
4
0
Fork 0
mirror of https://github.com/alibaba/arthas.git synced 2024-04-21 10:21:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use custom worker group to reduce the occurrence of http api block

Browse Source
This commit is contained in:
gongdewei
2020-05-27 11:44:21 +08:00
parent 80ce091342
commit 898da3da02
8 changed files with 64 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java
+27 -29
View File
@@ -48,27 +48,10 @@ import com.taobao.arthas.core.util.ArthasBanner;
import com.taobao.arthas.core.util.FileUtils;
import com.taobao.arthas.core.util.LogUtil;
import com.taobao.arthas.core.util.UserStatUtil;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import java.arthas.SpyAPI;
import java.io.File;
import java.io.IOException;
import java.lang.instrument.Instrumentation;
import java.lang.reflect.Method;
import java.net.URI;
import java.security.CodeSource;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* @author vlinux on 15/5/2.
@@ -78,9 +61,9 @@ public class ArthasBootstrap {
public static final String ARTHAS_HOME_PROPERTY = "arthas.home";
private static String ARTHAS_SHOME = null;
public static final String CONFIG_NAME_PROPERTY = "arthas.config.name";
public static final String CONFIG_NAME_PROPERTY = "arthas.config.name";
public static final String CONFIG_LOCATION_PROPERTY = "arthas.config.location";
public static final String CONFIG_OVERRIDE_ALL= "arthas.config.overrideAll";
public static final String CONFIG_OVERRIDE_ALL = "arthas.config.overrideAll";
private static ArthasBootstrap arthasBootstrap;
@@ -98,6 +81,7 @@ public class ArthasBootstrap {
private File arthasOutputDir;
private static LoggerContext loggerContext;
private EventExecutorGroup workerGroup;
private Timer timer = new Timer("arthas-timer", true);
@@ -258,14 +242,14 @@ public class ArthasBootstrap {
tunnelClient.setTunnelServerUrl(configure.getTunnelServer());
// ws://127.0.0.1:8563/ws
String host = "127.0.0.1";
if(configure.getIp() != null) {
if (configure.getIp() != null) {
host = configure.getIp();
}
URI uri = new URI("ws", null, host, configure.getHttpPort(), "/ws", null, null);
tunnelClient.setLocalServerUrl(uri.toString());
ChannelFuture channelFuture = tunnelClient.start();
channelFuture.await(10, TimeUnit.SECONDS);
if(channelFuture.isSuccess()) {
if (channelFuture.isSuccess()) {
agentId = tunnelClient.getId();
}
}
@@ -275,29 +259,34 @@ public class ArthasBootstrap {
try {
ShellServerOptions options = new ShellServerOptions()
.setInstrumentation(instrumentation)
.setPid(PidUtils.currentLongPid())
.setSessionTimeout(configure.getSessionTimeout() * 1000);
.setInstrumentation(instrumentation)
.setPid(PidUtils.currentLongPid())
.setSessionTimeout(configure.getSessionTimeout() * 1000);
if (agentId != null) {
Map<String, String> welcomeInfos = new HashMap<String, String>();
welcomeInfos.put("id", agentId);
options.setWelcomeMessage(ArthasBanner.welcome(welcomeInfos));
}
shellServer = new ShellServerImpl(options, this);
BuiltinCommandPack builtinCommands = new BuiltinCommandPack();
List<CommandResolver> resolvers = new ArrayList<CommandResolver>();
resolvers.add(builtinCommands);
//worker group
workerGroup = new NioEventLoopGroup(24);
// TODO: discover user provided command resolver
if (configure.getTelnetPort() > 0) {
shellServer.registerTermServer(new HttpTelnetTermServer(configure.getIp(), configure.getTelnetPort(),
options.getConnectionTimeout()));
options.getConnectionTimeout(), workerGroup));
} else {
logger().info("telnet port is {}, skip bind telnet server.", configure.getTelnetPort());
}
if (configure.getHttpPort() > 0) {
shellServer.registerTermServer(new HttpTermServer(configure.getIp(), configure.getHttpPort(),
options.getConnectionTimeout()));
options.getConnectionTimeout(), workerGroup));
} else {
logger().info("http port is {}, skip bind http server.", configure.getHttpPort());
}
@@ -321,7 +310,7 @@ public class ArthasBootstrap {
UserStatUtil.setStatUrl(configure.getStatUrl());
UserStatUtil.arthasStart();
logger().info("as-server started in {} ms", System.currentTimeMillis() - start );
logger().info("as-server started in {} ms", System.currentTimeMillis() - start);
} catch (Throwable e) {
logger().error("Error during bind to port " + configure.getTelnetPort(), e);
if (shellServer != null) {
@@ -330,11 +319,18 @@ public class ArthasBootstrap {
if (sessionManager != null){
sessionManager.close();
}
//shutdownWorkGroup();
shutdownWorkGroup();
throw e;
}
}
private void shutdownWorkGroup() {
if (workerGroup != null) {
workerGroup.shutdownGracefully(200, 200, TimeUnit.MILLISECONDS);
workerGroup = null;
}
}
/**
* 判断服务端是否已经启动
*
@@ -350,12 +346,13 @@ public class ArthasBootstrap {
try {
tunnelClient.stop();
} catch (Throwable e) {
logger().error("arthas", "stop tunnel client error", e);
logger().error("stop tunnel client error", e);
}
}
executorService.shutdownNow();
transformerManager.destroy();
UserStatUtil.destroy();
shutdownWorkGroup();
// clear the reference in Spy class.
cleanUpSpyReference();
try {
@@ -384,6 +381,7 @@ public class ArthasBootstrap {
}
return arthasBootstrap;
}
/**
* @return ArthasServer单例
*/
core/src/main/java/com/taobao/arthas/core/shell/term/impl/HttpTermServer.java
+5 -3
View File
@@ -7,7 +7,7 @@ import com.taobao.arthas.core.shell.handlers.Handler;
import com.taobao.arthas.core.shell.term.Term;
import com.taobao.arthas.core.shell.term.TermServer;
import com.taobao.arthas.core.shell.term.impl.http.NettyWebsocketTtyBootstrap;
import com.taobao.arthas.core.shell.term.impl.httptelnet.HttpTelnetTermServer;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.tty.TtyConnection;
@@ -25,11 +25,13 @@ public class HttpTermServer extends TermServer {
private String hostIp;
private int port;
private long connectionTimeout;
private EventExecutorGroup workerGroup;
public HttpTermServer(String hostIp, int port, long connectionTimeout) {
public HttpTermServer(String hostIp, int port, long connectionTimeout, EventExecutorGroup workerGroup) {
this.hostIp = hostIp;
this.port = port;
this.connectionTimeout = connectionTimeout;
this.workerGroup = workerGroup;
}
@Override
@@ -41,7 +43,7 @@ public class HttpTermServer extends TermServer {
@Override
public TermServer listen(Handler<Future<TermServer>> listenHandler) {
// TODO: charset and inputrc from options
bootstrap = new NettyWebsocketTtyBootstrap().setHost(hostIp).setPort(port);
bootstrap = new NettyWebsocketTtyBootstrap(workerGroup).setHost(hostIp).setPort(port);
try {
bootstrap.start(new Consumer<TtyConnection>() {
@Override
core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java
+5 -2
View File
@@ -10,6 +10,7 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ImmediateEventExecutor;
@@ -30,8 +31,10 @@ public class NettyWebsocketTtyBootstrap {
private int port;
private EventLoopGroup group;
private Channel channel;
private EventExecutorGroup workerGroup;
public NettyWebsocketTtyBootstrap() {
public NettyWebsocketTtyBootstrap(EventExecutorGroup workerGroup) {
this.workerGroup = workerGroup;
this.host = "localhost";
this.port = 8080;
}
@@ -59,7 +62,7 @@ public class NettyWebsocketTtyBootstrap {
ServerBootstrap b = new ServerBootstrap();
b.group(group).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new TtyServerInitializer(channelGroup, handler));
.childHandler(new TtyServerInitializer(channelGroup, handler, workerGroup));
final ChannelFuture f = b.bind(host, port);
f.addListener(new GenericFutureListener<Future<? super Void>>() {
core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/TtyServerInitializer.java
+8 -4
View File
@@ -1,7 +1,5 @@
package com.taobao.arthas.core.shell.term.impl.http;
import java.io.File;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.group.ChannelGroup;
@@ -10,9 +8,12 @@ import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.tty.TtyConnection;
import java.io.File;
/**
* @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
@@ -21,19 +22,22 @@ public class TtyServerInitializer extends ChannelInitializer<SocketChannel> {
private final ChannelGroup group;
private final Consumer<TtyConnection> handler;
private EventExecutorGroup workerGroup;
public TtyServerInitializer(ChannelGroup group, Consumer<TtyConnection> handler) {
public TtyServerInitializer(ChannelGroup group, Consumer<TtyConnection> handler, EventExecutorGroup workerGroup) {
this.group = group;
this.handler = handler;
this.workerGroup = workerGroup;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(64 * 1024));
pipeline.addLast(new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(workerGroup, "HttpRequestHandler", new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
pipeline.addLast(new TtyWebSocketFrameHandler(group, handler));
}
core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/HttpTelnetTermServer.java
+5 -2
View File
@@ -11,6 +11,7 @@ import com.taobao.arthas.core.shell.term.TermServer;
import com.taobao.arthas.core.shell.term.impl.Helper;
import com.taobao.arthas.core.shell.term.impl.TermImpl;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.tty.TtyConnection;
@@ -29,11 +30,13 @@ public class HttpTelnetTermServer extends TermServer {
private String hostIp;
private int port;
private long connectionTimeout;
private EventExecutorGroup workerGroup;
public HttpTelnetTermServer(String hostIp, int port, long connectionTimeout) {
public HttpTelnetTermServer(String hostIp, int port, long connectionTimeout, EventExecutorGroup workerGroup) {
this.hostIp = hostIp;
this.port = port;
this.connectionTimeout = connectionTimeout;
this.workerGroup = workerGroup;
}
@Override
@@ -45,7 +48,7 @@ public class HttpTelnetTermServer extends TermServer {
@Override
public TermServer listen(Handler<Future<TermServer>> listenHandler) {
// TODO: charset and inputrc from options
bootstrap = new NettyHttpTelnetTtyBootstrap().setHost(hostIp).setPort(port);
bootstrap = new NettyHttpTelnetTtyBootstrap(workerGroup).setHost(hostIp).setPort(port);
try {
bootstrap.start(new Consumer<TtyConnection>() {
@Override
core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetBootstrap.java
+5 -2
View File
@@ -11,6 +11,7 @@ import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ImmediateEventExecutor;
@@ -28,8 +29,10 @@ public class NettyHttpTelnetBootstrap extends TelnetBootstrap {
private EventLoopGroup group;
private ChannelGroup channelGroup;
private EventExecutorGroup workerGroup;
public NettyHttpTelnetBootstrap() {
public NettyHttpTelnetBootstrap(EventExecutorGroup workerGroup) {
this.workerGroup = workerGroup;
this.group = new NioEventLoopGroup();
this.channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
}
@@ -56,7 +59,7 @@ public class NettyHttpTelnetBootstrap extends TelnetBootstrap {
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ProtocolDetectHandler(channelGroup, handlerFactory, factory));
ch.pipeline().addLast(new ProtocolDetectHandler(channelGroup, handlerFactory, factory, workerGroup));
}
});
core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/NettyHttpTelnetTtyBootstrap.java
+3 -2
View File
@@ -2,6 +2,7 @@ package com.taobao.arthas.core.shell.term.impl.httptelnet;
import java.nio.charset.Charset;
import io.netty.util.concurrent.EventExecutorGroup;
import io.termd.core.function.Consumer;
import io.termd.core.function.Supplier;
import io.termd.core.telnet.TelnetHandler;
@@ -21,8 +22,8 @@ public class NettyHttpTelnetTtyBootstrap {
private boolean inBinary;
private Charset charset = Charset.forName("UTF-8");
public NettyHttpTelnetTtyBootstrap() {
this.httpTelnetTtyBootstrap = new NettyHttpTelnetBootstrap();
public NettyHttpTelnetTtyBootstrap(EventExecutorGroup workerGroup) {
this.httpTelnetTtyBootstrap = new NettyHttpTelnetBootstrap(workerGroup);
}
public String getHost() {
core/src/main/java/com/taobao/arthas/core/shell/term/impl/httptelnet/ProtocolDetectHandler.java
+6 -3
View File
@@ -4,8 +4,8 @@ import java.io.File;
import java.util.concurrent.TimeUnit;
import com.taobao.arthas.core.shell.term.impl.http.HttpRequestHandler;
import com.taobao.arthas.core.shell.term.impl.http.TtyWebSocketFrameHandler;
import com.taobao.arthas.core.shell.term.impl.http.TtyWebSocketFrameHandler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -15,6 +15,7 @@ import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.ScheduledFuture;
import io.termd.core.function.Consumer;
import io.termd.core.function.Supplier;
@@ -31,12 +32,14 @@ public class ProtocolDetectHandler extends ChannelInboundHandlerAdapter {
private ChannelGroup channelGroup;
private Supplier<TelnetHandler> handlerFactory;
private Consumer<TtyConnection> ttyConnectionFactory;
private EventExecutorGroup workerGroup;
public ProtocolDetectHandler(ChannelGroup channelGroup, final Supplier<TelnetHandler> handlerFactory,
Consumer<TtyConnection> ttyConnectionFactory) {
Consumer<TtyConnection> ttyConnectionFactory, EventExecutorGroup workerGroup) {
this.channelGroup = channelGroup;
this.handlerFactory = handlerFactory;
this.ttyConnectionFactory = ttyConnectionFactory;
this.workerGroup = workerGroup;
}
private ScheduledFuture<?> detectTelnetFuture;
@@ -82,7 +85,7 @@ public class ProtocolDetectHandler extends ChannelInboundHandlerAdapter {
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(64 * 1024));
pipeline.addLast(new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(workerGroup, "HttpRequestHandler", new HttpRequestHandler("/ws", new File("arthas-output")));
pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
pipeline.addLast(new TtyWebSocketFrameHandler(channelGroup, ttyConnectionFactory));
ctx.fireChannelActive();