You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ga...@apache.org on 2022/03/14 06:25:34 UTC
[dolphinscheduler] branch dev updated: [DS-7992][feat] Alert module judging strategy (#8636)
This is an automated email from the ASF dual-hosted git repository.
gaojun2048 pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 0d8079a [DS-7992][feat] Alert module judging strategy (#8636)
0d8079a is described below
commit 0d8079a816a0bdfca754ad12168d671a2e16195e
Author: wangyang <wa...@lqwang.net>
AuthorDate: Mon Mar 14 14:25:27 2022 +0800
[DS-7992][feat] Alert module judging strategy (#8636)
---
.../dolphinscheduler/alert/api/AlertConstants.java | 4 ++
.../dolphinscheduler/alert/api/AlertData.java | 29 +++++++--
.../dolphinscheduler/alert/AlertPluginManager.java | 23 +++++++-
.../alert/AlertRequestProcessor.java | 3 +-
.../apache/dolphinscheduler/alert/AlertSender.java | 68 ++++++++++++++++++----
.../alert/processor/AlertRequestProcessorTest.java | 3 +-
.../alert/runner/AlertSenderTest.java | 12 ++--
.../dolphinscheduler/common/enums/WarningType.java | 16 +++++
.../org/apache/dolphinscheduler/dao/AlertDao.java | 3 +
.../apache/dolphinscheduler/dao/entity/Alert.java | 22 +++++++
.../dolphinscheduler/dao/mapper/AlertMapper.xml | 6 +-
.../src/main/resources/sql/dolphinscheduler_h2.sql | 3 +-
.../main/resources/sql/dolphinscheduler_mysql.sql | 1 +
.../resources/sql/dolphinscheduler_postgresql.sql | 1 +
.../2.1.0_schema/mysql/dolphinscheduler_ddl.sql | 3 +-
.../postgresql/dolphinscheduler_ddl.sql | 2 +
.../dao/mapper/AlertMapperTest.java | 4 +-
.../command/alert/AlertSendRequestCommand.java | 13 ++++-
.../command/alert/AlertSendRequestCommandTest.java | 3 +-
.../service/alert/AlertClientService.java | 10 ++--
.../service/alert/ProcessAlertManager.java | 2 +
.../service/alert/AlertClientServiceTest.java | 15 ++---
.../server/worker/runner/TaskExecuteThread.java | 11 ++--
23 files changed, 211 insertions(+), 46 deletions(-)
diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertConstants.java b/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertConstants.java
index 9d85fd2..6c0848f 100644
--- a/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertConstants.java
+++ b/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertConstants.java
@@ -27,6 +27,10 @@ public final class AlertConstants {
public static final String NAME_SHOW_TYPE = "showType";
+ public static final String WARNING_TYPE = "warningType";
+
+ public static final String NAME_WARNING_TYPE = "WarningType";
+
private AlertConstants() {
throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
}
diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertData.java b/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertData.java
index 9d1db84..a920139 100644
--- a/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertData.java
+++ b/dolphinscheduler-alert/dolphinscheduler-alert-api/src/main/java/org/apache/dolphinscheduler/alert/api/AlertData.java
@@ -24,12 +24,14 @@ public class AlertData {
private String title;
private String content;
private String log;
+ private int warnType;
- public AlertData(int id, String title, String content, String log) {
+ public AlertData(int id, String title, String content, String log, int warnType) {
this.id = id;
this.title = title;
this.content = content;
this.log = log;
+ this.warnType = warnType;
}
public AlertData() {
@@ -75,6 +77,14 @@ public class AlertData {
return this;
}
+ public int getWarnType() {
+ return warnType;
+ }
+
+ public void setWarnType(int warnType) {
+ this.warnType = warnType;
+ }
+
public boolean equals(final Object o) {
if (o == this) {
return true;
@@ -89,6 +99,9 @@ public class AlertData {
if (this.getId() != other.getId()) {
return false;
}
+ if (this.getWarnType() != other.getWarnType()) {
+ return false;
+ }
final Object this$title = this.getTitle();
final Object other$title = other.getTitle();
if (this$title == null ? other$title != null : !this$title.equals(other$title)) {
@@ -115,6 +128,7 @@ public class AlertData {
final int PRIME = 59;
int result = 1;
result = result * PRIME + this.getId();
+ result = result * PRIME + this.getWarnType();
final Object $title = this.getTitle();
result = result * PRIME + ($title == null ? 43 : $title.hashCode());
final Object $content = this.getContent();
@@ -125,7 +139,7 @@ public class AlertData {
}
public String toString() {
- return "AlertData(id=" + this.getId() + ", title=" + this.getTitle() + ", content=" + this.getContent() + ", log=" + this.getLog() + ")";
+ return "AlertData(id=" + this.getId() + ", title=" + this.getTitle() + ", content=" + this.getContent() + ", log=" + this.getLog() + ", warnType=" + this.getWarnType() + ")";
}
public static class AlertDataBuilder {
@@ -133,6 +147,7 @@ public class AlertData {
private String title;
private String content;
private String log;
+ private int warnType;
AlertDataBuilder() {
}
@@ -157,12 +172,18 @@ public class AlertData {
return this;
}
+ public AlertDataBuilder warnType(int warnType) {
+ this.warnType = warnType;
+ return this;
+ }
+
public AlertData build() {
- return new AlertData(id, title, content, log);
+ return new AlertData(id, title, content, log, warnType);
}
public String toString() {
- return "AlertData.AlertDataBuilder(id=" + this.id + ", title=" + this.title + ", content=" + this.content + ", log=" + this.log + ")";
+ return "AlertData.AlertDataBuilder(id=" + this.id + ", title=" + this.title + ", content=" + this.content + ", log=" + this.log + ", warnType=" + this.warnType + ")";
}
+
}
}
diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertPluginManager.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertPluginManager.java
index 682a496..f590d31 100644
--- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertPluginManager.java
+++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertPluginManager.java
@@ -21,12 +21,18 @@ import static java.lang.String.format;
import org.apache.dolphinscheduler.alert.api.AlertChannel;
import org.apache.dolphinscheduler.alert.api.AlertChannelFactory;
+import org.apache.dolphinscheduler.alert.api.AlertConstants;
import org.apache.dolphinscheduler.common.enums.PluginType;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
+import org.apache.dolphinscheduler.spi.params.base.ParamsOptions;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
+import org.apache.dolphinscheduler.spi.params.base.Validate;
+import org.apache.dolphinscheduler.spi.params.radio.RadioParam;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -51,10 +57,23 @@ public final class AlertPluginManager {
private final Map<Integer, AlertChannel> channelKeyedById = new HashMap<>();
+ private final PluginParams warningTypeParams = getWarningTypeParams();
+
public AlertPluginManager(PluginDao pluginDao) {
this.pluginDao = pluginDao;
}
+ public PluginParams getWarningTypeParams() {
+ return
+ RadioParam.newBuilder(AlertConstants.NAME_WARNING_TYPE, AlertConstants.WARNING_TYPE)
+ .addParamsOptions(new ParamsOptions(WarningType.SUCCESS.getDescp(), WarningType.SUCCESS.getDescp(), false))
+ .addParamsOptions(new ParamsOptions(WarningType.FAILURE.getDescp(), WarningType.FAILURE.getDescp(), false))
+ .addParamsOptions(new ParamsOptions(WarningType.ALL.getDescp(), WarningType.ALL.getDescp(), false))
+ .setValue(WarningType.ALL.getDescp())
+ .addValidate(Validate.newBuilder().setRequired(true).build())
+ .build();
+ }
+
@EventListener
public void installPlugin(ApplicationReadyEvent readyEvent) {
final Set<String> names = new HashSet<>();
@@ -72,7 +91,9 @@ public final class AlertPluginManager {
logger.info("Registered alert plugin: {}", name);
- final List<PluginParams> params = factory.params();
+ final List<PluginParams> params = new ArrayList<>(factory.params());
+ params.add(0, warningTypeParams);
+
final String paramsJson = PluginParamsTransfer.transferParamsToJson(params);
final PluginDefine pluginDefine = new PluginDefine(name, PluginType.ALERT.getDesc(), paramsJson);
diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertRequestProcessor.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertRequestProcessor.java
index 31c690a..c85292f 100644
--- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertRequestProcessor.java
+++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertRequestProcessor.java
@@ -54,7 +54,8 @@ public final class AlertRequestProcessor implements NettyRequestProcessor {
AlertSendResponseCommand alertSendResponseCommand = alertSender.syncHandler(
alertSendRequestCommand.getGroupId(),
alertSendRequestCommand.getTitle(),
- alertSendRequestCommand.getContent());
+ alertSendRequestCommand.getContent(),
+ alertSendRequestCommand.getWarnType());
channel.writeAndFlush(alertSendResponseCommand.convert2Command(command.getOpaque()));
}
}
diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertSender.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertSender.java
index 9f7268f..2f15fb6 100644
--- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertSender.java
+++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertSender.java
@@ -18,10 +18,12 @@
package org.apache.dolphinscheduler.alert;
import org.apache.dolphinscheduler.alert.api.AlertChannel;
+import org.apache.dolphinscheduler.alert.api.AlertConstants;
import org.apache.dolphinscheduler.alert.api.AlertData;
import org.apache.dolphinscheduler.alert.api.AlertInfo;
import org.apache.dolphinscheduler.alert.api.AlertResult;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
@@ -66,13 +68,17 @@ public final class AlertSender {
alertData.setId(alert.getId())
.setContent(alert.getContent())
.setLog(alert.getLog())
- .setTitle(alert.getTitle());
+ .setTitle(alert.getTitle())
+ .setTitle(alert.getTitle())
+ .setWarnType(alert.getWarningType().getCode());
for (AlertPluginInstance instance : alertInstanceList) {
AlertResult alertResult = this.alertResultHandler(instance, alertData);
- AlertStatus alertStatus = Boolean.parseBoolean(String.valueOf(alertResult.getStatus())) ? AlertStatus.EXECUTION_SUCCESS : AlertStatus.EXECUTION_FAILURE;
- alertDao.updateAlert(alertStatus, alertResult.getMessage(), alert.getId());
+ if (alertResult != null) {
+ AlertStatus alertStatus = Boolean.parseBoolean(String.valueOf(alertResult.getStatus())) ? AlertStatus.EXECUTION_SUCCESS : AlertStatus.EXECUTION_FAILURE;
+ alertDao.updateAlert(alertStatus, alertResult.getMessage(), alert.getId());
+ }
}
}
@@ -86,11 +92,12 @@ public final class AlertSender {
* @param content content
* @return AlertSendResponseCommand
*/
- public AlertSendResponseCommand syncHandler(int alertGroupId, String title, String content) {
+ public AlertSendResponseCommand syncHandler(int alertGroupId, String title, String content , int warnType) {
List<AlertPluginInstance> alertInstanceList = alertDao.listInstanceByAlertGroupId(alertGroupId);
AlertData alertData = new AlertData();
alertData.setContent(content)
- .setTitle(title);
+ .setTitle(title)
+ .setWarnType(warnType);
boolean sendResponseStatus = true;
List<AlertSendResponseResult> sendResponseResults = new ArrayList<>();
@@ -107,10 +114,12 @@ public final class AlertSender {
for (AlertPluginInstance instance : alertInstanceList) {
AlertResult alertResult = this.alertResultHandler(instance, alertData);
- AlertSendResponseResult alertSendResponseResult = new AlertSendResponseResult(
- Boolean.parseBoolean(String.valueOf(alertResult.getStatus())), alertResult.getMessage());
- sendResponseStatus = sendResponseStatus && alertSendResponseResult.getStatus();
- sendResponseResults.add(alertSendResponseResult);
+ if (alertResult != null) {
+ AlertSendResponseResult alertSendResponseResult = new AlertSendResponseResult(
+ Boolean.parseBoolean(String.valueOf(alertResult.getStatus())), alertResult.getMessage());
+ sendResponseStatus = sendResponseStatus && alertSendResponseResult.getStatus();
+ sendResponseResults.add(alertSendResponseResult);
+ }
}
return new AlertSendResponseCommand(sendResponseStatus, sendResponseResults);
@@ -135,9 +144,48 @@ public final class AlertSender {
return alertResultExtend;
}
+ Map<String, String> paramsMap = JSONUtils.toMap(instance.getPluginInstanceParams());
+ String instanceWarnType = WarningType.ALL.getDescp();
+
+ if(paramsMap != null){
+ instanceWarnType = paramsMap.getOrDefault(AlertConstants.NAME_WARNING_TYPE, WarningType.ALL.getDescp());
+ }
+
+ WarningType warningType = WarningType.of(instanceWarnType);
+
+ if (warningType == null) {
+ String message = String.format("Alert Plugin %s send error : plugin warnType is null", pluginInstanceName);
+ alertResultExtend.setStatus(String.valueOf(false));
+ alertResultExtend.setMessage(message);
+ logger.error("Alert Plugin {} send error : plugin warnType is null", pluginInstanceName);
+ return alertResultExtend;
+ }
+
+ boolean sendWarning = false;
+ switch (warningType) {
+ case ALL:
+ sendWarning = true;
+ break;
+ case SUCCESS:
+ if (alertData.getWarnType() == WarningType.SUCCESS.getCode()) {
+ sendWarning = true;
+ }
+ break;
+ case FAILURE:
+ if (alertData.getWarnType() == WarningType.FAILURE.getCode()) {
+ sendWarning = true;
+ }
+ break;
+ default:
+ }
+
+ if (!sendWarning) {
+ logger.info("Alert Plugin {} send ignore warning type not match: plugin warning type is {}, alert data warning type is {}", pluginInstanceName, warningType.getCode(), alertData.getWarnType());
+ return null;
+ }
+
AlertInfo alertInfo = new AlertInfo();
alertInfo.setAlertData(alertData);
- Map<String, String> paramsMap = JSONUtils.toMap(instance.getPluginInstanceParams());
alertInfo.setAlertParams(paramsMap);
AlertResult alertResult;
try {
diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessorTest.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessorTest.java
index 20e6768..64e92a0 100644
--- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessorTest.java
+++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessorTest.java
@@ -21,6 +21,7 @@ import static org.mockito.Mockito.mock;
import org.apache.dolphinscheduler.alert.AlertRequestProcessor;
import org.apache.dolphinscheduler.alert.AlertSender;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
@@ -44,7 +45,7 @@ public class AlertRequestProcessorTest {
@Test
public void testProcess() {
Channel channel = mock(Channel.class);
- AlertSendRequestCommand alertSendRequestCommand = new AlertSendRequestCommand(1, "title", "content");
+ AlertSendRequestCommand alertSendRequestCommand = new AlertSendRequestCommand(1, "title", "content", WarningType.FAILURE.getCode());
Command reqCommand = alertSendRequestCommand.convert2Command();
Assert.assertEquals(CommandType.ALERT_SEND_REQUEST, reqCommand.getType());
alertRequestProcessor.process(channel, reqCommand);
diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java
index 160afeb..4060e46 100644
--- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java
+++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java
@@ -24,6 +24,7 @@ import org.apache.dolphinscheduler.alert.AlertPluginManager;
import org.apache.dolphinscheduler.alert.AlertSender;
import org.apache.dolphinscheduler.alert.api.AlertChannel;
import org.apache.dolphinscheduler.alert.api.AlertResult;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
@@ -69,7 +70,7 @@ public class AlertSenderTest {
//1.alert instance does not exist
when(alertDao.listInstanceByAlertGroupId(alertGroupId)).thenReturn(null);
- AlertSendResponseCommand alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
+ AlertSendResponseCommand alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@@ -88,7 +89,7 @@ public class AlertSenderTest {
PluginDefine pluginDefine = new PluginDefine(pluginName, "1", null);
when(pluginDao.getPluginDefineById(pluginDefineId)).thenReturn(pluginDefine);
- alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
+ alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@@ -98,7 +99,7 @@ public class AlertSenderTest {
when(alertChannelMock.process(Mockito.any())).thenReturn(null);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
- alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
+ alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@@ -110,7 +111,7 @@ public class AlertSenderTest {
when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
- alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
+ alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@@ -122,7 +123,7 @@ public class AlertSenderTest {
when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
- alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
+ alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertTrue(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@@ -139,6 +140,7 @@ public class AlertSenderTest {
alert.setAlertGroupId(alertGroupId);
alert.setTitle(title);
alert.setContent(content);
+ alert.setWarningType(WarningType.FAILURE);
alertList.add(alert);
alertSender = new AlertSender(alertDao, alertPluginManager);
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WarningType.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WarningType.java
index d3cbffd..d3feeba 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WarningType.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/WarningType.java
@@ -17,7 +17,13 @@
package org.apache.dolphinscheduler.common.enums;
+import static java.util.stream.Collectors.toMap;
+
+import java.util.Arrays;
+import java.util.Map;
+
import com.baomidou.mybatisplus.annotation.EnumValue;
+import com.google.common.base.Functions;
/**
* types for whether to send warning when process ending;
@@ -50,4 +56,14 @@ public enum WarningType {
public String getDescp() {
return descp;
}
+
+ private static final Map<String, WarningType> WARNING_TYPE_MAP =
+ Arrays.stream(WarningType.values()).collect(toMap(WarningType::getDescp, Functions.identity()));
+
+ public static WarningType of(String descp) {
+ if (WARNING_TYPE_MAP.containsKey(descp)) {
+ return WARNING_TYPE_MAP.get(descp);
+ }
+ return null;
+ }
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
index d0e8365..487d42a 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.dao;
import org.apache.dolphinscheduler.common.enums.AlertEvent;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.AlertWarnLevel;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
@@ -101,6 +102,7 @@ public class AlertDao {
Alert alert = new Alert();
alert.setTitle("Fault tolerance warning");
+ alert.setWarningType(WarningType.FAILURE);
alert.setAlertStatus(AlertStatus.WAIT_EXECUTION);
alert.setContent(content);
alert.setAlertGroupId(alertGroupId);
@@ -140,6 +142,7 @@ public class AlertDao {
private void saveTaskTimeoutAlert(Alert alert, String content, int alertGroupId) {
alert.setAlertGroupId(alertGroupId);
+ alert.setWarningType(WarningType.FAILURE);
alert.setContent(content);
alert.setCreateTime(new Date());
alert.setUpdateTime(new Date());
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
index 752ab9d..7b59f49 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
@@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import java.util.Date;
import java.util.HashMap;
@@ -52,6 +53,13 @@ public class Alert {
*/
@TableField(value = "alert_status")
private AlertStatus alertStatus;
+
+ /**
+ * warning_type
+ */
+ @TableField(value = "warning_type")
+ private WarningType warningType;
+
/**
* log
*/
@@ -151,6 +159,14 @@ public class Alert {
this.info = info;
}
+ public WarningType getWarningType() {
+ return warningType;
+ }
+
+ public void setWarningType(WarningType warningType) {
+ this.warningType = warningType;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -183,6 +199,9 @@ public class Alert {
if (!createTime.equals(alert.createTime)) {
return false;
}
+ if (warningType != alert.warningType) {
+ return false;
+ }
return updateTime.equals(alert.updateTime) && info.equals(alert.info);
}
@@ -193,6 +212,7 @@ public class Alert {
result = 31 * result + title.hashCode();
result = 31 * result + content.hashCode();
result = 31 * result + alertStatus.hashCode();
+ result = 31 * result + warningType.hashCode();
result = 31 * result + log.hashCode();
result = 31 * result + alertGroupId;
result = 31 * result + createTime.hashCode();
@@ -213,6 +233,8 @@ public class Alert {
+ '\''
+ ", alertStatus="
+ alertStatus
+ + ", warningType="
+ + warningType
+ ", log='"
+ log
+ '\''
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml
index 40f5383..912fba0 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml
@@ -21,7 +21,7 @@
<sql id="baseSql">
id
, title, content, alert_status, log,
- alertgroup_id, create_time, update_time
+ alertgroup_id, create_time, update_time, warning_type
</sql>
<select id="listAlertByStatus" resultType="org.apache.dolphinscheduler.dao.entity.Alert">
select
@@ -31,8 +31,8 @@
</select>
<insert id="insertAlertWhenServerCrash">
- insert into t_ds_alert(title, content, alert_status, log, alertgroup_id, create_time, update_time)
- SELECT #{alert.title}, #{alert.content}, #{alert.alertStatus.code}, #{alert.log}, #{alert.alertGroupId},
+ insert into t_ds_alert(title, content, alert_status, warning_type, log, alertgroup_id, create_time, update_time)
+ SELECT #{alert.title}, #{alert.content}, #{alert.alertStatus.code}, #{alert.warningType.code}, #{alert.log}, #{alert.alertGroupId},
#{alert.createTime}, #{alert.updateTime}
from t_ds_alert
where content = #{alert.content} and alert_status = #{alert.alertStatus.code}
diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql
index 1da410e..6e046ab 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql
@@ -274,6 +274,7 @@ CREATE TABLE t_ds_alert
title varchar(64) DEFAULT NULL,
content text,
alert_status tinyint(4) DEFAULT '0',
+ warning_type tinyint(4) DEFAULT '2',
log text,
alertgroup_id int(11) DEFAULT NULL,
create_time datetime DEFAULT NULL,
@@ -1908,4 +1909,4 @@ CREATE TABLE t_ds_k8s_namespace (
-- Records of t_ds_k8s_namespace
-- ----------------------------
INSERT INTO t_ds_k8s_namespace
-VALUES (1, 10000, 'default', 99, 'owner',1,NULL,1,'test',NULL,'default',null,null);
\ No newline at end of file
+VALUES (1, 10000, 'default', 99, 'owner',1,NULL,1,'test',NULL,'default',null,null);
diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql
index 00238b1..023f6f6 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql
@@ -281,6 +281,7 @@ CREATE TABLE `t_ds_alert` (
`title` varchar(64) DEFAULT NULL COMMENT 'title',
`content` text COMMENT 'Message content (can be email, can be SMS. Mail is stored in JSON map, and SMS is string)',
`alert_status` tinyint(4) DEFAULT '0' COMMENT '0:wait running,1:success,2:failed',
+ `warning_type` tinyint(4) DEFAULT '2' COMMENT '1 process is successfully, 2 process/task is failed',
`log` text COMMENT 'log',
`alertgroup_id` int(11) DEFAULT NULL COMMENT 'alert group id',
`create_time` datetime DEFAULT NULL COMMENT 'create time',
diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql
index 9d941f9..d537d6c 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql
@@ -210,6 +210,7 @@ CREATE TABLE t_ds_alert (
title varchar(64) DEFAULT NULL ,
content text ,
alert_status int DEFAULT '0' ,
+ warning_type int DEFAULT '2' ,
log text ,
alertgroup_id int DEFAULT NULL ,
create_time timestamp DEFAULT NULL ,
diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/mysql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/mysql/dolphinscheduler_ddl.sql
index c5a667a..ac08027 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/mysql/dolphinscheduler_ddl.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/mysql/dolphinscheduler_ddl.sql
@@ -30,6 +30,7 @@ alter table t_ds_task_definition add `task_group_id` int(11) DEFAULT NULL COMMEN
alter table t_ds_task_definition add `task_group_priority` int(11) DEFAULT '0' COMMENT 'task group id' AFTER `task_group_id`;
ALTER TABLE `t_ds_user` ADD COLUMN `time_zone` varchar(32) DEFAULT NULL COMMENT 'time zone';
+ALTER TABLE `t_ds_alert` ADD COLUMN `warning_type` tinyint(4) DEFAULT '2' COMMENT '1 process is successfully, 2 process/task is failed';
ALTER TABLE `t_ds_alert` ADD INDEX `idx_status` (`alert_status`) USING BTREE;
@@ -209,4 +210,4 @@ CREATE TABLE `t_ds_k8s_namespace` (
`update_time` datetime DEFAULT NULL COMMENT 'update time',
PRIMARY KEY (`id`),
UNIQUE KEY `k8s_namespace_unique` (`namespace`,`k8s`)
-) ENGINE= INNODB AUTO_INCREMENT= 1 DEFAULT CHARSET= utf8;
\ No newline at end of file
+) ENGINE= INNODB AUTO_INCREMENT= 1 DEFAULT CHARSET= utf8;
diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/postgresql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/postgresql/dolphinscheduler_ddl.sql
index ecfbc76..9e2459f 100644
--- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/2.1.0_schema/postgresql/dolphinscheduler_ddl.sql
@@ -46,6 +46,8 @@ EXECUTE 'CREATE INDEX IF NOT EXISTS idx_task_definition_log_code_version ON ' ||
EXECUTE 'ALTER TABLE ' || quote_ident(v_schema) ||'.t_ds_user ADD COLUMN IF NOT EXISTS "time_zone" varchar(32) DEFAULT NULL';
+EXECUTE 'ALTER TABLE ' || quote_ident(v_schema) ||'.t_ds_alert ADD COLUMN IF NOT EXISTS "warning_type" int DEFAULT 2';
+
EXECUTE 'CREATE TABLE IF NOT EXISTS' || quote_ident(v_schema) ||'."t_ds_dq_comparison_type" (
id serial NOT NULL,
"type" varchar NOT NULL,
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java
index 35159fe..b92d47e 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertMapperTest.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.Alert;
@@ -156,6 +157,7 @@ public class AlertMapperTest extends BaseDaoTest {
alert.setTitle("test alert");
alert.setContent("[{'type':'WORKER','host':'192.168.xx.xx','event':'server down','warning level':'serious'}]");
alert.setAlertStatus(alertStatus);
+ alert.setWarningType(WarningType.FAILURE);
alert.setLog("success");
alert.setCreateTime(DateUtils.getCurrentDate());
alert.setUpdateTime(DateUtils.getCurrentDate());
@@ -163,4 +165,4 @@ public class AlertMapperTest extends BaseDaoTest {
alertMapper.insert(alert);
return alert;
}
-}
\ No newline at end of file
+}
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommand.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommand.java
index da56b0d..ba37e22 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommand.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommand.java
@@ -31,6 +31,8 @@ public class AlertSendRequestCommand implements Serializable {
private String content;
+ private int warnType;
+
public int getGroupId() {
return groupId;
}
@@ -55,14 +57,23 @@ public class AlertSendRequestCommand implements Serializable {
this.content = content;
}
+ public int getWarnType() {
+ return warnType;
+ }
+
+ public void setWarnType(int warnType) {
+ this.warnType = warnType;
+ }
+
public AlertSendRequestCommand(){
}
- public AlertSendRequestCommand(int groupId, String title, String content) {
+ public AlertSendRequestCommand(int groupId, String title, String content, int warnType) {
this.groupId = groupId;
this.title = title;
this.content = content;
+ this.warnType = warnType;
}
/**
diff --git a/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommandTest.java b/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommandTest.java
index 79d2131..3de8908 100644
--- a/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommandTest.java
+++ b/dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/remote/command/alert/AlertSendRequestCommandTest.java
@@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.remote.command.alert;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
@@ -30,7 +31,7 @@ public class AlertSendRequestCommandTest {
int groupId = 1;
String title = "test-title";
String content = "test-content";
- AlertSendRequestCommand requestCommand = new AlertSendRequestCommand(groupId,title,content);
+ AlertSendRequestCommand requestCommand = new AlertSendRequestCommand(groupId,title,content,WarningType.FAILURE.getCode());
Command command = requestCommand.convert2Command();
Assert.assertEquals(CommandType.ALERT_SEND_REQUEST,command.getType());
AlertSendRequestCommand verifyCommand = new AlertSendRequestCommand();
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/AlertClientService.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/AlertClientService.java
index 0aeb25b..f5d7f93 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/AlertClientService.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/AlertClientService.java
@@ -84,8 +84,8 @@ public class AlertClientService implements AutoCloseable {
* @param content
* @return
*/
- public AlertSendResponseCommand sendAlert(int groupId, String title, String content) {
- return this.sendAlert(this.host,this.port,groupId,title,content);
+ public AlertSendResponseCommand sendAlert(int groupId, String title, String content, int strategy) {
+ return this.sendAlert(this.host,this.port,groupId,title,content,strategy);
}
/**
@@ -97,9 +97,9 @@ public class AlertClientService implements AutoCloseable {
* @param content content
* @return AlertSendResponseCommand
*/
- public AlertSendResponseCommand sendAlert(String host, int port, int groupId, String title, String content) {
- logger.info("sync alert send, host : {}, port : {}, groupId : {}, title : {} ", host, port, groupId, title);
- AlertSendRequestCommand request = new AlertSendRequestCommand(groupId, title, content);
+ public AlertSendResponseCommand sendAlert(String host, int port, int groupId, String title, String content, int strategy) {
+ logger.info("sync alert send, host : {}, port : {}, groupId : {}, title : {} , strategy : {} ", host, port, groupId, title, strategy);
+ AlertSendRequestCommand request = new AlertSendRequestCommand(groupId, title, content, strategy);
final Host address = new Host(host, port);
try {
Command command = request.convert2Command();
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ProcessAlertManager.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ProcessAlertManager.java
index 1a8cb8b..b679a6b 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ProcessAlertManager.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ProcessAlertManager.java
@@ -193,6 +193,7 @@ public class ProcessAlertManager {
alert.setTitle("worker fault tolerance");
String content = getWorkerToleranceContent(processInstance, toleranceTaskList);
alert.setContent(content);
+ alert.setWarningType(WarningType.FAILURE);
alert.setCreateTime(new Date());
alert.setAlertGroupId(processInstance.getWarningGroupId() == null ? 1 : processInstance.getWarningGroupId());
alertDao.addAlert(alert);
@@ -223,6 +224,7 @@ public class ProcessAlertManager {
String cmdName = getCommandCnName(processInstance.getCommandType());
String success = processInstance.getState().typeIsSuccess() ? "success" : "failed";
alert.setTitle(cmdName + " " + success);
+ alert.setWarningType(processInstance.getState().typeIsSuccess() ? WarningType.SUCCESS : WarningType.FAILURE);
String content = getContentProcessInstance(processInstance, taskInstances,projectUser);
alert.setContent(content);
alert.setAlertGroupId(processInstance.getWarningGroupId());
diff --git a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/AlertClientServiceTest.java b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/AlertClientServiceTest.java
index ba44191..0499e37 100644
--- a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/AlertClientServiceTest.java
+++ b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/AlertClientServiceTest.java
@@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.service.alert;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.remote.NettyRemotingClient;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendRequestCommand;
@@ -67,10 +68,10 @@ public class AlertClientServiceTest {
String content = "test-content";
//1.alter server does not exist
- AlertSendResponseCommand alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content);
+ AlertSendResponseCommand alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content, WarningType.FAILURE.getCode());
Assert.assertNull(alertSendResponseCommand);
- AlertSendRequestCommand alertSendRequestCommand = new AlertSendRequestCommand(groupId,title,content);
+ AlertSendRequestCommand alertSendRequestCommand = new AlertSendRequestCommand(groupId,title,content, WarningType.FAILURE.getCode());
Command reqCommand = alertSendRequestCommand.convert2Command();
boolean sendResponseStatus;
List<AlertSendResponseResult> sendResponseResults = new ArrayList<>();
@@ -86,7 +87,7 @@ public class AlertClientServiceTest {
Command resCommand = alertSendResponseCommandData.convert2Command(reqCommand.getOpaque());
PowerMockito.when(client.sendSync(Mockito.any(), Mockito.any(), Mockito.anyLong())).thenReturn(resCommand);
- alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content);
+ alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content, WarningType.FAILURE.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
@@ -100,7 +101,7 @@ public class AlertClientServiceTest {
alertSendResponseCommandData = new AlertSendResponseCommand(sendResponseStatus, sendResponseResults);
resCommand = alertSendResponseCommandData.convert2Command(reqCommand.getOpaque());
PowerMockito.when(client.sendSync(Mockito.any(), Mockito.any(), Mockito.anyLong())).thenReturn(resCommand);
- alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content);
+ alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content, WarningType.FAILURE.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
@@ -113,7 +114,7 @@ public class AlertClientServiceTest {
alertSendResponseCommandData = new AlertSendResponseCommand(sendResponseStatus, sendResponseResults);
resCommand = alertSendResponseCommandData.convert2Command(reqCommand.getOpaque());
PowerMockito.when(client.sendSync(Mockito.any(), Mockito.any(), Mockito.anyLong())).thenReturn(resCommand);
- alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content);
+ alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content, WarningType.FAILURE.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
@@ -125,7 +126,7 @@ public class AlertClientServiceTest {
alertSendResponseCommandData = new AlertSendResponseCommand(sendResponseStatus, sendResponseResults);
resCommand = alertSendResponseCommandData.convert2Command(reqCommand.getOpaque());
PowerMockito.when(client.sendSync(Mockito.any(), Mockito.any(), Mockito.anyLong())).thenReturn(resCommand);
- alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content);
+ alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content, WarningType.FAILURE.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
@@ -138,7 +139,7 @@ public class AlertClientServiceTest {
alertSendResponseCommandData = new AlertSendResponseCommand(sendResponseStatus, sendResponseResults);
resCommand = alertSendResponseCommandData.convert2Command(reqCommand.getOpaque());
PowerMockito.when(client.sendSync(Mockito.any(), Mockito.any(), Mockito.anyLong())).thenReturn(resCommand);
- alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content);
+ alertSendResponseCommand = alertClient.sendAlert(host, port, groupId, title, content, WarningType.FAILURE.getCode());
Assert.assertTrue(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java
index dfe411b..ff3c421 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java
@@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.Event;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.TaskType;
+import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.utils.CommonUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
@@ -189,12 +190,13 @@ public class TaskExecuteThread implements Runnable, Delayed {
// task handle
this.task.handle();
+ responseCommand.setStatus(this.task.getExitStatus().getCode());
+
// task result process
if (this.task.getNeedAlert()) {
- sendAlert(this.task.getTaskAlertInfo());
+ sendAlert(this.task.getTaskAlertInfo(), responseCommand.getStatus());
}
- responseCommand.setStatus(this.task.getExitStatus().getCode());
responseCommand.setEndTime(new Date());
responseCommand.setProcessId(this.task.getProcessId());
responseCommand.setAppIds(this.task.getAppIds());
@@ -215,8 +217,9 @@ public class TaskExecuteThread implements Runnable, Delayed {
}
}
- private void sendAlert(TaskAlertInfo taskAlertInfo) {
- alertClientService.sendAlert(taskAlertInfo.getAlertGroupId(), taskAlertInfo.getTitle(), taskAlertInfo.getContent());
+ private void sendAlert(TaskAlertInfo taskAlertInfo, int status) {
+ int strategy = status == ExecutionStatus.SUCCESS.getCode() ? WarningType.SUCCESS.getCode() : WarningType.FAILURE.getCode();
+ alertClientService.sendAlert(taskAlertInfo.getAlertGroupId(), taskAlertInfo.getTitle(), taskAlertInfo.getContent(), strategy);
}
/**