[modify] remove init and destory method in BasicProcessor, It's really redundant

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/common/utils/ContainerTemplateGenerator.java

@@ -30,7 +30,7 @@ public class ContainerTemplateGenerator {
      */
     public static File generate(String group, String artifact, String name, String packageName, Integer javaVersion) throws IOException {
 
-        String workerDir = OmsFileUtils.genTemporaryWorkePath();
+        String workerDir = OmsFileUtils.genTemporaryWorkPath();
         File originJar = new File(workerDir + "tmp.jar");
         String tmpPath = workerDir + "/unzip/";
 

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/common/utils/OmsFileUtils.java

@@ -48,7 +48,7 @@ public class OmsFileUtils {
      * 获取临时目录（随机目录，不会重复），用完记得删除
      * @return 临时目录
      */
-    public static String genTemporaryWorkePath() {
+    public static String genTemporaryWorkPath() {
         String uuid = StringUtils.replace(UUID.randomUUID().toString(), "-", "");
         return genTemporaryPath() + uuid + "/";
     }

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/common/utils/TimeUtils.java

@@ -27,8 +27,9 @@ public class TimeUtils {
     public static void check() throws TimeCheckException {
 
         NTPUDPClient timeClient = new NTPUDPClient();
-        timeClient.setDefaultTimeout((int) RemoteConstant.DEFAULT_TIMEOUT_MS);
 
+        try {
+            timeClient.setDefaultTimeout((int) RemoteConstant.DEFAULT_TIMEOUT_MS);
             for (String address : NTP_SERVER_LIST) {
                 try {
                     TimeInfo t = timeClient.getTime(InetAddress.getByName(address));
@@ -48,16 +49,15 @@ public class TimeUtils {
                     log.warn("[TimeUtils] ntp server: {} may down!", address);
                 }
             }
-
             throw new TimeCheckException("no available ntp server, maybe alibaba, sjtu and apple are both collapse");
+        }finally {
+            timeClient.close();
+        }
     }
 
     public static final class TimeCheckException extends RuntimeException {
         public TimeCheckException(String message) {
             super(message);
         }
-        public TimeCheckException(Throwable cause) {
-            super(cause);
-        }
     }
 }

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/persistence/core/repository/InstanceInfoRepository.java

@@ -39,19 +39,19 @@ public interface InstanceInfoRepository extends JpaRepository<InstanceInfoDO, Lo
     @Transactional
     @Modifying
     @CanIgnoreReturnValue
-    @Query(value = "update instance_log set status = ?2, running_times = ?3, actual_trigger_time = ?4, finished_time = ?5, task_tracker_address = ?6, result = ?7, instance_params = ?8, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
+    @Query(value = "update instance_info set status = ?2, running_times = ?3, actual_trigger_time = ?4, finished_time = ?5, task_tracker_address = ?6, result = ?7, instance_params = ?8, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
     int update4TriggerFailed(long instanceId, int status, long runningTimes, long actualTriggerTime, long finishedTime, String taskTrackerAddress, String result, String instanceParams);
 
     @Transactional
     @Modifying
     @CanIgnoreReturnValue
-    @Query(value = "update instance_log set status = ?2, running_times = ?3, actual_trigger_time = ?4, task_tracker_address = ?5, instance_params = ?6, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
+    @Query(value = "update instance_info set status = ?2, running_times = ?3, actual_trigger_time = ?4, task_tracker_address = ?5, instance_params = ?6, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
     int update4TriggerSucceed(long instanceId, int status, long runningTimes, long actualTriggerTime, String taskTrackerAddress, String instanceParams);
 
     @Modifying
     @Transactional
     @CanIgnoreReturnValue
-    @Query(value = "update instance_log set status = ?2, running_times = ?3, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
+    @Query(value = "update instance_info set status = ?2, running_times = ?3, gmt_modified = now() where instance_id = ?1", nativeQuery = true)
     int update4FrequentJob(long instanceId, int status, long runningTimes);
 
     // 状态检查三兄弟，对应 WAITING_DISPATCH 、 WAITING_WORKER_RECEIVE 和 RUNNING 三阶段
@@ -71,7 +71,7 @@ public interface InstanceInfoRepository extends JpaRepository<InstanceInfoDO, Lo
     long countByAppIdAndStatus(long appId, int status);
     long countByAppIdAndStatusAndGmtCreateAfter(long appId, int status, Date time);
 
-    @Query(value = "select job_id from instance_log where job_id in ?1 and status in ?2", nativeQuery = true)
+    @Query(value = "select job_id from instance_info where job_id in ?1 and status in ?2", nativeQuery = true)
     List<Long> findByJobIdInAndStatusIn(List<Long> jobIds, List<Integer> status);
 
 }

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

@@ -130,7 +130,7 @@ public class ContainerService {
      */
     public String uploadContainerJarFile(MultipartFile file) throws IOException {
 
-        String workerDirStr = OmsFileUtils.genTemporaryWorkePath();
+        String workerDirStr = OmsFileUtils.genTemporaryWorkPath();
         String tmpFileStr = workerDirStr + "tmp.jar";
 
         File workerDir = new File(workerDirStr);
@@ -141,11 +141,11 @@ public class ContainerService {
             FileUtils.forceMkdirParent(tmpFile);
             file.transferTo(tmpFile);
 
-            // 生成MD5
+            // 生成MD5，这兄弟耗时有点小严重
             String md5 = OmsFileUtils.md5(tmpFile);
             String fileName = genContainerJarName(md5);
 
-            // 上传到 mongoDB
+            // 上传到 mongoDB，这兄弟耗时也有点小严重，导致这个接口整体比较慢...不过也没必要开线程去处理
             gridFsManager.store(tmpFile, GridFsManager.CONTAINER_BUCKET, fileName);
 
             // 将文件拷贝到正确的路径
@@ -269,7 +269,7 @@ public class ContainerService {
         ContainerSourceType sourceType = ContainerSourceType.of(container.getSourceType());
         if (sourceType == ContainerSourceType.Git) {
 
-            String workerDirStr = OmsFileUtils.genTemporaryWorkePath();
+            String workerDirStr = OmsFileUtils.genTemporaryWorkPath();
             File workerDir = new File(workerDirStr);
             FileUtils.forceMkdir(workerDir);
 

oh-my-scheduler-server/src/main/java/com/github/kfcfans/oms/server/service/alarm/DefaultMailAlarmService.java

@@ -38,6 +38,8 @@ public class DefaultMailAlarmService implements Alarmable {
     @Override
     public void alarm(AlarmContent alarmContent, List<UserInfoDO> targetUserList) {
 
+        log.debug("[DefaultMailAlarmService] content: {}, user: {}", alarmContent, targetUserList);
+
         if (CollectionUtils.isEmpty(targetUserList)) {
             return;
         }

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

@@ -166,6 +166,14 @@ public class InstanceManager {
         // 告警
         if (instanceStatus == InstanceStatus.FAILED) {
 
+            if (jobInfo == null) {
+                jobInfo = fetchJobInfo(instanceId);
+            }
+            if (jobInfo == null) {
+                log.warn("[InstanceManager] can't find jobInfo by instanceId({}), alarm failed.", instanceId);
+                return;
+            }
+
             InstanceInfoDO instanceInfo = getInstanceInfoRepository().findByInstanceId(instanceId);
             AlarmContent content = new AlarmContent();
             BeanUtils.copyProperties(jobInfo, content);
@@ -180,7 +188,15 @@ public class InstanceManager {
     }
 
     public static JobInfoDO fetchJobInfo(Long instanceId) {
-        return instanceId2JobInfo.get(instanceId);
+        JobInfoDO jobInfo = instanceId2JobInfo.get(instanceId);
+        if (jobInfo != null) {
+            return jobInfo;
+        }
+        InstanceInfoDO instanceInfo = getInstanceInfoRepository().findByInstanceId(instanceId);
+        if (instanceInfo != null) {
+            return getJobInfoRepository().findById(instanceInfo.getJobId()).orElse(null);
+        }
+        return null;
     }
 
     private static InstanceInfoRepository getInstanceInfoRepository() {

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

@@ -3,7 +3,7 @@ logging.config=classpath:logback-dev.xml
 
 ####### 数据库配置 #######
 spring.datasource.core.driver-class-name=com.mysql.cj.jdbc.Driver
-spring.datasource.core.jdbc-url=jdbc:mysql://remotehost:3391/oms-daily?useUnicode=true&characterEncoding=UTF-8
+spring.datasource.core.jdbc-url=jdbc:mysql://remotehost:3391/oms-daily?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8
 spring.datasource.core.username=root
 spring.datasource.core.password=No1Bug2Please3!
 spring.datasource.core.hikari.maximum-pool-size=20

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

@@ -3,7 +3,7 @@ logging.config=classpath:logback-product.xml
 
 ####### 数据库配置 #######
 spring.datasource.core.driver-class-name=com.mysql.cj.jdbc.Driver
-spring.datasource.core.jdbc-url=jdbc:mysql://remotehost:3391/oms-pre?useUnicode=true&characterEncoding=UTF-8
+spring.datasource.core.jdbc-url=jdbc:mysql://remotehost:3391/oms-pre?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8
 spring.datasource.core.username=root
 spring.datasource.core.password=No1Bug2Please3!
 spring.datasource.core.hikari.maximum-pool-size=20

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

@@ -3,7 +3,7 @@ logging.config=classpath:logback-product.xml
 
 ####### 数据库配置 #######
 spring.datasource.core.driver-class-name=com.mysql.cj.jdbc.Driver
-spring.datasource.core.jdbc-url=jdbc:mysql://remotehost:3391/oms-product?useUnicode=true&characterEncoding=UTF-8
+spring.datasource.core.jdbc-url=jdbc:mysql://remotehost:3391/oms-product?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8
 spring.datasource.core.username=root
 spring.datasource.core.password=No1Bug2Please3!
 spring.datasource.core.hikari.maximum-pool-size=20

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

@@ -1,8 +1,8 @@
 package com.github.kfcfans.oms.server.test;
 
+import com.github.kfcfans.oms.common.TimeExpressionType;
 import com.github.kfcfans.oms.common.utils.NetUtils;
 import com.github.kfcfans.oms.server.common.constans.JobStatus;
-import com.github.kfcfans.oms.common.TimeExpressionType;
 import com.github.kfcfans.oms.server.persistence.core.model.InstanceInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.model.JobInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.model.OmsLockDO;
@@ -13,10 +13,10 @@ 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;
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -25,7 +25,7 @@ import java.util.List;
  * @author tjq
  * @since 2020/4/5
  */
-@ActiveProfiles("daily")
+//@ActiveProfiles("daily")
 @RunWith(SpringRunner.class)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class RepositoryTest {
@@ -73,4 +73,12 @@ public class RepositoryTest {
         instanceInfoRepository.update4FrequentJob(1586310419650L, 2, 200);
     }
 
+    @Test
+    public void testCheckQuery() {
+        Date time = new Date();
+        System.out.println(time);
+        final List<InstanceInfoDO> res = instanceInfoRepository.findByAppIdInAndStatusAndGmtModifiedBefore(Lists.newArrayList(1L), 3, time);
+        System.out.println(res);
+    }
+
 }

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

@@ -9,6 +9,7 @@ import org.junit.Test;
 
 import java.util.Date;
 import java.util.List;
+import java.util.TimeZone;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 
@@ -77,4 +78,8 @@ public class UtilsTest {
         System.out.println(s.length());
     }
 
+    @Test
+    public void testTZ() {
+        System.out.println(TimeZone.getDefault());
+    }
 }

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

@@ -0,0 +1,35 @@
+# http 服务端口
+server.port=7700
+
+spring.profiles.active=daily
+spring.jpa.open-in-view=false
+spring.jpa.show-sql=true
+spring.data.mongodb.repositories.type=none
+
+# 文件上传配置
+spring.servlet.multipart.enabled =true
+spring.servlet.multipart.file-size-threshold=0
+spring.servlet.multipart.max-file-size=209715200
+spring.servlet.multipart.max-request-size=209715200
+
+####### 数据库配置 #######
+spring.datasource.core.driver-class-name=com.mysql.cj.jdbc.Driver
+spring.datasource.core.jdbc-url=jdbc:mysql://remotehost:3391/oms-daily?useUnicode=true&characterEncoding=UTF-8
+spring.datasource.core.username=root
+spring.datasource.core.password=No1Bug2Please3!
+spring.datasource.core.hikari.maximum-pool-size=20
+spring.datasource.core.hikari.minimum-idle=5
+
+####### mongoDB配置，非核心依赖，可移除 #######
+spring.data.mongodb.uri=mongodb://remotehost:27017/oms-daily
+
+###### OhMyScheduler 自身配置（该配置只允许存在于 application.properties 文件中） ######
+# akka ActorSystem 服务端口
+oms.akka.port=10086
+# 报警服务 bean名称
+oms.alarm.bean.names=omsDefaultMailAlarmService
+####### 日志保留天数，单位天 #######
+oms.log.retention.local=0
+oms.log.retention.remote=0
+oms.container.retention.local=0
+oms.container.retention.remote=0

oh-my-scheduler-worker/src/main/java/com/github/kfcfans/oms/worker/container/OmsJarContainer.java

@@ -76,9 +76,7 @@ public class OmsJarContainer implements OmsContainer {
             // 直接实例化
             try {
                 Object obj = targetClass.getDeclaredConstructor().newInstance();
-                BasicProcessor processor = (BasicProcessor) obj;
+                return (BasicProcessor) obj;
-                processor.init();
-                return processor;
             } catch (Exception e) {
                 log.error("[OmsJarContainer-{}] load {} failed", name, className, e);
             }

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

@@ -34,10 +34,7 @@ public class ProcessorBeanFactory {
             try {
 
                 Class<?> clz = Class.forName(className);
-                BasicProcessor processor = (BasicProcessor) clz.getDeclaredConstructor().newInstance();
+                return (BasicProcessor) clz.getDeclaredConstructor().newInstance();
-                processor.init();
-
-                return processor;
 
             }catch (Exception e) {
                 log.error("[ProcessorBeanFactory] load local Processor(className = {}) failed.", className, e);

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

@@ -12,8 +12,6 @@ import java.io.*;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -29,7 +27,6 @@ public abstract class ScriptProcessor implements BasicProcessor {
     // 脚本绝对路径
     private final String scriptPath;
     private final long timeout;
-    private final ExecutorService threadPool;
 
     private static final Set<String> DOWNLOAD_PROTOCOL = Sets.newHashSet("http", "https", "ftp");
 
@@ -38,7 +35,6 @@ public abstract class ScriptProcessor implements BasicProcessor {
         this.instanceId = instanceId;
         this.scriptPath = OmsWorkerFileUtils.getScriptDir() + genScriptName(instanceId);
         this.timeout = timeout;
-        this.threadPool = Executors.newFixedThreadPool(2);
 
         File script = new File(scriptPath);
         if (script.exists()) {
@@ -82,8 +78,10 @@ public abstract class ScriptProcessor implements BasicProcessor {
         StringBuilder inputSB = new StringBuilder();
         StringBuilder errorSB = new StringBuilder();
 
-        threadPool.submit(() -> copyStream(process.getInputStream(), inputSB));
+        // 为了代码优雅而牺牲那么一点点点点点点点点性能
-        threadPool.submit(() -> copyStream(process.getErrorStream(), errorSB));
+        // 从外部传入线程池总感觉怪怪的...内部创建嘛又要考虑考虑资源释放问题，想来想去还是直接创建算了。
+        new Thread(() -> copyStream(process.getInputStream(), inputSB)).start();
+        new Thread(() -> copyStream(process.getErrorStream(), errorSB)).start();
 
         try {
             boolean s = process.waitFor(timeout, TimeUnit.MILLISECONDS);
@@ -95,8 +93,6 @@ public abstract class ScriptProcessor implements BasicProcessor {
             return new ProcessResult(true, result);
         }catch (InterruptedException ie) {
             return new ProcessResult(false, "Interrupted");
-        }finally {
-            threadPool.shutdownNow();
         }
     }
 

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

@@ -18,20 +18,4 @@ public interface BasicProcessor {
      * @throws Exception 异常，允许抛出异常，但不推荐，最好由业务开发者自己处理
      */
     ProcessResult process(TaskContext context) throws Exception;
-
-    /**
-     * 用于构造 Processor 对象，相当于构造方法
-     * @throws Exception 异常，抛出异常则视为处理器构造失败，任务直接失败
-     */
-    default void init() throws Exception {
-    }
-
-    /**
-     * 销毁 Processor 时的回调方法，暂时未被使用
-     * @throws Exception 异常
-     */
-    @Deprecated
-    default void destroy() throws Exception {
-    }
-
 }

oh-my-scheduler-worker/src/test/java/com/github/kfcfans/oms/processors/TestMapReduceProcessor.java

@@ -70,8 +70,4 @@ public class TestMapReduceProcessor extends MapReduceProcessor {
         private int age;
     }
 
-    @Override
-    public void init() throws Exception {
-        System.out.println("============== TestMapReduceProcessor#init ==============");
-    }
 }

others/doc/SystemInitGuide.md

@@ -7,9 +7,12 @@
 * mongoDB（可选）：任意支持GridFS的mongoDB版本（4.2.6测试通过，其余未经测试，仅从理论角度分析可用）
 
 #### 流程
+>注意，由于调度系统的特殊性，请务必**确保数据库和调度服务器处于同一个时区**。
+
 1. 部署数据库：由于任务调度中心的数据持久层基于`Spring Data Jpa`实现，**开发者仅需要完成数据库的创建**，即运行SQL`CREATE database if NOT EXISTS oms-product default character set utf8mb4 collate utf8mb4_unicode_ci;`。
     * 注1：任务调度中心支持多环境部署（日常、预发、线上），其分别对应三个数据库：oms-daily、oms-pre和oms-product。
     * 注2：手动建表SQL文件：[oms-sql.sql](../oms-sql.sql)
+    * 注3：部署完成后建议查看时区信息：`show variables like "%time_zone%";`，务必使`time_zone`代表的时区与JDBC连接URL中`serverTimezone`字段代表的时区一致！
     
 2. 部署调度服务器（OhMyScheduler-Server），需要先修改配置文件（同样为了支持多环境部署，采用了daily、pre和product3套配置文件），之后自行编译部署运行。
     * 注1：OhMyScheduler-Server支持集群部署，具备完全的水平扩展能力。建议部署多个实例以实现高可用&高性能。