diff --git a/common/src/main/java/com/taobao/arthas/common/ArthasConstants.java b/common/src/main/java/com/taobao/arthas/common/ArthasConstants.java new file mode 100644 index 000000000..2b0fc144f --- /dev/null +++ b/common/src/main/java/com/taobao/arthas/common/ArthasConstants.java @@ -0,0 +1,16 @@ +package com.taobao.arthas.common; + +/** + * + * @author hengyunabc 2020-09-02 + * + */ +public class ArthasConstants { + /** + * local address in VM communication + * + * @see io.netty.channel.local.LocalAddress + * @see io.netty.channel.local.LocalChannel + */ + public static final String NETTY_LOCAL_ADDRESS = "arthas-netty-LocalAddress"; +} diff --git a/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java b/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java index 9339981ed..863251d48 100644 --- a/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java +++ b/core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java @@ -62,6 +62,7 @@ import io.netty.util.concurrent.EventExecutorGroup; /** * @author vlinux on 15/5/2. + * @author hengyunabc */ public class ArthasBootstrap { private static final String ARTHAS_SPY_JAR = "arthas-spy.jar"; @@ -289,17 +290,10 @@ public class ArthasBootstrap { String agentId = null; try { - if (configure.getTunnelServer() != null && configure.getHttpPort() > 0) { + if (configure.getTunnelServer() != null) { tunnelClient = new TunnelClient(); tunnelClient.setId(configure.getAgentId()); tunnelClient.setTunnelServerUrl(configure.getTunnelServer()); - // ws://127.0.0.1:8563/ws - String host = "127.0.0.1"; - 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()) { @@ -341,6 +335,11 @@ public class ArthasBootstrap { shellServer.registerTermServer(new HttpTermServer(configure.getIp(), configure.getHttpPort(), options.getConnectionTimeout(), workerGroup)); } else { + // listen local address in VM communication + if (configure.getTunnelServer() != null) { + shellServer.registerTermServer(new HttpTermServer(configure.getIp(), configure.getHttpPort(), + options.getConnectionTimeout(), workerGroup)); + } logger().info("http port is {}, skip bind http server.", configure.getHttpPort()); } diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/LocalTtyServerInitializer.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/LocalTtyServerInitializer.java new file mode 100644 index 000000000..3ba279fb1 --- /dev/null +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/LocalTtyServerInitializer.java @@ -0,0 +1,46 @@ +package com.taobao.arthas.core.shell.term.impl.http; + +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelPipeline; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.local.LocalChannel; +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 hengyunabc 2020-09-02 + * + */ +public class LocalTtyServerInitializer extends ChannelInitializer { + + private final ChannelGroup group; + private final Consumer handler; + private EventExecutorGroup workerGroup; + + public LocalTtyServerInitializer(ChannelGroup group, Consumer handler, + EventExecutorGroup workerGroup) { + this.group = group; + this.handler = handler; + this.workerGroup = workerGroup; + } + + @Override + protected void initChannel(LocalChannel ch) throws Exception { + + ChannelPipeline pipeline = ch.pipeline(); + pipeline.addLast(new HttpServerCodec()); + pipeline.addLast(new ChunkedWriteHandler()); + pipeline.addLast(new HttpObjectAggregator(64 * 1024)); + pipeline.addLast(workerGroup, "HttpRequestHandler", new HttpRequestHandler("/ws", new File("arthas-output"))); + pipeline.addLast(new WebSocketServerProtocolHandler("/ws")); + pipeline.addLast(new TtyWebSocketFrameHandler(group, handler)); + } +} diff --git a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java index de6ee9ac8..15503b979 100644 --- a/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java +++ b/core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/NettyWebsocketTtyBootstrap.java @@ -1,11 +1,15 @@ package com.taobao.arthas.core.shell.term.impl.http; +import com.taobao.arthas.common.ArthasConstants; + import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.EventLoopGroup; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; +import io.netty.channel.local.LocalAddress; +import io.netty.channel.local.LocalServerChannel; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; @@ -61,22 +65,43 @@ public class NettyWebsocketTtyBootstrap { public void start(Consumer handler, final Consumer doneHandler) { group = new NioEventLoopGroup(new DefaultThreadFactory("arthas-NettyWebsocketTtyBootstrap", true)); - ServerBootstrap b = new ServerBootstrap(); - b.group(group).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)) - .childHandler(new TtyServerInitializer(channelGroup, handler, workerGroup)); + if (this.port > 0) { + ServerBootstrap b = new ServerBootstrap(); + b.group(group).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)) + .childHandler(new TtyServerInitializer(channelGroup, handler, workerGroup)); - final ChannelFuture f = b.bind(host, port); - f.addListener(new GenericFutureListener>() { - @Override - public void operationComplete(Future future) throws Exception { - if (future.isSuccess()) { - channel = f.channel(); - doneHandler.accept(null); - } else { - doneHandler.accept(future.cause()); + final ChannelFuture f = b.bind(host, port); + f.addListener(new GenericFutureListener>() { + @Override + public void operationComplete(Future future) throws Exception { + if (future.isSuccess()) { + channel = f.channel(); + doneHandler.accept(null); + } else { + doneHandler.accept(future.cause()); + } } - } - }); + }); + } + + // listen local address in VM communication + ServerBootstrap b2 = new ServerBootstrap(); + b2.group(group).channel(LocalServerChannel.class).handler(new LoggingHandler(LogLevel.INFO)) + .childHandler(new LocalTtyServerInitializer(channelGroup, handler, workerGroup)); + + ChannelFuture bindLocalFuture = b2.bind(new LocalAddress(ArthasConstants.NETTY_LOCAL_ADDRESS)); + if (this.port < 0) { // 保证回调doneHandler + bindLocalFuture.addListener(new GenericFutureListener>() { + @Override + public void operationComplete(Future future) throws Exception { + if (future.isSuccess()) { + doneHandler.accept(null); + } else { + doneHandler.accept(future.cause()); + } + } + }); + } } public CompletableFuture start(Consumer handler) { diff --git a/tunnel-client/pom.xml b/tunnel-client/pom.xml index 3bd73d4e1..2e343d399 100644 --- a/tunnel-client/pom.xml +++ b/tunnel-client/pom.xml @@ -11,6 +11,12 @@ arthas-tunnel-client + + com.taobao.arthas + arthas-common + ${project.version} + + org.slf4j slf4j-api diff --git a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClient.java b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClient.java index d09541f97..89eaa22ae 100644 --- a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClient.java +++ b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClient.java @@ -38,11 +38,9 @@ import io.netty.util.concurrent.DefaultThreadFactory; public class ForwardClient { private final static Logger logger = LoggerFactory.getLogger(ForwardClient.class); private URI tunnelServerURI; - private URI localServerURI; - public ForwardClient(URI tunnelServerURI, URI localServerURI) { + public ForwardClient(URI tunnelServerURI) { this.tunnelServerURI = tunnelServerURI; - this.localServerURI = localServerURI; } public void start() throws URISyntaxException, SSLException, InterruptedException { @@ -79,8 +77,7 @@ public class ForwardClient { WebSocketVersion.V13, null, true, new DefaultHttpHeaders()); final WebSocketClientProtocolHandler websocketClientHandler = new WebSocketClientProtocolHandler(newHandshaker); - final ForwardClientSocketClientHandler forwardClientSocketClientHandler = new ForwardClientSocketClientHandler( - localServerURI); + final ForwardClientSocketClientHandler forwardClientSocketClientHandler = new ForwardClientSocketClientHandler(); final EventLoopGroup group = new NioEventLoopGroup(1, new DefaultThreadFactory("arthas-ForwardClient", true)); ChannelFuture closeFuture = null; diff --git a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClientSocketClientHandler.java b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClientSocketClientHandler.java index 9abe8478c..ca488a2ce 100644 --- a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClientSocketClientHandler.java +++ b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/ForwardClientSocketClientHandler.java @@ -1,10 +1,13 @@ package com.alibaba.arthas.tunnel.client; import java.net.URI; +import java.net.URISyntaxException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.taobao.arthas.common.ArthasConstants; + import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; @@ -16,9 +19,9 @@ import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPromise; import io.netty.channel.EventLoopGroup; import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.local.LocalAddress; +import io.netty.channel.local.LocalChannel; import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.SocketChannel; -import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpObjectAggregator; @@ -39,11 +42,6 @@ public class ForwardClientSocketClientHandler extends SimpleChannelInboundHandle private static final Logger logger = LoggerFactory.getLogger(ForwardClientSocketClientHandler.class); private ChannelPromise handshakeFuture; - private final URI localServerURI; - - public ForwardClientSocketClientHandler(URI localServerURI) { - this.localServerURI = localServerURI; - } @Override public void channelActive(ChannelHandlerContext ctx) { @@ -67,12 +65,13 @@ public class ForwardClientSocketClientHandler extends SimpleChannelInboundHandle } } - private void connectLocalServer(final ChannelHandlerContext ctx) throws InterruptedException { + private void connectLocalServer(final ChannelHandlerContext ctx) throws InterruptedException, URISyntaxException { final EventLoopGroup group = new NioEventLoopGroup(1, new DefaultThreadFactory("arthas-forward-client-connect-local", true)); ChannelFuture closeFuture = null; try { logger.info("ForwardClientSocketClientHandler star connect local arthas server"); - WebSocketClientHandshaker newHandshaker = WebSocketClientHandshakerFactory.newHandshaker(localServerURI, + // 入参URI实际无意义，只为了程序不出错 + WebSocketClientHandshaker newHandshaker = WebSocketClientHandshakerFactory.newHandshaker(new URI("ws://127.0.0.1:8563/ws"), WebSocketVersion.V13, null, true, new DefaultHttpHeaders()); final WebSocketClientProtocolHandler websocketClientHandler = new WebSocketClientProtocolHandler( newHandshaker); @@ -80,17 +79,19 @@ public class ForwardClientSocketClientHandler extends SimpleChannelInboundHandle Bootstrap b = new Bootstrap(); b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000); - b.group(group).channel(NioSocketChannel.class) - .handler(new ChannelInitializer() { + b.group(group).channel(LocalChannel.class) + .handler(new ChannelInitializer() { @Override - protected void initChannel(SocketChannel ch) { + protected void initChannel(LocalChannel ch) { ChannelPipeline p = ch.pipeline(); p.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192), websocketClientHandler, localFrameHandler); } }); - Channel localChannel = b.connect(localServerURI.getHost(), localServerURI.getPort()).sync().channel(); + LocalAddress localAddress = new LocalAddress(ArthasConstants.NETTY_LOCAL_ADDRESS); + Channel localChannel = b.connect(localAddress).sync().channel(); + // Channel localChannel = b.connect(localServerURI.getHost(), localServerURI.getPort()).sync().channel(); this.handshakeFuture = localFrameHandler.handshakeFuture(); handshakeFuture.addListener(new GenericFutureListener() { @Override diff --git a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClient.java b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClient.java index 5b03d1c13..e6b22636f 100644 --- a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClient.java +++ b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClient.java @@ -10,7 +10,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import io.netty.bootstrap.Bootstrap; -import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelInitializer; @@ -43,8 +42,6 @@ public class TunnelClient { private String tunnelServerUrl; - private String localServerUrl = "ws://127.0.0.1:8563/ws"; - private int reconnectDelay = 5; // connect to proxy server @@ -148,14 +145,6 @@ public class TunnelClient { this.tunnelServerUrl = tunnelServerUrl; } - public String getLocalServerUrl() { - return localServerUrl; - } - - public void setLocalServerUrl(String localServerUrl) { - this.localServerUrl = localServerUrl; - } - public int getReconnectDelay() { return reconnectDelay; } diff --git a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClientSocketClientHandler.java b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClientSocketClientHandler.java index b4f19c5d5..11179ac0f 100644 --- a/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClientSocketClientHandler.java +++ b/tunnel-client/src/main/java/com/alibaba/arthas/tunnel/client/TunnelClientSocketClientHandler.java @@ -2,6 +2,7 @@ package com.alibaba.arthas.tunnel.client; import java.net.URI; +import java.net.URL; import java.util.List; import java.util.Map; import java.util.concurrent.Executors; @@ -80,12 +81,10 @@ public class TunnelClientSocketClientHandler extends SimpleChannelInboundHandler logger.info("start ForwardClient, uri: {}", forwardUri); try { - ForwardClient forwardClient = new ForwardClient(forwardUri, - new URI(tunnelClient.getLocalServerUrl())); + ForwardClient forwardClient = new ForwardClient(forwardUri); forwardClient.start(); } catch (Throwable e) { - logger.error("start ForwardClient error, forwardUri: {}, localServerUri: {}", forwardUri, - tunnelClient.getLocalServerUrl(), e); + logger.error("start ForwardClient error, forwardUri: {}", forwardUri, e); } }