[Fix-17308][Alert] Missing alert logic when workflow instance finished (#17309)

Author: SbloodyS

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WarningType.java

@@ -22,20 +22,22 @@
 import java.util.Arrays;
 import java.util.Map;
 
+import lombok.Getter;
+
 import com.baomidou.mybatisplus.annotation.EnumValue;
 import com.google.common.base.Functions;
 
 /**
- * types for whether to send warning when process ends;
+ * types for whether to send warning when workflow instance ends
  */
+@Getter
 public enum WarningType {
 
     /**
      * 0 do not send warning;
-     * 1 send if process success;
-     * 2 send if process failed;
-     * 3 send if process ends, whatever the result;
-     * 4 send global events;
+     * 1 send if workflow success;
+     * 2 send if workflow failed;
+     * 3 send if workflow ends, whatever the result;
      */
     NONE(0, "none"),
     SUCCESS(1, "success"),
@@ -51,14 +53,6 @@ public enum WarningType {
     private final int code;
     private final String descp;
 
-    public int getCode() {
-        return code;
-    }
-
-    public String getDescp() {
-        return descp;
-    }
-
     private static final Map<String, WarningType> WARNING_TYPE_MAP =
             Arrays.stream(WarningType.values()).collect(toMap(WarningType::getDescp, Functions.identity()));
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java

@@ -35,6 +35,7 @@
 import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowTopologyLogicalTransitionWithTaskFinishLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.utils.WorkflowInstanceUtils;
+import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager;
 
 import org.apache.commons.collections4.CollectionUtils;
 
@@ -61,6 +62,9 @@ public abstract class AbstractWorkflowStateAction implements IWorkflowStateActio
     @Autowired
     protected WorkflowEventBusCoordinator workflowEventBusCoordinator;
 
+    @Autowired
+    protected WorkflowAlertManager workflowAlertManager;
+
     /**
      * Try to trigger the tasks if the trigger condition is met.
      * <p> If all the given tasks trigger condition is not met then will try to emit workflow finish event.
@@ -190,6 +194,7 @@ protected void finalizeEventAction(final IWorkflowExecutionRunnable workflowExec
 
         workflowCacheRepository.remove(workflowExecutionRunnable.getId());
         workflowEventBusCoordinator.unRegisterWorkflowEventBus(workflowExecutionRunnable);
+        workflowAlertManager.sendAlertWorkflowInstance(workflowExecutionRunnable.getWorkflowInstance());
 
         log.info("Successfully finalize WorkflowExecuteRunnable: {}", workflowExecutionRunnable.getName());
     }

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java

@@ -24,14 +24,16 @@
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.dao.AlertDao;
 import org.apache.dolphinscheduler.dao.entity.Alert;
+import org.apache.dolphinscheduler.dao.entity.Project;
 import org.apache.dolphinscheduler.dao.entity.ProjectUser;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.entity.WorkflowAlertContent;
 import org.apache.dolphinscheduler.dao.entity.WorkflowDefinitionLog;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
-import org.apache.dolphinscheduler.dao.mapper.UserMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
+import org.apache.dolphinscheduler.dao.repository.ProjectDao;
+import org.apache.dolphinscheduler.dao.repository.UserDao;
+import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao;
 
 import java.util.ArrayList;
 import java.util.Date;
@@ -46,20 +48,20 @@
 @Slf4j
 public class WorkflowAlertManager {
 
-    /**
-     * alert dao
-     */
     @Autowired
     private AlertDao alertDao;
 
     @Autowired
-    private WorkflowDefinitionLogMapper workflowDefinitionLogMapper;
+    private WorkflowDefinitionLogDao workflowDefinitionLogDao;
+
+    @Autowired
+    private UserDao userDao;
 
     @Autowired
-    private UserMapper userMapper;
+    private ProjectDao projectDao;
 
     /**
-     * command type convert chinese
+     * convert command type to human-readable name
      *
      * @param commandType command type
      * @return command name
@@ -95,73 +97,41 @@ private String getCommandCnName(CommandType commandType) {
      * get workflow instance content
      *
      * @param workflowInstance workflow instance
-     * @param taskInstances task instance list
      * @return workflow instance format content
      */
     public String getContentWorkflowInstance(WorkflowInstance workflowInstance,
-                                             List<TaskInstance> taskInstances,
-                                             ProjectUser projectUser) {
+                                             Project project) {
 
-        String res = "";
-        WorkflowDefinitionLog workflowDefinitionLog = workflowDefinitionLogMapper
+        String res;
+        WorkflowDefinitionLog workflowDefinitionLog = workflowDefinitionLogDao
                 .queryByDefinitionCodeAndVersion(workflowInstance.getWorkflowDefinitionCode(),
                         workflowInstance.getWorkflowDefinitionVersion());
 
         String modifyBy = "";
         if (workflowDefinitionLog != null) {
-            User operator = userMapper.selectById(workflowDefinitionLog.getOperator());
+            User operator = userDao.queryById(workflowDefinitionLog.getOperator());
             modifyBy = operator == null ? "" : operator.getUserName();
         }
 
-        if (workflowInstance.getState().isSuccess()) {
-            List<WorkflowAlertContent> successTaskList = new ArrayList<>(1);
-            WorkflowAlertContent workflowAlertContent = WorkflowAlertContent.builder()
-                    .projectCode(projectUser.getProjectCode())
-                    .projectName(projectUser.getProjectName())
-                    .owner(projectUser.getUserName())
-                    .workflowInstanceId(workflowInstance.getId())
-                    .workflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode())
-                    .workflowInstanceName(workflowInstance.getName())
-                    .commandType(workflowInstance.getCommandType())
-                    .workflowExecutionStatus(workflowInstance.getState())
-                    .modifyBy(modifyBy)
-                    .recovery(workflowInstance.getRecovery())
-                    .runTimes(workflowInstance.getRunTimes())
-                    .workflowStartTime(workflowInstance.getStartTime())
-                    .workflowEndTime(workflowInstance.getEndTime())
-                    .workflowHost(workflowInstance.getHost())
-                    .build();
-            successTaskList.add(workflowAlertContent);
-            res = JSONUtils.toJsonString(successTaskList);
-        } else if (workflowInstance.getState().isFailure()) {
-
-            List<WorkflowAlertContent> failedTaskList = new ArrayList<>();
-            for (TaskInstance task : taskInstances) {
-                if (task.getState().isSuccess()) {
-                    continue;
-                }
-                WorkflowAlertContent workflowAlertContent = WorkflowAlertContent.builder()
-                        .projectCode(projectUser.getProjectCode())
-                        .projectName(projectUser.getProjectName())
-                        .owner(projectUser.getUserName())
-                        .workflowInstanceId(workflowInstance.getId())
-                        .workflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode())
-                        .workflowInstanceName(workflowInstance.getName())
-                        .modifyBy(modifyBy)
-                        .taskCode(task.getTaskCode())
-                        .taskName(task.getName())
-                        .taskType(task.getTaskType())
-                        .taskState(task.getState())
-                        .taskStartTime(task.getStartTime())
-                        .taskEndTime(task.getEndTime())
-                        .taskHost(task.getHost())
-                        .taskPriority(task.getTaskInstancePriority().getDescp())
-                        .logPath(task.getLogPath())
-                        .build();
-                failedTaskList.add(workflowAlertContent);
-            }
-            res = JSONUtils.toJsonString(failedTaskList);
-        }
+        List<WorkflowAlertContent> successTaskList = new ArrayList<>(1);
+        WorkflowAlertContent workflowAlertContent = WorkflowAlertContent.builder()
+                .projectCode(project.getCode())
+                .projectName(project.getName())
+                .owner(project.getUserName())
+                .workflowInstanceId(workflowInstance.getId())
+                .workflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode())
+                .workflowInstanceName(workflowInstance.getName())
+                .commandType(workflowInstance.getCommandType())
+                .workflowExecutionStatus(workflowInstance.getState())
+                .modifyBy(modifyBy)
+                .recovery(workflowInstance.getRecovery())
+                .runTimes(workflowInstance.getRunTimes())
+                .workflowStartTime(workflowInstance.getStartTime())
+                .workflowEndTime(workflowInstance.getEndTime())
+                .workflowHost(workflowInstance.getHost())
+                .build();
+        successTaskList.add(workflowAlertContent);
+        res = JSONUtils.toJsonString(successTaskList);
 
         return res;
     }
@@ -177,12 +147,12 @@ private String getWorkerToleranceContent(WorkflowInstance workflowInstance, List
 
         List<WorkflowAlertContent> toleranceTaskInstanceList = new ArrayList<>();
 
-        WorkflowDefinitionLog workflowDefinitionLog = workflowDefinitionLogMapper
+        WorkflowDefinitionLog workflowDefinitionLog = workflowDefinitionLogDao
                 .queryByDefinitionCodeAndVersion(workflowInstance.getWorkflowDefinitionCode(),
                         workflowInstance.getWorkflowDefinitionVersion());
         String modifyBy = "";
         if (workflowDefinitionLog != null) {
-            User operator = userMapper.selectById(workflowDefinitionLog.getOperator());
+            User operator = userDao.queryById(workflowDefinitionLog.getOperator());
             modifyBy = operator == null ? "" : operator.getUserName();
         }
 
@@ -232,24 +202,23 @@ public void sendAlertWorkerToleranceFault(WorkflowInstance workflowInstance, Lis
      * send workflow instance alert
      *
      * @param workflowInstance workflow instance
-     * @param taskInstances task instance list
      */
-    public void sendAlertWorkflowInstance(WorkflowInstance workflowInstance,
-                                          List<TaskInstance> taskInstances,
-                                          ProjectUser projectUser) {
+    public void sendAlertWorkflowInstance(WorkflowInstance workflowInstance) {
         if (!isNeedToSendWarning(workflowInstance)) {
             return;
         }
+        Project project = projectDao.queryByCode(workflowInstance.getProjectCode());
+
         Alert alert = new Alert();
         String cmdName = getCommandCnName(workflowInstance.getCommandType());
         String success = workflowInstance.getState().isSuccess() ? "success" : "failed";
         alert.setTitle(cmdName + " " + success);
         alert.setWarningType(workflowInstance.getState().isSuccess() ? WarningType.SUCCESS : WarningType.FAILURE);
-        String content = getContentWorkflowInstance(workflowInstance, taskInstances, projectUser);
+        String content = getContentWorkflowInstance(workflowInstance, project);
         alert.setContent(content);
         alert.setAlertGroupId(workflowInstance.getWarningGroupId());
         alert.setCreateTime(new Date());
-        alert.setProjectCode(projectUser.getProjectCode());
+        alert.setProjectCode(workflowInstance.getProjectCode());
         alert.setWorkflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode());
         alert.setWorkflowInstanceId(workflowInstance.getId());
         alert.setAlertType(workflowInstance.getState().isSuccess() ? AlertType.WORKFLOW_INSTANCE_SUCCESS
@@ -295,54 +264,4 @@ public void sendTaskTimeoutAlert(WorkflowInstance workflowInstance,
                                      ProjectUser projectUser) {
         alertDao.sendTaskTimeoutAlert(workflowInstance, taskInstance, projectUser);
     }
-
-    /**
-     *
-     * check node type and workflow blocking flag, then insert a block record into db
-     *
-     * @param workflowInstance workflow instance
-     * @param projectUser the project owner
-     */
-    public void sendWorkflowBlockingAlert(WorkflowInstance workflowInstance,
-                                          ProjectUser projectUser) {
-        Alert alert = new Alert();
-        String cmdName = getCommandCnName(workflowInstance.getCommandType());
-        List<WorkflowAlertContent> blockingNodeList = new ArrayList<>(1);
-
-        WorkflowDefinitionLog workflowDefinitionLog = workflowDefinitionLogMapper
-                .queryByDefinitionCodeAndVersion(workflowInstance.getWorkflowDefinitionCode(),
-                        workflowInstance.getWorkflowDefinitionVersion());
-
-        String modifyBy = "";
-        if (workflowDefinitionLog != null) {
-            User operator = userMapper.selectById(workflowDefinitionLog.getOperator());
-            modifyBy = operator == null ? "" : operator.getUserName();
-        }
-
-        WorkflowAlertContent workflowAlertContent = WorkflowAlertContent.builder()
-                .projectCode(projectUser.getProjectCode())
-                .projectName(projectUser.getProjectName())
-                .owner(projectUser.getUserName())
-                .workflowInstanceId(workflowInstance.getId())
-                .workflowInstanceName(workflowInstance.getName())
-                .commandType(workflowInstance.getCommandType())
-                .workflowExecutionStatus(workflowInstance.getState())
-                .modifyBy(modifyBy)
-                .runTimes(workflowInstance.getRunTimes())
-                .workflowStartTime(workflowInstance.getStartTime())
-                .workflowEndTime(workflowInstance.getEndTime())
-                .workflowHost(workflowInstance.getHost())
-                .build();
-        blockingNodeList.add(workflowAlertContent);
-        String content = JSONUtils.toJsonString(blockingNodeList);
-        alert.setTitle(cmdName + " Blocked");
-        alert.setContent(content);
-        alert.setAlertGroupId(workflowInstance.getWarningGroupId());
-        alert.setCreateTime(new Date());
-        alert.setProjectCode(projectUser.getProjectCode());
-        alert.setWorkflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode());
-        alert.setWorkflowInstanceId(workflowInstance.getId());
-        alert.setAlertType(AlertType.WORKFLOW_INSTANCE_BLOCKED);
-        alertDao.addAlert(alert);
-    }
 }

dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManagerTest.java

@@ -17,18 +17,13 @@
 
 package org.apache.dolphinscheduler.service.alert;
 
-import org.apache.dolphinscheduler.common.enums.CommandType;
-import org.apache.dolphinscheduler.common.enums.WarningType;
-import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.AlertDao;
-import org.apache.dolphinscheduler.dao.entity.ProjectUser;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
 import org.apache.dolphinscheduler.dao.mapper.UserMapper;
 import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
 
 import java.util.ArrayList;
-import java.util.Date;
 import java.util.List;
 
 import org.junit.jupiter.api.Test;
@@ -64,7 +59,6 @@ public class WorkflowAlertManagerTest {
      */
     @Test
     public void sendWarningWorkerToleranceFaultTest() {
-        // process instance
         WorkflowInstance workflowInstance = new WorkflowInstance();
         workflowInstance.setName("test");
 
@@ -77,50 +71,4 @@ public void sendWarningWorkerToleranceFaultTest() {
 
         workflowAlertManager.sendAlertWorkerToleranceFault(workflowInstance, taskInstanceList);
     }
-
-    /**
-     * send worker alert fault tolerance
-     */
-    @Test
-    public void sendWarnningOfProcessInstanceTest() {
-        // process instance
-        WorkflowInstance workflowInstance = new WorkflowInstance();
-        workflowInstance.setWarningType(WarningType.SUCCESS);
-        workflowInstance.setState(WorkflowExecutionStatus.SUCCESS);
-        workflowInstance.setCommandType(CommandType.COMPLEMENT_DATA);
-        workflowInstance.setWarningGroupId(1);
-        workflowInstance.setWorkflowDefinitionCode(1L);
-        workflowInstance.setWorkflowDefinitionVersion(1);
-
-        ProjectUser projectUser = new ProjectUser();
-        TaskInstance taskInstance = new TaskInstance();
-        List<TaskInstance> taskInstanceList = new ArrayList<>();
-        taskInstanceList.add(taskInstance);
-
-        workflowAlertManager.sendAlertWorkflowInstance(workflowInstance, taskInstanceList, projectUser);
-    }
-
-    /**
-     * send blocking alert
-     */
-    @Test
-    public void sendBlockingAlertTest() {
-        // process instance
-        WorkflowInstance workflowInstance = new WorkflowInstance();
-        workflowInstance.setId(1);
-        workflowInstance.setName("test-process-01");
-        workflowInstance.setCommandType(CommandType.START_PROCESS);
-        workflowInstance.setState(WorkflowExecutionStatus.RUNNING_EXECUTION);
-        workflowInstance.setRunTimes(0);
-        workflowInstance.setStartTime(new Date());
-        workflowInstance.setEndTime(new Date());
-        workflowInstance.setHost("127.0.0.1");
-        workflowInstance.setWarningGroupId(1);
-        workflowInstance.setWorkflowDefinitionCode(1L);
-        workflowInstance.setWorkflowDefinitionVersion(1);
-
-        ProjectUser projectUser = new ProjectUser();
-
-        workflowAlertManager.sendWorkflowBlockingAlert(workflowInstance, projectUser);
-    }
 }