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

[ server ] [ scheduler ]- script scheduler - 1

Browse Source
This commit is contained in:
zeaslity
2023-01-17 12:05:04 +08:00
parent 4812756408
commit 8ef3b271b1
26 changed files with 709 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
server/src/main/java/io/wdd/rpc/scheduler/beans/ScriptSchedulerDTO.java Normal file
View File

@@ -0,0 +1,22 @@
package io.wdd.rpc.scheduler.beans;
import io.swagger.annotations.ApiModel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import java.util.List;
@ApiModel("定时脚本任务-中间转换状态-实体类")
@Data
@AllArgsConstructor
@NoArgsConstructor
@SuperBuilder(toBuilder = true)
public class ScriptSchedulerDTO extends ScriptSchedulerVO{
List<List<String>> commandList;
List<String> targetMachineList;
}

114
server/src/main/java/io/wdd/rpc/scheduler/beans/ScriptSchedulerVO.java Normal file
View File

@@ -0,0 +1,114 @@
package io.wdd.rpc.scheduler.beans;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import io.swagger.annotations.ApiModel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import javax.annotation.Nullable;
import javax.validation.constraints.NotNull;
import java.time.LocalDateTime;
@Data
@AllArgsConstructor
@NoArgsConstructor
@SuperBuilder(toBuilder = true)
@ApiModel("定时脚本任务的VO实体类")
public class ScriptSchedulerVO {
public static final String SCHEDULE_MISSION_GROUP_NAME = "SCRIPT_SCHEDULER";
/**
*
*/
@TableId(value = "scheduler_uuid")
@Nullable
private String schedulerUuid;
/**
*
*/
@TableField(value = "name")
@NotNull
private String name;
/**
*
*/
@TableField(value = "cron_express")
@NotNull
private String cronExpress;
/**
*
*/
@TableField(value = "description")
@Nullable
private String description;
/**
* 脚本任务的内容
*/
@TableField(value = "script_content")
@NotNull
private String scriptContent;
/**
* 执行目标机器agent_topic_name列表，使用, 分隔
*/
@TableField(value = "target_machine")
@Nullable
private String targetMachine;
/**
* 与 execution_log表的主键对应,方便查询执行日志
*/
@TableField(value = "last_execution_id")
@Nullable
private Long lastExecutionId;
/**
* 与 execution_log表的 result_key 对应,方便查询执行日志
*/
@TableField(value = "last_execution_result_key")
@Nullable
private String lastExecutionResultKey;
/**
* 任务上次执行状态
*/
@TableField(value = "last_execution_status")
@Nullable
private Integer lastExecutionStatus;
/**
* 定时脚本任务创建时间
*/
@TableField(value = "create_time")
@NotNull
private LocalDateTime createTime;
/**
* 上次更新时间
*/
@TableField(value = "update_time")
@Nullable
private LocalDateTime updateTime;
/**
* 任务下次计划执行时间
*/
@TableField(value = "next_schedule_time")
@Nullable
private LocalDateTime nextScheduleTime;
/**
* 任务上次计划执行时间
*/
@TableField(value = "last_schedule_time")
@Nullable
private LocalDateTime lastScheduleTime;
}

6
server/src/main/java/io/wdd/rpc/scheduler/job/AgentRunMetricStatusJob.java
View File

@@ -1,6 +1,6 @@
package io.wdd.rpc.scheduler.job;
import io.wdd.rpc.status.AgentRuntimeMetricStatus;
import io.wdd.rpc.scheduler.service.status.AgentRuntimeMetricStatus;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
@@ -8,8 +8,8 @@ import org.springframework.scheduling.quartz.QuartzJobBean;
import javax.annotation.Resource;
import static io.wdd.rpc.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIMES_COUNT;
import static io.wdd.rpc.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIME_PINCH;
import static io.wdd.rpc.scheduler.service.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIMES_COUNT;
import static io.wdd.rpc.scheduler.service.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIME_PINCH;
public class AgentRunMetricStatusJob extends QuartzJobBean {

34
server/src/main/java/io/wdd/rpc/scheduler/job/AgentScriptSchedulerJob.java Normal file
View File

@@ -0,0 +1,34 @@
package io.wdd.rpc.scheduler.job;
import io.wdd.rpc.scheduler.beans.ScriptSchedulerDTO;
import io.wdd.rpc.scheduler.service.script.AgentApplyScheduledScript;
import io.wdd.server.beans.po.ScriptSchedulerPO;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import javax.annotation.Resource;
public class AgentScriptSchedulerJob extends QuartzJobBean {
@Resource
AgentApplyScheduledScript agentApplyScheduledScript;
@Override
protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
// 解析 Scheduler 模块传递过来的参数
JobDataMap jobDataMap = jobExecutionContext
.getJobDetail()
.getJobDataMap();
// ScriptScheduleDTO
ScriptSchedulerDTO scriptSchedulerDTO = (ScriptSchedulerDTO) jobDataMap.get("scriptSchedulerPO");
// 调用实际任务执行器
agentApplyScheduledScript.apply(scriptSchedulerDTO);
}
}

2
server/src/main/java/io/wdd/rpc/scheduler/job/AgentStatusMonitorJob.java
View File

@@ -1,7 +1,7 @@
package io.wdd.rpc.scheduler.job;
import io.wdd.rpc.scheduler.config.QuartzLogOperator;
import io.wdd.rpc.status.MonitorAllAgentStatus;
import io.wdd.rpc.scheduler.service.status.MonitorAllAgentStatus;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;

8
server/src/main/java/io/wdd/rpc/scheduler/service/BuildStatusScheduleTask.java
View File

@@ -14,8 +14,8 @@ import java.text.ParseException;
import java.util.Date;
import java.util.HashMap;
import static io.wdd.rpc.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIMES_COUNT;
import static io.wdd.rpc.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIME_PINCH;
import static io.wdd.rpc.scheduler.service.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIMES_COUNT;
import static io.wdd.rpc.scheduler.service.status.AgentRuntimeMetricStatus.METRIC_REPORT_TIME_PINCH;
@Component
@Slf4j
@@ -80,7 +80,7 @@ public class BuildStatusScheduleTask {
// build the Job 只发送一次消息，然后让Agent获取消息 （重复间隔，重复次数） 进行相应的处理！
// todo 解决创建太多对象的问题，需要缓存相应的内容
octopusQuartzService.addJob(
octopusQuartzService.addMission(
AgentRunMetricStatusJob.class,
"agentRunMetricStatusJob",
JOB_GROUP_NAME,
@@ -101,7 +101,7 @@ public class BuildStatusScheduleTask {
private void buildMonitorAllAgentStatusScheduleTask() {
// build the Job
octopusQuartzService.addJob(
octopusQuartzService.addMission(
AgentStatusMonitorJob.class,
"monitorAllAgentStatusJob",
JOB_GROUP_NAME,

27
server/src/main/java/io/wdd/rpc/scheduler/service/QuartzSchedulerService.java
View File

@@ -1,6 +1,7 @@
package io.wdd.rpc.scheduler.service;
import io.wdd.rpc.scheduler.beans.OctopusQuartzJob;
import io.wdd.rpc.scheduler.beans.ScriptSchedulerVO;
import org.quartz.Trigger;
import org.springframework.scheduling.quartz.QuartzJobBean;
import java.util.List;
@@ -10,7 +11,7 @@ import java.util.Map;
public interface QuartzSchedulerService {
boolean addJob(OctopusQuartzJob quartzJob);
boolean addMission(OctopusQuartzJob quartzJob);
/**
* 增加一个任务job
@@ -21,7 +22,7 @@ public interface QuartzSchedulerService {
* @param jobRunRepeatTimes 任务运行次数（若<0，则不限次数）
* @param jobData 任务参数
*/
void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int jobRunTimePinch, int jobRunRepeatTimes, Map jobData);
void addMission(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int jobRunTimePinch, int jobRunRepeatTimes, Map jobData);
/**
* 增加一个任务job
@@ -33,7 +34,7 @@ public interface QuartzSchedulerService {
* @param cronJobExpression 任务时间表达式
* @param jobData 任务参数
*/
void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int startTime, String cronJobExpression, Map jobData);
void addMission(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int startTime, String cronJobExpression, Map jobData);
/**
* 修改一个任务job
@@ -41,7 +42,7 @@ public interface QuartzSchedulerService {
* @param jobGroupName 任务组名
* @param jobTime cron时间表达式
*/
void updateJob(String jobName, String jobGroupName, String jobTime);
void updateMission(String jobName, String jobGroupName, String jobTime);
/**
@@ -51,21 +52,21 @@ public interface QuartzSchedulerService {
* @param jobGroupName
* @return
*/
boolean deleteJob(String jobName, String jobGroupName);
boolean deleteMission(String jobName, String jobGroupName);
/**
* 暂停一个任务job
* @param jobName
* @param jobGroupName
*/
void pauseJob(String jobName, String jobGroupName);
void pauseMission(String jobName, String jobGroupName);
/**
* 恢复一个任务job
* @param jobName
* @param jobGroupName
*/
void resumeJob(String jobName, String jobGroupName);
void resumeMission(String jobName, String jobGroupName);
/**
* 立即执行一个任务job
@@ -78,13 +79,13 @@ public interface QuartzSchedulerService {
* 获取所有任务job
* @return
*/
List<Map<String, Object>> queryAllJob();
List<Map<String, Object>> queryAllMission();
/**
* 获取正在运行的任务job
* @return
*/
List<Map<String, Object>> queryRunJob();
List<Map<String, Object>> queryRunMission();
/**
@@ -94,4 +95,12 @@ public interface QuartzSchedulerService {
List<Trigger> queryAllTrigger();
/**
* --------------------------------------------------------------
* 页面定时任务部分
* 应该只有脚本功能才可以定时，目前一阶段的功能
* */
void createScriptScheduledMission(ScriptSchedulerVO scriptSchedulerVO);
}

140
server/src/main/java/io/wdd/rpc/scheduler/service/QuartzSchedulerServiceImpl.java
View File

@@ -1,9 +1,19 @@
package io.wdd.rpc.scheduler.service;
import io.wdd.common.handler.MyRuntimeException;
import io.wdd.common.utils.FunctionReader;
import io.wdd.rpc.scheduler.beans.OctopusQuartzJob;
import io.wdd.rpc.scheduler.beans.ScriptSchedulerDTO;
import io.wdd.rpc.scheduler.beans.ScriptSchedulerVO;
import io.wdd.rpc.scheduler.job.AgentScriptSchedulerJob;
import io.wdd.server.beans.po.ScriptSchedulerPO;
import io.wdd.server.service.ScriptSchedulerService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.quartz.*;
import org.quartz.DateBuilder.IntervalUnit;
import org.quartz.impl.matchers.GroupMatcher;
@@ -12,9 +22,11 @@ import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;
import static io.wdd.rpc.scheduler.beans.ScriptSchedulerVO.SCHEDULE_MISSION_GROUP_NAME;
import static io.wdd.rpc.scheduler.service.BuildStatusScheduleTask.JOB_GROUP_NAME;
import static org.quartz.TriggerBuilder.newTrigger;
@@ -29,6 +41,113 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
@Autowired
private Scheduler scheduler;
@Resource
ScriptSchedulerService scriptSchedulerService;
/**
* --------------------------------------------------------------
* 页面定时任务部分
* 应该只有脚本功能才可以定时，目前一阶段的功能
* */
/**
*
*
* */
@Override
public void createScriptScheduledMission(ScriptSchedulerVO scriptSchedulerVO) {
// validate important value
ScriptSchedulerDTO scriptSchedulerDTO = validateAndConvertToScriptSchedulerDTO(scriptSchedulerVO);
// build the job
// build the trigger
// bind job and trigger
HashMap<String, Object> dataMap = new HashMap<>();
dataMap.put("scriptSchedulerDTO", scriptSchedulerDTO);
this.addMission(
AgentScriptSchedulerJob.class,
scriptSchedulerVO.getName(),
SCHEDULE_MISSION_GROUP_NAME,
1,
scriptSchedulerVO.getScriptContent(),
dataMap
);
// persistent the script scheduled mission
// todo dto should store more info
ScriptSchedulerPO scriptSchedulerPO = convertToScriptSchedulerPO(scriptSchedulerDTO);
scriptSchedulerService.save(scriptSchedulerPO);
}
/**
* 转换 中间层 --> 持久层
*
* @param dto 定时脚本任务-中间转换状态-实体类
* @return 定时脚本任务-持久化-实体类
*/
private ScriptSchedulerPO convertToScriptSchedulerPO(ScriptSchedulerDTO dto) {
// todo should be a static method
return ScriptSchedulerPO
.builder()
.cronExpress(dto.getCronExpress())
.schedulerUuid(dto.getSchedulerUuid())
.scriptContent(String.valueOf(dto.getCommandList()))
.targetMachine(String.valueOf(dto.getTargetMachineList()))
.build();
}
/**
* 转换 视图层 --> 中间层
* @param scriptSchedulerVO 定时脚本任务-前端页面-实体类
* @return 定时脚本任务-中间转换状态-实体类
*/
@SneakyThrows
private ScriptSchedulerDTO validateAndConvertToScriptSchedulerDTO(ScriptSchedulerVO scriptSchedulerVO) {
// 验证cron表达式
if (!CronExpression.isValidExpression(scriptSchedulerVO.getCronExpress())) {
throw new MyRuntimeException("cron express wrong !");
}
// 归一化 commandList
List<List<String>> commandList = FunctionReader.ReadStringToCommandList(scriptSchedulerVO.getScriptContent());
if (commandList.size() == 0) {
throw new MyRuntimeException("commandList parse wrong !");
}
// 执行机器目标归一化
String[] targetMachineSplit = scriptSchedulerVO
.getTargetMachine()
.split(",");
if (targetMachineSplit.length == 0) {
throw new MyRuntimeException("target machine wrong !");
}
ArrayList<String> targetMachineList = new ArrayList<>();
targetMachineList.stream().forEach(
targetMachineList::add
);
// 生成DTO对象
ScriptSchedulerDTO dto = new ScriptSchedulerDTO();
BeanUtils.copyProperties(dto,scriptSchedulerVO);
// 设置属性值
dto.setCommandList(null);
dto.setTargetMachineList(targetMachineList);
// 生成 scheduler uuid
String uuid = RandomStringUtils.randomAlphabetic(32);
dto.setSchedulerUuid(uuid);
return dto;
}
@PostConstruct
public void startScheduler() {
try {
@@ -39,7 +158,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
}
@Override
public boolean addJob(OctopusQuartzJob quartzJob) {
public boolean addMission(OctopusQuartzJob quartzJob) {
return false;
@@ -56,7 +175,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @param jobData 参数
*/
@Override
public void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int jobRunTimePinch, int jobRunRepeatTimes, Map jobData) {
public void addMission(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int jobRunTimePinch, int jobRunRepeatTimes, Map jobData) {
try {
// 任务名称和组构成任务key
JobDetail jobDetail = JobBuilder
@@ -119,7 +238,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @param jobData 参数
*/
@Override
public void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int startTime, String cronJobExpression, Map jobData) {
public void addMission(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int startTime, String cronJobExpression, Map jobData) {
try {
// 创建jobDetail实例，绑定Job实现类
// 指明job的名称，所在组的名称，以及绑定job类
@@ -188,7 +307,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @param jobTime
*/
@Override
public void updateJob(String jobName, String jobGroupName, String jobTime) {
public void updateMission(String jobName, String jobGroupName, String jobTime) {
try {
TriggerKey triggerKey = TriggerKey.triggerKey(
jobName,
@@ -223,7 +342,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @return
*/
@Override
public boolean deleteJob(String jobName, String jobGroupName) {
public boolean deleteMission(String jobName, String jobGroupName) {
try {
scheduler.deleteJob(new JobKey(
@@ -245,7 +364,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @param jobGroupName
*/
@Override
public void pauseJob(String jobName, String jobGroupName) {
public void pauseMission(String jobName, String jobGroupName) {
try {
JobKey jobKey = JobKey.jobKey(
jobName,
@@ -265,7 +384,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @param jobGroupName
*/
@Override
public void resumeJob(String jobName, String jobGroupName) {
public void resumeMission(String jobName, String jobGroupName) {
try {
JobKey jobKey = JobKey.jobKey(
jobName,
@@ -304,7 +423,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @return
*/
@Override
public List<Map<String, Object>> queryAllJob() {
public List<Map<String, Object>> queryAllMission() {
List<Map<String, Object>> jobList = null;
try {
GroupMatcher<JobKey> matcher = GroupMatcher.anyJobGroup();
@@ -362,7 +481,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
* @return
*/
@Override
public List<Map<String, Object>> queryRunJob() {
public List<Map<String, Object>> queryRunMission() {
List<Map<String, Object>> jobList = null;
try {
List<JobExecutionContext> executingJobs = scheduler.getCurrentlyExecutingJobs();
@@ -432,4 +551,7 @@ public class QuartzSchedulerServiceImpl implements QuartzSchedulerService {
}
}
}

39
server/src/main/java/io/wdd/rpc/scheduler/service/script/AgentApplyScheduledScript.java Normal file
View File

@@ -0,0 +1,39 @@
package io.wdd.rpc.scheduler.service.script;
import io.wdd.rpc.execute.service.CoreExecutionService;
import io.wdd.rpc.scheduler.beans.ScriptSchedulerDTO;
import io.wdd.server.beans.po.ScriptSchedulerPO;
import lombok.extern.log4j.Log4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.List;
/**
* 1.
*/
@Service
@Log4j
public class AgentApplyScheduledScript {
@Resource
CoreExecutionService coreExecutionService;
public void apply(ScriptSchedulerDTO scriptSchedulerDTO) {
List<List<String>> commandList = scriptSchedulerDTO.getCommandList();
List<String> targetMachineList = scriptSchedulerDTO.getTargetMachineList();
targetMachineList
.stream()
.map(
targetMachine -> {
coreExecutionService.SendCommandToAgent()
}
)
}
}

64
server/src/main/java/io/wdd/rpc/scheduler/service/status/AgentRuntimeMetricStatus.java Normal file
View File

@@ -0,0 +1,64 @@
package io.wdd.rpc.scheduler.service.status;
import io.wdd.common.beans.status.OctopusStatusMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;
import static io.wdd.common.beans.status.OctopusStatusMessage.METRIC_STATUS_MESSAGE_TYPE;
/**
* 收集OctopusAgent的运行Metric信息
* <p>
* CPU Memory AppStatus易变信息
*/
@Service
@Slf4j
public class AgentRuntimeMetricStatus {
public static List<String> ALL_HEALTHY_AGENT_TOPIC_NAMES;
public static final String METRIC_REPORT_TIME_PINCH = "metricRepeatPinch";
public static final String METRIC_REPORT_TIMES_COUNT = "metricRepeatCount";
@Resource
CollectAgentStatus collectAgentStatus;
public void collect(int metricRepeatCount, int metricRepeatPinch) {
// 检查基础信息
if (CollectionUtils.isEmpty(ALL_HEALTHY_AGENT_TOPIC_NAMES)) {
log.error("Metric Status Collect Failed ! no ALL_HEALTHY_AGENT_TOPIC_NAMES");
}
// 构建 OctopusMessage
// 只发送一次消息，让Agent循环定时执行任务
buildMetricStatusMessageAndSend(metricRepeatCount, metricRepeatPinch);
//
}
private void buildMetricStatusMessageAndSend(int metricRepeatCount, int metricRepeatPinch) {
List<OctopusStatusMessage> collect = ALL_HEALTHY_AGENT_TOPIC_NAMES.stream()
.map(
agentTopicName -> {
return OctopusStatusMessage.builder()
.type(METRIC_STATUS_MESSAGE_TYPE)
.metricRepeatCount(metricRepeatCount)
.metricRepeatPinch(metricRepeatPinch)
.agentTopicName(agentTopicName)
.build();
}
).collect(Collectors.toList());
// send to the next level
collectAgentStatus.statusMessageToAgent(collect);
}
}

84
server/src/main/java/io/wdd/rpc/scheduler/service/status/AgentStatusStreamReader.java Normal file
View File

@@ -0,0 +1,84 @@
package io.wdd.rpc.scheduler.service.status;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.wdd.common.beans.status.AgentStatus;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
@Getter
@Setter
@Slf4j
public class AgentStatusStreamReader implements StreamListener<String, MapRecord<String,String, String >> {
// https://medium.com/nerd-for-tech/event-driven-architecture-with-redis-streams-using-spring-boot-a81a1c9a4cde
//https://segmentfault.com/a/1190000040946712
//https://docs.spring.io/spring-data/redis/docs/2.5.5/reference/html/#redis.streams.receive.containers
/**
* 消费者类型：独立消费、消费组消费
*/
private String consumerType;
/**
* 消费组
*/
private String group;
/**
* 消费组中的某个消费者
*/
private String consumerName;
public AgentStatusStreamReader(String consumerType, String group, String consumerName) {
this.consumerType = consumerType;
this.group = group;
this.consumerName = consumerName;
}
@Override
public void onMessage(MapRecord<String, String, String> message) {
String streamKey = message.getStream();
RecordId messageId = message.getId();
String key = (String) message.getValue().keySet().toArray()[0];
String value = message.getValue().get(key);
log.info("Octopus Agent [ {} ] status of [ {} ] Time is [ {} ] stream recordId is [{}]", streamKey, key, key, messageId);
// print to console
printPrettyAgentStatus(value);
}
private void printPrettyAgentStatus(String valueString){
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true);
try {
String tmp = objectMapper.readValue(valueString, new TypeReference<String>() {
});
AgentStatus agentStatus = objectMapper.readValue(tmp, new TypeReference<AgentStatus>() {
});
System.out.println(objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(agentStatus));
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
}

72
server/src/main/java/io/wdd/rpc/scheduler/service/status/CollectAgentStatus.java Normal file
View File

@@ -0,0 +1,72 @@
package io.wdd.rpc.scheduler.service.status;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.wdd.common.beans.rabbitmq.OctopusMessage;
import io.wdd.common.beans.rabbitmq.OctopusMessageType;
import io.wdd.common.beans.status.OctopusStatusMessage;
import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.message.sender.ToAgentMessageSender;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;
/**
* 1. 定时任务
* 2. 向RabbitMQ中发送消息，STATUS类型的消息
* 3. 然后开始监听相应的Result StreamKey
*/
@Service
public class CollectAgentStatus {
@Resource
ToAgentMessageSender toAgentMessageSender;
@Resource
ObjectMapper objectMapper;
public void collectAgentStatus(OctopusStatusMessage statusMessage) {
this.statusMessageToAgent(List.of(statusMessage));
}
public void statusMessageToAgent(List<OctopusStatusMessage> statusMessageList) {
// build all the OctopusMessage
List<OctopusMessage> octopusMessageList = statusMessageList.stream().map(
statusMessage -> {
OctopusMessage octopusMessage = buildOctopusMessageStatus(statusMessage);
return octopusMessage;
}
).collect(Collectors.toList());
// batch send all messages to RabbitMQ
toAgentMessageSender.send(octopusMessageList);
// todo how to get result ?
}
private OctopusMessage buildOctopusMessageStatus(OctopusStatusMessage octopusStatusMessage) {
// must be like this or it will be deserialized as LinkedHashMap
String s;
try {
s = objectMapper.writeValueAsString(octopusStatusMessage);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
return OctopusMessage.builder()
.uuid(octopusStatusMessage.getAgentTopicName())
.type(OctopusMessageType.STATUS)
.init_time(TimeUtils.currentTime())
.content(s)
.build();
}
}

261
server/src/main/java/io/wdd/rpc/scheduler/service/status/MonitorAllAgentStatus.java Normal file
View File

@@ -0,0 +1,261 @@
package io.wdd.rpc.scheduler.service.status;
import io.wdd.common.beans.status.AgentHealthyStatusEnum;
import io.wdd.common.beans.status.OctopusStatusMessage;
import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.scheduler.service.BuildStatusScheduleTask;
import io.wdd.server.beans.vo.ServerInfoVO;
import io.wdd.server.coreService.CoreServerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static io.wdd.common.beans.status.OctopusStatusMessage.ALL_AGENT_STATUS_REDIS_KEY;
import static io.wdd.common.beans.status.OctopusStatusMessage.HEALTHY_STATUS_MESSAGE_TYPE;
import static io.wdd.rpc.scheduler.service.status.AgentRuntimeMetricStatus.ALL_HEALTHY_AGENT_TOPIC_NAMES;
/**
* 更新频率被类 BuildStatusScheduleTask.class控制
* <p>
* <p>
* 获取所有注册的Agent
* <p>
* 发送状态检查信息， agent需要update相应的HashMap的值
* redis --> all-agent-health-map agent-topic-name : 1
* todo 分布式问题，弱网环境，多线程操作同一个hashMap会不会出现冲突
* <p>
* 休眠 MAX_WAIT_AGENT_REPORT_STATUS_TIME 秒 等待agent的状态上报
* <p>
* 检查相应的 状态HashMap，然后全部置为零
*/
@Service
@Slf4j
public class MonitorAllAgentStatus {
/**
* 存储 状态对应Agent列表的Map
* Agent的状态描述为 AgentHealthyStatusEnum
* HEALTHY -> ["agentTopicName-1"， "agentTopicName-2"]
* FAILED -> ["agentTopicName-1"， "agentTopicName-2"]
*/
public static final Map<String, List<String>> HEALTHY_STATUS_AGENT_LIST_MAP = new HashMap<>();
/**
* 存储所有Agent状态的Map
* <p>
* 内容为 agentTopicName-健康状态
*/
public static final Map<String, String> ALL_AGENT_HEALTHY_STATUS_MAP = new HashMap<>();
/**
* 存储所有的AgentTopicName的缓存
*/
public static final Set<String> ALL_AGENT_TOPIC_NAME_SET = new HashSet<>();
private static final int MAX_WAIT_AGENT_REPORT_STATUS_TIME = 5;
@Resource
RedisTemplate redisTemplate;
@Resource
CollectAgentStatus collectAgentStatus;
@Resource
CoreServerService coreServerService;
@Resource
BuildStatusScheduleTask buildStatusScheduleTask;
private List<String> ALL_AGENT_TOPIC_NAME_LIST;
private HashMap<String, String> AGENT_HEALTHY_INIT_MAP;
public void go() {
try {
// 1. 获取所有注册的Agent
// todo need to cache this
List<ServerInfoVO> allAgentInfo = coreServerService.serverGetAll();
Assert.notEmpty(
allAgentInfo,
"not agent registered ! skip the agent healthy status check !"
);
ALL_AGENT_TOPIC_NAME_LIST = allAgentInfo
.stream()
.map(ServerInfoVO::getTopicName)
.collect(Collectors.toList());
// 2023-01-16
ALL_AGENT_TOPIC_NAME_SET.clear();
ALL_AGENT_TOPIC_NAME_SET.addAll(ALL_AGENT_TOPIC_NAME_LIST);
// 1.1 检查 Agent状态保存数据结构是否正常
checkOrCreateRedisHealthyKey();
// 2.发送状态检查信息， agent需要update相应的HashMap的值
buildAndSendAgentHealthMessage();
// 3. 休眠 MAX_WAIT_AGENT_REPORT_STATUS_TIME 秒 等待agent的状态上报
TimeUnit.SECONDS.sleep(MAX_WAIT_AGENT_REPORT_STATUS_TIME);
// 4.检查相应的 状态HashMap，然后全部置为零
// todo 存储到某个地方，目前只是打印日志
updateAllAgentHealthyStatus();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
private void checkOrCreateRedisHealthyKey() {
// must init the cached map && make sure the redis key existed!
if (null == AGENT_HEALTHY_INIT_MAP || !redisTemplate.hasKey(ALL_AGENT_STATUS_REDIS_KEY)) {
log.info("ALL_AGENT_STATUS_REDIS_KEY not existed , start to create");
// build the redis all agent healthy map struct
HashMap<String, String> initMap = new HashMap<>(32);
ALL_AGENT_TOPIC_NAME_LIST
.stream()
.forEach(
agentTopicName -> {
initMap.put(
agentTopicName,
"0"
);
}
);
initMap.put(
"updateTime",
TimeUtils.currentTimeString()
);
// cache this map struct
AGENT_HEALTHY_INIT_MAP = initMap;
// create the healthy redis structure
redisTemplate
.opsForHash()
.putAll(
ALL_AGENT_STATUS_REDIS_KEY,
initMap
);
}
}
private void buildAndSendAgentHealthMessage() {
List<OctopusStatusMessage> collect = ALL_AGENT_TOPIC_NAME_LIST
.stream()
.map(
agentTopicName -> OctopusStatusMessage
.builder()
.agentTopicName(agentTopicName)
.type(HEALTHY_STATUS_MESSAGE_TYPE)
.build()
)
.collect(Collectors.toList());
collectAgentStatus.statusMessageToAgent(collect);
}
private void updateAllAgentHealthyStatus() {
List statusList = redisTemplate
.opsForHash()
.multiGet(
ALL_AGENT_STATUS_REDIS_KEY,
ALL_AGENT_TOPIC_NAME_LIST
);
// current log to console is ok
// agent-topic-name : STATUS(healthy, failed, unknown)
HashMap<String, String> agentStatusMap = new HashMap<>(32);
for (int i = 0; i < ALL_AGENT_TOPIC_NAME_LIST.size(); i++) {
agentStatusMap.put(
ALL_AGENT_TOPIC_NAME_LIST.get(i),
uniformHealthyStatus(String.valueOf(statusList.get(i)))
);
}
String currentTimeString = TimeUtils.currentTimeString();
log.info(
"[ AGENT HEALTHY CHECK ] time is {} , result are => {}",
currentTimeString,
agentStatusMap
);
// 2023-01-16
ALL_AGENT_HEALTHY_STATUS_MAP.clear();
ALL_AGENT_HEALTHY_STATUS_MAP.putAll(agentStatusMap);
// 2023-01-16
Map<String, List<String>> statusAgentListMap = agentStatusMap
.entrySet()
.stream()
.collect(
Collectors.groupingBy(
Map.Entry::getValue
)
)
.entrySet()
.stream()
.collect(
Collectors.toMap(
entry -> entry.getKey(),
entry -> entry
.getValue()
.stream()
.map(
Map.Entry::getKey
)
.collect(Collectors.toList())
)
);
HEALTHY_STATUS_AGENT_LIST_MAP.putAll(statusAgentListMap);
log.debug("Agent存活状态 状态-Agent名称-Map 已经更新了");
// help gc
agentStatusMap = null;
// Trigger调用Agent Metric 任务
ArrayList<String> allHealthyAgentTopicNames = new ArrayList<>(32);
for (int i = 0; i < statusList.size(); i++) {
if (statusList
.get(i)
.equals("1")) {
allHealthyAgentTopicNames.add(ALL_AGENT_TOPIC_NAME_LIST.get(i));
}
}
ALL_HEALTHY_AGENT_TOPIC_NAMES = allHealthyAgentTopicNames;
// 执行Metric上报任务
buildStatusScheduleTask.buildAgentMetricScheduleTask();
// update time
AGENT_HEALTHY_INIT_MAP.put(
"updateTime",
currentTimeString
);
// init the healthy map
redisTemplate
.opsForHash()
.putAll(
ALL_AGENT_STATUS_REDIS_KEY,
AGENT_HEALTHY_INIT_MAP
);
}
private String uniformHealthyStatus(String agentStatus) {
switch (agentStatus) {
case "0":
return AgentHealthyStatusEnum.FAILED.getStatus();
case "1":
return AgentHealthyStatusEnum.HEALTHY.getStatus();
default:
return AgentHealthyStatusEnum.UNKNOWN.getStatus();
}
}
}