zeaslity/ProjectOctopus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[server-agent] accomplish the register procedure

Browse Source
This commit is contained in:
zeaslity
2022-12-03 20:29:00 +08:00
parent d89bc2947a
commit ae11bc28b4
24 changed files with 506 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
agent/src/main/java/io/wdd/agent/config/rabbitmq/OMHandlerInit.java
View File

@@ -1,22 +0,0 @@
package io.wdd.agent.config.rabbitmq;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.beans.rabbitmq.OctopusMessageType;
public class OMHandlerInit extends AbstractOctopusMessageHandler {
@Override
public boolean handle(OctopusMessage octopusMessage) {
if (!octopusMessage.getType().equals(OctopusMessageType.INIT)) {
this.getNextHandler().handle(octopusMessage);
}
// handle the init message
return false;
}
}

20
agent/src/main/java/io/wdd/agent/config/rabbitmq/config/OctopusRabbitMQAdminConfig.java Normal file
View File

@@ -0,0 +1,20 @@
package io.wdd.agent.config.rabbitmq.config;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class OctopusRabbitMQAdminConfig {
@Autowired
ConnectionFactory connectionFactory;
@Bean
public RabbitAdmin rabbitAdmin(){
return new RabbitAdmin(connectionFactory);
}
}

6
agent/src/main/java/io/wdd/agent/config/rabbitmq/AbstractOctopusMessageHandler.java → agent/src/main/java/io/wdd/agent/config/rabbitmq/handler/AbstractOctopusMessageHandler.java
View File

@@ -1,7 +1,8 @@
package io.wdd.agent.config.rabbitmq;
package io.wdd.agent.config.rabbitmq.handler;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
/**
@@ -10,10 +11,11 @@ import org.springframework.context.annotation.Configuration;
@Configuration
public abstract class AbstractOctopusMessageHandler {
private AbstractOctopusMessageHandler next;
protected AbstractOctopusMessageHandler next;
public void addHandler(AbstractOctopusMessageHandler handler) {
this.next = handler;
}
public AbstractOctopusMessageHandler getNextHandler() {

9
agent/src/main/java/io/wdd/agent/config/rabbitmq/OMHandlerAgent.java → agent/src/main/java/io/wdd/agent/config/rabbitmq/handler/OMHandlerAgent.java
View File

@@ -1,14 +1,17 @@
package io.wdd.agent.config.rabbitmq;
package io.wdd.agent.config.rabbitmq.handler;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.beans.rabbitmq.OctopusMessageType;
import org.springframework.stereotype.Component;
@Component
public class OMHandlerAgent extends AbstractOctopusMessageHandler {
@Override
public boolean handle(OctopusMessage octopusMessage) {
if (!octopusMessage.getType().equals(OctopusMessageType.AGENT)) {
this.getNextHandler().handle(octopusMessage);
return next.handle(octopusMessage);
}
return false;
return true;
}
}

2
agent/src/main/java/io/wdd/agent/config/rabbitmq/OMHandlerBlackHole.java → agent/src/main/java/io/wdd/agent/config/rabbitmq/handler/OMHandlerBlackHole.java
View File

@@ -1,4 +1,4 @@
package io.wdd.agent.config.rabbitmq;
package io.wdd.agent.config.rabbitmq.handler;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import lombok.extern.slf4j.Slf4j;

8
agent/src/main/java/io/wdd/agent/config/rabbitmq/OMHandlerExecutor.java → agent/src/main/java/io/wdd/agent/config/rabbitmq/handler/OMHandlerExecutor.java
View File

@@ -1,15 +1,17 @@
package io.wdd.agent.config.rabbitmq;
package io.wdd.agent.config.rabbitmq.handler;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.beans.rabbitmq.OctopusMessageType;
import org.springframework.stereotype.Component;
@Component
public class OMHandlerExecutor extends AbstractOctopusMessageHandler {
@Override
public boolean handle(OctopusMessage octopusMessage) {
if (!octopusMessage.getType().equals(OctopusMessageType.EXECUTOR)) {
this.getNextHandler().handle(octopusMessage);
return next.handle(octopusMessage);
}
return false;
return true;
}
}

58
agent/src/main/java/io/wdd/agent/config/rabbitmq/handler/OMHandlerInit.java Normal file
View File

@@ -0,0 +1,58 @@
package io.wdd.agent.config.rabbitmq.handler;
import io.wdd.agent.initialization.beans.AgentServerInfo;
import io.wdd.agent.initialization.rabbitmq.GenerateOctopusConnection;
import io.wdd.agent.message.ToServerMessage;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.beans.rabbitmq.OctopusMessageType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* handle the agent PassThroughTopicName info
* 1. generator the unique topic queue for agent itself
* 2. send PassThroughTopicName successful info to the server
*/
@Lazy
@Component
@Slf4j
public class OMHandlerInit extends AbstractOctopusMessageHandler {
@Resource
GenerateOctopusConnection generateOctopusConnection;
@Resource
ToServerMessage toServerMessage;
@Resource
AgentServerInfo agentServerInfo;
@Override
public boolean handle(OctopusMessage octopusMessage) {
if (!octopusMessage.getType().equals(OctopusMessageType.INIT)) {
return next.handle(octopusMessage);
}
// handle the PassThroughTopicName message
// 1. generator the unique topic queue for agent itself
// 1.1 initial the specific topic queue listener
generateOctopusConnection.ManualGenerate(octopusMessage);
// 2. send PassThroughTopicName successful info to the server
String success = String.format("Octopus Agent [ %s ] has successfully PassThroughTopicName with server [ %s ] !", agentServerInfo, octopusMessage);
octopusMessage.setResult(success);
log.info(success);
toServerMessage.send(octopusMessage);
return true;
}
}

8
agent/src/main/java/io/wdd/agent/config/rabbitmq/OMHandlerStatus.java → agent/src/main/java/io/wdd/agent/config/rabbitmq/handler/OMHandlerStatus.java
View File

@@ -1,16 +1,18 @@
package io.wdd.agent.config.rabbitmq;
package io.wdd.agent.config.rabbitmq.handler;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.beans.rabbitmq.OctopusMessageType;
import org.springframework.stereotype.Component;
@Component
public class OMHandlerStatus extends AbstractOctopusMessageHandler {
@Override
public boolean handle(OctopusMessage octopusMessage) {
if (!octopusMessage.getType().equals(OctopusMessageType.STATUS)) {
this.getNextHandler().handle(octopusMessage);
return next.handle(octopusMessage);
}
return false;
return true;
}
}

2
agent/src/main/java/io/wdd/agent/initialization/bootup/CollectSystemInfo.java
View File

@@ -100,7 +100,7 @@ public class CollectSystemInfo implements ApplicationContextAware {
// start to send message to Octopus Server
octopusAgentInitService.SendInfoToServer(agentServerInfo);
log.info("init server info has been send to octopus server !");
log.info("PassThroughTopicName server info has been send to octopus server !");
}

64
agent/src/main/java/io/wdd/agent/initialization/bootup/OctopusAgentInitService.java
View File

@@ -3,18 +3,14 @@ package io.wdd.agent.initialization.bootup;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import io.wdd.agent.initialization.beans.AgentServerInfo;
import io.wdd.agent.initialization.rabbitmq.InitialRabbitMqConnector;
import io.wdd.agent.message.ToServerMessage;
import io.wdd.agent.message.handler.OctopusMessageHandler;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.handler.MyRuntimeException;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
@@ -23,44 +19,36 @@ import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
@Service
@Lazy
@Slf4j
public class OctopusAgentInitService {
@Resource
RabbitTemplate rabbitTemplate;
@Resource
InitialRabbitMqConnector initialRabbitMqConnector;
@Autowired
ObjectMapper objectMapper;
ToServerMessage toServerMessage;
@Autowired
OctopusMessageHandler octopusMessageHandler;
@Resource
AgentServerInfo agentServerInfo;
@Value("${octopus.message.init_ttl}")
String defaultInitRegisterTimeOut;
@SneakyThrows
@Resource
ObjectMapper objectMapper;
public void SendInfoToServer(AgentServerInfo agentServerInfo) {
// set init agent register ttl
InitMessagePostProcessor initMessagePostProcessor = new InitMessagePostProcessor(defaultInitRegisterTimeOut);
log.info("send INIT AgentServerInfo to Server = {}", agentServerInfo);
// send the register server info to EXCHANGE:INIT_EXCHANGE QUEUE: init_to_server
rabbitTemplate.convertAndSend(initialRabbitMqConnector.INIT_EXCHANGE, initialRabbitMqConnector.INIT_TO_SERVER_KEY, objectMapper.writeValueAsString(agentServerInfo), initMessagePostProcessor);
toServerMessage.sendInitInfo(agentServerInfo, defaultInitRegisterTimeOut);
}
/**
* listen to the init queue from octopus server
* listen to the PassThroughTopicName queue from octopus server
*
* @param message 该方法不需要手动调用，Spring会自动运行这个监听方法
* <p>
@@ -82,11 +70,17 @@ public class OctopusAgentInitService {
)
public void ReceiveInitInfoFromServer(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
try {
OctopusMessage octopusMessage = objectMapper.readValue(message.getBody(), OctopusMessage.class);
// consider the multi-agents register situation
// judge the machineID begin
String[] split = octopusMessage.getUuid().split("-");
if (!agentServerInfo.getMachineId().startsWith(split[split.length - 1])) {
throw new MyRuntimeException("INIT Message not for this agent !");
}
// response chain to handle all kind of type of octopus message
if (!octopusMessageHandler.handle(octopusMessage)) {
throw new MyRuntimeException(" Handle Octopus Message Error !");
@@ -99,8 +93,8 @@ public class OctopusAgentInitService {
// long deliveryTag, boolean requeue
// channel.basicReject(deliveryTag,true);
Thread.sleep(1000); // 这里只是便于出现死循环时查看
// 这里只是便于出现死循环时查看
TimeUnit.SECONDS.sleep(5);
throw new MyRuntimeException("Octopus Agent Initialization Error, please check !");
}
@@ -109,25 +103,5 @@ public class OctopusAgentInitService {
channel.basicAck(deliveryTag, false);
}
private class InitMessagePostProcessor implements MessagePostProcessor {
private final String initMessageTTL;
public InitMessagePostProcessor(Long initMessageTTL) {
this.initMessageTTL = String.valueOf(initMessageTTL);
}
public InitMessagePostProcessor(String initMessageTTL) {
this.initMessageTTL = initMessageTTL;
}
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// set init register expiration time
MessageProperties messageProperties = message.getMessageProperties();
messageProperties.setExpiration(initMessageTTL);
return message;
}
}
}

113
agent/src/main/java/io/wdd/agent/initialization/rabbitmq/GenerateOctopusConnection.java Normal file
View File

@@ -0,0 +1,113 @@
package io.wdd.agent.initialization.rabbitmq;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.wdd.agent.message.handler.OctopusMessageHandler;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.handler.MyRuntimeException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueInformation;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@Lazy
@Component
@Slf4j
@RequiredArgsConstructor
public class GenerateOctopusConnection {
private final List<MessageListenerContainer> messageListenerContainerList = new ArrayList<>();
private final SimpleRabbitListenerContainerFactory containerFactory;
@Resource
RabbitAdmin rabbitAdmin;
@Resource
ObjectMapper objectMapper;
@Resource
OctopusMessageHandler octopusMessageHandler;
public void ManualGenerate(OctopusMessage octopusMessage) {
// generate the ne topic queue for unique agent
String agentTopicName = octopusMessage.getUuid();
// reboot judgyment of existing exchange
QueueInformation queueInfo = rabbitAdmin.getQueueInfo(agentTopicName);
if (ObjectUtils.isNotEmpty(queueInfo) && queueInfo.getConsumerCount() > 0 ) {
log.info("Octopus Agent Specific Topic Queue Already Existed ! == {}", agentTopicName);
return;
}
Queue queue = new Queue(agentTopicName, true, false, false);
Binding binding = new Binding(
agentTopicName,
Binding.DestinationType.QUEUE,
octopusMessage.getContent().toString(),
agentTopicName + "*",
null
);
// Exchange are created by Octopus Server at server BootUP
rabbitAdmin.declareQueue(queue);
rabbitAdmin.declareBinding(binding);
// create the listener
SimpleMessageListenerContainer listenerContainer = containerFactory.createListenerContainer();
listenerContainer.addQueues(queue);
listenerContainer.setMessageListener(this::AgentListenToSpecificTopicOctopusMessage);
listenerContainer.start();
log.info("Specific Octopus Topic Queue Generate Successfully !");
messageListenerContainerList.add(listenerContainer);
}
/**
* Maunally Generated Octopus Message Listener
*
* @param message Rabbitmq Message
*/
public void AgentListenToSpecificTopicOctopusMessage(Message message) {
OctopusMessage octopusMessage;
try {
octopusMessage = objectMapper.readValue(message.getBody(), OctopusMessage.class);
if (!octopusMessageHandler.handle(octopusMessage)) {
throw new MyRuntimeException("Octopus Message Handle Err");
}
} catch (IOException e) {
throw new MyRuntimeException("Octopus Message Wrong !");
}
}
@PreDestroy
public void destroy() {
messageListenerContainerList.forEach(Lifecycle::stop);
log.info("- stop all message listeners...");
}
}

13
agent/src/main/java/io/wdd/agent/initialization/rabbitmq/InitialRabbitMqConnector.java → agent/src/main/java/io/wdd/agent/initialization/rabbitmq/InitRabbitMQConnector.java
View File

@@ -9,12 +9,12 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* agent send server init info to octopus server
* agent send server PassThroughTopicName info to octopus server
* <p>
* agent boot up should send message to server using this queue
*/
@Configuration
public class InitialRabbitMqConnector {
public class InitRabbitMQConnector {
@Value("${octopus.message.init_exchange}")
public String INIT_EXCHANGE;
@@ -31,6 +31,15 @@ public class InitialRabbitMqConnector {
@Value("${octopus.message.init_to_server_key}")
public String INIT_TO_SERVER_KEY;
@Value("${octopus.message.octopus_exchange}")
public String OCTOPUS_EXCHANGE;
@Value("${octopus.message.octopus_to_server}")
public String OCTOPUS_TO_SERVER;
//
public static String SPECIFIC_AGENT_TOPIC_NAME;
@Bean
public DirectExchange initDirectExchange() {
return new DirectExchange(INIT_EXCHANGE);

101
agent/src/main/java/io/wdd/agent/message/ToServerMessage.java Normal file
View File

@@ -0,0 +1,101 @@
package io.wdd.agent.message;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.wdd.agent.initialization.beans.AgentServerInfo;
import io.wdd.agent.initialization.rabbitmq.InitRabbitMQConnector;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.handler.MyRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.LocalDateTime;
@Service
@Slf4j(topic = "To Octopus Server Message")
public class ToServerMessage {
@Resource
InitRabbitMQConnector initRabbitMqConnector;
@Resource
ObjectMapper objectMapper;
@Resource
RabbitTemplate rabbitTemplate;
public boolean send(OctopusMessage octopusMessage) {
octopusMessage.setAc_time(LocalDateTime.now());
// send to Queue -- InitToServer
log.info("send Message to Server = {}", octopusMessage);
try {
rabbitTemplate.convertAndSend(
initRabbitMqConnector.OCTOPUS_EXCHANGE,
initRabbitMqConnector.OCTOPUS_TO_SERVER,
objectMapper.writeValueAsBytes(octopusMessage)
);
} catch (JsonProcessingException e) {
log.error("Failed to send message to Serv er ! = {}", octopusMessage);
throw new MyRuntimeException(e);
}
return true;
}
public void sendInitInfo(AgentServerInfo agentServerInfo, String defaultInitRegisterTimeOut) {
// set PassThroughTopicName agent register ttl
InitMessagePostProcessor initMessagePostProcessor = new InitMessagePostProcessor(defaultInitRegisterTimeOut);
log.info("send INIT AgentServerInfo to Server = {}", agentServerInfo);
// send the register server info to EXCHANGE:INIT_EXCHANGE QUEUE: init_to_server
try {
rabbitTemplate.convertAndSend(initRabbitMqConnector.INIT_EXCHANGE, initRabbitMqConnector.INIT_TO_SERVER_KEY, objectMapper.writeValueAsBytes(agentServerInfo), initMessagePostProcessor);
} catch (JsonProcessingException e) {
log.error("Failed to send INIT message to Server ! = {}", agentServerInfo);
throw new RuntimeException(e);
}
}
private static class InitMessagePostProcessor implements MessagePostProcessor {
private final String initMessageTTL;
// public InitMessagePostProcessor(Long initMessageTTL) {
// this.initMessageTTL = String.valueOf(initMessageTTL);
// }
public InitMessagePostProcessor(String initMessageTTL) {
this.initMessageTTL = initMessageTTL;
}
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// set PassThroughTopicName register expiration time
MessageProperties messageProperties = message.getMessageProperties();
messageProperties.setExpiration(initMessageTTL);
return message;
}
}
}

23
agent/src/main/java/io/wdd/agent/message/handler/OctopusMessageHandler.java
View File

@@ -1,11 +1,12 @@
package io.wdd.agent.message.handler;
import io.wdd.agent.config.rabbitmq.*;
import io.wdd.agent.config.rabbitmq.handler.*;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
@Service
@@ -14,16 +15,28 @@ public class OctopusMessageHandler {
private AbstractOctopusMessageHandler octopusMessageHandler;
@Resource
OMHandlerAgent omHandlerAgent;
@Resource
OMHandlerExecutor omHandlerExecutor;
@Resource
OMHandlerInit omHandlerInit;
@Resource
OMHandlerStatus omHandlerStatus;
@PostConstruct
private void registerAllHandler() {
AbstractOctopusMessageHandler.Builder builder = new AbstractOctopusMessageHandler.Builder();
octopusMessageHandler = builder
.addHandler(new OMHandlerExecutor())
.addHandler(new OMHandlerAgent())
.addHandler(new OMHandlerStatus())
.addHandler(new OMHandlerInit())
.addHandler(omHandlerExecutor)
.addHandler(omHandlerAgent)
.addHandler(omHandlerStatus)
.addHandler(omHandlerInit)
.build();
}

14
agent/src/main/resources/application.yml
View File

@@ -1,7 +1,11 @@
server:
port: 8000
spring:
main:
allow-circular-references: true
allow-bean-definition-overriding: true
rabbitmq:
host: 127.0.0.1
port: 35672
@@ -22,4 +26,12 @@ octopus:
# agent boot up default common exchange routing key
init_from_server_key: InitFromServerKey
# initialization register time out (unit ms) default is 5 min
init_ttl: "3000000"
init_ttl: "3000000"
# Octopus Exchange Name == server comunicate with agent
octopus_exchange: OctopusExchange
# Octopus Message To Server == all agent send info to server queue and topic
octopus_to_server: OctopusToServer
logging:
level:
web: debug