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

[ Status ] 完成Agent Metric部分

Browse Source
This commit is contained in:
zeaslity
2023-07-10 16:24:36 +08:00
parent f5a3db2f56
commit 2f477fd1cc
43 changed files with 491 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
server/src/main/java/io/wdd/rpc/agent/OctopusAgentServiceImpl.java
View File

@@ -7,7 +7,7 @@ import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.message.OctopusMessage;
import io.wdd.rpc.message.OctopusMessageType;
import io.wdd.rpc.message.handler.async.AsyncWaitOctopusMessageResultService;
import io.wdd.rpc.message.handler.async.OctopusMessageAsyncReplayContend;
import io.wdd.rpc.message.handler.async.OctopusMessageSynScReplayContend;
import io.wdd.rpc.message.sender.OMessageToAgentSender;
import io.wdd.server.beans.vo.ServerInfoVO;
import io.wdd.server.config.ServerCommonPool;
@@ -26,10 +26,10 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_AGENT_TOPIC_NAME_SET;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_HEALTHY_AGENT_TOPIC_NAME_LIST;
import static io.wdd.rpc.message.handler.sync.OMessageHandlerServer.LATEST_VERSION;
import static io.wdd.rpc.message.handler.sync.OMessageHandlerServer.OCTOPUS_MESSAGE_FROM_AGENT;
import static io.wdd.rpc.status.CommonAndStatusCache.ALL_AGENT_TOPIC_NAME_SET;
import static io.wdd.rpc.status.CommonAndStatusCache.ALL_HEALTHY_AGENT_TOPIC_NAME_LIST;
@Service
@Slf4j
@@ -70,7 +70,7 @@ public class OctopusAgentServiceImpl implements OctopusAgentService {
);
// 构造 异步结果监听内容
OctopusMessageAsyncReplayContend agentReplayContend = OctopusMessageAsyncReplayContend.build(
OctopusMessageSynScReplayContend agentReplayContend = OctopusMessageSynScReplayContend.build(
ALL_HEALTHY_AGENT_TOPIC_NAME_LIST.size(),
CurrentAppOctopusMessageType,
currentTime
@@ -147,16 +147,16 @@ public class OctopusAgentServiceImpl implements OctopusAgentService {
);
// 构造结果
OctopusMessageAsyncReplayContend OctopusMessageAsyncReplayContend = io.wdd.rpc.message.handler.async.OctopusMessageAsyncReplayContend.build(
OctopusMessageSynScReplayContend OctopusMessageSynScReplayContend = OctopusMessageSynScReplayContend.build(
ALL_HEALTHY_AGENT_TOPIC_NAME_LIST.size(),
CurrentAppOctopusMessageType,
currentTime
);
CountDownLatch countDownLatch = OctopusMessageAsyncReplayContend.getCountDownLatch();
CountDownLatch countDownLatch = OctopusMessageSynScReplayContend.getCountDownLatch();
// 调用后台接收处理所有的Replay信息
asyncWaitOctopusMessageResultService.waitFor(OctopusMessageAsyncReplayContend);
asyncWaitOctopusMessageResultService.waitFor(OctopusMessageSynScReplayContend);
/* CompletableFuture<Void> getAllAgentCoreInfoFuture = waitCollectAllAgentCoreInfo(
result,
@@ -176,10 +176,10 @@ public class OctopusAgentServiceImpl implements OctopusAgentService {
// 超时,或者 全部信息已经收集
// 此处调用,即可中断 异步任务的收集工作
asyncWaitOctopusMessageResultService.stopWaiting(OctopusMessageAsyncReplayContend);
asyncWaitOctopusMessageResultService.stopWaiting(OctopusMessageSynScReplayContend);
// 处理结果
OctopusMessageAsyncReplayContend
OctopusMessageSynScReplayContend
.getReplayOMList()
.stream()
.forEach(
@@ -207,7 +207,7 @@ public class OctopusAgentServiceImpl implements OctopusAgentService {
);
// help gc
OctopusMessageAsyncReplayContend = null;
OctopusMessageSynScReplayContend = null;
}
return result;

4
server/src/main/java/io/wdd/rpc/controller/ExecutionController.java
View File

@@ -17,8 +17,8 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_AGENT_TOPIC_NAME_LIST;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_HEALTHY_AGENT_TOPIC_NAME_LIST;
import static io.wdd.rpc.status.CommonAndStatusCache.ALL_AGENT_TOPIC_NAME_LIST;
import static io.wdd.rpc.status.CommonAndStatusCache.ALL_HEALTHY_AGENT_TOPIC_NAME_LIST;
@RestController
@RequestMapping("/octopus/server/executor")

11
server/src/main/java/io/wdd/rpc/controller/StatusController.java
View File

@@ -5,7 +5,7 @@ import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.wdd.common.response.R;
import io.wdd.rpc.scheduler.service.status.AgentAliveStatusMonitorService;
import io.wdd.rpc.status.service.AsyncStatusService;
import io.wdd.rpc.status.service.SyncStatusService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -15,7 +15,7 @@ import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import static io.wdd.rpc.init.AgentStatusCacheService.*;
import static io.wdd.rpc.status.CommonAndStatusCache.*;
@RestController
@@ -24,7 +24,7 @@ import static io.wdd.rpc.init.AgentStatusCacheService.*;
public class StatusController {
@Resource
AsyncStatusService asyncStatusService;
SyncStatusService syncStatusService;
@Resource
AgentAliveStatusMonitorService agentAliveStatusMonitorService;
@@ -80,7 +80,10 @@ public class StatusController {
public R<Map<String, List<String>>> ManualUpdateAgentStatus() {
// 手动调用更新
Map<String, Boolean> agentAliveStatusMap = asyncStatusService.AsyncCollectAgentAliveStatus(ALL_AGENT_TOPIC_NAME_LIST, 5);
Map<String, Boolean> agentAliveStatusMap = syncStatusService.SyncCollectAgentAliveStatus(
ALL_AGENT_TOPIC_NAME_LIST,
5
);
agentAliveStatusMonitorService.updateAllAgentHealthyStatus(agentAliveStatusMap);

4
server/src/main/java/io/wdd/rpc/execute/service/AsyncExecutionServiceImpl.java
View File

@@ -3,7 +3,7 @@ package io.wdd.rpc.execute.service;
import io.wdd.rpc.message.OctopusMessage;
import io.wdd.rpc.message.OctopusMessageType;
import io.wdd.rpc.message.handler.async.AsyncWaitOctopusMessageResultService;
import io.wdd.rpc.message.handler.async.OctopusMessageAsyncReplayContend;
import io.wdd.rpc.message.handler.async.OctopusMessageSynScReplayContend;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -205,7 +205,7 @@ public class AsyncExecutionServiceImpl implements AsyncExecutionService {
}
// 构造回复信息的内容
OctopusMessageAsyncReplayContend executionReplayContent = OctopusMessageAsyncReplayContend.build(
OctopusMessageSynScReplayContend executionReplayContent = OctopusMessageSynScReplayContend.build(
commandCount,
CurrentAppOctopusMessageType,
initTime

2
server/src/main/java/io/wdd/rpc/execute/service/SyncExecutionServiceImpl.java
View File

@@ -19,7 +19,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.stream.Collectors;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_AGENT_TOPIC_NAME_SET;
import static io.wdd.rpc.status.CommonAndStatusCache.ALL_AGENT_TOPIC_NAME_SET;
@Service

6
server/src/main/java/io/wdd/rpc/init/AcceptAgentInitInfo.java
View File

@@ -9,7 +9,7 @@ import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.message.OctopusMessage;
import io.wdd.rpc.message.OctopusMessageType;
import io.wdd.rpc.message.sender.OMessageToAgentSender;
import io.wdd.rpc.status.AgentStatus;
import io.wdd.rpc.status.deprecate.AgentStatus;
import io.wdd.server.beans.vo.ServerInfoVO;
import io.wdd.server.utils.DaemonDatabaseOperator;
import lombok.SneakyThrows;
@@ -53,7 +53,9 @@ public class AcceptAgentInitInfo {
"London",
7,
"LosAngeles",
7
7,
"Beijing",
8
)
);
public static Set<String> ALL_SERVER_ARCH_INFO = new HashSet<>(

20
server/src/main/java/io/wdd/rpc/message/handler/async/AsyncWaitOctopusMessageResultService.java
View File

@@ -25,29 +25,29 @@ public class AsyncWaitOctopusMessageResultService {
/**
* 为了避免线程不安全的问题,增加一层缓存,仅仅由当前类操作此部分
* KEY -> replayMatchKey
* VALUE -> OctopusMessageAsyncReplayContend - 包含countDownLatch 和 result
* VALUE -> OctopusMessageSynScReplayContend - 包含countDownLatch 和 result
*/
private static final HashMap<String, OctopusMessageAsyncReplayContend> OM_REPLAY_WAITING_TARGET_MAP = new HashMap<>();
private static final HashMap<String, OctopusMessageSynScReplayContend> OM_REPLAY_WAITING_TARGET_MAP = new HashMap<>();
public void waitFor(OctopusMessageAsyncReplayContend OctopusMessageAsyncReplayContend) {
public void waitFor(OctopusMessageSynScReplayContend OctopusMessageSynScReplayContend) {
// 向 REPLAY_CACHE_MAP中写入 Key
OM_REPLAY_WAITING_TARGET_MAP.put(
OctopusMessageAsyncReplayContend.getReplayMatchKey(),
OctopusMessageAsyncReplayContend
OctopusMessageSynScReplayContend.getReplayMatchKey(),
OctopusMessageSynScReplayContend
);
// 在调用线程的countDownLunch结束之后,关闭
// 清除 REPLAY_CACHE_MAP 中的队列
}
public void stopWaiting(OctopusMessageAsyncReplayContend OctopusMessageAsyncReplayContend) {
public void stopWaiting(OctopusMessageSynScReplayContend OctopusMessageSynScReplayContend) {
// 在调用线程的countDownLunch结束之后,关闭 清除 REPLAY_CACHE_MAP 中的队列
OctopusMessageAsyncReplayContend contend = OM_REPLAY_WAITING_TARGET_MAP.get(OctopusMessageAsyncReplayContend.getReplayMatchKey());
OctopusMessageSynScReplayContend contend = OM_REPLAY_WAITING_TARGET_MAP.get(OctopusMessageSynScReplayContend.getReplayMatchKey());
// 移除该内容
OM_REPLAY_WAITING_TARGET_MAP.remove(OctopusMessageAsyncReplayContend.getReplayMatchKey());
OM_REPLAY_WAITING_TARGET_MAP.remove(OctopusMessageSynScReplayContend.getReplayMatchKey());
// help gc
contend = null;
@@ -88,7 +88,7 @@ public class AsyncWaitOctopusMessageResultService {
OctopusMessage replayOMessage = OCTOPUS_MESSAGE_FROM_AGENT.poll();
// 构造 replayMatchKey
String matchKey = OctopusMessageAsyncReplayContend.generateMatchKey(
String matchKey = OctopusMessageSynScReplayContend.generateMatchKey(
replayOMessage.getType(),
replayOMessage.getInit_time()
);
@@ -105,7 +105,7 @@ public class AsyncWaitOctopusMessageResultService {
}
// Map中包含有Key,那么放置进去
OctopusMessageAsyncReplayContend replayContend = OM_REPLAY_WAITING_TARGET_MAP.get(matchKey);
OctopusMessageSynScReplayContend replayContend = OM_REPLAY_WAITING_TARGET_MAP.get(matchKey);
replayContend
.getReplayOMList()
.add(replayOMessage);

6
server/src/main/java/io/wdd/rpc/message/handler/async/OctopusMessageAsyncReplayContend.java → server/src/main/java/io/wdd/rpc/message/handler/async/OctopusMessageSynScReplayContend.java
View File

@@ -19,7 +19,7 @@ import java.util.concurrent.CountDownLatch;
@NoArgsConstructor
@SuperBuilder(toBuilder = true)
@ApiModel("众多业务调用RPC,异步等待需要确定返回消息是谁的")
public class OctopusMessageAsyncReplayContend {
public class OctopusMessageSynScReplayContend {
@ApiModelProperty("rpc消息的类型")
OctopusMessageType type;
@@ -54,14 +54,14 @@ public class OctopusMessageAsyncReplayContend {
*
* @return
*/
public static OctopusMessageAsyncReplayContend build(int waitForReplayNum, OctopusMessageType currentOMType, LocalDateTime currentTime) {
public static OctopusMessageSynScReplayContend build(int waitForReplayNum, OctopusMessageType currentOMType, LocalDateTime currentTime) {
CountDownLatch latch = null;
if (waitForReplayNum != 0) {
latch = new CountDownLatch(waitForReplayNum);
}
return new OctopusMessageAsyncReplayContend(
return new OctopusMessageSynScReplayContend(
currentOMType,
currentTime,
generateMatchKey(

2
server/src/main/java/io/wdd/rpc/scheduler/job/ScheduleUpdateDBInfoJob.java → server/src/main/java/io/wdd/rpc/scheduler/dto/ScheduleUpdateDBInfoJob.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.scheduler.job;
package io.wdd.rpc.scheduler.dto;
import org.quartz.JobExecutionContext;

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

@@ -23,7 +23,7 @@ public class AgentAliveStatusMonitorJob extends QuartzJobBean {
//JobDataMap jobDataMap = jobExecutionContext.getJobDetail().getJobDataMap();
// actually execute the monitor service
agentAliveStatusMonitorService.go();
agentAliveStatusMonitorService.collectAllAgentAliveStatus();
// log to somewhere
quartzLogOperator.save();

29
server/src/main/java/io/wdd/rpc/scheduler/job/AgentMetricStatusJob.java Normal file
View File

@@ -0,0 +1,29 @@
package io.wdd.rpc.scheduler.job;
import io.wdd.rpc.scheduler.service.status.AgentMetricStatusCollectService;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import javax.annotation.Resource;
public class AgentMetricStatusJob extends QuartzJobBean {
@Resource
AgentMetricStatusCollectService agentMetricStatusCollectService;
@Override
protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
// 从JobDetailContext中获取相应的信息
// JobDataMap jobDataMap = jobExecutionContext
// .getJobDetail()
// .getJobDataMap();
// 执行Agent Metric 状态收集任务
agentMetricStatusCollectService.collectHealthyAgentMetric();
}
}

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

@@ -1,38 +0,0 @@
package io.wdd.rpc.scheduler.job;
import io.wdd.rpc.scheduler.service.status.AgentMetricStatusCollectService;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import javax.annotation.Resource;
import static io.wdd.rpc.scheduler.service.status.AgentMetricStatusCollectService.METRIC_REPORT_TIMES_COUNT;
import static io.wdd.rpc.scheduler.service.status.AgentMetricStatusCollectService.METRIC_REPORT_TIME_PINCH;
public class AgentRunMetricStatusJob extends QuartzJobBean {
@Resource
AgentMetricStatusCollectService agentMetricStatusCollectService;
@Override
protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
// 从JobDetailContext中获取相应的信息
JobDataMap jobDataMap = jobExecutionContext
.getJobDetail()
.getJobDataMap();
// 执行Agent Metric 状态收集任务
agentMetricStatusCollectService.collect(
(Integer) jobDataMap.get(METRIC_REPORT_TIMES_COUNT),
(Integer) jobDataMap.get(METRIC_REPORT_TIME_PINCH)
);
// todo 机构设计状态会被存储至 Redis Stream Key 中
// AgentTopicName-Metric
}
}

1
server/src/main/java/io/wdd/rpc/scheduler/job/AgentScriptSchedulerJob.java
View File

@@ -18,6 +18,7 @@ import static io.wdd.rpc.scheduler.service.QuartzSchedulerServiceImpl.SCRIPT_SCH
* 定时脚本任务核心类，Quartz框架定时调用该类
*/
@Slf4j
@Deprecated
public class AgentScriptSchedulerJob extends QuartzJobBean {
@Resource

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

@@ -2,52 +2,39 @@ package io.wdd.rpc.scheduler.service;
import io.wdd.rpc.scheduler.job.AgentAliveStatusMonitorJob;
import io.wdd.rpc.scheduler.job.AgentRunMetricStatusJob;
import io.wdd.rpc.scheduler.job.AgentMetricStatusJob;
import lombok.extern.slf4j.Slf4j;
import org.quartz.CronExpression;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.text.ParseException;
import java.util.Date;
import java.util.HashMap;
import static io.wdd.rpc.scheduler.service.status.AgentMetricStatusCollectService.METRIC_REPORT_TIMES_COUNT;
import static io.wdd.rpc.scheduler.service.status.AgentMetricStatusCollectService.METRIC_REPORT_TIME_PINCH;
@Component
@Slf4j
public class BuildStatusScheduleTask {
public static final String JOB_GROUP_NAME = "OctopusAgent";
@Resource
QuartzSchedulerService octopusQuartzService;
@Value(value = "${octopus.status.healthy.cron}")
String healthyCronTimeExpress;
// todo 此时间可以更新Nacos配置更新 自动进行任务更新
@Value(value = "${octopus.status.healthy.start-delay}")
int healthyCheckStartDelaySeconds;
// @Value(value = "${octopus.status.metric.cron}")
// int metricReportCronExpress;
@Value(value = "${octopus.status.metric.pinch}")
int metricReportTimePinch;
public static final String JOB_GROUP_NAME = "OctopusAgent";
@Value(value = "${octopus.status.metric.cron}")
int metricReportCronExpress;
@Value(value = "${octopus.status.metric.start-delay}")
int metricReportStartDelaySeconds;
@PostConstruct
private void buildAll() {
private void buildAllPreScheduledTask() {
// Agent存活健康状态检查
buildMonitorAllAgentAliveStatusScheduleTask();
// Agent运行信息检查 Metric
// Agent全部信息检查 All
buildAgentMetricScheduleTask();
}
@@ -55,11 +42,23 @@ public class BuildStatusScheduleTask {
* Agent运行信息检查 Metric
* 【调用】应该由 健康状态检查结果 调用 ==> 所有存活节点需要进行Metric信息汇报
* 【间隔】存活间隔内，间隔一定的时间汇报Metric
* <p>
* 2023年7月10日 更改为按照cron表达式进行执行
*/
public void buildAgentMetricScheduleTask() {
// 2023年7月10日 更改为按照cron表达式进行执行
octopusQuartzService.addMission(
AgentMetricStatusJob.class,
"agentRunMetricStatusJob",
JOB_GROUP_NAME,
metricReportStartDelaySeconds,
metricReportCronExpress,
null
);
// 计算 Metric检测的时间间隔
int metricReportTimesCount = 19;
/*int metricReportTimesCount = 19;
try {
CronExpression cronExpression = new CronExpression(healthyCronTimeExpress);
@@ -68,8 +67,8 @@ public class BuildStatusScheduleTask {
long totalSeconds = (nextValidTime.getTime() - now.getTime()) / 1000;
metricReportTimesCount = (int) (totalSeconds / metricReportTimePinch) - 1;
/*System.out.println("totalSeconds = " + totalSeconds);
System.out.println("metricReportTimesCount = " + metricReportTimesCount);*/
*//*System.out.println("totalSeconds = " + totalSeconds);
System.out.println("metricReportTimesCount = " + metricReportTimesCount);*//*
} catch (ParseException e) {
throw new RuntimeException(e);
@@ -77,18 +76,10 @@ public class BuildStatusScheduleTask {
HashMap<String, Integer> map = new HashMap<String, Integer>();
map.put(METRIC_REPORT_TIME_PINCH, metricReportTimePinch);
map.put(METRIC_REPORT_TIMES_COUNT, metricReportTimesCount);
map.put(METRIC_REPORT_TIMES_COUNT, metricReportTimesCount);*/
//
// build the Job 只发送一次消息，然后让Agent获取消息 （重复间隔，重复次数） 进行相应的处理！
// todo 解决创建太多对象的问题，需要缓存相应的内容
octopusQuartzService.addMission(
AgentRunMetricStatusJob.class,
"agentRunMetricStatusJob",
JOB_GROUP_NAME,
metricReportTimePinch,
1,
map
);
}

1
server/src/main/java/io/wdd/rpc/scheduler/service/script/AgentApplyScheduledScript.java
View File

@@ -17,6 +17,7 @@ import java.util.List;
*/
@Service
@Slf4j
@Deprecated
public class AgentApplyScheduledScript {
@Resource

61
server/src/main/java/io/wdd/rpc/scheduler/service/status/AgentAliveStatusMonitorService.java
View File

@@ -1,9 +1,8 @@
package io.wdd.rpc.scheduler.service.status;
import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.init.AgentStatusCacheService;
import io.wdd.rpc.scheduler.service.BuildStatusScheduleTask;
import io.wdd.rpc.status.service.AsyncStatusService;
import io.wdd.rpc.status.CommonAndStatusCache;
import io.wdd.rpc.status.service.SyncStatusService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.context.annotation.Lazy;
@@ -14,47 +13,36 @@ import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_AGENT_TOPIC_NAME_LIST;
import static io.wdd.rpc.status.CommonAndStatusCache.*;
import static io.wdd.rpc.status.OctopusStatusMessage.ALL_AGENT_STATUS_REDIS_KEY;
/**
* 更新频率被类 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，然后全部置为零
* 定时任务 检测所有的Agent的 存活状态 的实际执行类
*/
@Service
@Slf4j
@Lazy
public class AgentAliveStatusMonitorService {
private static final int MAX_WAIT_AGENT_REPORT_STATUS_TIME = 5;
@Resource
RedisTemplate redisTemplate;
@Resource
AgentStatusCacheService agentStatusCacheService;
CommonAndStatusCache commonAndStatusCache;
@Resource
BuildStatusScheduleTask buildStatusScheduleTask;
@Resource
AsyncStatusService asyncStatusService;
SyncStatusService syncStatusService;
private HashMap<String, Boolean> AGENT_HEALTHY_INIT_MAP;
public void go() {
/**
* 收集所有Agent的存活状态
* 实际的定时任务的执行类,
*/
public void collectAllAgentAliveStatus() {
// 1. 获取所有注册的Agent 手动更新
agentStatusCacheService.updateAllAgentTopicNameCache();
commonAndStatusCache.updateAllAgentTopicNameCache();
if (CollectionUtils.isEmpty(ALL_AGENT_TOPIC_NAME_LIST)) {
log.warn("[Scheduler] No Agent Registered ! End Up Status Monitor !");
return;
@@ -63,17 +51,16 @@ public class AgentAliveStatusMonitorService {
// 1.1 检查 Agent状态保存数据结构是否正常
checkOrCreateRedisHealthyKey();
// 2.发送状态检查信息， agent需要update相应的HashMap的值
// 2023年6月14日 2. 发送ping等待所有的Agent返回PONG, 然后进行redis的状态修改
// 使用同步更新的策略
Map<String, Boolean> agentAliveStatusMap = asyncStatusService.AsyncCollectAgentAliveStatus(
// 同步的方法, 超时等待所有主机的存活状态
Map<String, Boolean> agentAliveStatusMap = syncStatusService.SyncCollectAgentAliveStatus(
ALL_AGENT_TOPIC_NAME_LIST,
5
);
// 更新Agent的状态
updateAllAgentHealthyStatus(agentAliveStatusMap);
// 更新Agent的状态 一级和二级缓存 同时更新 write-through的方式
updateAllAgentStatusCache(agentAliveStatusMap);
}
/**
@@ -112,21 +99,18 @@ public class AgentAliveStatusMonitorService {
.opsForHash()
.put(
ALL_AGENT_STATUS_REDIS_KEY,
"initTime",
STATUS_INIT_TIME_KEY,
TimeUtils.currentTimeString()
);
}
public void updateAllAgentHealthyStatus(Map<String, Boolean> agentAliveStatusMap) {
public void updateAllAgentStatusCache(Map<String, Boolean> agentAliveStatusMap) {
String currentTimeString = TimeUtils.currentTimeString();
// 更新所有的缓存状态
agentStatusCacheService.updateAgentStatusMapCache(agentAliveStatusMap);
// 执行Metric上报定时任务
// buildStatusScheduleTask.buildAgentMetricScheduleTask();
// 更新 二级缓存
commonAndStatusCache.updateAgentStatusCache(agentAliveStatusMap);
log.debug(
"[存活状态] - 当前时间为 [ %s ] , 所有的Agent存活状态为=> %s",
@@ -134,12 +118,13 @@ public class AgentAliveStatusMonitorService {
agentAliveStatusMap
);
// 更新 一级缓存
// 这里仅仅是更新时间
redisTemplate
.opsForHash()
.put(
ALL_AGENT_STATUS_REDIS_KEY,
"updateTime",
STATUS_UPDATE_TIME_KEY,
currentTimeString
);

61
server/src/main/java/io/wdd/rpc/scheduler/service/status/AgentMetricStatusCollectService.java
View File

@@ -1,74 +1,55 @@
package io.wdd.rpc.scheduler.service.status;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.message.OctopusMessage;
import io.wdd.rpc.message.sender.OMessageToAgentSender;
import io.wdd.rpc.status.beans.AgentStatus;
import io.wdd.rpc.status.service.SyncStatusService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.stream.Collectors;
import java.util.Map;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_HEALTHY_AGENT_TOPIC_NAME_LIST;
import static io.wdd.rpc.status.OctopusStatusMessage.ConstructAgentStatusMessage;
import static io.wdd.rpc.status.OctopusStatusMessage.METRIC_STATUS_MESSAGE_TYPE;
import static io.wdd.rpc.status.CommonAndStatusCache.ALL_HEALTHY_AGENT_TOPIC_NAME_LIST;
/**
* 收集OctopusAgent的运行Metric信息
* <p>
* CPU Memory AppStatus易变信息
* 定时任务 收集Agent的运行Metric的实际执行类
*/
@Service
@Slf4j
public class AgentMetricStatusCollectService {
public static final String METRIC_REPORT_TIME_PINCH = "metricRepeatPinch";
public static final String METRIC_REPORT_TIMES_COUNT = "metricRepeatCount";
@Resource
OMessageToAgentSender oMessageToAgentSender;
@Resource
ObjectMapper objectMapper;
SyncStatusService syncStatusService;
public void collect(int metricRepeatCount, int metricRepeatPinch) {
/**
* 收集所有健康主机的运行数据
*/
public void collectHealthyAgentMetric() {
// 检查基础信息
// 检查是否存在健康的主机
if (CollectionUtils.isEmpty(ALL_HEALTHY_AGENT_TOPIC_NAME_LIST)) {
log.error("Metric Status Collect Failed ! no ALL_HEALTHY_AGENT_TOPIC_NAMES");
return;
}
buildMetricStatusMessageAndSend(
metricRepeatCount,
metricRepeatPinch
// 调用核心的服务
Map<String, AgentStatus> agentMetricStatusMap = syncStatusService.SyncCollectAgentMetricStatus(
ALL_HEALTHY_AGENT_TOPIC_NAME_LIST,
10
);
}
// todo 需要进行存储或者咋滴
log.info(
"[Agent Metric] - 所有主机的状态为 => %s",
agentMetricStatusMap
);
private void buildMetricStatusMessageAndSend(int metricRepeatCount, int metricRepeatPinch) {
LocalDateTime currentTime = TimeUtils.currentFormatTime();
List<OctopusMessage> octopusStatusMessageList = ALL_HEALTHY_AGENT_TOPIC_NAME_LIST
.stream()
.map(
agentTopicName -> ConstructAgentStatusMessage(
METRIC_STATUS_MESSAGE_TYPE,
agentTopicName,
currentTime
)
)
.collect(Collectors.toList());
// batch send all messages to RabbitMQ
oMessageToAgentSender.send(octopusStatusMessageList);
}

3
server/src/main/java/io/wdd/rpc/scheduler/service/status/AgentStatusStreamReader.java
View File

@@ -4,7 +4,7 @@ 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.rpc.status.AgentStatus;
import io.wdd.rpc.status.deprecate.AgentStatus;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
@@ -15,6 +15,7 @@ import org.springframework.data.redis.stream.StreamListener;
@Getter
@Setter
@Slf4j
@Deprecated
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

17
server/src/main/java/io/wdd/rpc/scheduler/定时任务说明.md Normal file
View File

@@ -0,0 +1,17 @@
# 定时框架Quartz的说明
# 核心为 QuartzSchedulerService
1. addMission()方法
2. Scheduler.scheduleJob()方法就可以设置一个定时任务
# 项目创建 固定-定时任务的入口为 BuildStatusScheduleTask
1. 创建的方法为 buildAllPreScheduledTask()
# 需要将定时任务包装为一个个的Job
1. 需要继承 QuartzJobBean
2. 然后调用实际定时任务的Service
1. AgentAliveStatusMonitorJob extends QuartzJobBean
2.

2
server/src/main/java/io/wdd/rpc/status/AgentHealthyStatusEnum.java
View File

@@ -26,6 +26,4 @@ public enum AgentHealthyStatusEnum {
}
}

59
server/src/main/java/io/wdd/rpc/init/AgentStatusCacheService.java → server/src/main/java/io/wdd/rpc/status/CommonAndStatusCache.java
View File

@@ -1,12 +1,12 @@
package io.wdd.rpc.init;
package io.wdd.rpc.status;
import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.status.AgentHealthyStatusEnum;
import io.wdd.server.beans.vo.ServerInfoVO;
import io.wdd.server.coreService.CoreServerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
@@ -15,17 +15,19 @@ import java.util.*;
import java.util.stream.Collectors;
/**
* Server启动或者运行的时候，需要初 缓存一系列的信息
* <p>
* 所有Agent的TopicName ALL_AGENT_TOPIC_NAME_SET
* <p>
* 2023年7月10日 此部分应该初始化全部为 False状态
* Agent状态信息的两个Map STATUS_AGENT_LIST_MAP ALL_AGENT_STATUS_MAP
* <p>
* 2023年7月10日 -- 此部分作为Redis中存储的 二级缓存部分 应该严格遵循
*/
@Service
@Slf4j
public class AgentStatusCacheService {
public class CommonAndStatusCache {
/**
* 存储所有的AgentTopicName的缓存
@@ -36,8 +38,6 @@ public class AgentStatusCacheService {
* 存储所有的AgentTopicName的缓存
*/
public static final List<String> ALL_AGENT_TOPIC_NAME_LIST = new ArrayList<>();
/**
* 存储 状态对应Agent列表的Map
* Agent的状态描述为 AgentHealthyStatusEnum
@@ -51,34 +51,52 @@ public class AgentStatusCacheService {
* 内容为 agentTopicName- True代表健康 False代表不健康
*/
public static final Map<String, Boolean> ALL_AGENT_STATUS_MAP = new HashMap<>();
/**
* 保存所有健康运行的Agent Topic Name
*/
public static final List<String> ALL_HEALTHY_AGENT_TOPIC_NAME_LIST = new ArrayList<>();
/**
* 记录状态信息缓存的更新时间
*/
private static final String STATUS_UPDATE_TIME_KEY = "UPDATE_TIME";
public static final String STATUS_UPDATE_TIME_KEY = "UPDATE_TIME";
/**
* 记录状态信息缓存的初始化时间
*/
public static final String STATUS_INIT_TIME_KEY = "INIT_TIME";
/**
* AgentTopicName 在Redis中緩存的Key
*/
private static final String ALL_AGENT_TOPIC_NAME_REDIS_KEY = "ALL_AGENT_TOPIC_NAME";
@Resource
CoreServerService coreServerService;
@Resource
RedisTemplate redisTemplate;
@PostConstruct
public void GenerateAllCache() {
public void InitToGenerateAllStatusCache() {
//所有Agent的TopicName ALL_AGENT_TOPIC_NAME_SET
updateAllAgentTopicNameCache();
// Agent状态信息的两个Map
// updateAgentStatusMapCache(agentAliveStatusMap);
// 初始化 默认创建全部失败的Map
Map<String, Boolean> initAgentFalseStatusMap = ALL_AGENT_TOPIC_NAME_LIST
.stream()
.collect(Collectors.toMap(
topicName -> topicName,
topicName -> Boolean.FALSE
));
updateAgentStatusCache(initAgentFalseStatusMap);
}
/**
* 从数据库中获取所有注册过的Agent名称
* <p>
* 2023年7月10日 写入Redis中保存一份
*/
public void updateAllAgentTopicNameCache() {
@@ -98,23 +116,28 @@ public class AgentStatusCacheService {
.map(ServerInfoVO::getTopicName)
.collect(Collectors.toList());
ALL_AGENT_TOPIC_NAME_LIST.addAll(collect);
ALL_AGENT_TOPIC_NAME_SET.addAll(collect);
// 2023年7月10日 同步缓存至Redis中
redisTemplate
.opsForSet()
.add(
ALL_AGENT_TOPIC_NAME_REDIS_KEY,
ALL_AGENT_TOPIC_NAME_SET
);
}
/**
* 从redis中获取信息，更新Agent状态信息的全局缓存
* 根据传入的状态Map更新二级缓存的两个状态Map和健康主机的列表
* ALL_AGENT_STATUS_MAP
* ALL_HEALTHY_AGENT_TOPIC_NAME_LIST
* STATUS_AGENT_LIST_MAP
* <p>
* 由定时任务或者初始化服务触发
* 2023-02-21 前端接口，手动更新
* ALL_HEALTHY_AGENT_TOPIC_NAME_LIST
*/
public void updateAgentStatusMapCache(Map<String, Boolean> agentAliveStatusMap) {
public void updateAgentStatusCache(Map<String, Boolean> agentAliveStatusMap) {
// 检查，排除没有节点的情况
if (CollectionUtils.isEmpty(ALL_AGENT_TOPIC_NAME_LIST)) {
@@ -122,12 +145,10 @@ public class AgentStatusCacheService {
return;
}
// 2023年6月15日 更新状态缓存
ALL_AGENT_STATUS_MAP.clear();
ALL_AGENT_STATUS_MAP.putAll(agentAliveStatusMap);
// 2023-01-16
// 更新 状态-Agent容器 内容为
// HEALTHY -> ["agentTopicName-1"， "agentTopicName-2"]

3
server/src/main/java/io/wdd/rpc/status/OctopusStatusMessage.java
View File

@@ -34,9 +34,6 @@ public class OctopusStatusMessage {
*/
String statusType;
int metricRepeatCount;
int metricRepeatPinch;
public static OctopusMessage ConstructAgentStatusMessage(String statusType, String agentTopicName, LocalDateTime currentTime) {

2
server/src/main/java/io/wdd/rpc/status/AgentStatus.java → server/src/main/java/io/wdd/rpc/status/deprecate/AgentStatus.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import lombok.AllArgsConstructor;

2
server/src/main/java/io/wdd/rpc/status/AgentSystemInfo.java → server/src/main/java/io/wdd/rpc/status/deprecate/AgentSystemInfo.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import io.wdd.common.utils.TimeUtils;
import lombok.AllArgsConstructor;

2
server/src/main/java/io/wdd/rpc/status/AppStatusEnum.java → server/src/main/java/io/wdd/rpc/status/deprecate/AppStatusEnum.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
public enum AppStatusEnum {

2
server/src/main/java/io/wdd/rpc/status/AppStatusInfo.java → server/src/main/java/io/wdd/rpc/status/deprecate/AppStatusInfo.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import lombok.AllArgsConstructor;
import lombok.Data;

2
server/src/main/java/io/wdd/rpc/status/CpuInfo.java → server/src/main/java/io/wdd/rpc/status/deprecate/CpuInfo.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import lombok.AllArgsConstructor;
import lombok.Data;

2
server/src/main/java/io/wdd/rpc/status/CpuTicks.java → server/src/main/java/io/wdd/rpc/status/deprecate/CpuTicks.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import lombok.AllArgsConstructor;
import lombok.Data;

2
server/src/main/java/io/wdd/rpc/status/DiskInfo.java → server/src/main/java/io/wdd/rpc/status/deprecate/DiskInfo.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import io.wdd.common.utils.FormatUtils;
import lombok.AllArgsConstructor;

2
server/src/main/java/io/wdd/rpc/status/MemoryInfo.java → server/src/main/java/io/wdd/rpc/status/deprecate/MemoryInfo.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import io.wdd.common.utils.FormatUtils;
import lombok.AllArgsConstructor;

2
server/src/main/java/io/wdd/rpc/status/MetricStatus.java → server/src/main/java/io/wdd/rpc/status/deprecate/MetricStatus.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import lombok.Data;

2
server/src/main/java/io/wdd/rpc/status/NetworkInfo.java → server/src/main/java/io/wdd/rpc/status/deprecate/NetworkInfo.java
View File

@@ -1,4 +1,4 @@
package io.wdd.rpc.status;
package io.wdd.rpc.status.deprecate;
import io.wdd.common.utils.FormatUtils;

16
server/src/main/java/io/wdd/rpc/status/service/AsyncStatusService.java
View File

@@ -1,16 +0,0 @@
package io.wdd.rpc.status.service;
import java.util.List;
import java.util.Map;
public interface AsyncStatusService {
/**
* 应该是同步收集 agentTopicNameList 的节点的存活状态，并返回所有的状态存活结果
*
* @param agentTopicNameList
* @param aliveStatusWaitMaxTime
* @return
*/
Map<String, Boolean> AsyncCollectAgentAliveStatus(List<String> agentTopicNameList, int aliveStatusWaitMaxTime);
}

119
server/src/main/java/io/wdd/rpc/status/service/AsyncStatusServiceImpl.java
View File

@@ -1,119 +0,0 @@
package io.wdd.rpc.status.service;
import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.message.OctopusMessage;
import io.wdd.rpc.message.OctopusMessageType;
import io.wdd.rpc.message.handler.async.AsyncWaitOctopusMessageResultService;
import io.wdd.rpc.message.handler.async.OctopusMessageAsyncReplayContend;
import io.wdd.rpc.message.sender.OMessageToAgentSender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static io.wdd.rpc.init.AgentStatusCacheService.ALL_AGENT_TOPIC_NAME_LIST;
import static io.wdd.rpc.status.OctopusStatusMessage.ConstructAgentStatusMessage;
import static io.wdd.rpc.status.OctopusStatusMessage.HEALTHY_STATUS_MESSAGE_TYPE;
@Slf4j
@Service
public class AsyncStatusServiceImpl implements AsyncStatusService {
private static final OctopusMessageType CurrentAppOctopusMessageType = OctopusMessageType.STATUS;
@Resource
OMessageToAgentSender oMessageToAgentSender;
@Resource
AsyncWaitOctopusMessageResultService asyncWaitOctopusMessageResultService;
@Override
public Map<String, Boolean> AsyncCollectAgentAliveStatus(List<String> agentTopicNameList, int aliveStatusWaitMaxTime) {
// 构造最后的结果Map
Map<String, Boolean> agentAliveStatusMap = agentTopicNameList
.stream()
.collect(
Collectors.toMap(
agentTopicName -> agentTopicName,
agentTopicName -> Boolean.FALSE
));
LocalDateTime currentTime = TimeUtils.currentFormatTime();
// 构造OctopusMessage - StatusMessage结构体, 下发所有的消息
buildAndSendAgentAliveOctopusMessage(currentTime);
// 异步收集消息
OctopusMessageAsyncReplayContend statusAsyncReplayContend = OctopusMessageAsyncReplayContend.build(
agentTopicNameList.size(),
CurrentAppOctopusMessageType,
currentTime
);
asyncWaitOctopusMessageResultService.waitFor(statusAsyncReplayContend);
// 解析结果
CountDownLatch countDownLatch = statusAsyncReplayContend.getCountDownLatch();
// 等待状态返回的结果
boolean agentAliveStatusCollectResult = false;
try {
agentAliveStatusCollectResult = countDownLatch.await(
aliveStatusWaitMaxTime,
TimeUnit.SECONDS
);
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
if (!agentAliveStatusCollectResult) {
log.debug("Agent存活状态检查，没有检查到全部的Agent！");
}
// 移除等待队列
asyncWaitOctopusMessageResultService.stopWaiting(statusAsyncReplayContend);
// 处理结果
statusAsyncReplayContend
.getReplayOMList()
.stream()
.forEach(
statusOMessage -> {
if (statusOMessage.getResult() != null) {
agentAliveStatusMap.put(
statusOMessage.getUuid(),
Boolean.TRUE
);
}
}
);
}
// 返回Agent的存活状态内容
return agentAliveStatusMap;
}
private void buildAndSendAgentAliveOctopusMessage(LocalDateTime currentTime) {
List<OctopusMessage> octopusStatusMessageList = ALL_AGENT_TOPIC_NAME_LIST
.stream()
.map(
agentTopicName -> ConstructAgentStatusMessage(
HEALTHY_STATUS_MESSAGE_TYPE,
agentTopicName,
currentTime
)
)
.collect(Collectors.toList());
// 发送信息
oMessageToAgentSender.send(octopusStatusMessageList);
}
}

27
server/src/main/java/io/wdd/rpc/status/service/SyncStatusService.java Normal file
View File

@@ -0,0 +1,27 @@
package io.wdd.rpc.status.service;
import io.wdd.rpc.status.beans.AgentStatus;
import java.util.List;
import java.util.Map;
public interface SyncStatusService {
/**
* 同步收集 agentTopicNameList 的节点的存活状态，并返回所有的状态存活结果
*
* @param agentTopicNameList
* @param aliveStatusWaitMaxTime
* @return
*/
Map<String, Boolean> SyncCollectAgentAliveStatus(List<String> agentTopicNameList, int aliveStatusWaitMaxTime);
/**
* 同步收集 节点的运行状态
*
* @param agentTopicNameList
* @param collectMetricWaitMaxTime
* @return
*/
Map<String, AgentStatus> SyncCollectAgentMetricStatus(List<String> agentTopicNameList, int collectMetricWaitMaxTime);
}

214
server/src/main/java/io/wdd/rpc/status/service/SyncStatusServiceImpl.java Normal file
View File

@@ -0,0 +1,214 @@
package io.wdd.rpc.status.service;
import com.fasterxml.jackson.core.JsonProcessingException;
import io.wdd.common.utils.TimeUtils;
import io.wdd.rpc.message.OctopusMessage;
import io.wdd.rpc.message.OctopusMessageType;
import io.wdd.rpc.message.handler.async.AsyncWaitOctopusMessageResultService;
import io.wdd.rpc.message.handler.async.OctopusMessageSynScReplayContend;
import io.wdd.rpc.message.sender.OMessageToAgentSender;
import io.wdd.rpc.status.beans.AgentStatus;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static io.wdd.common.utils.OctopusObjectMapperConfig.OctopusObjectMapper;
import static io.wdd.rpc.status.OctopusStatusMessage.*;
@Slf4j
@Service
public class SyncStatusServiceImpl implements SyncStatusService {
private static final OctopusMessageType CurrentAppOctopusMessageType = OctopusMessageType.STATUS;
@Resource
OMessageToAgentSender oMessageToAgentSender;
@Resource
AsyncWaitOctopusMessageResultService asyncWaitOctopusMessageResultService;
@Override
public Map<String, Boolean> SyncCollectAgentAliveStatus(List<String> agentTopicNameList, int aliveStatusWaitMaxTime) {
// 构造最后的结果Map
Map<String, Boolean> agentAliveStatusMap = agentTopicNameList
.stream()
.collect(
Collectors.toMap(
agentTopicName -> agentTopicName,
agentTopicName -> Boolean.FALSE
));
// 当前的时间
LocalDateTime currentTime = TimeUtils.currentFormatTime();
// 构造OctopusMessage - StatusMessage结构体, 下发所有的消息
buildAndSendAgentStatusOctopusMessage(
agentTopicNameList,
HEALTHY_STATUS_MESSAGE_TYPE,
currentTime
);
// 同步收集消息
OctopusMessageSynScReplayContend statusSyncReplayContend = OctopusMessageSynScReplayContend.build(
agentTopicNameList.size(),
CurrentAppOctopusMessageType,
currentTime
);
asyncWaitOctopusMessageResultService.waitFor(statusSyncReplayContend);
// 解析结果
CountDownLatch countDownLatch = statusSyncReplayContend.getCountDownLatch();
// 等待状态返回的结果
boolean agentAliveStatusCollectResult = false;
try {
agentAliveStatusCollectResult = countDownLatch.await(
aliveStatusWaitMaxTime,
TimeUnit.SECONDS
);
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
if (!agentAliveStatusCollectResult) {
log.debug("Agent存活状态检查，没有检查到全部的Agent！");
}
// 移除等待队列
asyncWaitOctopusMessageResultService.stopWaiting(statusSyncReplayContend);
// 处理结果
statusSyncReplayContend
.getReplayOMList()
.stream()
.forEach(
statusOMessage -> {
if (statusOMessage.getResult() != null) {
agentAliveStatusMap.put(
statusOMessage.getUuid(),
Boolean.TRUE
);
}
}
);
}
// 返回Agent的存活状态内容
return agentAliveStatusMap;
}
@Override
public Map<String, AgentStatus> SyncCollectAgentMetricStatus(List<String> agentTopicNameList, int collectMetricWaitMaxTime) {
// 状态的结果Map
HashMap<String, AgentStatus> metricMap = new HashMap<>();
// 当前的时间
LocalDateTime currentTime = TimeUtils.currentFormatTime();
// 构造所有的Metric的OM并且下发
buildAndSendAgentStatusOctopusMessage(
agentTopicNameList,
METRIC_STATUS_MESSAGE_TYPE,
currentTime
);
// 同步等待结果, 并且解析结果
OctopusMessageSynScReplayContend metricSyncReplayContend = OctopusMessageSynScReplayContend.build(
agentTopicNameList.size(),
CurrentAppOctopusMessageType,
currentTime
);
asyncWaitOctopusMessageResultService.waitFor(metricSyncReplayContend);
// 解析结果
CountDownLatch countDownLatch = metricSyncReplayContend.getCountDownLatch();
// 等待状态返回的结果
boolean agentAliveStatusCollectResult = false;
try {
agentAliveStatusCollectResult = countDownLatch.await(
collectMetricWaitMaxTime,
TimeUnit.SECONDS
);
} catch (InterruptedException e) {
log.error("[Agent Metric] - 收集Agent的运行状态失败!");
throw new RuntimeException(e);
} finally {
if (!agentAliveStatusCollectResult) {
log.debug("Agent存活状态检查，没有检查到全部的Agent！");
}
// 移除等待队列
asyncWaitOctopusMessageResultService.stopWaiting(metricSyncReplayContend);
// 处理结果
metricSyncReplayContend
.getReplayOMList()
.stream()
.forEach(
statusOMessage -> {
if (statusOMessage.getResult() != null) {
// 解析Result对象为 AgentStatus
try {
AgentStatus agentStatus = OctopusObjectMapper.readValue(
(String) statusOMessage.getResult(),
AgentStatus.class
);
// 保存结果
metricMap.put(
statusOMessage.getUuid(),
agentStatus
);
} catch (JsonProcessingException e) {
log.error("[Agent Metric] - 解析AgentStatus失败!");
throw new RuntimeException(e);
}
}
}
);
}
return metricMap;
}
/**
* 2023年7月10日 通用的底层构造方法 Status类型的Octopus Message
*
* @param agentTopicNameList
* @param statusType
* @param currentTime
*/
private void buildAndSendAgentStatusOctopusMessage(List<String> agentTopicNameList, String statusType, LocalDateTime currentTime) {
List<OctopusMessage> octopusStatusMessageList = agentTopicNameList
.stream()
.map(
agentTopicName -> ConstructAgentStatusMessage(
statusType,
agentTopicName,
currentTime
)
)
.collect(Collectors.toList());
// 发送信息
oMessageToAgentSender.send(octopusStatusMessageList);
}
}

4
server/src/main/resources/application.yml
View File

@@ -121,7 +121,9 @@ octopus:
cron: 10 * * * * ? *
start-delay: 30
metric:
pinch: 20
type: cron
cron: 30 * * * * ? *
start-delay: 40
oss:
# 这里只是因为需要一个层级，不一定下面的都是oracle