Merge pull request #32 from qq254963746/develop

Develop

Author: qq254963746

lts-admin/pom.xml

@@ -56,6 +56,21 @@
             <artifactId>lts-queue-mongo</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.lts</groupId>
+            <artifactId>lts-logger-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.lts</groupId>
+            <artifactId>lts-logger-mysql</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.lts</groupId>
+            <artifactId>lts-logger-mongo</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>log4j</groupId>
             <artifactId>log4j</artifactId>

lts-admin/src/main/java/com/lts/job/web/cluster/AdminApplication.java

@@ -1,5 +1,6 @@
 package com.lts.job.web.cluster;
 
+import com.lts.job.biz.logger.JobLogger;
 import com.lts.job.core.Application;
 import com.lts.job.core.cluster.Node;
 import com.lts.job.queue.*;
@@ -14,6 +15,7 @@ public class AdminApplication extends Application {
     private ExecutingJobQueue executingJobQueue;
     private JobFeedbackQueue jobFeedbackQueue;
     private NodeGroupStore nodeGroupStore;
+    private JobLogger jobLogger;
     private Node node;
 
     public Node getNode() {
@@ -63,4 +65,12 @@ public JobFeedbackQueue getJobFeedbackQueue() {
     public void setJobFeedbackQueue(JobFeedbackQueue jobFeedbackQueue) {
         this.jobFeedbackQueue = jobFeedbackQueue;
     }
+
+    public JobLogger getJobLogger() {
+        return jobLogger;
+    }
+
+    public void setJobLogger(JobLogger jobLogger) {
+        this.jobLogger = jobLogger;
+    }
 }

lts-admin/src/main/java/com/lts/job/web/cluster/RegistryService.java

@@ -6,9 +6,9 @@
 import com.lts.job.core.registry.NotifyListener;
 import com.lts.job.core.registry.Registry;
 import com.lts.job.core.registry.RegistryFactory;
-import com.lts.job.core.util.CollectionUtils;
-import com.lts.job.core.util.ConcurrentHashSet;
-import com.lts.job.core.util.StringUtils;
+import com.lts.job.core.commons.utils.CollectionUtils;
+import com.lts.job.core.commons.collect.ConcurrentHashSet;
+import com.lts.job.core.commons.utils.StringUtils;
 import com.lts.job.web.request.NodeRequest;
 import com.lts.job.web.support.AppConfigurer;
 import com.lts.job.web.support.memorydb.NodeMemoryDB;

lts-admin/src/main/java/com/lts/job/web/controller/api/JobQueueApiController.java

@@ -1,22 +1,22 @@
 package com.lts.job.web.controller.api;
 
+import com.lts.job.biz.logger.domain.JobLogPo;
+import com.lts.job.biz.logger.domain.JobLoggerRequest;
+import com.lts.job.core.commons.utils.*;
 import com.lts.job.core.domain.JobQueueRequest;
 import com.lts.job.core.domain.PageResponse;
-import com.lts.job.core.exception.CronException;
 import com.lts.job.core.support.CronExpression;
-import com.lts.job.core.util.CollectionUtils;
-import com.lts.job.core.util.StringUtils;
 import com.lts.job.queue.domain.JobPo;
+import com.lts.job.queue.exception.DuplicateJobException;
 import com.lts.job.web.cluster.AdminApplication;
 import com.lts.job.web.controller.AbstractController;
 import com.lts.job.web.vo.RestfulResponse;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
-import java.util.List;
-import java.util.Map;
+import java.text.ParseException;
+import java.util.Date;
 
 /**
  * @author Robert HG (254963746@qq.com) on 6/6/15.
@@ -63,7 +63,14 @@ public RestfulResponse cronJobUpdate(JobQueueRequest request) {
         // 检查参数
         // 1. 检测 cronExpression是否是正确的
         if (StringUtils.isNotEmpty(request.getCronExpression())) {
-            if (!CronExpression.isValidExpression(request.getCronExpression())) {
+            try {
+                CronExpression expression = new CronExpression(request.getCronExpression());
+                if (expression.getTimeAfter(new Date()) == null) {
+                    response.setSuccess(false);
+                    response.setMsg(StringUtils.format("该CronExpression={} 已经没有执行时间点!", request.getCronExpression()));
+                    return response;
+                }
+            } catch (ParseException e) {
                 response.setSuccess(false);
                 response.setMsg("请输入正确的 CronExpression!");
                 return response;
@@ -87,4 +94,95 @@ public RestfulResponse cronJobDelete(JobQueueRequest request) {
         return response;
     }
 
+    @RequestMapping("/job-logger/job-logger-get")
+    public RestfulResponse jobLoggerGet(JobLoggerRequest request) {
+        RestfulResponse response = new RestfulResponse();
+
+        try {
+            Assert.hasLength(request.getTaskId(), "taskId不能为空!");
+            Assert.hasLength(request.getTaskTrackerNodeGroup(), "taskTrackerNodeGroup不能为空!");
+        } catch (IllegalArgumentException e) {
+            response.setSuccess(false);
+            response.setMsg(e.getMessage());
+            return response;
+        }
+
+        PageResponse<JobLogPo> pageResponse = application.getJobLogger().search(request);
+        response.setResults(pageResponse.getResults());
+        response.setRows(pageResponse.getRows());
+
+        response.setSuccess(true);
+        return response;
+    }
+
+    @RequestMapping("/job-queue/job-add")
+    public RestfulResponse jobAdd(JobQueueRequest request) {
+        RestfulResponse response = new RestfulResponse();
+        // 表单check
+
+        Long triggerTime = null;
+        try {
+            Assert.hasLength(request.getTaskId(), "taskId不能为空!");
+            Assert.hasLength(request.getTaskTrackerNodeGroup(), "taskTrackerNodeGroup不能为空!");
+            Assert.hasLength(request.getSubmitNodeGroup(), "submitNodeGroup不能为空!");
+
+            if (StringUtils.isNotEmpty(request.getCronExpression())) {
+                try {
+                    CronExpression expression = new CronExpression(request.getCronExpression());
+                    Date nextTime = expression.getTimeAfter(new Date());
+                    if (nextTime == null) {
+                        response.setSuccess(false);
+                        response.setMsg(StringUtils.format("该CronExpression={} 已经没有执行时间点!", request.getCronExpression()));
+                        return response;
+                    } else {
+                        triggerTime = nextTime.getTime();
+                    }
+                } catch (ParseException e) {
+                    response.setSuccess(false);
+                    response.setMsg("请输入正确的 CronExpression!");
+                    return response;
+                }
+            }
+
+        } catch (IllegalArgumentException e) {
+            response.setSuccess(false);
+            response.setMsg(e.getMessage());
+            return response;
+        }
+
+        addJob(request, triggerTime);
+
+        response.setSuccess(true);
+        return response;
+    }
+
+    private void addJob(JobQueueRequest request, Long triggerTime) {
+        JobPo jobPo = new JobPo();
+        // 这里暂时用UUID来代替
+        jobPo.setJobId(StringUtils.generateUUID());
+        jobPo.setCronExpression(request.getCronExpression());
+        jobPo.setExtParams(request.getExtParams());
+        jobPo.setGmtCreated(DateUtils.currentTimeMillis());
+        jobPo.setGmtModified(jobPo.getGmtCreated());
+        jobPo.setNeedFeedback(request.getNeedFeedback());
+        jobPo.setPriority(request.getPriority());
+        jobPo.setTaskId(request.getTaskId());
+        jobPo.setSubmitNodeGroup(request.getSubmitNodeGroup());
+        jobPo.setTaskTrackerNodeGroup(request.getTaskTrackerNodeGroup());
+        if (request.getTriggerTime() != null) {
+            jobPo.setTriggerTime(request.getTriggerTime().getTime());
+        }
+
+        if (jobPo.isSchedule()) {
+            application.getCronJobQueue().add(jobPo);
+            if (triggerTime != null) {
+                jobPo.setTriggerTime(triggerTime);
+            }
+        }
+        if (jobPo.getTriggerTime() == null) {
+            jobPo.setTriggerTime(DateUtils.currentTimeMillis());
+        }
+
+        application.getExecutableJobQueue().add(jobPo);
+    }
 }

lts-admin/src/main/java/com/lts/job/web/controller/api/NodeApiController.java

@@ -1,8 +1,8 @@
 package com.lts.job.web.controller.api;
 
 import com.lts.job.core.cluster.Node;
-import com.lts.job.core.util.CollectionUtils;
-import com.lts.job.core.util.StringUtils;
+import com.lts.job.core.commons.utils.CollectionUtils;
+import com.lts.job.core.commons.utils.StringUtils;
 import com.lts.job.web.cluster.RegistryService;
 import com.lts.job.web.controller.AbstractController;
 import com.lts.job.web.request.NodeRequest;

lts-admin/src/main/java/com/lts/job/web/controller/ui/UIController.java

@@ -1,6 +1,7 @@
 package com.lts.job.web.controller.ui;
 
 import com.lts.job.core.cluster.NodeType;
+import com.lts.job.core.commons.utils.DateUtils;
 import com.lts.job.queue.domain.NodeGroupPo;
 import com.lts.job.web.cluster.AdminApplication;
 import com.lts.job.web.cluster.RegistryService;
@@ -9,6 +10,7 @@
 import org.springframework.ui.Model;
 import org.springframework.web.bind.annotation.RequestMapping;
 
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -29,6 +31,29 @@ public String nodeManagerUI(Model model) {
         return "node-manager";
     }
 
+    @RequestMapping("job-queue/job-add-ui")
+    public String addJobUI(Model model) {
+        setAttr(model);
+        return "job-add";
+    }
+
+    @RequestMapping("job-logger/job-logger")
+    public String jobLoggerUI(Model model, String taskId, String taskTrackerNodeGroup,
+                              Date startTimestamp, Date endTimestamp) {
+        model.addAttribute("taskId", taskId);
+        model.addAttribute("taskTrackerNodeGroup", taskTrackerNodeGroup);
+        if (startTimestamp == null) {
+            startTimestamp = DateUtils.addDay(new Date(), -3);
+        }
+        model.addAttribute("startTimestamp", DateUtils.formatYMD_HMS(startTimestamp));
+        if (endTimestamp == null) {
+            endTimestamp = new Date();
+        }
+        model.addAttribute("endTimestamp", DateUtils.formatYMD_HMS(endTimestamp));
+        setAttr(model);
+        return "job-logger";
+    }
+
     @RequestMapping("job-queue/cron-job-queue")
     public String cronJobQueueUI(Model model) {
         setAttr(model);
@@ -47,10 +72,11 @@ public String executingJobQueueUI(Model model) {
         return "executing-job-queue";
     }
 
-    private void setAttr(Model model){
+    private void setAttr(Model model) {
         List<NodeGroupPo> jobClientNodeGroups = application.getNodeGroupStore().getNodeGroup(NodeType.JOB_CLIENT);
         model.addAttribute("jobClientNodeGroups", jobClientNodeGroups);
         List<NodeGroupPo> taskTrackerNodeGroups = application.getNodeGroupStore().getNodeGroup(NodeType.TASK_TRACKER);
         model.addAttribute("taskTrackerNodeGroups", taskTrackerNodeGroups);
     }
+
 }

lts-admin/src/main/java/com/lts/job/web/support/AdminAppFactoryBean.java

@@ -1,10 +1,11 @@
 package com.lts.job.web.support;
 
+import com.lts.job.biz.logger.JobLoggerFactory;
 import com.lts.job.core.cluster.Config;
 import com.lts.job.core.cluster.Node;
 import com.lts.job.core.cluster.NodeType;
 import com.lts.job.core.extension.ExtensionLoader;
-import com.lts.job.core.util.StringUtils;
+import com.lts.job.core.commons.utils.StringUtils;
 import com.lts.job.queue.*;
 import com.lts.job.web.cluster.AdminApplication;
 import org.springframework.beans.factory.FactoryBean;
@@ -25,6 +26,8 @@ public class AdminAppFactoryBean implements FactoryBean<AdminApplication>, Initi
             ExecutingJobQueueFactory.class).getAdaptiveExtension();
     NodeGroupStoreFactory nodeGroupStoreFactory = ExtensionLoader.getExtensionLoader(
             NodeGroupStoreFactory.class).getAdaptiveExtension();
+    JobLoggerFactory jobLoggerFactory = ExtensionLoader.getExtensionLoader(
+            JobLoggerFactory.class).getAdaptiveExtension();
 
     private AdminApplication application;
 
@@ -70,6 +73,7 @@ public void afterPropertiesSet() throws Exception {
         application.setExecutableJobQueue(executableJobQueueFactory.getQueue(config));
         application.setExecutingJobQueue(executingJobQueueFactory.getQueue(config));
         application.setNodeGroupStore(nodeGroupStoreFactory.getStore(config));
+        application.setJobLogger(jobLoggerFactory.getJobLogger(config));
     }
 
 }

lts-admin/src/main/java/com/lts/job/web/support/DateEditor.java

@@ -1,6 +1,6 @@
 package com.lts.job.web.support;
 
-import com.lts.job.core.util.DateUtils;
+import com.lts.job.core.commons.utils.DateUtils;
 import org.springframework.util.StringUtils;
 
 import java.beans.PropertyEditorSupport;

lts-admin/src/main/java/com/lts/job/web/support/ExceptionHandlerResolver.java

@@ -1,6 +1,6 @@
 package com.lts.job.web.support;
 
-import com.lts.job.core.util.JSONUtils;
+import com.lts.job.core.commons.utils.JSONUtils;
 import com.lts.job.web.vo.RestfulResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

lts-admin/src/main/java/com/lts/job/web/support/MapEditor.java

@@ -1,7 +1,7 @@
 package com.lts.job.web.support;
 
 import com.alibaba.fastjson.TypeReference;
-import com.lts.job.core.util.JSONUtils;
+import com.lts.job.core.commons.utils.JSONUtils;
 import org.springframework.util.StringUtils;
 
 import java.beans.PropertyEditorSupport;