[dev] finished workflow alarm service

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

@@ -1,17 +1,12 @@
 package com.github.kfcfans.oms.server.persistence.core.model;
 
 
-import com.google.common.base.Splitter;
-import com.google.common.collect.Lists;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.springframework.util.StringUtils;
 
 import javax.persistence.*;
 import java.util.Date;
-import java.util.List;
-import java.util.stream.Collectors;
 
 /**
  * 任务信息表
@@ -94,10 +89,4 @@ public class JobInfoDO {
     private Date gmtCreate;
     private Date gmtModified;
 
-    public List<Long> fetchNotifyUserIds() {
-        if (StringUtils.isEmpty(notifyUserIds)) {
-            return Lists.newLinkedList();
-        }
-        return Splitter.on(",").splitToList(notifyUserIds).stream().map(Long::valueOf).collect(Collectors.toList());
-    }
 }

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

@@ -0,0 +1,57 @@
+package com.github.kfcfans.oms.server.service;
+
+import com.github.kfcfans.oms.server.persistence.core.model.UserInfoDO;
+import com.github.kfcfans.oms.server.persistence.core.repository.UserInfoRepository;
+import com.github.kfcfans.oms.server.web.request.ModifyUserInfoRequest;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Lists;
+import org.springframework.beans.BeanUtils;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import javax.annotation.Resource;
+import java.util.Date;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * 用户服务
+ *
+ * @author tjq
+ * @since 2020/6/12
+ */
+@Service
+public class UserService {
+
+    @Resource
+    private UserInfoRepository userInfoRepository;
+
+    /**
+     * 保存/修改 用户
+     * @param request 请求
+     */
+    public void save(ModifyUserInfoRequest request) {
+        UserInfoDO userInfoDO = new UserInfoDO();
+        BeanUtils.copyProperties(request, userInfoDO);
+        userInfoDO.setGmtCreate(new Date());
+        userInfoDO.setGmtModified(userInfoDO.getGmtCreate());
+        userInfoRepository.saveAndFlush(userInfoDO);
+    }
+
+    /**
+     * 根据用户ID字符串获取用户信息详细列表
+     * @param userIds 逗号分割的用户ID信息
+     * @return 用户信息详细列表
+     */
+    public List<UserInfoDO> fetchNotifyUserList(String userIds) {
+        if (StringUtils.isEmpty(userIds)) {
+            return Lists.newLinkedList();
+        }
+        // 去重
+        Set<Long> userIdList = Splitter.on(",").splitToList(userIds).stream().map(Long::valueOf).collect(Collectors.toSet());
+        List<UserInfoDO> res = userInfoRepository.findByIdIn(Lists.newLinkedList(userIdList));
+        res.forEach(x -> x.setPassword(null));
+        return res;
+    }
+}

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

@@ -12,5 +12,17 @@ import java.util.List;
  */
 public interface Alarmable {
 
-    void alarm(AlarmContent alarmContent, List<UserInfoDO> targetUserList);
+    /**
+     * 任务执行失败报警
+     * @param content 任务实例相关信息
+     * @param targetUserList 目标用户列表
+     */
+    void onJobInstanceFailed(JobInstanceAlarmContent content, List<UserInfoDO> targetUserList);
+
+    /**
+     * 工作流执行失败报警
+     * @param content 工作流实例相关信息
+     * @param targetUserList 目标用户列表
+     */
+    void onWorkflowInstanceFailed(WorkflowInstanceAlarmContent content, List<UserInfoDO> targetUserList);
 }

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

@@ -27,8 +27,9 @@ public class DefaultMailAlarmService implements Alarmable {
     @Value("${spring.mail.username}")
     private String from;
 
-    private static final String MAIL_TITLE = "OhMyScheduler 任务执行失败报警";
+    private static final String MAIL_TITLE = "OhMyScheduler AlarmService";
-    private static final String MAIL_CONTENT_PATTERN = "任务运行失败，详细信息如下：%s";
+    private static final String JOB_INSTANCE_FAILED_CONTENT_PATTERN = "Job run failed, detail is: %s";
+    private static final String WF_INSTANCE_FAILED_CONTENT_PATTERN = "Workflow run failed, detail is: %s";
 
     @Autowired(required = false)
     public DefaultMailAlarmService(JavaMailSender javaMailSender) {
@@ -36,24 +37,35 @@ public class DefaultMailAlarmService implements Alarmable {
     }
 
     @Override
-    public void alarm(AlarmContent alarmContent, List<UserInfoDO> targetUserList) {
+    public void onJobInstanceFailed(JobInstanceAlarmContent content, List<UserInfoDO> targetUserList) {
+        String msg = String.format(JOB_INSTANCE_FAILED_CONTENT_PATTERN, JsonUtils.toJSONString(content));
+        sendMail(msg, targetUserList);
+    }
 
-        log.debug("[DefaultMailAlarmService] content: {}, user: {}", alarmContent, targetUserList);
+    @Override
+    public void onWorkflowInstanceFailed(WorkflowInstanceAlarmContent content, List<UserInfoDO> targetUserList) {
+        String msg = String.format(WF_INSTANCE_FAILED_CONTENT_PATTERN, JsonUtils.toJSONString(content));
+        sendMail(msg, targetUserList);
+    }
 
-        if (CollectionUtils.isEmpty(targetUserList)) {
+    private void sendMail(String msg, List<UserInfoDO> targetUserList) {
+
+        log.debug("[OmsMailAlarmService] msg: {}, to: {}", msg, targetUserList);
+
+        if (CollectionUtils.isEmpty(targetUserList) || javaMailSender == null) {
             return;
         }
 
         SimpleMailMessage sm = new SimpleMailMessage();
-        sm.setFrom(from);
-        sm.setTo(targetUserList.stream().map(UserInfoDO::getEmail).toArray(String[]::new));
-        sm.setSubject(MAIL_TITLE);
-        sm.setText(String.format(MAIL_CONTENT_PATTERN, JsonUtils.toJSONString(alarmContent)));
-
         try {
+            sm.setFrom(from);
+            sm.setTo(targetUserList.stream().map(UserInfoDO::getEmail).toArray(String[]::new));
+            sm.setSubject(MAIL_TITLE);
+            sm.setText(msg);
+
             javaMailSender.send(sm);
         }catch (Exception e) {
-            log.error("[DefaultMailAlarmService] send mail({}) failed.", sm, e);
+            log.error("[OmsMailAlarmService] send mail({}) failed.", sm, e);
         }
     }
 }

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

@@ -3,13 +3,13 @@ package com.github.kfcfans.oms.server.service.alarm;
 import lombok.Data;
 
 /**
- * 告警对象
+ * 任务执行失败告警对象
  *
  * @author tjq
  * @since 2020/4/30
  */
 @Data
-public class AlarmContent {
+public class JobInstanceAlarmContent {
     // 应用ID
     private long appId;
     // 任务ID

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

@@ -6,13 +6,11 @@ import com.google.common.base.Splitter;
 import com.google.common.collect.Lists;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * 报警服务
@@ -36,11 +34,24 @@ public class OmsCenterAlarmService implements Alarmable {
 
     @Async("omsCommonPool")
     @Override
-    public void alarm(AlarmContent alarmContent, List<UserInfoDO> targetUserList) {
+    public void onJobInstanceFailed(JobInstanceAlarmContent content, List<UserInfoDO> targetUserList) {
         init();
         alarmableList.forEach(alarmable -> {
             try {
-                alarmable.alarm(alarmContent, targetUserList);
+                alarmable.onJobInstanceFailed(content, targetUserList);
+            }catch (Exception e) {
+                log.warn("[OmsCenterAlarmService] alarm failed.", e);
+            }
+        });
+    }
+
+    @Async("omsCommonPool")
+    @Override
+    public void onWorkflowInstanceFailed(WorkflowInstanceAlarmContent content, List<UserInfoDO> targetUserList) {
+        init();
+        alarmableList.forEach(alarmable -> {
+            try {
+                alarmable.onWorkflowInstanceFailed(content, targetUserList);
             }catch (Exception e) {
                 log.warn("[OmsCenterAlarmService] alarm failed.", e);
             }

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

@@ -0,0 +1,37 @@
+package com.github.kfcfans.oms.server.service.alarm;
+
+import com.github.kfcfans.oms.common.model.PEWorkflowDAG;
+import lombok.Data;
+
+/**
+ * 工作流执行失败告警对象
+ *
+ * @author tjq
+ * @since 2020/6/12
+ */
+@Data
+public class WorkflowInstanceAlarmContent {
+
+    private String workflowName;
+
+    // 任务所属应用的ID，冗余提高查询效率
+    private Long appId;
+    private Long workflowId;
+    // workflowInstanceId（任务实例表都使用单独的ID作为主键以支持潜在的分表需求）
+    private Long wfInstanceId;
+    // workflow 状态（WorkflowInstanceStatus）
+    private Integer status;
+
+    private PEWorkflowDAG peWorkflowDAG;
+    private String result;
+
+    // 实际触发时间
+    private Long actualTriggerTime;
+    // 结束时间
+    private Long finishedTime;
+
+    // 时间表达式类型（CRON/API/FIX_RATE/FIX_DELAY）
+    private Integer timeExpressionType;
+    // 时间表达式，CRON/NULL/LONG/LONG
+    private String timeExpression;
+}

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

@@ -1,19 +1,19 @@
 package com.github.kfcfans.oms.server.service.instance;
 
 import com.github.kfcfans.oms.common.InstanceStatus;
-import com.github.kfcfans.oms.common.request.TaskTrackerReportInstanceStatusReq;
 import com.github.kfcfans.oms.common.TimeExpressionType;
+import com.github.kfcfans.oms.common.request.TaskTrackerReportInstanceStatusReq;
 import com.github.kfcfans.oms.server.common.utils.SpringUtils;
 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.UserInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.repository.InstanceInfoRepository;
 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.alarm.AlarmContent;
+import com.github.kfcfans.oms.server.service.UserService;
 import com.github.kfcfans.oms.server.service.alarm.Alarmable;
+import com.github.kfcfans.oms.server.service.alarm.JobInstanceAlarmContent;
 import com.github.kfcfans.oms.server.service.timing.schedule.HashedWheelTimerHolder;
 import com.github.kfcfans.oms.server.service.workflow.WorkflowInstanceManager;
 import com.google.common.collect.Maps;
@@ -185,15 +185,12 @@ public class InstanceManager {
             }
 
             InstanceInfoDO instanceInfo = getInstanceInfoRepository().findByInstanceId(instanceId);
-            AlarmContent content = new AlarmContent();
+            JobInstanceAlarmContent content = new JobInstanceAlarmContent();
             BeanUtils.copyProperties(jobInfo, content);
             BeanUtils.copyProperties(instanceInfo, content);
 
-            List<Long> userIds = jobInfo.fetchNotifyUserIds();
+            List<UserInfoDO> userList = SpringUtils.getBean(UserService.class).fetchNotifyUserList(jobInfo.getNotifyUserIds());
-            List<UserInfoDO> userList = SpringUtils.getBean(UserInfoRepository.class).findByIdIn(userIds);
+            getAlarmService().onJobInstanceFailed(content, userList);
-            userList.forEach(x -> x.setPassword(null));
-
-            getAlarmService().alarm(content, userList);
         }
     }
 

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

@@ -11,11 +11,16 @@ import com.github.kfcfans.oms.common.utils.SegmentLock;
 import com.github.kfcfans.oms.server.common.constans.SwitchableStatus;
 import com.github.kfcfans.oms.server.common.utils.WorkflowDAGUtils;
 import com.github.kfcfans.oms.server.persistence.core.model.JobInfoDO;
+import com.github.kfcfans.oms.server.persistence.core.model.UserInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.model.WorkflowInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.model.WorkflowInstanceInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.repository.JobInfoRepository;
+import com.github.kfcfans.oms.server.persistence.core.repository.WorkflowInfoRepository;
 import com.github.kfcfans.oms.server.persistence.core.repository.WorkflowInstanceInfoRepository;
 import com.github.kfcfans.oms.server.service.DispatchService;
+import com.github.kfcfans.oms.server.service.UserService;
+import com.github.kfcfans.oms.server.service.alarm.Alarmable;
+import com.github.kfcfans.oms.server.service.alarm.WorkflowInstanceAlarmContent;
 import com.github.kfcfans.oms.server.service.id.IdGenerateService;
 import com.github.kfcfans.oms.server.service.instance.InstanceService;
 import com.google.common.collect.LinkedListMultimap;
@@ -23,6 +28,7 @@ import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimap;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
@@ -50,8 +56,15 @@ public class WorkflowInstanceManager {
     @Resource
     private JobInfoRepository jobInfoRepository;
     @Resource
+    private UserService userService;
+    @Resource
+    private WorkflowInfoRepository workflowInfoRepository;
+    @Resource
     private WorkflowInstanceInfoRepository workflowInstanceInfoRepository;
 
+    @Resource(name = "omsCenterAlarmService")
+    private Alarmable omsCenterAlarmService;
+
     private final SegmentLock segmentLock = new SegmentLock(16);
 
     /**
@@ -86,12 +99,10 @@ public class WorkflowInstanceManager {
 
         if (dbNum < allJobIds.size()) {
             log.warn("[Workflow-{}|{}] this workflow need {} jobs, but just find {} jobs in database, maybe you delete or disable some job!", wfId, wfInstanceId, needNum, dbNum);
-            newWfInstance.setStatus(WorkflowInstanceStatus.FAILED.getV());
+            onWorkflowInstanceFailed(SystemInstanceResult.CAN_NOT_FIND_JOB, newWfInstance);
-            newWfInstance.setFinishedTime(System.currentTimeMillis());
+        }else {
-            newWfInstance.setResult(SystemInstanceResult.CAN_NOT_FIND_JOB);
+            workflowInstanceInfoRepository.save(newWfInstance);
         }
-
-        workflowInstanceInfoRepository.save(newWfInstance);
         return wfInstanceId;
     }
 
@@ -118,11 +129,7 @@ public class WorkflowInstanceManager {
         // 并发度控制
         int instanceConcurrency = workflowInstanceInfoRepository.countByWorkflowIdAndStatusIn(wfInfo.getId(), WorkflowInstanceStatus.generalizedRunningStatus);
         if (instanceConcurrency > wfInfo.getMaxWfInstanceNum()) {
-            wfInstanceInfo.setStatus(WorkflowInstanceStatus.FAILED.getV());
+            onWorkflowInstanceFailed(String.format(SystemInstanceResult.TOO_MUCH_INSTANCE, instanceConcurrency, wfInfo.getMaxWfInstanceNum()), wfInstanceInfo);
-            wfInstanceInfo.setResult(String.format(SystemInstanceResult.TOO_MUCH_INSTANCE, instanceConcurrency, wfInfo.getMaxWfInstanceNum()));
-            wfInstanceInfo.setFinishedTime(System.currentTimeMillis());
-
-            workflowInstanceInfoRepository.saveAndFlush(wfInstanceInfo);
             return;
         }
 
@@ -151,13 +158,8 @@ public class WorkflowInstanceManager {
             roots.forEach(root -> runInstance(root.getJobId(), root.getInstanceId(), wfInstanceId, null));
         }catch (Exception e) {
 
-            wfInstanceInfo.setStatus(WorkflowInstanceStatus.FAILED.getV());
-            wfInstanceInfo.setResult(e.getMessage());
-            wfInstanceInfo.setFinishedTime(System.currentTimeMillis());
-
             log.error("[Workflow-{}|{}] submit workflow: {} failed.", wfInfo.getId(), wfInstanceId, wfInfo, e);
-
+            onWorkflowInstanceFailed(e.getMessage(), wfInstanceInfo);
-            workflowInstanceInfoRepository.saveAndFlush(wfInstanceInfo);
         }
     }
 
@@ -222,12 +224,8 @@ public class WorkflowInstanceManager {
 
                 // 任务失败，DAG流程被打断，整体失败
                 if (status == InstanceStatus.FAILED) {
-                    wfInstance.setStatus(WorkflowInstanceStatus.FAILED.getV());
-                    wfInstance.setResult(SystemInstanceResult.MIDDLE_JOB_FAILED);
-                    wfInstance.setFinishedTime(System.currentTimeMillis());
-                    workflowInstanceInfoRepository.saveAndFlush(wfInstance);
-
                     log.warn("[Workflow-{}|{}] workflow instance process failed because middle task(instanceId={}) failed", wfId, wfInstanceId, instanceId);
+                    onWorkflowInstanceFailed(SystemInstanceResult.MIDDLE_JOB_FAILED, wfInstance);
                     return;
                 }
 
@@ -297,11 +295,7 @@ public class WorkflowInstanceManager {
                 jobId2InstanceId.forEach((jobId, newInstanceId) -> runInstance(jobId, newInstanceId, wfInstanceId, jobId2InstanceParams.get(jobId)));
 
             }catch (Exception e) {
-                wfInstance.setStatus(WorkflowInstanceStatus.FAILED.getV());
+                onWorkflowInstanceFailed("MOVE NEXT STEP FAILED: " + e.getMessage(), wfInstance);
-                wfInstance.setResult("MOVE NEXT STEP FAILED: " + e.getMessage());
-                wfInstance.setFinishedTime(System.currentTimeMillis());
-                workflowInstanceInfoRepository.saveAndFlush(wfInstance);
-
                 log.error("[Workflow-{}|{}] update failed.", wfId, wfInstanceId, e);
             }
 
@@ -326,4 +320,28 @@ public class WorkflowInstanceManager {
         dispatchService.dispatch(jobInfo, instanceId, 0, instanceParams, wfInstanceId);
     }
 
+    private void onWorkflowInstanceFailed(String result, WorkflowInstanceInfoDO wfInstance) {
+
+        wfInstance.setStatus(WorkflowInstanceStatus.FAILED.getV());
+        wfInstance.setResult(result);
+        wfInstance.setFinishedTime(System.currentTimeMillis());
+        wfInstance.setGmtModified(new Date());
+
+        workflowInstanceInfoRepository.saveAndFlush(wfInstance);
+
+        // 报警
+        try {
+            workflowInfoRepository.findById(wfInstance.getWorkflowId()).ifPresent(wfInfo -> {
+                WorkflowInstanceAlarmContent content = new WorkflowInstanceAlarmContent();
+
+                BeanUtils.copyProperties(wfInfo, content);
+                BeanUtils.copyProperties(wfInstance, content);
+                content.setResult(result);
+
+                List<UserInfoDO> userList = userService.fetchNotifyUserList(wfInfo.getNotifyUserIds());
+                omsCenterAlarmService.onWorkflowInstanceFailed(content, userList);
+            });
+        }catch (Exception ignore) {
+        }
+    }
 }

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

@@ -3,18 +3,17 @@ package com.github.kfcfans.oms.server.web.controller;
 import com.github.kfcfans.oms.common.response.ResultDTO;
 import com.github.kfcfans.oms.server.persistence.core.model.UserInfoDO;
 import com.github.kfcfans.oms.server.persistence.core.repository.UserInfoRepository;
+import com.github.kfcfans.oms.server.service.UserService;
 import com.github.kfcfans.oms.server.web.request.ModifyUserInfoRequest;
 import com.google.common.collect.Lists;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
-import org.springframework.beans.BeanUtils;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.*;
 
 import javax.annotation.Resource;
-import java.util.Date;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -28,16 +27,14 @@ import java.util.stream.Collectors;
 @RequestMapping("/user")
 public class UserInfoController {
 
+    @Resource
+    private UserService userService;
     @Resource
     private UserInfoRepository userInfoRepository;
 
     @PostMapping("save")
     public ResultDTO<Void> save(@RequestBody ModifyUserInfoRequest request) {
-        UserInfoDO userInfoDO = new UserInfoDO();
+        userService.save(request);
-        BeanUtils.copyProperties(request, userInfoDO);
-        userInfoDO.setGmtCreate(new Date());
-        userInfoDO.setGmtModified(userInfoDO.getGmtCreate());
-        userInfoRepository.saveAndFlush(userInfoDO);
         return ResultDTO.success(null);
     }
 

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

@@ -13,7 +13,7 @@ spring.datasource.core.hikari.minimum-idle=5
 spring.data.mongodb.uri=mongodb://remotehost:27017/oms-daily
 
 ####### 邮件配置（启用邮件报警则需要） #######
-spring.mail.host=smtp.qq.com
+spring.mail.host=smtp.163.com
 spring.mail.username=zqq
 spring.mail.password=qqz
 spring.mail.properties.mail.smtp.auth=true