support listen at netty LocalAddress, and tunnel client connect to LocalAddress. #1467

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";
+}

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());
             }
 

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<LocalChannel> {
+
+    private final ChannelGroup group;
+    private final Consumer<TtyConnection> handler;
+    private EventExecutorGroup workerGroup;
+
+    public LocalTtyServerInitializer(ChannelGroup group, Consumer<TtyConnection> 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));
+    }
+}

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<TtyConnection> handler, final Consumer<Throwable> 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<Future<? super Void>>() {
-            @Override
-            public void operationComplete(Future<? super Void> 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<Future<? super Void>>() {
+                @Override
+                public void operationComplete(Future<? super Void> 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<Future<? super Void>>() {
+                @Override
+                public void operationComplete(Future<? super Void> future) throws Exception {
+                    if (future.isSuccess()) {
+                        doneHandler.accept(null);
+                    } else {
+                        doneHandler.accept(future.cause());
+                    }
+                }
+            });
+        }
     }
 
     public CompletableFuture<Void> start(Consumer<TtyConnection> handler) {

tunnel-client/pom.xml

@@ -11,6 +11,12 @@
 	<name>arthas-tunnel-client</name>
 
 	<dependencies>
+        <dependency>
+            <groupId>com.taobao.arthas</groupId>
+            <artifactId>arthas-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
 		<dependency>
 			<groupId>org.slf4j</groupId>
 			<artifactId>slf4j-api</artifactId>

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;

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<SocketChannel>() {
+            b.group(group).channel(LocalChannel.class)
+                    .handler(new ChannelInitializer<LocalChannel>() {
                         @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<ChannelFuture>() {
                         @Override

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;
     }

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);
                 }
             }