Browse Source

[Feature] [ALERT-9406]enrich alert info when insert new row into DB (#9445)

* enrich alert info when insert new row into DB

* fix comments
3.0.0/version-upgrade
Tq 2 years ago committed by GitHub
parent
commit
fb0139e959
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 14
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java
  2. 12
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
  3. 9
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
  4. 18
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ProcessAlertManager.java

14
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java

@ -24,13 +24,17 @@ import com.baomidou.mybatisplus.annotation.EnumValue;
*/ */
public enum AlertType { public enum AlertType {
/** /**
* 0 process instance failure; 1 process instance success, 2 fault tolerance warning, 3 task failure, 4 task success * 0 process instance failure, 1 process instance success, 2 process instance blocked, 3 process instance timeout, 4 fault tolerance warning,
*/ * 5 task failure, 6 task success, 7 task timeout
*/
PROCESS_INSTANCE_FAILURE(0, "process instance failure"), PROCESS_INSTANCE_FAILURE(0, "process instance failure"),
PROCESS_INSTANCE_SUCCESS(1, "process instance success"), PROCESS_INSTANCE_SUCCESS(1, "process instance success"),
FAULT_TOLERANCE_WARNING(2, "fault tolerance warning"), PROCESS_INSTANCE_BLOCKED(2, "process instance blocked"),
TASK_FAILURE(3, "task failure"), PROCESS_INSTANCE_TIMEOUT(3, "process instance timeout"),
TASK_SUCCESS(4, "task success"); FAULT_TOLERANCE_WARNING(4, "fault tolerance warning"),
TASK_FAILURE(5, "task failure"),
TASK_SUCCESS(6, "task success"),
TASK_TIMEOUT(7, "task timeout"),;
AlertType(int code, String descp) { AlertType(int code, String descp) {
this.code = code; this.code = code;

12
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java

@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.enums.AlertEvent;
import org.apache.dolphinscheduler.common.enums.AlertStatus; import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.AlertWarnLevel; import org.apache.dolphinscheduler.common.enums.AlertWarnLevel;
import org.apache.dolphinscheduler.common.enums.WarningType; import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.Alert; import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance; import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
@ -73,7 +74,7 @@ public class AlertDao {
} }
/** /**
* update alert * update alert sending(execution) status
* *
* @param alertStatus alertStatus * @param alertStatus alertStatus
* @param log log * @param log log
@ -131,6 +132,7 @@ public class AlertDao {
alert.setAlertGroupId(alertGroupId); alert.setAlertGroupId(alertGroupId);
alert.setCreateTime(new Date()); alert.setCreateTime(new Date());
alert.setUpdateTime(new Date()); alert.setUpdateTime(new Date());
alert.setAlertType(AlertType.FAULT_TOLERANCE_WARNING);
// we use this method to avoid insert duplicate alert(issue #5525) // we use this method to avoid insert duplicate alert(issue #5525)
alertMapper.insertAlertWhenServerCrash(alert); alertMapper.insertAlertWhenServerCrash(alert);
} }
@ -163,6 +165,10 @@ public class AlertDao {
processAlertContentList.add(processAlertContent); processAlertContentList.add(processAlertContent);
String content = JSONUtils.toJsonString(processAlertContentList); String content = JSONUtils.toJsonString(processAlertContentList);
alert.setTitle("Process Timeout Warn"); alert.setTitle("Process Timeout Warn");
alert.setProjectCode(projectUser.getProjectCode());
alert.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
alert.setProcessInstanceId(processInstance.getId());
alert.setAlertType(AlertType.PROCESS_INSTANCE_TIMEOUT);
saveTaskTimeoutAlert(alert, content, alertGroupId); saveTaskTimeoutAlert(alert, content, alertGroupId);
} }
@ -203,6 +209,10 @@ public class AlertDao {
processAlertContentList.add(processAlertContent); processAlertContentList.add(processAlertContent);
String content = JSONUtils.toJsonString(processAlertContentList); String content = JSONUtils.toJsonString(processAlertContentList);
alert.setTitle("Task Timeout Warn"); alert.setTitle("Task Timeout Warn");
alert.setProjectCode(projectUser.getProjectCode());
alert.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
alert.setProcessInstanceId(processInstance.getId());
alert.setAlertType(AlertType.TASK_TIMEOUT);
saveTaskTimeoutAlert(alert, content, processInstance.getWarningGroupId()); saveTaskTimeoutAlert(alert, content, processInstance.getWarningGroupId());
} }

9
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java

@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.dao.entity; package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.AlertStatus; import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.enums.WarningType; import org.apache.dolphinscheduler.common.enums.WarningType;
import java.util.Date; import java.util.Date;
@ -66,7 +67,7 @@ public class Alert {
@TableField(value = "log") @TableField(value = "log")
private String log; private String log;
/** /**g
* alertgroup_id * alertgroup_id
*/ */
@TableField("alertgroup_id") @TableField("alertgroup_id")
@ -105,7 +106,7 @@ public class Alert {
* alert_type * alert_type
*/ */
@TableField("alert_type") @TableField("alert_type")
private int alertType; private AlertType alertType;
@TableField(exist = false) @TableField(exist = false)
private Map<String, Object> info = new HashMap<>(); private Map<String, Object> info = new HashMap<>();
@ -217,11 +218,11 @@ public class Alert {
this.processInstanceId = processInstanceId; this.processInstanceId = processInstanceId;
} }
public int getAlertType() { public AlertType getAlertType() {
return alertType; return alertType;
} }
public void setAlertType(int alertType) { public void setAlertType(AlertType alertType) {
this.alertType = alertType; this.alertType = alertType;
} }

18
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ProcessAlertManager.java

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.service.alert; package org.apache.dolphinscheduler.service.alert;
import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.enums.CommandType; import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag; import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.WarningType; import org.apache.dolphinscheduler.common.enums.WarningType;
@ -194,6 +195,7 @@ public class ProcessAlertManager {
alert.setWarningType(WarningType.FAILURE); alert.setWarningType(WarningType.FAILURE);
alert.setCreateTime(new Date()); alert.setCreateTime(new Date());
alert.setAlertGroupId(processInstance.getWarningGroupId() == null ? 1 : processInstance.getWarningGroupId()); alert.setAlertGroupId(processInstance.getWarningGroupId() == null ? 1 : processInstance.getWarningGroupId());
alert.setAlertType(AlertType.FAULT_TOLERANCE_WARNING);
alertDao.addAlert(alert); alertDao.addAlert(alert);
logger.info("add alert to db , alert : {}", alert); logger.info("add alert to db , alert : {}", alert);
@ -227,6 +229,10 @@ public class ProcessAlertManager {
alert.setContent(content); alert.setContent(content);
alert.setAlertGroupId(processInstance.getWarningGroupId()); alert.setAlertGroupId(processInstance.getWarningGroupId());
alert.setCreateTime(new Date()); alert.setCreateTime(new Date());
alert.setProjectCode(projectUser.getProjectCode());
alert.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
alert.setProcessInstanceId(processInstance.getId());
alert.setAlertType(processInstance.getState().typeIsSuccess() ? AlertType.PROCESS_INSTANCE_SUCCESS : AlertType.PROCESS_INSTANCE_FAILURE);
alertDao.addAlert(alert); alertDao.addAlert(alert);
logger.info("add alert to db , alert: {}", alert); logger.info("add alert to db , alert: {}", alert);
} }
@ -285,6 +291,11 @@ public class ProcessAlertManager {
alert.setContent(content); alert.setContent(content);
alert.setAlertGroupId(processInstance.getWarningGroupId()); alert.setAlertGroupId(processInstance.getWarningGroupId());
alert.setCreateTime(new Date()); alert.setCreateTime(new Date());
alert.setProjectCode(result.getProjectCode());
alert.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
alert.setProcessInstanceId(processInstance.getId());
//might need to change to data quality status
alert.setAlertType(processInstance.getState().typeIsSuccess() ? AlertType.PROCESS_INSTANCE_SUCCESS : AlertType.PROCESS_INSTANCE_FAILURE);
alertDao.addAlert(alert); alertDao.addAlert(alert);
logger.info("add alert to db , alert: {}", alert); logger.info("add alert to db , alert: {}", alert);
} }
@ -299,6 +310,9 @@ public class ProcessAlertManager {
alert.setContent(content); alert.setContent(content);
alert.setAlertGroupId(processInstance.getWarningGroupId()); alert.setAlertGroupId(processInstance.getWarningGroupId());
alert.setCreateTime(new Date()); alert.setCreateTime(new Date());
alert.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
alert.setProcessInstanceId(processInstance.getId());
alert.setAlertType(AlertType.TASK_FAILURE);
alertDao.addAlert(alert); alertDao.addAlert(alert);
logger.info("add alert to db , alert: {}", alert); logger.info("add alert to db , alert: {}", alert);
} }
@ -392,6 +406,10 @@ public class ProcessAlertManager {
alert.setContent(content); alert.setContent(content);
alert.setAlertGroupId(processInstance.getWarningGroupId()); alert.setAlertGroupId(processInstance.getWarningGroupId());
alert.setCreateTime(new Date()); alert.setCreateTime(new Date());
alert.setProjectCode(projectUser.getProjectCode());
alert.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
alert.setProcessInstanceId(processInstance.getId());
alert.setAlertType(AlertType.PROCESS_INSTANCE_BLOCKED);
alertDao.addAlert(alert); alertDao.addAlert(alert);
logger.info("add alert to db, alert: {}",alert); logger.info("add alert to db, alert: {}",alert);
} }

Loading…
Cancel
Save