[dev] opt clean strategy

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

@@ -2,6 +2,11 @@ package com.github.kfcfans.oms.server.persistence.core.repository;
 
 import com.github.kfcfans.oms.server.persistence.core.model.WorkflowInstanceInfoDO;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+
+import javax.transaction.Transactional;
+import java.util.Date;
 
 /**
  * 工作流运行实例数据操作
@@ -10,4 +15,11 @@ import org.springframework.data.jpa.repository.JpaRepository;
  * @since 2020/5/26
  */
 public interface WorkflowInstanceInfoRepository extends JpaRepository<WorkflowInstanceInfoDO, Long> {
+
+    // 删除历史数据，JPA自带的删除居然是根据ID循环删，2000条数据删了几秒，也太拉垮了吧...
+    // 结果只能用 int 接收
+    @Modifying
+    @Transactional
+    @Query(value = "delete from workflow_instance_info where gmt_modified < ?1", nativeQuery = true)
+    int deleteAllByGmtModifiedBefore(Date time);
 }

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

@@ -36,9 +36,9 @@ import java.util.concurrent.TimeUnit;
 public class InstanceManager {
 
     // 存储 instanceId 对应的 Job 信息，便于重试
-    private static final Map<Long, JobInfoDO> instanceId2JobInfo = Maps.newConcurrentMap();
+    private static Map<Long, JobInfoDO> instanceId2JobInfo = Maps.newConcurrentMap();
     // 存储 instance 的状态（暂时只用到了 lastReportTime）
-    private static final Map<Long, InstanceStatusHolder> instanceId2StatusHolder = Maps.newConcurrentMap();
+    private static Map<Long, InstanceStatusHolder> instanceId2StatusHolder = Maps.newConcurrentMap();
 
     // Spring Bean
     private static DispatchService dispatchService;
@@ -206,6 +206,14 @@ public class InstanceManager {
         return null;
     }
 
+    /**
+     * 释放本地缓存，防止内存泄漏
+     */
+    public static void releaseInstanceInfos() {
+        instanceId2JobInfo = Maps.newConcurrentMap();
+        instanceId2StatusHolder = Maps.newConcurrentMap();
+    }
+
     private static InstanceInfoRepository getInstanceInfoRepository() {
         while (instanceInfoRepository == null) {
             try {

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

@@ -2,8 +2,10 @@ package com.github.kfcfans.oms.server.service.timing;
 
 import com.github.kfcfans.oms.server.common.utils.OmsFileUtils;
 import com.github.kfcfans.oms.server.persistence.core.repository.InstanceInfoRepository;
+import com.github.kfcfans.oms.server.persistence.core.repository.WorkflowInstanceInfoRepository;
 import com.github.kfcfans.oms.server.persistence.mongodb.GridFsManager;
 import com.github.kfcfans.oms.server.service.ha.WorkerManagerService;
+import com.github.kfcfans.oms.server.service.instance.InstanceManager;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Stopwatch;
 import lombok.extern.slf4j.Slf4j;
@@ -31,6 +33,8 @@ public class CleanService {
     private GridFsManager gridFsManager;
     @Resource
     private InstanceInfoRepository instanceInfoRepository;
+    @Resource
+    private WorkflowInstanceInfoRepository workflowInstanceInfoRepository;
 
     @Value("${oms.log.retention.local}")
     private int localLogRetentionDay;
@@ -54,14 +58,20 @@ public class CleanService {
     @Scheduled(cron = CLEAN_TIME_EXPRESSION)
     public void timingClean() {
 
+        // 释放本地缓存
         WorkerManagerService.releaseContainerInfos();
+        InstanceManager.releaseInstanceInfos();
 
+        // 删除数据库运行记录
         cleanInstanceLog();
+        cleanWorkflowInstanceLog();
 
+        // 释放磁盘空间
         cleanLocal(OmsFileUtils.genLogDirPath(), localLogRetentionDay);
         cleanLocal(OmsFileUtils.genContainerJarPath(), localContainerRetentionDay);
         cleanLocal(OmsFileUtils.genTemporaryPath(), TEMPORARY_RETENTION_DAY);
 
+        // 删除 GridFS 过期文件
         cleanRemote(GridFsManager.LOG_BUCKET, remoteLogRetentionDay);
         cleanRemote(GridFsManager.CONTAINER_BUCKET, remoteContainerRetentionDay);
     }
@@ -127,4 +137,15 @@ public class CleanService {
         }
     }
 
+    @VisibleForTesting
+    public void cleanWorkflowInstanceLog() {
+        try {
+            Date t = DateUtils.addDays(new Date(), -instanceInfoRetentionDay);
+            int num = workflowInstanceInfoRepository.deleteAllByGmtModifiedBefore(t);
+            log.info("[CleanService] deleted {} workflow instanceInfo records whose modify time before {}.", num, t);
+        }catch (Exception e) {
+            log.warn("[CleanService] clean workflow instanceInfo failed.", e);
+        }
+    }
+
 }