diff --git a/channel/channel-server-app/pom.xml b/channel/channel-server-app/pom.xml new file mode 100644 index 000000000..3472104c4 --- /dev/null +++ b/channel/channel-server-app/pom.xml @@ -0,0 +1,76 @@ + + + + arthas-all + com.taobao.arthas + 3.3.10-SNAPSHOT + ../../pom.xml + + 4.0.0 + + arthas-channel-server-app + arthas-channel-server-app + + + 1.8 + 1.8 + + + + + + + org.springframework.boot + spring-boot-dependencies + ${spring-boot.version} + pom + import + + + + + + + org.springframework.boot + spring-boot-starter-webflux + + + org.springframework.boot + spring-boot-starter-web + + + com.taobao.arthas + arthas-channel-server-starter + ${project.version} + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + true + + + + + org.springframework.boot + spring-boot-maven-plugin + 2.1.7.RELEASE + + + package + + repackage + + + + + + + + \ No newline at end of file diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/ArthasChannelApplication.java b/channel/channel-server-app/src/main/java/com/alibaba/arthas/channel/server/ArthasChannelApplication.java similarity index 95% rename from channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/ArthasChannelApplication.java rename to channel/channel-server-app/src/main/java/com/alibaba/arthas/channel/server/ArthasChannelApplication.java index 4c8e93119..0669970ba 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/ArthasChannelApplication.java +++ b/channel/channel-server-app/src/main/java/com/alibaba/arthas/channel/server/ArthasChannelApplication.java @@ -4,7 +4,6 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableScheduling; -@EnableScheduling @SpringBootApplication public class ArthasChannelApplication { diff --git a/channel/channel-server/src/main/resources/application-memory.properties b/channel/channel-server-app/src/main/resources/application-memory.properties similarity index 100% rename from channel/channel-server/src/main/resources/application-memory.properties rename to channel/channel-server-app/src/main/resources/application-memory.properties diff --git a/channel/channel-server/src/main/resources/application-redis.properties b/channel/channel-server-app/src/main/resources/application-redis.properties similarity index 100% rename from channel/channel-server/src/main/resources/application-redis.properties rename to channel/channel-server-app/src/main/resources/application-redis.properties diff --git a/channel/channel-server/src/main/resources/application.properties b/channel/channel-server-app/src/main/resources/application.properties similarity index 100% rename from channel/channel-server/src/main/resources/application.properties rename to channel/channel-server-app/src/main/resources/application.properties diff --git a/channel/channel-server/src/main/resources/static/bootstrap-4.2.1.min.css b/channel/channel-server-app/src/main/resources/static/bootstrap-4.2.1.min.css similarity index 100% rename from channel/channel-server/src/main/resources/static/bootstrap-4.2.1.min.css rename to channel/channel-server-app/src/main/resources/static/bootstrap-4.2.1.min.css diff --git a/channel/channel-server/src/main/resources/static/bootstrap-4.2.1.min.js b/channel/channel-server-app/src/main/resources/static/bootstrap-4.2.1.min.js similarity index 100% rename from channel/channel-server/src/main/resources/static/bootstrap-4.2.1.min.js rename to channel/channel-server-app/src/main/resources/static/bootstrap-4.2.1.min.js diff --git a/channel/channel-server/src/main/resources/static/fullsc.png b/channel/channel-server-app/src/main/resources/static/fullsc.png similarity index 100% rename from channel/channel-server/src/main/resources/static/fullsc.png rename to channel/channel-server-app/src/main/resources/static/fullsc.png diff --git a/channel/channel-server/src/main/resources/static/index.html b/channel/channel-server-app/src/main/resources/static/index.html similarity index 100% rename from channel/channel-server/src/main/resources/static/index.html rename to channel/channel-server-app/src/main/resources/static/index.html diff --git a/channel/channel-server/src/main/resources/static/jquery-3.3.1.slim.min.js b/channel/channel-server-app/src/main/resources/static/jquery-3.3.1.slim.min.js similarity index 100% rename from channel/channel-server/src/main/resources/static/jquery-3.3.1.slim.min.js rename to channel/channel-server-app/src/main/resources/static/jquery-3.3.1.slim.min.js diff --git a/channel/channel-server/src/main/resources/static/logo.png b/channel/channel-server-app/src/main/resources/static/logo.png similarity index 100% rename from channel/channel-server/src/main/resources/static/logo.png rename to channel/channel-server-app/src/main/resources/static/logo.png diff --git a/channel/channel-server/src/main/resources/static/main.css b/channel/channel-server-app/src/main/resources/static/main.css similarity index 100% rename from channel/channel-server/src/main/resources/static/main.css rename to channel/channel-server-app/src/main/resources/static/main.css diff --git a/channel/channel-server/src/main/resources/static/popper-1.14.6.min.js b/channel/channel-server-app/src/main/resources/static/popper-1.14.6.min.js similarity index 100% rename from channel/channel-server/src/main/resources/static/popper-1.14.6.min.js rename to channel/channel-server-app/src/main/resources/static/popper-1.14.6.min.js diff --git a/channel/channel-server/src/main/resources/static/web-console.js b/channel/channel-server-app/src/main/resources/static/web-console.js similarity index 100% rename from channel/channel-server/src/main/resources/static/web-console.js rename to channel/channel-server-app/src/main/resources/static/web-console.js diff --git a/channel/channel-server/src/main/resources/static/xterm.css b/channel/channel-server-app/src/main/resources/static/xterm.css similarity index 100% rename from channel/channel-server/src/main/resources/static/xterm.css rename to channel/channel-server-app/src/main/resources/static/xterm.css diff --git a/channel/channel-server/src/main/resources/static/xterm.js b/channel/channel-server-app/src/main/resources/static/xterm.js similarity index 100% rename from channel/channel-server/src/main/resources/static/xterm.js rename to channel/channel-server-app/src/main/resources/static/xterm.js diff --git a/channel/channel-server-starter/pom.xml b/channel/channel-server-starter/pom.xml new file mode 100644 index 000000000..e003b76a6 --- /dev/null +++ b/channel/channel-server-starter/pom.xml @@ -0,0 +1,51 @@ + + + + arthas-all + com.taobao.arthas + 3.3.10-SNAPSHOT + ../../pom.xml + + 4.0.0 + + arthas-channel-server-starter + arthas-channel-server-starter + + + 1.8 + 1.8 + + + + + + + org.springframework.boot + spring-boot-dependencies + ${spring-boot.version} + pom + import + + + + + + + org.springframework.boot + spring-boot-autoconfigure + + + org.springframework.boot + spring-boot-configuration-processor + true + + + com.taobao.arthas + arthas-channel-server + ${project.version} + + + + \ No newline at end of file diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/configuration/ChannelServerConfiguration.java b/channel/channel-server-starter/src/main/java/com/alibaba/arthas/channel/server/autoconfigure/ChannelServerAutoConfiguration.java similarity index 71% rename from channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/configuration/ChannelServerConfiguration.java rename to channel/channel-server-starter/src/main/java/com/alibaba/arthas/channel/server/autoconfigure/ChannelServerAutoConfiguration.java index c378f0007..461b8d555 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/configuration/ChannelServerConfiguration.java +++ b/channel/channel-server-starter/src/main/java/com/alibaba/arthas/channel/server/autoconfigure/ChannelServerAutoConfiguration.java @@ -1,5 +1,6 @@ -package com.alibaba.arthas.channel.server.configuration; +package com.alibaba.arthas.channel.server.autoconfigure; +import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig; import com.alibaba.arthas.channel.server.grpc.ArthasServiceGrpcImpl; import com.alibaba.arthas.channel.server.grpc.ChannelServer; import com.alibaba.arthas.channel.server.message.MessageExchangeService; @@ -7,13 +8,15 @@ import com.alibaba.arthas.channel.server.message.impl.MessageExchangeServiceImpl import com.alibaba.arthas.channel.server.redis.RedisAgentManageServiceImpl; import com.alibaba.arthas.channel.server.redis.RedisMessageExchangeServiceImpl; import com.alibaba.arthas.channel.server.service.AgentBizSerivce; +import com.alibaba.arthas.channel.server.service.AgentCleaner; import com.alibaba.arthas.channel.server.service.AgentManageService; import com.alibaba.arthas.channel.server.service.ApiActionDelegateService; import com.alibaba.arthas.channel.server.service.impl.AgentBizServiceImpl; import com.alibaba.arthas.channel.server.service.impl.AgentManageServiceImpl; import com.alibaba.arthas.channel.server.service.impl.ApiActionDelegateServiceImpl; import com.alibaba.arthas.channel.server.ws.WebSocketServer; -import org.springframework.beans.factory.annotation.Value; +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.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -29,16 +32,16 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadFactory; -/** - * @author gongdewei 2020/8/14 - */ + @Configuration +@ConditionalOnClass(ChannelServer.class) @EnableConfigurationProperties(ChannelServerProperties.class) -public class ChannelServerConfiguration { +public class ChannelServerAutoConfiguration { @Bean - public ScheduledExecutorService executorService() { - // 设置较大的corePoolSize，避免长时间运行的task阻塞调度队列 (https://developer.aliyun.com/article/5897 "1.2 线程数量控制") + @ConditionalOnMissingBean + public ScheduledExecutorConfig scheduledExecutorConfig() { + // 设置较大的corePoolSize，避免并发运行的task阻塞调度队列 (https://developer.aliyun.com/article/5897 "1.2 线程数量控制") int corePoolSize = 10; ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(corePoolSize, new ThreadFactory() { @Override @@ -49,12 +52,15 @@ public class ChannelServerConfiguration { } }); - //ScheduledThreadPoolExecutor为无界队列，设置MaximumPoolSize无效 -// if (executorService instanceof ThreadPoolExecutor) { -// ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executorService; -// threadPoolExecutor.setMaximumPoolSize(50); -// } - return executorService; + ScheduledExecutorConfig scheduledExecutorConfig = new ScheduledExecutorConfig(); + scheduledExecutorConfig.setExecutorService(executorService); + return scheduledExecutorConfig; + } + + @Bean + @ConditionalOnMissingBean + public AgentBizSerivce agentBizSerivce() { + return new AgentBizServiceImpl(); } @Bean @@ -62,25 +68,33 @@ public class ChannelServerConfiguration { return new ApiActionDelegateServiceImpl(); } - @Bean - public AgentBizSerivce agentBizSerivce() { - return new AgentBizServiceImpl(); - } - @Bean public ArthasServiceGrpcImpl arthasServiceGrpc() { return new ArthasServiceGrpcImpl(); } @Bean(initMethod = "start", destroyMethod = "stop") + public AgentCleaner agentCleaner(ScheduledExecutorConfig scheduledExecutorConfig, ChannelServerProperties serverProperties) { + ChannelServerProperties.Agent agentConfig = serverProperties.getAgent(); + AgentCleaner agentCleaner = new AgentCleaner(scheduledExecutorConfig); + agentCleaner.setCleanIntervalMills(agentConfig.getCleanIntervalMills()); + agentCleaner.setRemovingTimeout(agentConfig.getRemovingTimeoutMills()); + agentCleaner.setDownTimeout(agentConfig.getDownTimeoutMills()); + agentCleaner.setOutOfServiceTimeout(agentConfig.getOutOfServiceTimeoutMills()); + return agentCleaner; + } + + @Bean(initMethod = "start", destroyMethod = "stop") + @ConditionalOnMissingBean @ConditionalOnProperty(value = "channel.server.backend.enabled", havingValue = "true", matchIfMissing = false) - public ChannelServer channelServer(@Value("${channel.server.backend.port}") int port) { + public ChannelServer channelServer(ChannelServerProperties serverProperties) { ChannelServer channelServer = new ChannelServer(); - channelServer.setPort(port); + channelServer.setPort(serverProperties.getBackend().getPort()); return channelServer; } @Bean(initMethod = "start", destroyMethod = "stop") + @ConditionalOnMissingBean @ConditionalOnProperty(value = "channel.server.websocket.enabled", havingValue = "true", matchIfMissing = false) public WebSocketServer webSocketServer(ChannelServerProperties serverProperties) { WebSocketServer server = new WebSocketServer(); @@ -96,11 +110,13 @@ public class ChannelServerConfiguration { static class StandaloneConfiguration { @Bean + @ConditionalOnMissingBean public AgentManageService agentManageService() { return new AgentManageServiceImpl(); } @Bean + @ConditionalOnMissingBean public MessageExchangeService messageExchangeService() { return new MessageExchangeServiceImpl(); } @@ -111,16 +127,19 @@ public class ChannelServerConfiguration { @Configuration static class RedisConfiguration { @Bean + @ConditionalOnMissingBean public AgentManageService agentManageService() { return new RedisAgentManageServiceImpl(); } @Bean + @ConditionalOnMissingBean public MessageExchangeService messageExchangeService() { return new RedisMessageExchangeServiceImpl(); } @Bean + @ConditionalOnMissingBean public ReactiveRedisTemplate reactiveRedisTemplate(ReactiveRedisConnectionFactory redisConnectionFactory) { ReactiveRedisTemplate template = new ReactiveRedisTemplate (redisConnectionFactory, RedisSerializationContext .newSerializationContext() @@ -133,6 +152,7 @@ public class ChannelServerConfiguration { } @Bean + @ConditionalOnMissingBean public ReactiveStringRedisTemplate reactiveStringRedisTemplate(ReactiveRedisConnectionFactory redisConnectionFactory) { ReactiveStringRedisTemplate template = new ReactiveStringRedisTemplate(redisConnectionFactory); return template; diff --git a/channel/channel-server-starter/src/main/java/com/alibaba/arthas/channel/server/autoconfigure/ChannelServerProperties.java b/channel/channel-server-starter/src/main/java/com/alibaba/arthas/channel/server/autoconfigure/ChannelServerProperties.java new file mode 100644 index 000000000..448858758 --- /dev/null +++ b/channel/channel-server-starter/src/main/java/com/alibaba/arthas/channel/server/autoconfigure/ChannelServerProperties.java @@ -0,0 +1,115 @@ +package com.alibaba.arthas.channel.server.autoconfigure; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "channel.server") +public class ChannelServerProperties { + + private Server websocket = new Server(); + private Server backend = new Server(); + + private Agent agent = new Agent(); + + public Server getWebsocket() { + return websocket; + } + + public void setWebsocket(Server websocket) { + this.websocket = websocket; + } + + public Server getBackend() { + return backend; + } + + public void setBackend(Server backend) { + this.backend = backend; + } + + public Agent getAgent() { + return agent; + } + + public void setAgent(Agent agent) { + this.agent = agent; + } + + public static class Server { + private String host; + private int port; + private boolean ssl; + private boolean enabled; + + public String getHost() { + return host; + } + + public void setHost(String host) { + this.host = host; + } + + public int getPort() { + return port; + } + + public void setPort(int port) { + this.port = port; + } + + public boolean isSsl() { + return ssl; + } + + public void setSsl(boolean ssl) { + this.ssl = ssl; + } + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + } + + public static class Agent { + private int cleanIntervalMills = 5000; + private int outOfServiceTimeoutMills = 15*1000; + private int downTimeoutMills = 30*1000; + private int removingTimeoutMills = 600*1000; + + public int getCleanIntervalMills() { + return cleanIntervalMills; + } + + public void setCleanIntervalMills(int cleanIntervalMills) { + this.cleanIntervalMills = cleanIntervalMills; + } + + public int getOutOfServiceTimeoutMills() { + return outOfServiceTimeoutMills; + } + + public void setOutOfServiceTimeoutMills(int outOfServiceTimeoutMills) { + this.outOfServiceTimeoutMills = outOfServiceTimeoutMills; + } + + public int getDownTimeoutMills() { + return downTimeoutMills; + } + + public void setDownTimeoutMills(int downTimeoutMills) { + this.downTimeoutMills = downTimeoutMills; + } + + public int getRemovingTimeoutMills() { + return removingTimeoutMills; + } + + public void setRemovingTimeoutMills(int removingTimeoutMills) { + this.removingTimeoutMills = removingTimeoutMills; + } + } + +} diff --git a/channel/channel-server-starter/src/main/resources/MEAT-INF/spring.factories b/channel/channel-server-starter/src/main/resources/MEAT-INF/spring.factories new file mode 100644 index 000000000..8431d887b --- /dev/null +++ b/channel/channel-server-starter/src/main/resources/MEAT-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +com.alibaba.arthas.channel.server.autoconfigure.ChannelServerAutoConfiguration \ No newline at end of file diff --git a/channel/channel-server/pom.xml b/channel/channel-server/pom.xml index d9e05872e..1970773f4 100644 --- a/channel/channel-server/pom.xml +++ b/channel/channel-server/pom.xml @@ -102,20 +102,6 @@ true - - - org.springframework.boot - spring-boot-maven-plugin - 2.1.7.RELEASE - - - package - - repackage - - - - diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/conf/ScheduledExecutorConfig.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/conf/ScheduledExecutorConfig.java new file mode 100644 index 000000000..dc7a5a2bd --- /dev/null +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/conf/ScheduledExecutorConfig.java @@ -0,0 +1,20 @@ +package com.alibaba.arthas.channel.server.conf; + +import java.util.concurrent.ScheduledExecutorService; + +/** + * ScheduledExecutorService config for channel server, avoiding bean injecting conflicts + * @author gongdewei 2020/9/8 + */ +public class ScheduledExecutorConfig { + + private ScheduledExecutorService executorService; + + public ScheduledExecutorService getExecutorService() { + return executorService; + } + + public void setExecutorService(ScheduledExecutorService executorService) { + this.executorService = executorService; + } +} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/configuration/ChannelServerProperties.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/configuration/ChannelServerProperties.java deleted file mode 100644 index 209b500c3..000000000 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/configuration/ChannelServerProperties.java +++ /dev/null @@ -1,66 +0,0 @@ -package com.alibaba.arthas.channel.server.configuration; - -import org.springframework.boot.context.properties.ConfigurationProperties; - -@ConfigurationProperties(prefix = "channel.server") -public class ChannelServerProperties { - - private Server websocket; - private Server backend; - - public Server getWebsocket() { - return websocket; - } - - public void setWebsocket(Server websocket) { - this.websocket = websocket; - } - - public Server getBackend() { - return backend; - } - - public void setBackend(Server backend) { - this.backend = backend; - } - - public static class Server { - private String host; - private int port; - private boolean ssl; - private boolean enabled = true; - - public String getHost() { - return host; - } - - public void setHost(String host) { - this.host = host; - } - - public int getPort() { - return port; - } - - public void setPort(int port) { - this.port = port; - } - - public boolean isSsl() { - return ssl; - } - - public void setSsl(boolean ssl) { - this.ssl = ssl; - } - - public boolean isEnabled() { - return enabled; - } - - public void setEnabled(boolean enabled) { - this.enabled = enabled; - } - } - -} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/grpc/ChannelServer.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/grpc/ChannelServer.java index 7abc24fc8..94cac8858 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/grpc/ChannelServer.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/grpc/ChannelServer.java @@ -1,5 +1,6 @@ package com.alibaba.arthas.channel.server.grpc; +import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig; import io.grpc.Server; import io.grpc.ServerBuilder; import org.slf4j.Logger; @@ -20,14 +21,14 @@ public class ChannelServer { private ArthasServiceGrpcImpl arthasServiceGrpc; @Autowired - private ScheduledExecutorService executorService; + private ScheduledExecutorConfig executorServiceConfig; private Server server; private int port = 7700; public void start() throws Exception { - executorService.submit(new Runnable() { + executorServiceConfig.getExecutorService().submit(new Runnable() { @Override public void run() { try { diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java index 0ac2dccca..455f4ef4b 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java @@ -1,5 +1,6 @@ package com.alibaba.arthas.channel.server.message.impl; +import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig; import com.alibaba.arthas.channel.server.message.MessageExchangeException; import com.alibaba.arthas.channel.server.message.MessageExchangeService; import com.alibaba.arthas.channel.server.message.topic.Topic; @@ -14,7 +15,6 @@ import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; /** @@ -28,7 +28,7 @@ public class MessageExchangeServiceImpl implements MessageExchangeService { private Map topicMap = new ConcurrentHashMap(); @Autowired - private ScheduledExecutorService executorService; + private ScheduledExecutorConfig executorServiceConfig; @Override public void createTopic(Topic topic) throws MessageExchangeException { @@ -108,7 +108,7 @@ public class MessageExchangeServiceImpl implements MessageExchangeService { topicData.setTimeout(timeout); final TopicData finalTopicData = topicData; - executorService.submit(new Runnable() { + executorServiceConfig.getExecutorService().submit(new Runnable() { @Override public void run() { while (true) { diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/redis/RedisMessageExchangeServiceImpl.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/redis/RedisMessageExchangeServiceImpl.java index eeee1c4b3..5bf701b73 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/redis/RedisMessageExchangeServiceImpl.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/redis/RedisMessageExchangeServiceImpl.java @@ -1,5 +1,6 @@ package com.alibaba.arthas.channel.server.redis; +import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig; import com.alibaba.arthas.channel.server.message.MessageExchangeException; import com.alibaba.arthas.channel.server.message.MessageExchangeService; import com.alibaba.arthas.channel.server.message.topic.Topic; @@ -13,7 +14,6 @@ import reactor.core.publisher.Mono; import java.time.Duration; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.ScheduledExecutorService; /** * Message exchange for cluster channel server @@ -30,7 +30,7 @@ public class RedisMessageExchangeServiceImpl implements MessageExchangeService { private ReactiveRedisTemplate redisTemplate; @Autowired - private ScheduledExecutorService executorService; + private ScheduledExecutorConfig executorServiceConfig; @Override public void createTopic(Topic topic) throws MessageExchangeException { @@ -82,7 +82,7 @@ public class RedisMessageExchangeServiceImpl implements MessageExchangeService { Mono mono = redisTemplate.opsForList().rightPop(topic.getTopic(), Duration.ofMillis(timeout)); mono.doOnSuccess(messageBytes -> { //schedule running, avoid blocking redis reactive - executorService.submit(() -> { + executorServiceConfig.getExecutorService().submit(() -> { if (messageBytes != null) { boolean next = messageHandler.onMessage(messageBytes); if (next) { @@ -96,7 +96,7 @@ public class RedisMessageExchangeServiceImpl implements MessageExchangeService { } }); }).doOnError(throwable -> { - executorService.submit(() -> { + executorServiceConfig.getExecutorService().submit(() -> { if (throwable instanceof QueryTimeoutException) { //ignore Redis command timed out subscribe(topic, timeout, messageHandler); diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/AgentBizSerivce.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/AgentBizSerivce.java index ded5abdd1..81155c070 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/AgentBizSerivce.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/AgentBizSerivce.java @@ -9,7 +9,5 @@ public interface AgentBizSerivce { void heartbeat(String agentId, String agentStatus, String agentVersion); - void cleanOutdatedAgents(); - void compareAndUpdateAgentStatus(String agentId, AgentStatus expectedStatus, AgentStatus newStatus); } diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/AgentCleaner.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/AgentCleaner.java new file mode 100644 index 000000000..21f1e545c --- /dev/null +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/AgentCleaner.java @@ -0,0 +1,124 @@ +package com.alibaba.arthas.channel.server.service; + +import com.alibaba.arthas.channel.proto.AgentStatus; +import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig; +import com.alibaba.arthas.channel.server.model.AgentVO; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import reactor.core.publisher.Mono; + +import java.util.List; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +public class AgentCleaner { + + private static final Logger logger = LoggerFactory.getLogger(AgentCleaner.class); + private int removingTimeout = 60000; + private int downTimeout = 30000; + private int outOfServiceTimeout = 15000; + private int cleanIntervalMills = 5000; + + @Autowired + private AgentManageService agentManageService; + private ScheduledExecutorConfig scheduledExecutorConfig; + private ScheduledFuture scheduledFuture; + + public AgentCleaner(ScheduledExecutorConfig scheduledExecutorConfig) { + this.scheduledExecutorConfig = scheduledExecutorConfig; + } + + public void start() { + + if (downTimeout < outOfServiceTimeout) { + throw new IllegalArgumentException(" agent down timeout must not be less than out of service timeout"); + } + + if (removingTimeout > 0 && removingTimeout < downTimeout) { + throw new IllegalArgumentException(" agent removing timeout must not be less than down timeout or -1"); + } + + scheduledFuture = scheduledExecutorConfig.getExecutorService().scheduleWithFixedDelay(new Runnable() { + @Override + public void run() { + doClean(); + } + }, cleanIntervalMills, cleanIntervalMills, TimeUnit.MILLISECONDS); + } + + public void stop() { + if (scheduledFuture != null) { + scheduledFuture.cancel(true); + } + } + + public void doClean() { + long now = System.currentTimeMillis(); + Mono> agentsMono = agentManageService.listAgents(); + agentsMono.doOnSuccess(agents -> { + for (AgentVO agent : agents) { + long heartbeatDelay = now - agent.getHeartbeatTime(); + if (heartbeatDelay > removingTimeout && removingTimeout > 0) { + logger.info("clean up dead agent: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay); + agentManageService.removeAgentById(agent.getAgentId()); + } else if (heartbeatDelay > downTimeout) { + if (!AgentStatus.DOWN.name().equals(agent.getAgentStatus())) { + logger.info("Mark agent status as DOWN, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay); + agent.setAgentStatus(AgentStatus.DOWN.name()); + agentManageService.updateAgent(agent); + } + } else if (heartbeatDelay > outOfServiceTimeout) { + if (!AgentStatus.OUT_OF_SERVICE.name().equals(agent.getAgentStatus())) { + logger.info("Mark agent status as OUT_OF_SERVICE, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay); + agent.setAgentStatus(AgentStatus.OUT_OF_SERVICE.name()); + agentManageService.updateAgent(agent); + } + } + } + }).doOnError(throwable -> { + logger.error("clean agent error", throwable); + }).subscribe(); + } + + public int getRemovingTimeout() { + return removingTimeout; + } + + public void setRemovingTimeout(int removingTimeout) { + this.removingTimeout = removingTimeout; + } + + public int getDownTimeout() { + return downTimeout; + } + + public void setDownTimeout(int downTimeout) { + if (downTimeout < 5000) { + throw new IllegalArgumentException("agent down timeout must be not be less than 5000 mills"); + } + this.downTimeout = downTimeout; + } + + public int getOutOfServiceTimeout() { + return outOfServiceTimeout; + } + + public void setOutOfServiceTimeout(int outOfServiceTimeout) { + if (outOfServiceTimeout < 5000) { + throw new IllegalArgumentException("agent out of service timeout must not be less than 5000 mills"); + } + this.outOfServiceTimeout = outOfServiceTimeout; + } + + public int getCleanIntervalMills() { + return cleanIntervalMills; + } + + public void setCleanIntervalMills(int cleanIntervalMills) { + if (cleanIntervalMills < 2000) { + throw new IllegalArgumentException("agent clean interval must not be less than 2000 mills"); + } + this.cleanIntervalMills = cleanIntervalMills; + } +} diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/AgentBizServiceImpl.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/AgentBizServiceImpl.java index cad18d8a9..99c6bb0e5 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/AgentBizServiceImpl.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/service/impl/AgentBizServiceImpl.java @@ -8,7 +8,6 @@ import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.scheduling.annotation.Scheduled; import reactor.core.publisher.Mono; import java.util.List; @@ -43,38 +42,6 @@ public class AgentBizServiceImpl implements AgentBizSerivce { }).subscribe(); } - @Override - @Scheduled(fixedDelayString = "5000") - public void cleanOutdatedAgents() { - long now = System.currentTimeMillis(); - Mono> agentsMono = agentManageService.listAgents(); - agentsMono.doOnSuccess(new Consumer>() { - @Override - public void accept(List agents) { - for (AgentVO agent : agents) { - long heartbeatDelay = now - agent.getHeartbeatTime(); - if (heartbeatDelay > 60000) { - logger.info("clean up dead agent: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay); - agentManageService.removeAgentById(agent.getAgentId()); - } else if (heartbeatDelay > 30000) { - if (!AgentStatus.DOWN.name().equals(agent.getAgentStatus())) { - logger.info("Mark agent status as DOWN, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay); - agent.setAgentStatus(AgentStatus.DOWN.name()); - agentManageService.updateAgent(agent); - } - } else if (heartbeatDelay > 15000) { - if (!AgentStatus.OUT_OF_SERVICE.name().equals(agent.getAgentStatus())) { - logger.info("Mark agent status as OUT_OF_SERVICE, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay); - agent.setAgentStatus(AgentStatus.OUT_OF_SERVICE.name()); - agentManageService.updateAgent(agent); - } - } - } - } - }).subscribe(); - - } - @Override public void compareAndUpdateAgentStatus(String agentId, AgentStatus expectedStatus, AgentStatus newStatus) { agentManageService.findAgentById(agentId).flatMap(optionalAgentVO -> { diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java index fb12b286a..93dccc305 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/AgentController.java @@ -4,6 +4,7 @@ import com.alibaba.arthas.channel.proto.ActionResponse; import com.alibaba.arthas.channel.proto.ExecuteResult; import com.alibaba.arthas.channel.proto.ResponseStatus; import com.alibaba.arthas.channel.server.api.ApiException; +import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig; import com.alibaba.arthas.channel.server.model.AgentVO; import com.alibaba.arthas.channel.server.api.ApiRequest; import com.alibaba.arthas.channel.server.api.ApiResponse; @@ -24,7 +25,6 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; -import org.springframework.web.client.HttpClientErrorException; import org.springframework.web.server.ResponseStatusException; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import reactor.core.publisher.Mono; @@ -32,7 +32,6 @@ import reactor.core.publisher.Mono; import java.io.IOException; import java.util.List; import java.util.Optional; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; /** @@ -50,7 +49,7 @@ public class AgentController { private ApiActionDelegateService apiActionDelegateService; @Autowired - private ScheduledExecutorService executorService; + private ScheduledExecutorConfig executorServiceConfig; @RequestMapping("/agents") @@ -174,7 +173,7 @@ public class AgentController { @RequestBody final String requestBody, @RequestParam(value = "timeout", defaultValue = "300000") final int timeout) { final SseEmitter emitter = new SseEmitter(Long.valueOf(timeout)); - executorService.submit(new Runnable() { + executorServiceConfig.getExecutorService().submit(new Runnable() { public void run() { ApiResponse apiResponse; try { diff --git a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/ws/WebSocketFrameHandler.java b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/ws/WebSocketFrameHandler.java index 12541405e..cbb9253c5 100644 --- a/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/ws/WebSocketFrameHandler.java +++ b/channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/ws/WebSocketFrameHandler.java @@ -29,9 +29,7 @@ import java.util.Collections; import java.util.List; import java.util.Optional; -/** - * @author gongdewei 2020/9/2 - */ + public class WebSocketFrameHandler extends SimpleChannelInboundHandler { private static final Logger logger = LoggerFactory.getLogger(WebSocketFrameHandler.class); diff --git a/pom.xml b/pom.xml index c1c4bf3f8..4899c78b5 100644 --- a/pom.xml +++ b/pom.xml @@ -74,6 +74,8 @@ channel/channel-common channel/channel-server channel/channel-client + channel/channel-server-starter + channel/channel-server-app