github
/
DolphinScheduler
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Browse Source

Merge remote-tracking branch 'upstream/dev' into json-split

pull/3/MERGE
lenboo 4 years ago
parent
d67b33514b 3dc4bdc4dd
commit
d1a0d585cc
7 changed files with 86 additions and 10 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 28
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessAlertContent.java
  2. 15
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProjectMapper.java
  3. 12
      dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProjectMapper.xml
  4. 4
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
  5. 15
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java
  6. 10
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/AlertManagerTest.java
  7. 10
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java

28
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ProcessAlertContent.java
Unescape View File

@ -33,6 +33,12 @@ import com.fasterxml.jackson.annotation.JsonProperty;
@JsonInclude(Include.NON_NULL)
public class ProcessAlertContent implements Serializable {
@JsonProperty("projectId")
private int projectId;
@JsonProperty("projectName")
private String projectName;
@JsonProperty("owner")
private String owner;
@JsonProperty("processId")
private int processId;
@JsonProperty("processName")
@ -79,6 +85,9 @@ public class ProcessAlertContent implements Serializable {
private String logPath;
private ProcessAlertContent(Builder builder) {
this.projectId = builder.projectId;
this.projectName = builder.projectName;
this.owner = builder.owner;
this.processId = builder.processId;
this.processName = builder.processName;
this.processType = builder.processType;
@ -107,7 +116,9 @@ public class ProcessAlertContent implements Serializable {
}
public static class Builder {
private int projectId;
private String projectName;
private String owner;
private int processId;
private String processName;
private CommandType processType;
@ -129,6 +140,21 @@ public class ProcessAlertContent implements Serializable {
private String taskHost;
private String logPath;
public Builder projectId(int projectId) {
this.projectId = projectId;
return this;
}
public Builder projectName(String projectName) {
this.projectName = projectName;
return this;
}
public Builder owner(String owner) {
this.owner = owner;
return this;
}
public Builder processId(int processId) {
this.processId = processId;
return this;

15
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProjectMapper.java
Unescape View File

@ -14,15 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.dao.entity.Project;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.ibatis.annotations.Param;
import java.util.List;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
/**
* project mapper interface
*/
@ -81,4 +85,11 @@ public interface ProjectMapper extends BaseMapper<Project> {
*/
List<Project> queryProjectCreatedAndAuthorizedByUserId(@Param("userId") int userId);
/**
* query project name and user name by processInstanceId.
* @param processInstanceId processInstanceId
* @return projectName and userName
*/
ProjectUser queryProjectWithUserByProcessInstanceId(@Param("processInstanceId") int processInstanceId);
}

12
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProjectMapper.xml
Unescape View File

@ -96,4 +96,16 @@
(select project_id from t_ds_relation_project_user where user_id=#{userId}
union select id as project_id from t_ds_project where user_id=#{userId})
</select>
<select id="queryProjectWithUserByProcessInstanceId" resultType="org.apache.dolphinscheduler.dao.entity.ProjectUser">
select
dp.id projectId,
dp.name projectName,
u.user_name userName
from t_ds_process_instance di
join t_ds_process_definition dpd on di.process_definition_id = dpd.id
join t_ds_project dp on dpd.project_id = dp.id
join t_ds_user u on dp.user_id = u.id
where di.id = #{processInstanceId};
</select>
</mapper>

4
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
Unescape View File

@ -47,6 +47,7 @@ import org.apache.dolphinscheduler.common.utils.ParameterUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.common.utils.VarPoolUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.utils.DagHelper;
@ -371,7 +372,8 @@ public class MasterExecThread implements Runnable {
processService.createRecoveryWaitingThreadCommand(null, processInstance);
}
List<TaskInstance> taskInstances = processService.findValidTaskListByProcessId(processInstance.getId());
alertManager.sendAlertProcessInstance(processInstance, taskInstances);
ProjectUser projectUser = processService.queryProjectWithUserByProcessInstanceId(processInstance.getId());
alertManager.sendAlertProcessInstance(processInstance, taskInstances, projectUser);
}
/**

15
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java
Unescape View File

@ -29,6 +29,7 @@ import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.dao.entity.ProcessAlertContent;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import java.util.ArrayList;
@ -94,12 +95,16 @@ public class AlertManager {
* @return process instance format content
*/
public String getContentProcessInstance(ProcessInstance processInstance,
List<TaskInstance> taskInstances) {
List<TaskInstance> taskInstances,
ProjectUser projectUser) {
String res = "";
if (processInstance.getState().typeIsSuccess()) {
List<ProcessAlertContent> successTaskList = new ArrayList<>(1);
ProcessAlertContent processAlertContent = ProcessAlertContent.newBuilder()
.projectId(projectUser.getProjectId())
.projectName(projectUser.getProjectName())
.owner(projectUser.getUserName())
.processId(processInstance.getId())
.processName(processInstance.getName())
.processType(processInstance.getCommandType())
@ -120,6 +125,9 @@ public class AlertManager {
continue;
}
ProcessAlertContent processAlertContent = ProcessAlertContent.newBuilder()
.projectId(projectUser.getProjectId())
.projectName(projectUser.getProjectName())
.owner(projectUser.getUserName())
.processId(processInstance.getId())
.processName(processInstance.getName())
.taskId(task.getId())
@ -196,7 +204,8 @@ public class AlertManager {
* @param taskInstances task instance list
*/
public void sendAlertProcessInstance(ProcessInstance processInstance,
List<TaskInstance> taskInstances) {
List<TaskInstance> taskInstances,
ProjectUser projectUser) {
if (Flag.YES == processInstance.getIsSubProcess()) {
return;
@ -231,7 +240,7 @@ public class AlertManager {
alert.setTitle(cmdName + " " + success);
ShowType showType = processInstance.getState().typeIsSuccess() ? ShowType.TEXT : ShowType.TABLE;
alert.setShowType(showType);
String content = getContentProcessInstance(processInstance, taskInstances);
String content = getContentProcessInstance(processInstance, taskInstances, projectUser);
alert.setContent(content);
alert.setAlertType(AlertType.EMAIL);
alert.setAlertGroupId(processInstance.getWarningGroupId());

10
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/AlertManagerTest.java
Unescape View File

@ -19,9 +19,11 @@ package org.apache.dolphinscheduler.server.master;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.server.utils.AlertManager;
import org.junit.Ignore;
@ -51,6 +53,9 @@ public class AlertManagerTest {
@Autowired
TaskInstanceMapper taskInstanceMapper;
@Autowired
ProjectMapper projectMapper;
AlertManager alertManager;
/**
@ -90,7 +95,6 @@ public class AlertManagerTest {
ProcessDefinition processDefinition = processDefinitionMapper.selectById(47);
processInstance.setProcessDefinition(processDefinition);
// fault task instance
TaskInstance toleranceTask1 = taskInstanceMapper.selectById(5038);
toleranceTask1.setState(ExecutionStatus.FAILURE);
@ -101,7 +105,9 @@ public class AlertManagerTest {
toleranceTaskList.add(toleranceTask1);
toleranceTaskList.add(toleranceTask2);
alertManager.sendAlertProcessInstance(processInstance, toleranceTaskList);
ProjectUser projectUser = projectMapper.queryProjectWithUserByProcessInstanceId(processInstance.getId());
alertManager.sendAlertProcessInstance(processInstance, toleranceTaskList, projectUser);
}
}

10
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
Unescape View File

@ -56,6 +56,7 @@ import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessInstanceMap;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.dolphinscheduler.dao.entity.Resource;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
@ -1859,6 +1860,15 @@ public class ProcessService {
return queue;
}
/**
* query project name and user name by processInstanceId.
* @param processInstanceId processInstanceId
* @return projectName and userName
*/
public ProjectUser queryProjectWithUserByProcessInstanceId(int processInstanceId) {
return projectMapper.queryProjectWithUserByProcessInstanceId(processInstanceId);
}
/**
* get task worker group
*

Write Preview
Loading…
Cancel
Save