[dev] add instanceLog Cleaner

.gitignore

@@ -34,3 +34,4 @@ build/
 *.jar
 *.log
 */.DS_Store
+.DS_Store

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/akka/actors/ServerActor.java

@@ -7,7 +7,7 @@ import com.github.kfcfans.oms.common.request.WorkerHeartbeat;
 import com.github.kfcfans.oms.common.request.WorkerLogReportReq;
 import com.github.kfcfans.oms.common.response.AskResponse;
 import com.github.kfcfans.oms.server.common.utils.SpringUtils;
-import com.github.kfcfans.oms.server.service.InstanceLogService;
+import com.github.kfcfans.oms.server.service.log.InstanceLogService;
 import com.github.kfcfans.oms.server.service.instance.InstanceManager;
 import com.github.kfcfans.oms.server.service.ha.WorkerManagerService;
 import lombok.extern.slf4j.Slf4j;

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/instance/InstanceManager.java

@@ -11,7 +11,7 @@ import com.github.kfcfans.oms.server.persistence.core.repository.InstanceInfoRep
 import com.github.kfcfans.oms.server.persistence.core.repository.JobInfoRepository;
 import com.github.kfcfans.oms.server.persistence.core.repository.UserInfoRepository;
 import com.github.kfcfans.oms.server.service.DispatchService;
-import com.github.kfcfans.oms.server.service.InstanceLogService;
+import com.github.kfcfans.oms.server.service.log.InstanceLogService;
 import com.github.kfcfans.oms.server.service.alarm.AlarmContent;
 import com.github.kfcfans.oms.server.service.alarm.Alarmable;
 import com.github.kfcfans.oms.server.service.timing.schedule.HashedWheelTimerHolder;

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/log/InstanceLogCleanService.java

@@ -0,0 +1,109 @@
+package com.github.kfcfans.oms.server.service.log;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Stopwatch;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.time.DateUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.gridfs.GridFsTemplate;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.io.File;
+import java.util.Date;
+
+/**
+ * 定时清理任务实例运行日志，包括本地和MongoDB
+ *
+ * @author tjq
+ * @since 2020/5/11
+ */
+@Slf4j
+@Service
+public class InstanceLogCleanService {
+
+    @Value("${oms.log.retention.local}")
+    private int localRetentionDay;
+    @Value("${oms.log.retention.remote}")
+    private int remoteRetentionDay;
+
+    // 直接操作 mongoDB 文件系统
+    private GridFsTemplate gridFsTemplate;
+    // 每天凌晨3点定时清理
+    private static final String CLEAN_TIME_EXPRESSION = "0 0 3 * * ?";
+
+    @Async("omsTimingPool")
+    @Scheduled(cron = CLEAN_TIME_EXPRESSION)
+    public void timingClean() {
+        cleanRemote();
+        cleanLocal();
+    }
+
+    @VisibleForTesting
+    public void cleanLocal() {
+
+        if (localRetentionDay < 0) {
+            log.info("[InstanceLogCleanService] won't clean up local logs because of localRetentionDay <= 0.");
+            return;
+        }
+
+        Stopwatch stopwatch = Stopwatch.createStarted();
+        String path = InstanceLogService.genLogDirPath();
+        File dir = new File(path);
+        if (!dir.exists()) {
+            return;
+        }
+        File[] logFiles = dir.listFiles();
+        if (logFiles == null || logFiles.length == 0) {
+            return;
+        }
+
+        // 计算最大偏移量
+        long maxOffset = localRetentionDay * 24 * 60 * 60 * 1000;
+
+        for (File f : logFiles) {
+            long offset = System.currentTimeMillis() - f.lastModified();
+            if (offset >= maxOffset) {
+                if (!f.delete()) {
+                    log.warn("[InstanceLogCleanService] delete local log({}) failed.", f.getName());
+                }else {
+                    log.info("[InstanceLogCleanService] delete local log({}) successfully.", f.getName());
+                }
+            }
+        }
+        log.info("[InstanceLogCleanService] clean local instance log file successfully, using {}.", stopwatch.stop());
+    }
+
+    @VisibleForTesting
+    public void cleanRemote() {
+
+        if (remoteRetentionDay < 0) {
+            log.info("[InstanceLogCleanService] won't clean up remote logs because of remoteRetentionDay <= 0.");
+            return;
+        }
+        if (gridFsTemplate == null) {
+            return;
+        }
+
+        Stopwatch stopwatch = Stopwatch.createStarted();
+        try {
+            // 计算时间
+            Date date = DateUtils.addDays(new Date(), -remoteRetentionDay);
+            Query mongoQuery = Query.query(Criteria.where("uploadDate").lt(date));
+            gridFsTemplate.delete(mongoQuery);
+        }catch (Exception e) {
+            log.warn("[InstanceLogCleanService] clean remote log failed.", e);
+        }
+        log.info("[InstanceLogCleanService] clean remote instance log file finished, using {}.", stopwatch.stop());
+    }
+
+    @Autowired(required = false)
+    public void setGridFsTemplate(GridFsTemplate gridFsTemplate) {
+        this.gridFsTemplate = gridFsTemplate;
+    }
+
+}

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/log/InstanceLogService.java

@@ -1,4 +1,4 @@
-package com.github.kfcfans.oms.server.service;
+package com.github.kfcfans.oms.server.service.log;
 
 import com.github.kfcfans.oms.common.TimeExpressionType;
 import com.github.kfcfans.oms.common.model.InstanceLogContent;
@@ -216,6 +216,13 @@ public class InstanceLogService {
                 if (f.exists() && (System.currentTimeMillis() - f.lastModified()) < EXPIRE_INTERVAL_MS) {
                     return f;
                 }
+
+                // 创建父文件夹（文件在开流时自动会被创建）
+                if (!f.getParentFile().exists()) {
+                    if (!f.getParentFile().mkdirs()) {
+                        throw new RuntimeException("create dir failed");
+                    }
+                }
                 // 重新构建文件
                 try (Stream<LocalInstanceLogDO> allLogStream = localInstanceLogRepository.findByInstanceIdOrderByLogTime(instanceId)) {
                     stream2File(allLogStream, f);
@@ -234,6 +241,13 @@ public class InstanceLogService {
                     return f;
                 }
 
+                // 创建父文件夹（文件在开流时自动会被创建）
+                if (!f.getParentFile().exists()) {
+                    if (!f.getParentFile().mkdirs()) {
+                        throw new RuntimeException("create dir failed");
+                    }
+                }
+
                 // 本地存在数据，从本地持久化（对应 SYNC 的情况）
                 if (instanceId2LastReportTime.containsKey(instanceId)) {
                     try (Stream<LocalInstanceLogDO> allLogStream = localInstanceLogRepository.findByInstanceIdOrderByLogTime(instanceId)) {
@@ -266,12 +280,6 @@ public class InstanceLogService {
      * @param logFile 目标日志文件
      */
     private void stream2File(Stream<LocalInstanceLogDO> stream, File logFile) {
-        if (!logFile.getParentFile().exists()) {
-            if (!logFile.getParentFile().mkdirs()) {
-                log.warn("[InstanceLogService] create dir for instanceLog failed, path is {}.", logFile.getPath());
-                return;
-            }
-        }
         try (FileWriter fw = new FileWriter(logFile); BufferedWriter bfw = new BufferedWriter(fw)) {
             stream.forEach(instanceLog -> {
                 try {
@@ -354,12 +362,15 @@ public class InstanceLogService {
         // 2. 删除长时间未 REPORT 的日志
     }
 
+    public static String genLogDirPath() {
+        return USER_HOME + "/oms-server/online_log/";
+    }
 
     private static String genLogFilePath(long instanceId, boolean stable) {
         if (stable) {
-            return USER_HOME + "/oms-server/online_log/" + String.format("%d-stable.log", instanceId);
+            return genLogDirPath() + String.format("%d-stable.log", instanceId);
         }else {
-            return USER_HOME + "/oms-server/online_log/" + String.format("%d-temporary.log", instanceId);
+            return genLogDirPath() + String.format("%d-temporary.log", instanceId);
         }
     }
     private static String genMongoFileName(long instanceId) {

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/web/controller/InstanceController.java

@@ -11,27 +11,22 @@ import com.github.kfcfans.oms.server.persistence.core.model.InstanceInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.repository.AppInfoRepository;
 import com.github.kfcfans.oms.server.persistence.core.repository.InstanceInfoRepository;
 import com.github.kfcfans.oms.server.service.CacheService;
-import com.github.kfcfans.oms.server.service.InstanceLogService;
+import com.github.kfcfans.oms.server.service.log.InstanceLogService;
 import com.github.kfcfans.oms.server.service.instance.InstanceService;
 import com.github.kfcfans.oms.server.web.request.QueryInstanceRequest;
 import com.github.kfcfans.oms.server.web.response.InstanceLogVO;
 import org.apache.commons.lang3.time.DateFormatUtils;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.core.io.InputStreamResource;
-import org.springframework.core.io.UrlResource;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 
 import javax.annotation.Resource;
 import javax.servlet.http.HttpServletResponse;
 import java.io.*;
 import java.net.URLEncoder;
-import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;

oh-my-scheduler-server/src/main/resources/application-daily.properties

@@ -18,4 +18,8 @@ spring.mail.username=zqq
 spring.mail.password=qqz
 spring.mail.properties.mail.smtp.auth=true
 spring.mail.properties.mail.smtp.starttls.enable=true
-spring.mail.properties.mail.smtp.starttls.required=true
+spring.mail.properties.mail.smtp.starttls.required=true
+
+####### 日志保留天数，单位天 #######
+oms.log.retention.local=0
+oms.log.retention.remote=0

oh-my-scheduler-server/src/main/resources/application-pre.properties

@@ -18,4 +18,8 @@ spring.mail.username=zqq
 spring.mail.password=qqz
 spring.mail.properties.mail.smtp.auth=true
 spring.mail.properties.mail.smtp.starttls.enable=true
-spring.mail.properties.mail.smtp.starttls.required=true
+spring.mail.properties.mail.smtp.starttls.required=true
+
+####### 日志保留天数，单位天 #######
+oms.log.retention.local=3
+oms.log.retention.remote=3

oh-my-scheduler-server/src/main/resources/application-product.properties

@@ -18,4 +18,8 @@ spring.mail.username=zqq
 spring.mail.password=qqz
 spring.mail.properties.mail.smtp.auth=true
 spring.mail.properties.mail.smtp.starttls.enable=true
-spring.mail.properties.mail.smtp.starttls.required=true
+spring.mail.properties.mail.smtp.starttls.required=true
+
+####### 日志保留天数，单位天 #######
+oms.log.retention.local=3
+oms.log.retention.remote=3

oh-my-scheduler-server/src/main/resources/application.properties

@@ -1,5 +1,6 @@
 server.port=7700
 
+spring.profiles.active=daily
 spring.jpa.open-in-view=false
 spring.data.mongodb.repositories.type=none
 

oh-my-scheduler-server/src/main/resources/static/css/chunk-4947264f.596fc436.css

@@ -1 +1 @@
-.wrap[data-v-9781a346]{background:#fff;display:flex;text-align:center;justify-content:space-around;align-items:center;margin:10px;box-shadow:0 2px 12px 0 rgba(0,0,0,.2);font-size:1.5rem;font-weight:bolder;height:131px}.el-table .warning-row{background:#fdf5e6}.el-table .success-row{background:#8fbc8f}.el-table .error-row{background:#ff5831}
+.wrap[data-v-bbc92dba]{background:#fff;display:flex;text-align:center;justify-content:space-around;align-items:center;margin:10px;box-shadow:0 2px 12px 0 rgba(0,0,0,.2);font-size:1.5rem;font-weight:bolder;height:131px}.el-table .warning-row{background:#fdf5e6}.el-table .success-row{background:#8fbc8f}.el-table .error-row{background:#ff5831}

oh-my-scheduler-server/src/main/resources/static/index.html

@@ -1 +1 @@
-<!DOCTYPE html><html lang=en><head><meta charset=utf-8><meta http-equiv=X-UA-Compatible content="IE=edge"><meta name=viewport content="width=device-width,initial-scale=1"><link rel=icon href=/favicon.ico><title>oms-console</title><link href=/css/chunk-29e8f75a.57228328.css rel=prefetch><link href=/css/chunk-7602a25e.b9e33c01.css rel=prefetch><link href=/js/chunk-29e8f75a.a9aa0f8f.js rel=prefetch><link href=/js/chunk-2d0c76e2.50e949d3.js rel=prefetch><link href=/js/chunk-2d21772a.415195dc.js rel=prefetch><link href=/js/chunk-7602a25e.165fb79d.js rel=prefetch><link href=/css/app.46de59ae.css rel=preload as=style><link href=/js/app.65e5a27b.js rel=preload as=script><link href=/js/chunk-vendors.193746e8.js rel=preload as=script><link href=/css/app.46de59ae.css rel=stylesheet></head><body><noscript><strong>We're sorry but oms-console doesn't work properly without JavaScript enabled. Please enable it to continue.</strong></noscript><div id=app></div><script src=/js/chunk-vendors.193746e8.js></script><script src=/js/app.65e5a27b.js></script></body></html>
+<!DOCTYPE html><html lang=en><head><meta charset=utf-8><meta http-equiv=X-UA-Compatible content="IE=edge"><meta name=viewport content="width=device-width,initial-scale=1"><link rel=icon href=/favicon.ico><title>oms-console</title><link href=/css/chunk-4947264f.596fc436.css rel=prefetch><link href=/css/chunk-7602a25e.b9e33c01.css rel=prefetch><link href=/js/chunk-2d0c76e2.d0248019.js rel=prefetch><link href=/js/chunk-2d21772a.415195dc.js rel=prefetch><link href=/js/chunk-4947264f.10340ff7.js rel=prefetch><link href=/js/chunk-7602a25e.165fb79d.js rel=prefetch><link href=/css/app.46de59ae.css rel=preload as=style><link href=/js/app.1ed552fe.js rel=preload as=script><link href=/js/chunk-vendors.193746e8.js rel=preload as=script><link href=/css/app.46de59ae.css rel=stylesheet></head><body><noscript><strong>We're sorry but oms-console doesn't work properly without JavaScript enabled. Please enable it to continue.</strong></noscript><div id=app></div><script src=/js/chunk-vendors.193746e8.js></script><script src=/js/app.1ed552fe.js></script></body></html>

oh-my-scheduler-server/src/main/resources/static/js/chunk-29e8f75a.a9aa0f8f.js

@@ -1,2 +0,0 @@
-(window["webpackJsonp"]=window["webpackJsonp"]||[]).push([["chunk-29e8f75a"],{6337:function(t,s,a){"use strict";var e=a("ffdc"),n=a.n(e);n.a},"7d8a":function(t,s,a){"use strict";a.r(s);var e=function(){var t=this,s=t.$createElement,a=t._self._c||s;return a("div",{attrs:{id:"home"}},[a("el-row",{attrs:{gutter:24}},[a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("任务总数")]),a("div",{staticClass:"text"},[t._v(t._s(t.systemInfo.jobCount))])]),a("i",{staticClass:"el-icon-s-custom"})])]),a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("当前运行实例数")]),a("div",{staticClass:"text"},[t._v(t._s(t.systemInfo.runningInstanceCount))])]),a("i",{staticClass:"el-icon-s-custom"})])]),a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("近期失败任务数")]),a("div",{staticClass:"text"},[t._v(t._s(t.systemInfo.failedInstanceCount))])]),a("i",{staticClass:"el-icon-s-custom"})])]),a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("集群机器数")]),a("div",{staticClass:"text"},[t._v(t._s(t.activeWorkerCount))])]),a("i",{staticClass:"el-icon-s-custom"})])])],1),a("el-row",[a("el-col",{attrs:{span:24}},[a("el-table",{staticStyle:{width:"100%"},attrs:{data:t.workerList,height:"400px","row-class-name":t.workerTableRowClassName}},[a("el-table-column",{attrs:{prop:"address",label:"机器地址"}}),a("el-table-column",{attrs:{prop:"cpuLoad",label:"CPU占用"}}),a("el-table-column",{attrs:{prop:"memoryLoad",label:"内存占用"}}),a("el-table-column",{attrs:{prop:"diskLoad",label:"磁盘占用"}})],1)],1)],1)],1)},n=[],i={name:"Home",data:function(){return{systemInfo:{jobCount:"N/A",runningInstanceCount:"N/A",failedInstanceCount:"N/A"},activeWorkerCount:"N/A",workerList:[]}},methods:{workerTableRowClassName:function(t){var s=t.row;switch(s.status){case 1:return"success-row";case 2:return"warning-row";case 3:return"error-row"}}},mounted:function(){var t=this,s=t.$store.state.appInfo.id;t.axios.get("/system/overview?appId="+s).then((function(s){return t.systemInfo=s})),t.axios.get("/system/listWorker?appId="+s).then((function(s){t.workerList=s,t.activeWorkerCount=t.workerList.length}))}},o=i,r=(a("ba4d"),a("6337"),a("2877")),c=Object(r["a"])(o,e,n,!1,null,"9781a346",null);s["default"]=c.exports},ba4d:function(t,s,a){"use strict";var e=a("ff73"),n=a.n(e);n.a},ff73:function(t,s,a){},ffdc:function(t,s,a){}}]);
-//# sourceMappingURL=chunk-29e8f75a.a9aa0f8f.js.map

oh-my-scheduler-server/src/main/resources/static/js/chunk-4947264f.10340ff7.js

@@ -0,0 +1,2 @@
+(window["webpackJsonp"]=window["webpackJsonp"]||[]).push([["chunk-4947264f"],{"05fd":function(t,s,a){"use strict";var e=a("ac51"),n=a.n(e);n.a},6337:function(t,s,a){"use strict";var e=a("ffdc"),n=a.n(e);n.a},"7d8a":function(t,s,a){"use strict";a.r(s);var e=function(){var t=this,s=t.$createElement,a=t._self._c||s;return a("div",{attrs:{id:"home"}},[a("el-row",{attrs:{gutter:24}},[a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("任务总数")]),a("div",{staticClass:"text"},[t._v(t._s(t.systemInfo.jobCount))])]),a("i",{staticClass:"el-icon-orange"})])]),a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("当前运行实例数")]),a("div",{staticClass:"text"},[t._v(t._s(t.systemInfo.runningInstanceCount))])]),a("i",{staticClass:"el-icon-loading"})])]),a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("近期失败任务数")]),a("div",{staticClass:"text"},[t._v(t._s(t.systemInfo.failedInstanceCount))])]),a("i",{staticClass:"el-icon-bell"})])]),a("el-col",{attrs:{span:6}},[a("div",{staticClass:"wrap"},[a("div",{staticClass:"grid-content bg-purple"},[a("div",{staticClass:"text"},[t._v("集群机器数")]),a("div",{staticClass:"text"},[t._v(t._s(t.activeWorkerCount))])]),a("i",{staticClass:"el-icon-grape"})])])],1),a("el-row",[a("el-col",{attrs:{span:24}},[a("el-table",{staticStyle:{width:"100%"},attrs:{data:t.workerList,height:"400px","row-class-name":t.workerTableRowClassName}},[a("el-table-column",{attrs:{prop:"address",label:"机器地址"}}),a("el-table-column",{attrs:{prop:"cpuLoad",label:"CPU占用"}}),a("el-table-column",{attrs:{prop:"memoryLoad",label:"内存占用"}}),a("el-table-column",{attrs:{prop:"diskLoad",label:"磁盘占用"}})],1)],1)],1)],1)},n=[],r={name:"Home",data:function(){return{systemInfo:{jobCount:"N/A",runningInstanceCount:"N/A",failedInstanceCount:"N/A"},activeWorkerCount:"N/A",workerList:[]}},methods:{workerTableRowClassName:function(t){var s=t.row;switch(s.status){case 1:return"success-row";case 2:return"warning-row";case 3:return"error-row"}}},mounted:function(){var t=this,s=t.$store.state.appInfo.id;t.axios.get("/system/overview?appId="+s).then((function(s){return t.systemInfo=s})),t.axios.get("/system/listWorker?appId="+s).then((function(s){t.workerList=s,t.activeWorkerCount=t.workerList.length}))}},i=r,o=(a("05fd"),a("6337"),a("2877")),l=Object(o["a"])(i,e,n,!1,null,"bbc92dba",null);s["default"]=l.exports},ac51:function(t,s,a){},ffdc:function(t,s,a){}}]);
+//# sourceMappingURL=chunk-4947264f.10340ff7.js.map

oh-my-scheduler-server/src/test/java/com/github/kfcfans/oms/server/test/OmsLogTest.java

@@ -0,0 +1,58 @@
+package com.github.kfcfans.oms.server.test;
+
+import com.github.kfcfans.oms.server.service.log.InstanceLogCleanService;
+import com.mongodb.client.gridfs.model.GridFSFile;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.gridfs.GridFsTemplate;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import javax.annotation.Resource;
+import java.util.Date;
+import java.util.function.Consumer;
+
+/**
+ * 在线日志测试
+ *
+ * @author tjq
+ * @since 2020/5/11
+ */
+
+@ActiveProfiles("daily")
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+public class OmsLogTest {
+
+    @Resource
+    private MongoTemplate mongoTemplate;
+    @Resource
+    private GridFsTemplate gridFsTemplate;
+    @Resource
+    private InstanceLogCleanService instanceLogCleanService;
+
+    @Test
+    public void testLocalLogCleaner() {
+        instanceLogCleanService.cleanLocal();
+    }
+
+    @Test
+    public void testRemoteLogCleaner() {
+        instanceLogCleanService.cleanRemote();
+    }
+
+    @Test
+    public void testGridFsQuery() {
+        Query mongoQuery = Query.query(Criteria.where("uploadDate").gt(new Date()));
+        gridFsTemplate.find(mongoQuery).forEach(new Consumer<GridFSFile>() {
+            @Override
+            public void accept(GridFSFile gridFSFile) {
+                System.out.println(gridFSFile.getFilename());
+            }
+        });
+    }
+}

oh-my-scheduler-server/src/test/java/com/github/kfcfans/oms/server/test/RepositoryTest.java

@@ -13,6 +13,7 @@ import org.assertj.core.util.Lists;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import javax.annotation.Resource;
@@ -24,6 +25,7 @@ import java.util.List;
  * @author tjq
  * @since 2020/4/5
  */
+@ActiveProfiles("daily")
 @RunWith(SpringRunner.class)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class RepositoryTest {

oh-my-scheduler-server/src/test/java/com/github/kfcfans/oms/server/test/ServiceTest.java

@@ -6,6 +6,7 @@ import org.assertj.core.util.Lists;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import javax.annotation.Resource;
@@ -17,6 +18,7 @@ import java.util.List;
  * @author tjq
  * @since 2020/4/2
  */
+@ActiveProfiles("daily")
 @RunWith(SpringRunner.class)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class ServiceTest {

oh-my-scheduler-worker-samples/src/main/java/com/github/kfcfans/oms/server/OhMySchedulerConfig.java

@@ -21,13 +21,14 @@ public class OhMySchedulerConfig {
     @Bean
     public OhMyWorker initOMS() throws Exception {
 
-        List<String> serverAddress = Lists.newArrayList("192.168.1.6:7700", "127.0.0.1:7700");
+        List<String> serverAddress = Lists.newArrayList("127.0.0.1:7700", "127.0.0.1:7701");
 
         // 1. 创建配置文件
         OhMyConfig config = new OhMyConfig();
         config.setAppName("oms-test");
         config.setServerAddress(serverAddress);
         // 如果没有大型 Map/MapReduce 的需求，建议使用内存来加速计算
+        // 为了本地模拟多个实例，只能使用 MEMORY 启动（文件只能由一个应用占有）
         config.setStoreStrategy(StoreStrategy.MEMORY);
 
         // 2. 创建 Worker 对象，设置配置文件

oh-my-scheduler-worker-samples/src/main/java/com/github/kfcfans/oms/server/processors/MapReduceProcessorDemo.java

@@ -1,5 +1,6 @@
 package com.github.kfcfans.oms.server.processors;
 
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.github.kfcfans.oms.common.utils.JsonUtils;
 import com.github.kfcfans.oms.worker.core.processor.ProcessResult;
@@ -29,11 +30,6 @@ import java.util.concurrent.ThreadLocalRandom;
 @Component
 public class MapReduceProcessorDemo extends MapReduceProcessor {
 
-    // 每一批发送任务大小
-    private static final int batchSize = 100;
-    // 发送的批次
-    private static final int batchNum = 2;
-
     @Override
     public ProcessResult process(TaskContext context) throws Exception {
 
@@ -43,6 +39,12 @@ public class MapReduceProcessorDemo extends MapReduceProcessor {
         System.out.println("isRootTask:" + isRootTask());
         System.out.println("taskContext:" + JsonUtils.toJSONString(context));
 
+        // 根据控制台参数获取MR批次及子任务大小
+        final JSONObject jobParams = JSONObject.parseObject(context.getJobParams());
+
+        Integer batchSize = (Integer) jobParams.getOrDefault("batchSize", 100);
+        Integer batchNum = (Integer) jobParams.getOrDefault("batchNum", 10);
+
         if (isRootTask()) {
             System.out.println("==== MAP ====");
             omsLogger.info("[DemoMRProcessor] start root task~");
@@ -60,7 +62,7 @@ public class MapReduceProcessorDemo extends MapReduceProcessor {
             return new ProcessResult(true, "MAP_SUCCESS");
         }else {
             System.out.println("==== NORMAL_PROCESS ====");
-            omsLogger.info("[DemoMRProcessor] normal process~");
+            omsLogger.info("[DemoMRProcessor] process subTask: {}.", JSON.toJSONString(context.getSubTask()));
             System.out.println("subTask: " + JsonUtils.toJSONString(context.getSubTask()));
             Thread.sleep(1000);
             if (context.getCurrentRetryTimes() == 0) {

oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/actors/TaskTrackerActor.java

@@ -48,10 +48,10 @@ public class TaskTrackerActor extends AbstractActor {
     private void onReceiveProcessorReportTaskStatusReq(ProcessorReportTaskStatusReq req) {
 
         TaskTracker taskTracker = TaskTrackerPool.getTaskTrackerPool(req.getInstanceId());
+        // 手动停止 TaskTracker 的情况下会出现这种情况
         if (taskTracker == null) {
             log.warn("[TaskTrackerActor] receive ProcessorReportTaskStatusReq({}) but system can't find TaskTracker.", req);
         } else {
-
             taskTracker.updateTaskStatus(req.getTaskId(), req.getStatus(), req.getReportTime(), req.getResult());
         }
     }

oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/background/OmsLogHandler.java

@@ -76,8 +76,10 @@ public class OmsLogHandler {
                 String serverPath = AkkaUtils.getAkkaServerPath(RemoteConstant.SERVER_ACTOR_NAME);
                 // 当前无可用 Server
                 if (StringUtils.isEmpty(serverPath)) {
-                    logQueue.clear();
+                    if (!logQueue.isEmpty()) {
-                    log.warn("[OmsLogHandler] because there is no available server to report logs which leads to queue accumulation, oms discarded all logs.");
+                        logQueue.clear();
+                        log.warn("[OmsLogHandler] because there is no available server to report logs which leads to queue accumulation, oms discarded all logs.");
+                    }
                     return;
                 }
 

oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/core/executor/ProcessorRunnable.java

@@ -43,7 +43,7 @@ public class ProcessorRunnable implements Runnable {
     private final BasicProcessor processor;
     private final OmsLogger omsLogger;
 
-    public void innerRun() {
+    public void innerRun() throws InterruptedException {
 
         String taskId = task.getTaskId();
         Long instanceId = task.getInstanceId();
@@ -176,6 +176,7 @@ public class ProcessorRunnable implements Runnable {
     public void run() {
         try {
             innerRun();
+        }catch (InterruptedException ignore) {
         }catch (Exception e) {
             log.error("[ProcessorRunnable-{}] execute failed, please fix this bug @tjq!", task.getInstanceId(), e);
         }finally {

oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/core/tracker/processor/ProcessorTracker.java

@@ -77,12 +77,12 @@ public class ProcessorTracker {
 
             this.omsLogger = new OmsServerLogger(instanceId);
 
+            // 初始化 线程池，TimingPool 启动的任务会检查 ThreadPool，所以必须先初始化线程池，否则NPE
+            initThreadPool();
             // 初始化定时任务
             initTimingJob();
             // 初始化 Processor
             initProcessor();
-            // 初始化 线程池，如果处理器创建失败则不执行
-            initThreadPool();
 
             log.info("[ProcessorTracker-{}] ProcessorTracker was successfully created!", instanceId);
         }catch (Exception e) {
@@ -221,6 +221,7 @@ public class ProcessorTracker {
             long waitingNum = threadPool.getQueue().size();
             ProcessorTrackerStatusReportReq req = new ProcessorTrackerStatusReportReq(instanceId, waitingNum);
             taskTrackerActorRef.tell(req, null);
+            log.debug("[ProcessorTracker-{}] send heartbeat to TaskTracker, current waiting task num is {}.", instanceId, waitingNum);
         }
 
     }

oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/persistence/ConnectionFactory.java

@@ -44,7 +44,6 @@ public class ConnectionFactory {
                 // 池中最大连接数量
                 config.setMaximumPoolSize(32);
                 dataSource = new HikariDataSource(config);
-
             }
         }
         return dataSource;