tunnel server support save agent info to redis; websocket url support config path. #1558

2026-10-21 04:12:40 +00:00 · 2020-10-29 00:50:22 +08:00
parent 20b507cd8c
commit 80b568404f
13 changed files with 374 additions and 36 deletions
@@ -55,6 +55,11 @@
 			<artifactId>spring-boot-starter-actuator</artifactId>
 		</dependency>

+         <dependency>
+             <groupId>org.springframework.boot</groupId>
+             <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
+
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-security</artifactId>
@@ -0,0 +1,95 @@
+package com.alibaba.arthas.tunnel.server;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+/**
+ * 
+ * @author hengyunabc 2020-10-27
+ *
+ */
+public class RedisTunnelClusterStore implements TunnelClusterStore {
+
+    private String prefix = "arthas-tunnel-agent-";
+
+    private StringRedisTemplate redisTemplate;
+
+    @Override
+    public String findHost(String agentId) {
+        ValueOperations<String, String> opsForValue = redisTemplate.opsForValue();
+
+        return opsForValue.get(prefix + agentId);
+    }
+
+    @Override
+    public void removeAgent(String agentId) {
+        ValueOperations<String, String> opsForValue = redisTemplate.opsForValue();
+        opsForValue.getOperations().delete(prefix + agentId);
+    }
+
+    @Override
+    public void addHost(String agentId, String host, long timeout, TimeUnit timeUnit) {
+        ValueOperations<String, String> opsForValue = redisTemplate.opsForValue();
+
+        opsForValue.set(prefix + agentId, host, timeout, timeUnit);
+    }
+
+    public StringRedisTemplate getRedisTemplate() {
+        return redisTemplate;
+    }
+
+    public void setRedisTemplate(StringRedisTemplate redisTemplate) {
+        this.redisTemplate = redisTemplate;
+    }
+
+    @Override
+    public Collection<String> allAgentIds() {
+        ValueOperations<String, String> opsForValue = redisTemplate.opsForValue();
+
+        Set<String> result = new HashSet<String>();
+
+        int length = prefix.length();
+        for (String value : opsForValue.getOperations().keys(prefix + "*")) {
+            result.add(value.substring(length));
+
+        }
+        return result;
+    }
+
+    @Override
+    public Collection<Pair<String, String>> agentInfo(String appName) {
+        ValueOperations<String, String> opsForValue = redisTemplate.opsForValue();
+
+        Set<String> keys = new HashSet<String>();
+
+        String prefixWithAppName = prefix + appName + "|";
+        
+        for (String value : opsForValue.getOperations().keys(prefixWithAppName + "*")) {
+            keys.add(value);
+
+        }
+
+        List<String> values = opsForValue.getOperations().opsForValue().multiGet(keys);
+
+        Collection<Pair<String, String>> result = new HashSet<>();
+
+        Iterator<String> iterator = values.iterator();
+
+        for (String key : keys) {
+            String host = iterator.next();
+            String agentId = key.substring(prefix.length());
+            result.add(Pair.of(agentId, host));
+        }
+
+        return result;
+    }
+
+}
@@ -0,0 +1,24 @@
+package com.alibaba.arthas.tunnel.server;
+
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * 保存agentId连接到哪个具体的 tunnel server，集群部署时使用
+ * 
+ * @author hengyunabc 2020-10-27
+ *
+ */
+public interface TunnelClusterStore {
+    public void addHost(String agentId, String host, long expire, TimeUnit timeUnit);
+
+    public String findHost(String agentId);
+
+    public void removeAgent(String agentId);
+
+    public Collection<String> allAgentIds();
+
+    public Collection<Pair<String, String>> agentInfo(String appName);
+}
@@ -1,6 +1,7 @@
 package com.alibaba.arthas.tunnel.server;

 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -10,6 +11,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.alibaba.arthas.tunnel.common.SimpleHttpResponse;
+import com.alibaba.arthas.tunnel.server.utils.InetAddressUtil;

 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
@@ -31,10 +33,12 @@ import io.netty.util.concurrent.Promise;
 */
 public class TunnelServer {
    private final static Logger logger = LoggerFactory.getLogger(TunnelServer.class);
+    public static final String DEFAULT_WEBSOCKET_PATH = "/ws";

    private boolean ssl;
    private String host;
    private int port;
+    private String path = DEFAULT_WEBSOCKET_PATH;

    private Map<String, AgentInfo> agentInfoMap = new ConcurrentHashMap<String, AgentInfo>();

@@ -50,6 +54,16 @@ public class TunnelServer {

    private Channel channel;

+    /**
+     * 在集群部署时，保存agentId和host关系
+     */
+    private TunnelClusterStore tunnelClusterStore;
+    
+    /**
+     * 集群部署时外部连接的host
+     */
+    private String clientConnectHost;
+
    public void start() throws Exception {
        // Configure SSL.
        final SslContext sslCtx;
@@ -78,6 +92,17 @@ public class TunnelServer {
                agentInfoMap.entrySet().removeIf(e -> !e.getValue().getChannelHandlerContext().channel().isActive());
                clientConnectionInfoMap.entrySet()
                        .removeIf(e -> !e.getValue().getChannelHandlerContext().channel().isActive());
+                
+                // 更新集群key信息
+                if (tunnelClusterStore != null && clientConnectHost != null) {
+                    try {
+                        for (Entry<String, AgentInfo> entry : agentInfoMap.entrySet()) {
+                            tunnelClusterStore.addHost(entry.getKey(), clientConnectHost, 60 * 60, TimeUnit.SECONDS);
+                        }
+                    } catch (Throwable t) {
+                        logger.error("update tunnel info error", t);
+                    }
+                }
            }

        }, 60, 60, TimeUnit.SECONDS);
@@ -97,10 +122,17 @@ public class TunnelServer {

    public void addAgent(String id, AgentInfo agentInfo) {
        agentInfoMap.put(id, agentInfo);
+        if (this.tunnelClusterStore != null) {
+            this.tunnelClusterStore.addHost(id, clientConnectHost, 60 * 60, TimeUnit.SECONDS);
+        }
    }

    public AgentInfo removeAgent(String id) {
-        return agentInfoMap.remove(id);
+        AgentInfo agentInfo = agentInfoMap.remove(id);
+        if (this.tunnelClusterStore != null) {
+            this.tunnelClusterStore.removeAgent(id);
+        }
+        return agentInfo;
    }
    
    public Optional<ClientConnectionInfo> findClientConnection(String id) {
@@ -175,4 +207,28 @@ public class TunnelServer {
    public void setClientConnectionInfoMap(Map<String, ClientConnectionInfo> clientConnectionInfoMap) {
        this.clientConnectionInfoMap = clientConnectionInfoMap;
    }
+
+    public TunnelClusterStore getTunnelClusterStore() {
+        return tunnelClusterStore;
+    }
+
+    public void setTunnelClusterStore(TunnelClusterStore tunnelClusterStore) {
+        this.tunnelClusterStore = tunnelClusterStore;
+    }
+
+    public String getClientConnectHost() {
+        return clientConnectHost;
+    }
+
+    public void setClientConnectHost(String clientConnectHost) {
+        this.clientConnectHost = clientConnectHost;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public void setPath(String path) {
+        this.path = path;
+    }
 }
@@ -18,8 +18,6 @@ import io.netty.handler.ssl.SslContext;
 */
 public class TunnelSocketServerInitializer extends ChannelInitializer<SocketChannel> {

-    private static final String WEBSOCKET_PATH = "/ws";
-
    private final SslContext sslCtx;

    private TunnelServer tunnelServer;
@@ -38,7 +36,7 @@ public class TunnelSocketServerInitializer extends ChannelInitializer<SocketChan
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new WebSocketServerCompressionHandler());
-        pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true, ArthasConstants.MAX_HTTP_CONTENT_LENGTH, false, true, 10000L));
+        pipeline.addLast(new WebSocketServerProtocolHandler(tunnelServer.getPath(), null, true, ArthasConstants.MAX_HTTP_CONTENT_LENGTH, false, true, 10000L));

        pipeline.addLast(new TunnelSocketFrameHandler(tunnelServer));
    }
@@ -1,25 +0,0 @@
-package com.alibaba.arthas.tunnel.server.app;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-import com.alibaba.arthas.tunnel.server.TunnelServer;
-
-@Configuration
-public class TunnelServerConfiguration {
-
-    @Autowired
-    ArthasProperties arthasProperties;
-
-    @Bean(initMethod = "start", destroyMethod = "stop")
-    public TunnelServer tunnelServer() {
-        TunnelServer tunnelServer = new TunnelServer();
-
-        tunnelServer.setHost(arthasProperties.getServer().getHost());
-        tunnelServer.setPort(arthasProperties.getServer().getPort());
-        tunnelServer.setSsl(arthasProperties.getServer().isSsl());
-        return tunnelServer;
-    }
-
-}
@@ -1,8 +1,11 @@
-package com.alibaba.arthas.tunnel.server.app;
+package com.alibaba.arthas.tunnel.server.app.configuration;

 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Component;

+import com.alibaba.arthas.tunnel.server.TunnelServer;
+import com.alibaba.arthas.tunnel.server.utils.InetAddressUtil;
+
 /**
 * 
 * @author hengyunabc 2019-08-29
@@ -23,9 +26,18 @@ public class ArthasProperties {
    }

    public static class Server {
+        /**
+         * tunnel server listen host
+         */
        private String host;
        private int port;
        private boolean ssl;
+        private String path = TunnelServer.DEFAULT_WEBSOCKET_PATH;
+
+        /**
+         * 客户端连接的地址。也用于保存到redis里，当部署tunnel server集群里需要。不配置则会自动获取
+         */
+        private String clientConnectHost = InetAddressUtil.getInetAddress();

        public String getHost() {
            return host;
@@ -51,6 +63,22 @@ public class ArthasProperties {
            this.ssl = ssl;
        }

+        public String getClientConnectHost() {
+            return clientConnectHost;
+        }
+
+        public void setClientConnectHost(String clientConnectHost) {
+            this.clientConnectHost = clientConnectHost;
+        }
+
+        public String getPath() {
+            return path;
+        }
+
+        public void setPath(String path) {
+            this.path = path;
+        }
+
    }

 }
@@ -0,0 +1,57 @@
+package com.alibaba.arthas.tunnel.server.app.configuration;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import com.alibaba.arthas.tunnel.server.RedisTunnelClusterStore;
+import com.alibaba.arthas.tunnel.server.TunnelClusterStore;
+import com.alibaba.arthas.tunnel.server.TunnelServer;
+
+/**
+ * 
+ * @author hengyunabc 2020-10-27
+ *
+ */
+@Configuration
+@AutoConfigureAfter(RedisAutoConfiguration.class)
+public class TunnelServerConfiguration {
+
+    @Autowired
+    ArthasProperties arthasProperties;
+
+    @Bean
+//    @ConditionalOnBean(StringRedisTemplate.class)
+    @ConditionalOnClass(StringRedisTemplate.class)
+    @ConditionalOnProperty("spring.redis.host")
+    @ConditionalOnMissingBean
+    public TunnelClusterStore tunnelClusterStore(@Autowired StringRedisTemplate redisTemplate) {
+        RedisTunnelClusterStore redisTunnelClusterStore = new RedisTunnelClusterStore();
+        redisTunnelClusterStore.setRedisTemplate(redisTemplate);
+        return redisTunnelClusterStore;
+    }
+
+    @Bean(initMethod = "start", destroyMethod = "stop")
+    @ConditionalOnMissingBean
+    public TunnelServer tunnelServer(@Autowired(required = false) TunnelClusterStore tunnelClusterStore) {
+        TunnelServer tunnelServer = new TunnelServer();
+
+        tunnelServer.setHost(arthasProperties.getServer().getHost());
+        tunnelServer.setPort(arthasProperties.getServer().getPort());
+        tunnelServer.setSsl(arthasProperties.getServer().isSsl());
+        tunnelServer.setPath(arthasProperties.getServer().getPath());
+        tunnelServer.setClientConnectHost(arthasProperties.getServer().getClientConnectHost());
+        if (tunnelClusterStore != null) {
+            tunnelServer.setTunnelClusterStore(tunnelClusterStore);
+        }
+        return tunnelServer;
+    }
+
+}
@@ -0,0 +1,44 @@
+package com.alibaba.arthas.tunnel.server.app.web;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+
+import com.alibaba.arthas.tunnel.server.TunnelClusterStore;
+import com.alibaba.arthas.tunnel.server.TunnelServer;
+
+/**
+ * 
+ * @author hengyunabc 2020-10-27
+ *
+ */
+@Controller
+public class ClusterController {
+    private final static Logger logger = LoggerFactory.getLogger(ClusterController.class);
+
+    @Autowired
+    TunnelServer tunnelServer;
+
+    @RequestMapping(value = "/api/cluster/findHost")
+    @ResponseBody
+    public String execute(@RequestParam(value = "agentId", required = true) String agentId) {
+        TunnelClusterStore tunnelClusterStore = tunnelServer.getTunnelClusterStore();
+
+        String host = null;
+        if (tunnelClusterStore != null) {
+            host = tunnelClusterStore.findHost(agentId);
+        }
+
+        if (host == null) {
+            host = "";
+        }
+
+        logger.info("arthas cluster findHost, agentId: {}, host: {}", agentId, host);
+
+        return host;
+    }
+}
@@ -6,7 +6,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;

-import com.alibaba.arthas.tunnel.server.app.ArthasProperties;
+import com.alibaba.arthas.tunnel.server.app.configuration.ArthasProperties;

@EnableConfigurationProperties(ArthasProperties.class)
@Configuration
@@ -8,7 +8,7 @@ import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
 import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;

 import com.alibaba.arthas.tunnel.server.TunnelServer;
-import com.alibaba.arthas.tunnel.server.app.ArthasProperties;
+import com.alibaba.arthas.tunnel.server.app.configuration.ArthasProperties;

@Endpoint(id = "arthas")
 public class ArthasEndpoint {
@@ -0,0 +1,51 @@
+package com.alibaba.arthas.tunnel.server.utils;
+
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.Enumeration;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 
+ * @author hengyunabc 2020-10-27
+ *
+ */
+public class InetAddressUtil {
+    private final static Logger logger = LoggerFactory.getLogger(InetAddressUtil.class);
+
+    /**
+     * 获得本机IP。
+     * <p>
+     * 在超过一块网卡时会有问题，因为这里每次都只是取了第一块网卡绑定的IP地址
+     */
+    public static String getInetAddress() {
+        try {
+            Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
+            InetAddress address = null;
+            while (interfaces.hasMoreElements()) {
+                NetworkInterface ni = interfaces.nextElement();
+                Enumeration<InetAddress> addresses = ni.getInetAddresses();
+                while (addresses.hasMoreElements()) {
+                    address = addresses.nextElement();
+                    if (isValidAddress(address)) {
+                        return address.getHostAddress();
+                    }
+                }
+            }
+            logger.warn("Can not get the server IP address");
+            return null;
+        } catch (Throwable t) {
+            logger.error("Can not get the server IP address", t);
+            return null;
+        }
+    }
+
+    public static boolean isValidAddress(InetAddress address) {
+        return address != null && !address.isLoopbackAddress() // filter 127.x.x.x
+                && !address.isAnyLocalAddress() // filter 0.0.0.0
+                && !address.isLinkLocalAddress() // filter 169.254.0.0/16
+                && !address.getHostAddress().contains(":");// filter IPv6
+    }
+}
@@ -77,10 +77,10 @@ function getTerminalSize () {
 }

 /** init websocket **/
-function initWs (ip, port, agentId) {
+function initWs (ip, port, path, agentId) {
    var protocol= location.protocol === 'https:'  ? 'wss://' : 'ws://';
-    var path = protocol + ip + ':' + port + '/ws?method=connectArthas&id=' + agentId;
-    ws = new WebSocket(path);
+    var uri = protocol + ip + ':' + port + '/' + path + '?method=connectArthas&id=' + agentId;
+    ws = new WebSocket(uri);
 }

 /** init xterm **/
@@ -114,8 +114,13 @@ function startConnect (silent) {
        alert('Already connected');
        return;
    }
+    
+    var path = getUrlParam('path');
+    if (path == null) {
+        path = "ws";
+    }
    // init webSocket
-    initWs(ip, port, agentId);
+    initWs(ip, port, path, agentId);
    ws.onerror = function () {
        ws.close();
        ws = null;