[ server ] - monitor all agent status - 2

agent/src/main/java/io/wdd/agent/config/message/handler/OMHandlerStatus.java

@@ -1,11 +1,27 @@
 package io.wdd.agent.config.message.handler;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.wdd.agent.status.HealthyReporter;
 import io.wdd.common.beans.rabbitmq.OctopusMessage;
 import io.wdd.common.beans.rabbitmq.OctopusMessageType;
+import io.wdd.common.beans.status.OctopusStatusMessage;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
+
+import static io.wdd.common.beans.status.OctopusStatusMessage.HEALTHY_STATUS_MESSAGE_TYPE;
+
 @Component
 public class OMHandlerStatus extends AbstractOctopusMessageHandler {
+
+    @Resource
+    ObjectMapper objectMapper;
+
+    @Resource
+    HealthyReporter healthyReporter;
+
     @Override
     public boolean handle(OctopusMessage octopusMessage) {
 
@@ -13,6 +29,25 @@ public class OMHandlerStatus extends AbstractOctopusMessageHandler {
             return next.handle(octopusMessage);
         }
 
+        // handle about the status kind
+        try {
+
+            OctopusStatusMessage statusMessage = objectMapper.readValue((String) octopusMessage.getContent(), new TypeReference<OctopusStatusMessage>() {
+            });
+
+            String statusType = statusMessage.getType();
+
+            if (statusType.equals(HEALTHY_STATUS_MESSAGE_TYPE)) {
+                // healthy check
+                healthyReporter.report();
+            }
+
+
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+
+
         return true;
     }
 }

agent/src/main/java/io/wdd/agent/status/HealthyReporter.java

@@ -0,0 +1,40 @@
+package io.wdd.agent.status;
+
+import io.wdd.agent.config.beans.init.AgentServerInfo;
+import io.wdd.common.utils.TimeUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+import static io.wdd.common.beans.status.OctopusStatusMessage.ALL_AGENT_STATUS_REDIS_KEY;
+
+/**
+ * 1. modify the redis key => ALL_AGENT_STATUS_REDIS_KEY
+ * 2. the hashmap struct  key => agentTopicName
+ * 3. modify the key to "1"
+ */
+@Service
+@Slf4j
+public class HealthyReporter {
+
+    @Resource
+    RedisTemplate redisTemplate;
+
+    @Resource
+    AgentServerInfo agentServerInfo;
+
+    public void report(){
+
+        redisTemplate.opsForHash().put(
+                ALL_AGENT_STATUS_REDIS_KEY,
+                agentServerInfo.getAgentTopicName(),
+                "1"
+        );
+
+        log.debug("Agent Healthy Check Complete ! Time is => {}", TimeUtils.currentTimeString());
+    }
+
+
+}

common/src/main/java/io/wdd/common/beans/status/OctopusStatusMessage.java

@@ -11,6 +11,10 @@ import lombok.experimental.SuperBuilder;
 @SuperBuilder(toBuilder = true)
 public class OctopusStatusMessage {
 
+    // below two will be used by both server and agent
+    public static final String ALL_AGENT_STATUS_REDIS_KEY = "ALL_AGENT_STATUS";
+    public static final String HEALTHY_STATUS_MESSAGE_TYPE = "ping";
+
     /**
     * which kind of status should be return
      * short => short time message

server/src/main/java/io/wdd/rpc/scheduler/call/CallAgentQuartzService.java

@@ -1,18 +0,0 @@
-package io.wdd.rpc.scheduler.call;
-
-import io.wdd.common.beans.status.OctopusStatusMessage;
-import io.wdd.rpc.status.CollectAgentStatus;
-import org.springframework.stereotype.Service;
-
-import javax.annotation.Resource;
-
-@Service
-public class CallAgentQuartzService {
-
-    @Resource
-    CollectAgentStatus collectAgentStatus;
-
-
-
-
-}

server/src/main/java/io/wdd/rpc/scheduler/config/ExecutionJob.java

@@ -13,7 +13,7 @@ import org.springframework.util.ReflectionUtils;
 import java.lang.reflect.Method;
 
 
-@Async
+@Deprecated
 public class ExecutionJob extends QuartzJobBean {
 
     @Override

server/src/main/java/io/wdd/rpc/scheduler/config/QuartzLogOperator.java

@@ -0,0 +1,22 @@
+package io.wdd.rpc.scheduler.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ *  save the octopus quartz log to database
+ */
+@Service
+@Slf4j
+public class QuartzLogOperator {
+
+    public boolean save(){
+
+        log.info("QuartzLogOperator pretend to have saved the log !");
+
+        return true;
+
+    }
+
+
+}

server/src/main/java/io/wdd/rpc/scheduler/config/QuartzRunnable.java

@@ -7,6 +7,7 @@ import org.springframework.util.ReflectionUtils;
 import java.lang.reflect.Method;
 import java.util.concurrent.Callable;
 
+@Deprecated
 public class QuartzRunnable implements Callable<Object> {
 
     private final Object target;

server/src/main/java/io/wdd/rpc/scheduler/job/MonitorAllAgentStatusJob.java

@@ -0,0 +1,34 @@
+package io.wdd.rpc.scheduler.job;
+
+import io.wdd.rpc.scheduler.config.QuartzLogOperator;
+import io.wdd.rpc.status.MonitorAllAgentStatus;
+import org.quartz.JobDataMap;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.springframework.scheduling.quartz.QuartzJobBean;
+
+import javax.annotation.Resource;
+import java.nio.file.AccessMode;
+
+public class MonitorAllAgentStatusJob extends QuartzJobBean {
+
+    @Resource
+    MonitorAllAgentStatus monitorAllAgentStatus;
+
+    @Resource
+    QuartzLogOperator quartzLogOperator;
+
+    @Override
+    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
+
+        // get the jobMetaMap
+        //JobDataMap jobDataMap = jobExecutionContext.getJobDetail().getJobDataMap();
+
+        // actually execute the monitor service
+        monitorAllAgentStatus.go();
+
+        // log to somewhere
+        quartzLogOperator.save();
+
+    }
+}

server/src/main/java/io/wdd/rpc/scheduler/service/OctopusQuartzServiceImpl.java

@@ -38,35 +38,9 @@ public class OctopusQuartzServiceImpl implements OctopusQuartzService {
 
     @Override
     public boolean addJob(OctopusQuartzJob quartzJob) {
-        try {
-            // 构建jobDetail,并与PrintHelloJob类绑定(Job执行内容)
-            JobDetail jobDetail = JobBuilder
-                    .newJob(PrintHelloJob.class)
-                    .withIdentity(quartzJob.getUuid())
-                    .build();
 
-            //通过触发器名和cron表达式创建Trigger
-            Trigger cronTrigger = newTrigger()
-                    .withIdentity(quartzJob.getUuid())
-                    .startNow()
-                    .withSchedule(CronScheduleBuilder.cronSchedule(quartzJob.getCronExpression()))
-                    .build();
 
-            //把job信息放入jobDataMap中 job_key为标识
+        return false;
-            cronTrigger.getJobDataMap().put(OctopusQuartzJob.JOB_KEY, quartzJob);
-
-            //重置启动时间
-            ((CronTriggerImpl)cronTrigger).setStartTime(new Date());
-
-            //执行定时任务
-            scheduler.scheduleJob(jobDetail, cronTrigger);
-
-        } catch (Exception e) {
-            log.error("【创建定时任务失败】 定时任务id：{}", quartzJob.getId());
-            return false;
-        }
-
-        return true;
     }
 
     /**
@@ -86,16 +60,19 @@ public class OctopusQuartzServiceImpl implements OctopusQuartzService {
      *            参数
      */
     @Override
-    public void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int jobTime,
+    public void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, int jobTime, int jobTimes, Map jobData) {
-                       int jobTimes, Map jobData) {
         try {
             // 任务名称和组构成任务key
-            JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobName, jobGroupName)
+            JobDetail jobDetail = JobBuilder
+                    .newJob(jobClass)
+                    .withIdentity(jobName, jobGroupName)
                     .build();
+
             // 设置job参数
             if(jobData!= null && jobData.size()>0){
                 jobDetail.getJobDataMap().putAll(jobData);
             }
+
             // 使用simpleTrigger规则
             Trigger trigger = null;
             if (jobTimes < 0) {
@@ -104,7 +81,8 @@ public class OctopusQuartzServiceImpl implements OctopusQuartzService {
                         .startNow().build();
             } else {
                 trigger = newTrigger().withIdentity(jobName, jobGroupName).withSchedule(SimpleScheduleBuilder
-                                .repeatSecondlyForever(1).withIntervalInSeconds(jobTime).withRepeatCount(jobTimes))
+                                .repeatSecondlyForever(1)
+                                .withIntervalInSeconds(jobTime).withRepeatCount(jobTimes))
                         .startNow().build();
             }
             log.info("jobDataMap: {}", jobDetail.getJobDataMap().getWrappedMap());
@@ -124,13 +102,13 @@ public class OctopusQuartzServiceImpl implements OctopusQuartzService {
      *            任务名称(建议唯一)
      * @param jobGroupName
      *            任务组名
-     * @param jobTime
+     * @param cronJobExpression
      *            时间表达式 （如：0/5 * * * * ? ）
      * @param jobData
      *            参数
      */
     @Override
-    public void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, String jobTime, Map jobData) {
+    public void addJob(Class<? extends QuartzJobBean> jobClass, String jobName, String jobGroupName, String cronJobExpression, Map jobData) {
         try {
             // 创建jobDetail实例，绑定Job实现类
             // 指明job的名称，所在组的名称，以及绑定job类
@@ -141,15 +119,21 @@ public class OctopusQuartzServiceImpl implements OctopusQuartzService {
             if(jobData!= null && jobData.size()>0){
                 jobDetail.getJobDataMap().putAll(jobData);
             }
+
             // 定义调度触发规则
             // 使用cornTrigger规则
             // 触发器key
-            Trigger trigger = newTrigger().withIdentity(jobName, jobGroupName)
+            Trigger trigger = newTrigger()
+                    .withIdentity(jobName, jobGroupName)
                     .startAt(DateBuilder.futureDate(1, IntervalUnit.SECOND))
-                    .withSchedule(CronScheduleBuilder.cronSchedule(jobTime)).startNow().build();
+                    .withSchedule(CronScheduleBuilder.cronSchedule(cronJobExpression))
+                    .startNow()
+                    .build();
+
             // 把作业和触发器注册到任务调度中
             scheduler.scheduleJob(jobDetail, trigger);
             log.info("jobDataMap: {}", jobDetail.getJobDataMap());
+
         } catch (Exception e) {
             e.printStackTrace();
             throw new MyRuntimeException("add job error!");

server/src/main/java/io/wdd/rpc/status/MonitorAllAgentStatus.java

@@ -15,6 +15,9 @@ import java.util.List;
 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;
+
 /**
  * 获取所有注册的Agent
  * <p>
@@ -31,8 +34,6 @@ import java.util.stream.Collectors;
 public class MonitorAllAgentStatus {
 
     private static final int MAX_WAIT_AGENT_REPORT_STATUS_TIME = 5;
-    private static final String ALL_AGENT_STATUS_REDIS_KEY = "ALL_AGENT_STATUS";
-    private static final String STATUS_MESSAGE_TYPE = "ping";
 
     private HashMap<String, String> AGENT_HEALTHY_INIT_MAP;
     private List<String> ALL_AGENT_TOPICNAME_LIST;
@@ -97,7 +98,7 @@ public class MonitorAllAgentStatus {
         List<OctopusStatusMessage> collect = ALL_AGENT_TOPICNAME_LIST.stream().map(
                 agentTopicName -> OctopusStatusMessage.builder()
                         .agentTopicName(agentTopicName)
-                        .type(STATUS_MESSAGE_TYPE)
+                        .type(HEALTHY_STATUS_MESSAGE_TYPE)
                         .build()
         ).collect(Collectors.toList());
         collectAgentStatus.collectAgentStatusList(collect);
@@ -108,17 +109,21 @@ public class MonitorAllAgentStatus {
                 ALL_AGENT_STATUS_REDIS_KEY,
                 ALL_AGENT_TOPICNAME_LIST);
 
+
+        // current log to console is ok
         HashMap<String, String> tmp = new HashMap<>(32);
         for (int i = 0; i < ALL_AGENT_TOPICNAME_LIST.size(); i++) {
-            tmp.put(ALL_AGENT_TOPICNAME_LIST.get(i),
+            tmp.put(
+                    ALL_AGENT_TOPICNAME_LIST.get(i),
                     uniformHealthyStatus(String.valueOf(statusList.get(i)))
             );
         }
-        // current log to console is ok
-
         String currentTimeString = TimeUtils.currentTimeString();
         log.info("[ AGENT HEALTHY CHECK ] time is {} ,  result are => {}", currentTimeString, tmp);
 
+        // help gc
+        tmp = null;
+
         // update time
         AGENT_HEALTHY_INIT_MAP.put("updateTime", currentTimeString);
         // init the healthy map