natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Feature-14832][Listener]Implementation of Listener Mechanism (#14981) 

* first commit

* 1. sql: sync ddl
2. front-end: change to ternary expression
3. back-end: correct license header in ListenerEvent.java

* test case

* frontend: remove unnecessary console

* fix unit test

* remove log depends on user-provided value

* fix dolphinscheduler_postgresql.sql

* sync database schema

* fix unit test

* fix unit test

* fix some NIT.

* extract GLOBAL_ALERT_GROUP_ID into variable

* fix ddl bug

* add column task_type in t_ds_fav_task in upgrade/3.2.0_schema

* add unit test
This commit is contained in:
Wei Xiaonan 2023-10-29 20:28:05 +08:00 committed by GitHub
parent 99214d0fc3
commit c0ed68121d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
75 changed files with 2929 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/AlertServer.java
View File

@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
import org.apache.dolphinscheduler.alert.registry.AlertRegistryClient;
import org.apache.dolphinscheduler.alert.rpc.AlertRpcServer;
import org.apache.dolphinscheduler.alert.service.AlertBootstrapService;
import org.apache.dolphinscheduler.alert.service.ListenerEventPostService;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.lifecycle.ServerLifeCycleManager;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
@ -44,6 +45,8 @@ public class AlertServer {
@Autowired
private AlertBootstrapService alertBootstrapService;
@Autowired
private ListenerEventPostService listenerEventPostService;
@Autowired
private AlertRpcServer alertRpcServer;
@Autowired
private AlertPluginManager alertPluginManager;
@ -61,6 +64,7 @@ public class AlertServer {
alertPluginManager.start();
alertRegistryClient.start();
alertBootstrapService.start();
listenerEventPostService.start();
alertRpcServer.start();
log.info("Alert server is started ...");
}
@ -88,6 +92,7 @@ public class AlertServer {
try (
AlertRpcServer closedAlertRpcServer = alertRpcServer;
AlertBootstrapService closedAlertBootstrapService = alertBootstrapService;
ListenerEventPostService closedListenerEventPostService = listenerEventPostService;
AlertRegistryClient closedAlertRegistryClient = alertRegistryClient) {
// close resource
}

19
dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/plugin/AlertPluginManager.java
View File

@ -19,16 +19,11 @@ package org.apache.dolphinscheduler.alert.plugin;
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 org.apache.dolphinscheduler.spi.plugin.PrioritySPIFactory;
import java.util.ArrayList;
@ -76,8 +71,6 @@ public final class AlertPluginManager {
}
private void installAlertPlugin() {
final PluginParams warningTypeParams = getWarningTypeParams();
PrioritySPIFactory<AlertChannelFactory> prioritySPIFactory =
new PrioritySPIFactory<>(AlertChannelFactory.class);
for (Map.Entry<String, AlertChannelFactory> entry : prioritySPIFactory.getSPIMap().entrySet()) {
@ -91,7 +84,6 @@ public final class AlertPluginManager {
log.info("Registered alert plugin: {} - {}", name, factory.getClass());
final List<PluginParams> params = new ArrayList<>(factory.params());
params.add(0, warningTypeParams);
final String paramsJson = PluginParamsTransfer.transferParamsToJson(params);
@ -102,15 +94,4 @@ public final class AlertPluginManager {
}
}
private 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();
}
}

263
dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/ListenerEventPostService.java Normal file
View File

@ -0,0 +1,263 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.alert.service;
import org.apache.dolphinscheduler.alert.api.AlertChannel;
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.alert.config.AlertConfig;
import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
import org.apache.dolphinscheduler.common.constants.Constants;
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.lifecycle.ServerLifeCycleManager;
import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.AlertSendStatus;
import org.apache.dolphinscheduler.dao.entity.ListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.AbstractListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionCreatedListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionDeletedListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionUpdatedListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessEndListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessFailListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessStartListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ServerDownListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.TaskEndListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.TaskFailListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.TaskStartListenerEvent;
import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ListenerEventMapper;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.curator.shaded.com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
@Service
@Slf4j
public final class ListenerEventPostService extends BaseDaemonThread implements AutoCloseable {
@Value("${alert.query_alert_threshold:100}")
private Integer QUERY_ALERT_THRESHOLD;
@Autowired
private ListenerEventMapper listenerEventMapper;
@Autowired
private AlertPluginInstanceMapper alertPluginInstanceMapper;
@Autowired
private AlertPluginManager alertPluginManager;
@Autowired
private AlertConfig alertConfig;
public ListenerEventPostService() {
super("ListenerEventPostService");
}
@Override
public void run() {
log.info("listener event post thread started");
while (!ServerLifeCycleManager.isStopped()) {
try {
List<ListenerEvent> listenerEvents = listenerEventMapper
.listingListenerEventByStatus(AlertStatus.WAIT_EXECUTION, QUERY_ALERT_THRESHOLD);
if (CollectionUtils.isEmpty(listenerEvents)) {
log.debug("There is no waiting listener events");
continue;
}
this.send(listenerEvents);
} catch (Exception e) {
log.error("listener event post thread meet an exception", e);
} finally {
ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS * 5L);
}
}
log.info("listener event post thread stopped");
}
public void send(List<ListenerEvent> listenerEvents) {
for (ListenerEvent listenerEvent : listenerEvents) {
int eventId = listenerEvent.getId();
List<AlertPluginInstance> globalAlertInstanceList =
alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList();
if (CollectionUtils.isEmpty(globalAlertInstanceList)) {
log.error("post listener event fail,no bind global plugin instance.");
listenerEventMapper.updateListenerEvent(eventId, AlertStatus.EXECUTION_FAILURE,
"no bind plugin instance", new Date());
continue;
}
AbstractListenerEvent event = generateEventFromContent(listenerEvent);
if (event == null) {
log.error("parse listener event to abstract listener event fail.ed {}", listenerEvent.getContent());
listenerEventMapper.updateListenerEvent(eventId, AlertStatus.EXECUTION_FAILURE,
"parse listener event to abstract listener event failed", new Date());
continue;
}
List<AbstractListenerEvent> events = Lists.newArrayList(event);
AlertData alertData = AlertData.builder()
.id(eventId)
.content(JSONUtils.toJsonString(events))
.log(listenerEvent.getLog())
.title(event.getTitle())
.warnType(WarningType.GLOBAL.getCode())
.alertType(event.getEventType().getCode())
.build();
int sendSuccessCount = 0;
List<AlertSendStatus> failedPostResults = new ArrayList<>();
for (AlertPluginInstance instance : globalAlertInstanceList) {
AlertResult alertResult = this.alertResultHandler(instance, alertData);
if (alertResult != null) {
AlertStatus sendStatus = Boolean.parseBoolean(alertResult.getStatus())
? AlertStatus.EXECUTION_SUCCESS
: AlertStatus.EXECUTION_FAILURE;
if (AlertStatus.EXECUTION_SUCCESS.equals(sendStatus)) {
sendSuccessCount++;
} else {
AlertSendStatus alertSendStatus = AlertSendStatus.builder()
.alertId(eventId)
.alertPluginInstanceId(instance.getId())
.sendStatus(sendStatus)
.log(JSONUtils.toJsonString(alertResult))
.createTime(new Date())
.build();
failedPostResults.add(alertSendStatus);
}
}
}
if (sendSuccessCount == globalAlertInstanceList.size()) {
listenerEventMapper.deleteById(eventId);
} else {
AlertStatus alertStatus =
sendSuccessCount == 0 ? AlertStatus.EXECUTION_FAILURE : AlertStatus.EXECUTION_PARTIAL_SUCCESS;
listenerEventMapper.updateListenerEvent(eventId, alertStatus, JSONUtils.toJsonString(failedPostResults),
new Date());
}
}
}
/**
* alert result handler
*
* @param instance instance
* @param alertData alertData
* @return AlertResult
*/
private @Nullable AlertResult alertResultHandler(AlertPluginInstance instance, AlertData alertData) {
String pluginInstanceName = instance.getInstanceName();
int pluginDefineId = instance.getPluginDefineId();
Optional<AlertChannel> alertChannelOptional = alertPluginManager.getAlertChannel(instance.getPluginDefineId());
if (!alertChannelOptional.isPresent()) {
String message =
String.format("Global Alert Plugin %s send error: the channel doesn't exist, pluginDefineId: %s",
pluginInstanceName,
pluginDefineId);
log.error("Global Alert Plugin {} send error : not found plugin {}", pluginInstanceName, pluginDefineId);
return new AlertResult("false", message);
}
AlertChannel alertChannel = alertChannelOptional.get();
Map<String, String> paramsMap = JSONUtils.toMap(instance.getPluginInstanceParams());
AlertInfo alertInfo = AlertInfo.builder()
.alertData(alertData)
.alertParams(paramsMap)
.alertPluginInstanceId(instance.getId())
.build();
int waitTimeout = alertConfig.getWaitTimeout();
try {
AlertResult alertResult;
if (waitTimeout <= 0) {
if (alertData.getAlertType() == AlertType.CLOSE_ALERT.getCode()) {
alertResult = alertChannel.closeAlert(alertInfo);
} else {
alertResult = alertChannel.process(alertInfo);
}
} else {
CompletableFuture<AlertResult> future;
if (alertData.getAlertType() == AlertType.CLOSE_ALERT.getCode()) {
future = CompletableFuture.supplyAsync(() -> alertChannel.closeAlert(alertInfo));
} else {
future = CompletableFuture.supplyAsync(() -> alertChannel.process(alertInfo));
}
alertResult = future.get(waitTimeout, TimeUnit.MILLISECONDS);
}
if (alertResult == null) {
throw new RuntimeException("Alert result cannot be null");
}
return alertResult;
} catch (InterruptedException e) {
log.error("post listener event error alert data id :{},", alertData.getId(), e);
Thread.currentThread().interrupt();
return new AlertResult("false", e.getMessage());
} catch (Exception e) {
log.error("post listener event error alert data id :{},", alertData.getId(), e);
return new AlertResult("false", e.getMessage());
}
}
private AbstractListenerEvent generateEventFromContent(ListenerEvent listenerEvent) {
String content = listenerEvent.getContent();
switch (listenerEvent.getEventType()) {
case SERVER_DOWN:
return JSONUtils.parseObject(content, ServerDownListenerEvent.class);
case PROCESS_DEFINITION_CREATED:
return JSONUtils.parseObject(content, ProcessDefinitionCreatedListenerEvent.class);
case PROCESS_DEFINITION_UPDATED:
return JSONUtils.parseObject(content, ProcessDefinitionUpdatedListenerEvent.class);
case PROCESS_DEFINITION_DELETED:
return JSONUtils.parseObject(content, ProcessDefinitionDeletedListenerEvent.class);
case PROCESS_START:
return JSONUtils.parseObject(content, ProcessStartListenerEvent.class);
case PROCESS_END:
return JSONUtils.parseObject(content, ProcessEndListenerEvent.class);
case PROCESS_FAIL:
return JSONUtils.parseObject(content, ProcessFailListenerEvent.class);
case TASK_START:
return JSONUtils.parseObject(content, TaskStartListenerEvent.class);
case TASK_END:
return JSONUtils.parseObject(content, TaskEndListenerEvent.class);
case TASK_FAIL:
return JSONUtils.parseObject(content, TaskFailListenerEvent.class);
default:
return null;
}
}
@Override
public void close() {
log.info("Closed ListenerEventPostService...");
}
}

1
dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/resources/application.yaml
View File

@ -81,6 +81,7 @@ alert:
# Define value is (0 = infinite), and alert server would be waiting alert result.
wait-timeout: 0
heartbeat-interval: 60s
query_alert_threshold: 100
registry:
type: zookeeper

154
dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/ListenerEventPostServiceTest.java Normal file
View File

@ -0,0 +1,154 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.alert.runner;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import org.apache.dolphinscheduler.alert.api.AlertChannel;
import org.apache.dolphinscheduler.alert.api.AlertResult;
import org.apache.dolphinscheduler.alert.config.AlertConfig;
import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
import org.apache.dolphinscheduler.alert.service.ListenerEventPostService;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.ListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ServerDownListenerEvent;
import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ListenerEventMapper;
import org.apache.commons.codec.digest.DigestUtils;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ListenerEventPostServiceTest {
private static final Logger logger = LoggerFactory.getLogger(ListenerEventPostServiceTest.class);
@Mock
private ListenerEventMapper listenerEventMapper;
@Mock
private AlertPluginInstanceMapper alertPluginInstanceMapper;
@Mock
private AlertPluginManager alertPluginManager;
@Mock
private AlertConfig alertConfig;
@InjectMocks
private ListenerEventPostService listenerEventPostService;
@BeforeEach
public void before() {
MockitoAnnotations.initMocks(this);
}
@Test
public void testSendServerDownEventSuccess() {
List<ListenerEvent> events = new ArrayList<>();
ServerDownListenerEvent serverDownListenerEvent = new ServerDownListenerEvent();
serverDownListenerEvent.setEventTime(new Date());
serverDownListenerEvent.setType("WORKER");
serverDownListenerEvent.setHost("192.168.*.*");
ListenerEvent successEvent = new ListenerEvent();
successEvent.setId(1);
successEvent.setPostStatus(AlertStatus.WAIT_EXECUTION);
successEvent.setContent(JSONUtils.toJsonString(serverDownListenerEvent));
successEvent.setSign(DigestUtils.sha256Hex(successEvent.getContent()));
successEvent.setEventType(ListenerEventType.SERVER_DOWN);
successEvent.setCreateTime(new Date());
successEvent.setUpdateTime(new Date());
events.add(successEvent);
int pluginDefineId = 1;
String pluginInstanceParams =
"{\"User\":\"xx\",\"receivers\":\"xx\",\"sender\":\"xx\",\"smtpSslTrust\":\"*\",\"enableSmtpAuth\":\"true\",\"receiverCcs\":null,\"showType\":\"table\",\"starttlsEnable\":\"false\",\"serverPort\":\"25\",\"serverHost\":\"xx\",\"Password\":\"xx\",\"sslEnable\":\"false\"}";
String pluginInstanceName = "alert-instance-mail";
List<AlertPluginInstance> alertInstanceList = new ArrayList<>();
AlertPluginInstance alertPluginInstance = new AlertPluginInstance(
pluginDefineId, pluginInstanceParams, pluginInstanceName);
alertPluginInstance.setInstanceType(AlertPluginInstanceType.GLOBAL);
alertPluginInstance.setId(1);
alertInstanceList.add(alertPluginInstance);
when(alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList()).thenReturn(alertInstanceList);
AlertResult sendResult = new AlertResult();
sendResult.setStatus(String.valueOf(true));
sendResult.setMessage(String.format("Alert Plugin %s send success", pluginInstanceName));
AlertChannel alertChannelMock = mock(AlertChannel.class);
when(alertChannelMock.process(Mockito.any())).thenReturn(sendResult);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
Assertions.assertTrue(Boolean.parseBoolean(sendResult.getStatus()));
when(listenerEventMapper.deleteById(1)).thenReturn(1);
listenerEventPostService.send(events);
}
@Test
public void testSendServerDownEventFailed() {
List<ListenerEvent> events = new ArrayList<>();
ServerDownListenerEvent serverDownListenerEvent = new ServerDownListenerEvent();
serverDownListenerEvent.setEventTime(new Date());
serverDownListenerEvent.setType("WORKER");
serverDownListenerEvent.setHost("192.168.*.*");
ListenerEvent successEvent = new ListenerEvent();
successEvent.setId(1);
successEvent.setPostStatus(AlertStatus.WAIT_EXECUTION);
successEvent.setContent(JSONUtils.toJsonString(serverDownListenerEvent));
successEvent.setSign(DigestUtils.sha1Hex(successEvent.getContent()));
successEvent.setEventType(ListenerEventType.SERVER_DOWN);
successEvent.setCreateTime(new Date());
successEvent.setUpdateTime(new Date());
events.add(successEvent);
int pluginDefineId = 1;
String pluginInstanceParams =
"{\"User\":\"xx\",\"receivers\":\"xx\",\"sender\":\"xx\",\"smtpSslTrust\":\"*\",\"enableSmtpAuth\":\"true\",\"receiverCcs\":null,\"showType\":\"table\",\"starttlsEnable\":\"false\",\"serverPort\":\"25\",\"serverHost\":\"xx\",\"Password\":\"xx\",\"sslEnable\":\"false\"}";
String pluginInstanceName = "alert-instance-mail";
List<AlertPluginInstance> alertInstanceList = new ArrayList<>();
AlertPluginInstance alertPluginInstance = new AlertPluginInstance(
pluginDefineId, pluginInstanceParams, pluginInstanceName);
alertPluginInstance.setInstanceType(AlertPluginInstanceType.GLOBAL);
alertPluginInstance.setId(1);
alertInstanceList.add(alertPluginInstance);
when(alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList()).thenReturn(alertInstanceList);
AlertResult sendResult = new AlertResult();
sendResult.setStatus(String.valueOf(false));
sendResult.setMessage(String.format("Alert Plugin %s send failed", pluginInstanceName));
AlertChannel alertChannelMock = mock(AlertChannel.class);
when(alertChannelMock.process(Mockito.any())).thenReturn(sendResult);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
Assertions.assertFalse(Boolean.parseBoolean(sendResult.getStatus()));
listenerEventPostService.send(events);
}
}

16
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/AlertGroupController.java
View File

@ -109,6 +109,22 @@ public class AlertGroupController extends BaseController {
return returnDataList(result);
}
/**
* normal alert group list
*
* @param loginUser login user
* @return normal alert group list
*/
@Operation(summary = "listNormalAlertgroupById", description = "QUERY_ALERT_GROUP_LIST_NOTES")
@GetMapping(value = "/normal-list")
@ResponseStatus(HttpStatus.OK)
@ApiException(QUERY_ALL_ALERTGROUP_ERROR)
public Result normalAlertGroupList(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser) {
Map<String, Object> result = alertGroupService.queryNormalAlertgroup(loginUser);
return returnDataList(result);
}
/**
* paging query alarm group list
*

10
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/AlertPluginInstanceController.java
View File

@ -29,6 +29,8 @@ import org.apache.dolphinscheduler.api.exceptions.ApiException;
import org.apache.dolphinscheduler.api.service.AlertPluginInstanceService;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
@ -88,9 +90,12 @@ public class AlertPluginInstanceController extends BaseController {
public Result createAlertPluginInstance(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@RequestParam(value = "pluginDefineId") int pluginDefineId,
@RequestParam(value = "instanceName") String instanceName,
@RequestParam(value = "instanceType") AlertPluginInstanceType instanceType,
@RequestParam(value = "warningType") WarningType warningType,
@RequestParam(value = "pluginInstanceParams") String pluginInstanceParams) {
Map<String, Object> result =
alertPluginInstanceService.create(loginUser, pluginDefineId, instanceName, pluginInstanceParams);
alertPluginInstanceService.create(loginUser, pluginDefineId, instanceName, instanceType, warningType,
pluginInstanceParams);
return returnDataList(result);
}
@ -115,9 +120,10 @@ public class AlertPluginInstanceController extends BaseController {
public Result updateAlertPluginInstance(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@PathVariable(value = "id") int id,
@RequestParam(value = "instanceName") String instanceName,
@RequestParam(value = "warningType") WarningType warningType,
@RequestParam(value = "pluginInstanceParams") String pluginInstanceParams) {
Map<String, Object> result =
alertPluginInstanceService.update(loginUser, id, instanceName, pluginInstanceParams);
alertPluginInstanceService.update(loginUser, id, instanceName, warningType, pluginInstanceParams);
return returnDataList(result);
}

1
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
View File

@ -539,6 +539,7 @@ public enum Status {
NOT_ALLOW_TO_DISABLE_OWN_ACCOUNT(130020, "Not allow to disable your own account", "不能停用自己的账号"),
NOT_ALLOW_TO_DELETE_DEFAULT_ALARM_GROUP(130030, "Not allow to delete the default alarm group ", "不能删除默认告警组"),
TIME_ZONE_ILLEGAL(130031, "time zone [{0}] is illegal", "时区参数 [{0}] 不合法"),
NOT_ALLOW_TO_UPDATE_GLOBAL_ALARM_GROUP(130032, "Not allow to update the global alert group ", "不能更新全局告警组"),
QUERY_K8S_NAMESPACE_LIST_PAGING_ERROR(1300001, "login user query k8s namespace list paging error",
"分页查询k8s名称空间列表错误"),

8
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertGroupService.java
View File

@ -35,6 +35,14 @@ public interface AlertGroupService {
*/
Map<String, Object> queryAlertgroup(User loginUser);
/**
* query normal alert group list
*
* @param loginUser
* @return alert group list
*/
Map<String, Object> queryNormalAlertgroup(User loginUser);
/**
* query alert group by id
*

8
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/AlertPluginInstanceService.java
View File

@ -18,6 +18,8 @@
package org.apache.dolphinscheduler.api.service;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.entity.User;
import java.util.Map;
@ -36,7 +38,9 @@ public interface AlertPluginInstanceService {
* @param pluginInstanceParams plugin instance params
* @return result
*/
Map<String, Object> create(User loginUser, int pluginDefineId, String instanceName, String pluginInstanceParams);
Map<String, Object> create(User loginUser, int pluginDefineId, String instanceName,
AlertPluginInstanceType instanceType, WarningType warningType,
String pluginInstanceParams);
/**
* update
@ -47,7 +51,7 @@ public interface AlertPluginInstanceService {
* @return result
*/
Map<String, Object> update(User loginUser, int alertPluginInstanceId, String instanceName,
String pluginInstanceParams);
WarningType warningType, String pluginInstanceParams);
/**
* delete alert plugin instance

30
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertGroupServiceImpl.java
View File

@ -42,6 +42,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
@ -90,6 +91,28 @@ public class AlertGroupServiceImpl extends BaseServiceImpl implements AlertGroup
return result;
}
@Override
public Map<String, Object> queryNormalAlertgroup(User loginUser) {
HashMap<String, Object> result = new HashMap<>();
List<AlertGroup> alertGroups;
if (loginUser.getUserType().equals(UserType.ADMIN_USER)) {
alertGroups = alertGroupMapper.queryAllGroupList();
} else {
Set<Integer> ids = resourcePermissionCheckService
.userOwnedResourceIdsAcquisition(AuthorizationType.ALERT_GROUP, loginUser.getId(), log);
if (ids.isEmpty()) {
result.put(Constants.DATA_LIST, Collections.emptyList());
putMsg(result, Status.SUCCESS);
return result;
}
alertGroups = alertGroupMapper.selectBatchIds(ids);
}
alertGroups = alertGroups.stream().filter(alertGroup -> alertGroup.getId() != 2).collect(Collectors.toList());
result.put(Constants.DATA_LIST, alertGroups);
putMsg(result, Status.SUCCESS);
return result;
}
/**
* query alert group by id
*
@ -223,6 +246,11 @@ public class AlertGroupServiceImpl extends BaseServiceImpl implements AlertGroup
public Map<String, Object> updateAlertgroup(User loginUser, int id, String groupName, String desc,
String alertInstanceIds) {
Map<String, Object> result = new HashMap<>();
// don't allow to update global alert group
if (id == 2) {
putMsg(result, Status.NOT_ALLOW_TO_UPDATE_GLOBAL_ALARM_GROUP);
return result;
}
if (!canOperatorPermissions(loginUser, new Object[]{id}, AuthorizationType.ALERT_GROUP, ALERT_GROUP_UPDATE)) {
putMsg(result, Status.USER_NO_OPERATION_PERM);
@ -282,7 +310,7 @@ public class AlertGroupServiceImpl extends BaseServiceImpl implements AlertGroup
}
// Not allow to delete the default alarm group ,because the module of service need to use it.
if (id == 1) {
if (id == 1 || id == 2) {
log.warn("Not allow to delete the default alarm group.");
putMsg(result, Status.NOT_ALLOW_TO_DELETE_DEFAULT_ALARM_GROUP);
return result;

58
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java
View File

@ -28,8 +28,11 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.api.vo.AlertPluginInstanceVO;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.dao.entity.User;
@ -39,6 +42,7 @@ import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.Arrays;
@ -76,6 +80,8 @@ public class AlertPluginInstanceServiceImpl extends BaseServiceImpl implements A
@Autowired
private AlertGroupMapper alertGroupMapper;
private final Integer GLOBAL_ALERT_GROUP_ID = 2;
/**
* creat alert plugin instance
*
@ -86,12 +92,15 @@ public class AlertPluginInstanceServiceImpl extends BaseServiceImpl implements A
*/
@Override
public Map<String, Object> create(User loginUser, int pluginDefineId, String instanceName,
AlertPluginInstanceType instanceType, WarningType warningType,
String pluginInstanceParams) {
AlertPluginInstance alertPluginInstance = new AlertPluginInstance();
String paramsMapJson = parsePluginParamsMap(pluginInstanceParams);
alertPluginInstance.setPluginInstanceParams(paramsMapJson);
alertPluginInstance.setInstanceName(instanceName);
alertPluginInstance.setPluginDefineId(pluginDefineId);
alertPluginInstance.setInstanceType(instanceType);
alertPluginInstance.setWarningType(warningType);
Map<String, Object> result = new HashMap<>();
if (!canOperatorPermissions(loginUser, null, AuthorizationType.ALERT_PLUGIN_INSTANCE, ALART_INSTANCE_CREATE)) {
@ -108,6 +117,20 @@ public class AlertPluginInstanceServiceImpl extends BaseServiceImpl implements A
int i = alertPluginInstanceMapper.insert(alertPluginInstance);
if (i > 0) {
log.info("Create alert plugin instance complete, name:{}", alertPluginInstance.getInstanceName());
// global instance will be added into global alert group automatically
if (instanceType == AlertPluginInstanceType.GLOBAL) {
AlertGroup globalAlertGroup = alertGroupMapper.selectById(GLOBAL_ALERT_GROUP_ID);
if (StringUtils.isEmpty(globalAlertGroup.getAlertInstanceIds())) {
globalAlertGroup.setAlertInstanceIds(String.valueOf(alertPluginInstance.getId()));
} else {
List<Integer> ids = Arrays.stream(globalAlertGroup.getAlertInstanceIds().split(","))
.map(s -> Integer.parseInt(s.trim()))
.collect(Collectors.toList());
ids.add(alertPluginInstance.getId());
globalAlertGroup.setAlertInstanceIds(StringUtils.join(ids, ","));
}
alertGroupMapper.updateById(globalAlertGroup);
}
result.put(Constants.DATA_LIST, alertPluginInstance);
putMsg(result, Status.SUCCESS);
return result;
@ -127,11 +150,11 @@ public class AlertPluginInstanceServiceImpl extends BaseServiceImpl implements A
*/
@Override
public Map<String, Object> update(User loginUser, int pluginInstanceId, String instanceName,
String pluginInstanceParams) {
WarningType warningType, String pluginInstanceParams) {
String paramsMapJson = parsePluginParamsMap(pluginInstanceParams);
AlertPluginInstance alertPluginInstance =
new AlertPluginInstance(pluginInstanceId, paramsMapJson, instanceName, new Date());
new AlertPluginInstance(pluginInstanceId, paramsMapJson, instanceName, warningType, new Date());
Map<String, Object> result = new HashMap<>();
@ -163,12 +186,26 @@ public class AlertPluginInstanceServiceImpl extends BaseServiceImpl implements A
@Override
public Map<String, Object> delete(User loginUser, int id) {
Map<String, Object> result = new HashMap<>();
// check if there is an associated alert group
boolean hasAssociatedAlertGroup = checkHasAssociatedAlertGroup(String.valueOf(id));
if (hasAssociatedAlertGroup) {
log.warn("Delete alert plugin failed because alert group is using it, pluginId:{}.", id);
putMsg(result, Status.DELETE_ALERT_PLUGIN_INSTANCE_ERROR_HAS_ALERT_GROUP_ASSOCIATED);
return result;
AlertPluginInstance alertPluginInstance = alertPluginInstanceMapper.selectById(id);
if (alertPluginInstance.getInstanceType() == AlertPluginInstanceType.GLOBAL) {
// global instance will be removed from global alert group automatically
AlertGroup globalAlertGroup = alertGroupMapper.selectById(GLOBAL_ALERT_GROUP_ID);
List<Integer> ids = Arrays.stream(globalAlertGroup.getAlertInstanceIds().split(","))
.map(s -> Integer.parseInt(s.trim()))
.collect(Collectors.toList());
ids = ids.stream().filter(x -> x != id).collect(Collectors.toList());
globalAlertGroup.setAlertInstanceIds(StringUtils.join(ids, ","));
alertGroupMapper.updateById(globalAlertGroup);
log.info("Remove global alert plugin instance from global alert group automatically, name:{}",
alertPluginInstance.getInstanceName());
} else {
// check if there is an associated alert group
boolean hasAssociatedAlertGroup = checkHasAssociatedAlertGroup(String.valueOf(id));
if (hasAssociatedAlertGroup) {
log.warn("Delete alert plugin failed because alert group is using it, pluginId:{}.", id);
putMsg(result, Status.DELETE_ALERT_PLUGIN_INSTANCE_ERROR_HAS_ALERT_GROUP_ASSOCIATED);
return result;
}
}
if (!canOperatorPermissions(loginUser, null, AuthorizationType.ALERT_PLUGIN_INSTANCE, ALERT_PLUGIN_DELETE)) {
putMsg(result, Status.USER_NO_OPERATION_PERM);
@ -255,12 +292,15 @@ public class AlertPluginInstanceServiceImpl extends BaseServiceImpl implements A
pluginDefineList.stream().collect(Collectors.toMap(PluginDefine::getId, Function.identity()));
alertPluginInstances.forEach(alertPluginInstance -> {
AlertPluginInstanceVO alertPluginInstanceVO = new AlertPluginInstanceVO();
alertPluginInstanceVO.setCreateTime(alertPluginInstance.getCreateTime());
alertPluginInstanceVO.setUpdateTime(alertPluginInstance.getUpdateTime());
alertPluginInstanceVO.setPluginDefineId(alertPluginInstance.getPluginDefineId());
alertPluginInstanceVO.setInstanceName(alertPluginInstance.getInstanceName());
alertPluginInstanceVO.setId(alertPluginInstance.getId());
alertPluginInstanceVO.setInstanceType(alertPluginInstance.getInstanceType().getDescp());
if (alertPluginInstance.getWarningType() != null) {
alertPluginInstanceVO.setWarningType(alertPluginInstance.getWarningType().getDescp().toUpperCase());
}
PluginDefine pluginDefine = pluginDefineMap.get(alertPluginInstance.getPluginDefineId());
// FIXME When the user removes the plug-in, this will happen. At this time, maybe we should add a new field
// to indicate that the plug-in has expired?

21
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
View File

@ -118,6 +118,7 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.plugin.task.api.parameters.ParametersNode;
import org.apache.dolphinscheduler.plugin.task.api.parameters.SqlParameters;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
import org.apache.dolphinscheduler.service.cron.CronUtils;
import org.apache.dolphinscheduler.service.model.TaskNode;
import org.apache.dolphinscheduler.service.process.ProcessService;
@ -251,6 +252,9 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
@Autowired
private MetricsCleanUpService metricsCleanUpService;
@Autowired
private ListenerEventAlertManager listenerEventAlertManager;
/**
* create process definition
*
@ -306,7 +310,13 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
globalParams, locations, timeout, loginUser.getId());
processDefinition.setExecutionType(executionType);
return createDagDefine(loginUser, taskRelationList, processDefinition, taskDefinitionLogs, otherParamsJson);
result = createDagDefine(loginUser, taskRelationList, processDefinition, taskDefinitionLogs, otherParamsJson);
if (result.get(Constants.STATUS) == Status.SUCCESS) {
listenerEventAlertManager.publishProcessDefinitionCreatedListenerEvent(loginUser, processDefinition,
taskDefinitionLogs,
taskRelationList);
}
return result;
}
private void createWorkflowValid(User user, ProcessDefinition processDefinition) {
@ -805,8 +815,14 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
JSONUtils.parseObject(JSONUtils.toJsonString(processDefinition), ProcessDefinition.class);
processDefinition.set(projectCode, name, description, globalParams, locations, timeout);
processDefinition.setExecutionType(executionType);
return updateDagDefine(loginUser, taskRelationList, processDefinition, processDefinitionDeepCopy,
result = updateDagDefine(loginUser, taskRelationList, processDefinition, processDefinitionDeepCopy,
taskDefinitionLogs, otherParamsJson);
if (result.get(Constants.STATUS) == Status.SUCCESS) {
listenerEventAlertManager.publishProcessDefinitionUpdatedListenerEvent(loginUser, processDefinition,
taskDefinitionLogs,
taskRelationList);
}
return result;
}
/**
@ -1071,6 +1087,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
processDefinitionDao.deleteByWorkflowDefinitionCode(processDefinition.getCode());
metricsCleanUpService.cleanUpWorkflowMetricsByDefinitionCode(code);
log.info("Success delete workflow definition workflowDefinitionCode: {}", code);
listenerEventAlertManager.publishProcessDefinitionDeletedListenerEvent(loginUser, project, processDefinition);
}
/**

10
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/vo/AlertPluginInstanceVO.java
View File

@ -39,6 +39,16 @@ public class AlertPluginInstanceVO {
*/
private String instanceName;
/**
* alert plugin instance type
*/
private String instanceType;
/**
* alert plugin instance warning type
*/
private String warningType;
/**
* plugin_instance_params
*/

46
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/AlertGroupControllerTest.java
View File

@ -217,4 +217,50 @@ public class AlertGroupControllerTest extends AbstractControllerTest {
Assertions.assertEquals(Status.NOT_ALLOW_TO_DELETE_DEFAULT_ALARM_GROUP.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
@Test
public void test100DelAlertGroupById() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
MvcResult mvcResult = mockMvc.perform(delete("/alert-groups/2")
.header("sessionId", sessionId)
.params(paramsMap))
.andExpect(status().isOk())
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assertions.assertEquals(Status.NOT_ALLOW_TO_DELETE_DEFAULT_ALARM_GROUP.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
@Test
public void test110UpdateAlertGroupById() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("groupName", defaultTestAlertGroupName);
paramsMap.add("groupType", "email");
paramsMap.add("description", "update alter group");
paramsMap.add("alertInstanceIds", "");
MvcResult mvcResult = mockMvc.perform(put("/alert-groups/2")
.header("sessionId", sessionId)
.params(paramsMap))
.andExpect(status().isOk())
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assertions.assertEquals(Status.NOT_ALLOW_TO_UPDATE_GLOBAL_ALARM_GROUP.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
@Test
public void test120QueryNormalAlertGroupList() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
MvcResult mvcResult = mockMvc.perform(get("/alert-groups/normal-list")
.header("sessionId", sessionId)
.params(paramsMap))
.andExpect(status().isOk())
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assertions.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
}

11
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/AlertPluginInstanceControllerTest.java
View File

@ -32,6 +32,8 @@ import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.AlertPluginInstanceService;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.User;
@ -52,6 +54,8 @@ public class AlertPluginInstanceControllerTest extends AbstractControllerTest {
private static final int pluginDefineId = 1;
private static final String instanceName = "instanceName";
private static final String pluginInstanceParams = "pluginInstanceParams";
private static final AlertPluginInstanceType pluginInstanceType = AlertPluginInstanceType.NORMAL;
private static final WarningType warningType = WarningType.ALL;
private static final Result expectResponseContent = JSONUtils.parseObject(
"{\"code\":0,\"msg\":\"success\",\"data\":\"Test Data\",\"success\":true,\"failed\":false}", Result.class);
private static final ImmutableMap<String, Object> alertPluginInstanceServiceResult =
@ -66,10 +70,12 @@ public class AlertPluginInstanceControllerTest extends AbstractControllerTest {
final MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("pluginDefineId", String.valueOf(pluginDefineId));
paramsMap.add("instanceName", instanceName);
paramsMap.add("instanceType", pluginInstanceType.getDescp());
paramsMap.add("warningType", warningType.name());
paramsMap.add("pluginInstanceParams", pluginInstanceParams);
when(alertPluginInstanceService.create(any(User.class), eq(pluginDefineId), eq(instanceName),
eq(pluginInstanceParams)))
eq(pluginInstanceType), eq(warningType), eq(pluginInstanceParams)))
.thenReturn(alertPluginInstanceServiceResult);
// When
@ -92,10 +98,11 @@ public class AlertPluginInstanceControllerTest extends AbstractControllerTest {
final MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("pluginDefineId", String.valueOf(pluginDefineId));
paramsMap.add("instanceName", instanceName);
paramsMap.add("warningType", warningType.name());
paramsMap.add("pluginInstanceParams", pluginInstanceParams);
when(alertPluginInstanceService.update(any(User.class), eq(pluginDefineId), eq(instanceName),
eq(pluginInstanceParams)))
eq(warningType), eq(pluginInstanceParams)))
.thenReturn(alertPluginInstanceServiceResult);
// When

76
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertGroupServiceTest.java
View File

@ -39,10 +39,8 @@ import org.apache.dolphinscheduler.dao.mapper.AlertGroupMapper;
import org.apache.commons.collections4.CollectionUtils;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@ -87,13 +85,23 @@ public class AlertGroupServiceTest {
Map<String, Object> result = alertGroupService.queryAlertgroup(getLoginUser());
logger.info(result.toString());
List<AlertGroup> alertGroups = (List<AlertGroup>) result.get(Constants.DATA_LIST);
Assertions.assertTrue(CollectionUtils.isNotEmpty(alertGroups));
Assertions.assertEquals(alertGroups.size(), 2);
}
@Test
public void testQueryNormalAlertGroup() {
Mockito.when(alertGroupMapper.queryAllGroupList()).thenReturn(getList());
Map<String, Object> result = alertGroupService.queryNormalAlertgroup(getLoginUser());
logger.info(result.toString());
List<AlertGroup> alertGroups = (List<AlertGroup>) result.get(Constants.DATA_LIST);
Assertions.assertEquals(alertGroups.size(), 1);
}
@Test
public void testListPaging() {
IPage<AlertGroup> page = new Page<>(1, 10);
page.setTotal(1L);
page.setTotal(2L);
page.setRecords(getList());
Mockito.when(alertGroupMapper.queryAlertGroupPage(any(Page.class), eq(groupName))).thenReturn(page);
User user = new User();
@ -101,8 +109,6 @@ public class AlertGroupServiceTest {
user.setUserType(UserType.GENERAL_USER);
user.setId(88);
Set<Integer> ids = new HashSet<>();
ids.add(1);
Result result = alertGroupService.listPaging(user, groupName, 1, 10);
logger.info(result.toString());
Assertions.assertEquals(Status.SUCCESS.getCode(), (int) result.getCode());
@ -119,7 +125,7 @@ public class AlertGroupServiceTest {
@Test
public void testCreateAlertgroup() {
Mockito.when(alertGroupMapper.insert(any(AlertGroup.class))).thenReturn(2);
Mockito.when(alertGroupMapper.insert(any(AlertGroup.class))).thenReturn(3);
User user = new User();
user.setId(0);
// no operate
@ -179,9 +185,9 @@ public class AlertGroupServiceTest {
Assertions.assertEquals(Status.ALERT_GROUP_NOT_EXIST, result.get(Constants.STATUS));
// success
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_GROUP,
new Object[]{2}, user.getId(), baseServiceLogger)).thenReturn(true);
Mockito.when(alertGroupMapper.selectById(2)).thenReturn(getEntity());
result = alertGroupService.updateAlertgroup(user, 2, groupName, groupName, null);
new Object[]{3}, user.getId(), baseServiceLogger)).thenReturn(true);
Mockito.when(alertGroupMapper.selectById(3)).thenReturn(getEntity());
result = alertGroupService.updateAlertgroup(user, 3, groupName, groupName, null);
logger.info(result.toString());
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
@ -195,17 +201,28 @@ public class AlertGroupServiceTest {
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.ALERT_GROUP,
user.getId(), ALERT_GROUP_UPDATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_GROUP,
new Object[]{2}, user.getId(), baseServiceLogger)).thenReturn(true);
Mockito.when(alertGroupMapper.selectById(2)).thenReturn(getEntity());
new Object[]{3}, user.getId(), baseServiceLogger)).thenReturn(true);
Mockito.when(alertGroupMapper.selectById(3)).thenReturn(getEntity());
Mockito.when(alertGroupMapper.updateById(Mockito.any()))
.thenThrow(new DuplicateKeyException("group name exist"));
Map<String, Object> result = alertGroupService.updateAlertgroup(user, 2, groupName, groupName, null);
Map<String, Object> result = alertGroupService.updateAlertgroup(user, 3, groupName, groupName, null);
Assertions.assertEquals(Status.ALERT_GROUP_EXIST, result.get(Constants.STATUS));
}
@Test
public void testDelAlertgroupById() {
public void testUpdateGlobalAlertgroup() {
User user = new User();
user.setId(0);
user.setUserType(UserType.ADMIN_USER);
AlertGroup globalAlertGroup = new AlertGroup();
globalAlertGroup.setId(2);
globalAlertGroup.setGroupName("global alert group");
Map<String, Object> result = alertGroupService.updateAlertgroup(user, 2, groupName, groupName, null);
Assertions.assertEquals(Status.NOT_ALLOW_TO_UPDATE_GLOBAL_ALARM_GROUP, result.get(Constants.STATUS));
}
@Test
public void testDelAlertgroupById() {
User user = new User();
user.setId(0);
// no operate
@ -222,18 +239,30 @@ public class AlertGroupServiceTest {
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.ALERT_GROUP,
user.getId(), ALERT_GROUP_DELETE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_GROUP,
new Object[]{2}, 0, baseServiceLogger)).thenReturn(true);
result = alertGroupService.delAlertgroupById(user, 2);
new Object[]{3}, 0, baseServiceLogger)).thenReturn(true);
result = alertGroupService.delAlertgroupById(user, 3);
logger.info(result.toString());
Assertions.assertEquals(Status.ALERT_GROUP_NOT_EXIST, result.get(Constants.STATUS));
// success
// not allowed1
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_GROUP,
new Object[]{1}, 0, baseServiceLogger)).thenReturn(true);
result = alertGroupService.delAlertgroupById(user, 1);
logger.info(result.toString());
Assertions.assertEquals(Status.NOT_ALLOW_TO_DELETE_DEFAULT_ALARM_GROUP, result.get(Constants.STATUS));
// not allowed2
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_GROUP,
new Object[]{2}, 0, baseServiceLogger)).thenReturn(true);
Mockito.when(alertGroupMapper.selectById(2)).thenReturn(getEntity());
result = alertGroupService.delAlertgroupById(user, 2);
logger.info(result.toString());
Assertions.assertEquals(Status.NOT_ALLOW_TO_DELETE_DEFAULT_ALARM_GROUP, result.get(Constants.STATUS));
// success
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_GROUP,
new Object[]{4}, 0, baseServiceLogger)).thenReturn(true);
Mockito.when(alertGroupMapper.selectById(4)).thenReturn(getEntity());
result = alertGroupService.delAlertgroupById(user, 4);
logger.info(result.toString());
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
@ -264,7 +293,14 @@ public class AlertGroupServiceTest {
*/
private List<AlertGroup> getList() {
List<AlertGroup> alertGroups = new ArrayList<>();
alertGroups.add(getEntity());
AlertGroup defaultAdminWarningGroup = new AlertGroup();
defaultAdminWarningGroup.setId(1);
defaultAdminWarningGroup.setGroupName("default admin warning group");
alertGroups.add(defaultAdminWarningGroup);
AlertGroup globalAlertGroup = new AlertGroup();
globalAlertGroup.setId(2);
globalAlertGroup.setGroupName("global alert group");
alertGroups.add(globalAlertGroup);
return alertGroups;
}

62
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertPluginInstanceServiceTest.java
View File

@ -26,8 +26,11 @@ import org.apache.dolphinscheduler.api.permission.ResourcePermissionCheckService
import org.apache.dolphinscheduler.api.service.impl.AlertPluginInstanceServiceImpl;
import org.apache.dolphinscheduler.api.service.impl.BaseServiceImpl;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.dao.entity.User;
@ -79,6 +82,12 @@ public class AlertPluginInstanceServiceTest {
private User user;
private final AlertPluginInstanceType normalInstanceType = AlertPluginInstanceType.NORMAL;
private final AlertPluginInstanceType globalInstanceType = AlertPluginInstanceType.GLOBAL;
private final WarningType warningType = WarningType.ALL;
private String uiParams = "[\n"
+ " {\n"
+ " \"field\":\"userParams\",\n"
@ -160,11 +169,7 @@ public class AlertPluginInstanceServiceTest {
user = new User();
user.setUserType(UserType.ADMIN_USER);
user.setId(1);
AlertPluginInstance alertPluginInstance = new AlertPluginInstance();
alertPluginInstance.setPluginInstanceParams("test1");
alertPluginInstance.setPluginDefineId(1);
alertPluginInstance.setId(1);
alertPluginInstance.setPluginInstanceParams("test");
AlertPluginInstance alertPluginInstance = getAlertPluginInstance(1, normalInstanceType, "test");
alertPluginInstances = new ArrayList<>();
alertPluginInstances.add(alertPluginInstance);
}
@ -176,29 +181,44 @@ public class AlertPluginInstanceServiceTest {
1, ALART_INSTANCE_CREATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_PLUGIN_INSTANCE,
null, 0, baseServiceLogger)).thenReturn(true);
Map<String, Object> result = alertPluginInstanceService.create(user, 1, "test", uiParams);
Map<String, Object> result =
alertPluginInstanceService.create(user, 1, "test", normalInstanceType, warningType, uiParams);
Assertions.assertEquals(Status.PLUGIN_INSTANCE_ALREADY_EXISTS, result.get(Constants.STATUS));
Mockito.when(alertPluginInstanceMapper.insert(Mockito.any())).thenReturn(1);
result = alertPluginInstanceService.create(user, 1, "test1", uiParams);
result = alertPluginInstanceService.create(user, 1, "test1", normalInstanceType, warningType, uiParams);
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
Assertions.assertNotNull(result.get(Constants.DATA_LIST));
}
@Test
public void testDelete() {
List<String> ids = Arrays.asList("11,2,3", null, "98,1");
List<String> ids = Arrays.asList("11,2,3", "5,96", null, "98,1");
Mockito.when(alertGroupMapper.queryInstanceIdsList()).thenReturn(ids);
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.ALERT_PLUGIN_INSTANCE,
1, ALERT_PLUGIN_DELETE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_PLUGIN_INSTANCE,
null, 0, baseServiceLogger)).thenReturn(true);
Map<String, Object> result = alertPluginInstanceService.delete(user, 1);
AlertPluginInstance normalInstanceWithId1 = getAlertPluginInstance(1, normalInstanceType, "test1");
AlertPluginInstance normalInstanceWithId9 = getAlertPluginInstance(9, normalInstanceType, "test9");
AlertPluginInstance globalInstanceWithId5 = getAlertPluginInstance(5, globalInstanceType, "test5");
Mockito.when(alertPluginInstanceMapper.selectById(1)).thenReturn(normalInstanceWithId1);
Mockito.when(alertPluginInstanceMapper.selectById(9)).thenReturn(normalInstanceWithId9);
Mockito.when(alertPluginInstanceMapper.selectById(5)).thenReturn(globalInstanceWithId5);
AlertGroup globalAlertGroup = new AlertGroup();
globalAlertGroup.setId(2);
globalAlertGroup.setAlertInstanceIds("5,96");
Mockito.when(alertGroupMapper.selectById(2)).thenReturn(globalAlertGroup);
Mockito.when(alertGroupMapper.updateById(Mockito.any())).thenReturn(1);
Map<String, Object> result;
result = alertPluginInstanceService.delete(user, 1);
Assertions.assertEquals(Status.DELETE_ALERT_PLUGIN_INSTANCE_ERROR_HAS_ALERT_GROUP_ASSOCIATED,
result.get(Constants.STATUS));
Mockito.when(alertPluginInstanceMapper.deleteById(9)).thenReturn(1);
result = alertPluginInstanceService.delete(user, 9);
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
Mockito.when(alertPluginInstanceMapper.deleteById(5)).thenReturn(1);
result = alertPluginInstanceService.delete(user, 5);
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
@ -208,20 +228,16 @@ public class AlertPluginInstanceServiceTest {
1, ALERT_PLUGIN_UPDATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.ALERT_PLUGIN_INSTANCE,
null, 0, baseServiceLogger)).thenReturn(true);
Map<String, Object> result = alertPluginInstanceService.update(user, 1, "testUpdate", uiParams);
Map<String, Object> result = alertPluginInstanceService.update(user, 1, "testUpdate", warningType, uiParams);
Assertions.assertEquals(Status.SAVE_ERROR, result.get(Constants.STATUS));
Mockito.when(alertPluginInstanceMapper.updateById(Mockito.any())).thenReturn(1);
result = alertPluginInstanceService.update(user, 1, "testUpdate", uiParams);
result = alertPluginInstanceService.update(user, 1, "testUpdate", warningType, uiParams);
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
public void testQueryAll() {
AlertPluginInstance alertPluginInstance = new AlertPluginInstance();
alertPluginInstance.setId(1);
alertPluginInstance.setPluginDefineId(1);
alertPluginInstance.setPluginInstanceParams(paramsMap);
alertPluginInstance.setInstanceName("test");
AlertPluginInstance alertPluginInstance = getAlertPluginInstance(1, normalInstanceType, "test");
PluginDefine pluginDefine = new PluginDefine("script", "script", uiParams);
pluginDefine.setId(1);
List<PluginDefine> pluginDefines = Collections.singletonList(pluginDefine);
@ -232,4 +248,16 @@ public class AlertPluginInstanceServiceTest {
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
private AlertPluginInstance getAlertPluginInstance(int id, AlertPluginInstanceType instanceType,
String instanceName) {
AlertPluginInstance alertPluginInstance = new AlertPluginInstance();
alertPluginInstance.setId(id);
alertPluginInstance.setPluginDefineId(1);
alertPluginInstance.setInstanceType(instanceType);
alertPluginInstance.setWarningType(warningType);
alertPluginInstance.setPluginInstanceParams(paramsMap);
alertPluginInstance.setInstanceName(instanceName);
return alertPluginInstance;
}
}

4
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionServiceTest.java
View File

@ -70,6 +70,7 @@ import org.apache.dolphinscheduler.dao.model.PageListingResult;
import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.ProcessDefinitionLogDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionLogDao;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.spi.enums.DbType;
@ -184,6 +185,9 @@ public class ProcessDefinitionServiceTest extends BaseServiceTestTool {
@Mock
private UserMapper userMapper;
@Mock
private ListenerEventAlertManager listenerEventAlertManager;
protected User user;
protected Exception exception;
protected final static long projectCode = 1L;

43
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertPluginInstanceType.java Normal file
View File

@ -0,0 +1,43 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.common.enums;
import com.baomidou.mybatisplus.annotation.EnumValue;
public enum AlertPluginInstanceType {
NORMAL(0, "NORMAL"),
GLOBAL(1, "GLOBAL");
AlertPluginInstanceType(int code, String descp) {
this.code = code;
this.descp = descp;
}
@EnumValue
private final int code;
private final String descp;
public int getCode() {
return code;
}
public String getDescp() {
return descp;
}
}

66
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ListenerEventType.java Normal file
View File

@ -0,0 +1,66 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.common.enums;
import java.util.HashMap;
import java.util.Map;
import lombok.Getter;
import com.baomidou.mybatisplus.annotation.EnumValue;
@Getter
public enum ListenerEventType {
SERVER_DOWN(0, "SERVER_DOWN"),
PROCESS_DEFINITION_CREATED(1, "PROCESS_DEFINITION_CREATED"),
PROCESS_DEFINITION_UPDATED(2, "PROCESS_DEFINITION_UPDATED"),
PROCESS_DEFINITION_DELETED(3, "PROCESS_DEFINITION_DELETED"),
PROCESS_START(4, "PROCESS_START"),
PROCESS_END(5, "PROCESS_INSTANCE_END"),
PROCESS_FAIL(6, "PROCESS_FAIL"),
TASK_START(10, "TASK_START"),
TASK_END(11, "TASK_END"),
TASK_FAIL(12, "TASK_FAIL");
private static final Map<Integer, ListenerEventType> CODE_MAP = new HashMap<>();
static {
for (ListenerEventType listenerEventType : ListenerEventType.values()) {
CODE_MAP.put(listenerEventType.getCode(), listenerEventType);
}
}
@EnumValue
private final int code;
private final String descp;
ListenerEventType(int code, String descp) {
this.code = code;
this.descp = descp;
}
public static ListenerEventType of(int code) {
ListenerEventType listenerEventType = CODE_MAP.get(code);
if (listenerEventType == null) {
throw new IllegalArgumentException(String.format("The task execution status code: %s is invalid",
code));
}
return listenerEventType;
}
}

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

@ -35,11 +35,13 @@ public enum WarningType {
* 1 send if process success;
* 2 send if process failed;
* 3 send if process ends, whatever the result;
* 4 send global events;
*/
NONE(0, "none"),
SUCCESS(1, "success"),
FAILURE(2, "failure"),
ALL(3, "all");
ALL(3, "all"),
GLOBAL(4, "global");
WarningType(int code, String descp) {
this.code = code;

43
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/enums/ListenerEventTypeTest.java Normal file
View File

@ -0,0 +1,43 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.common.enums;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class ListenerEventTypeTest {
@Test
public void testGetCode() {
Assertions.assertEquals(0, ListenerEventType.SERVER_DOWN.getCode());
Assertions.assertEquals(1, ListenerEventType.PROCESS_DEFINITION_CREATED.getCode());
}
@Test
public void testGetDesp() {
Assertions.assertEquals("PROCESS_DEFINITION_UPDATED", ListenerEventType.PROCESS_DEFINITION_UPDATED.getDescp());
Assertions.assertEquals("PROCESS_DEFINITION_DELETED", ListenerEventType.PROCESS_DEFINITION_DELETED.getDescp());
}
@Test
public void testGetListenerEventTypeByCode() {
Assertions.assertEquals(ListenerEventType.PROCESS_START, ListenerEventType.of(4));
Assertions.assertNotEquals(ListenerEventType.PROCESS_END, ListenerEventType.of(6));
Assertions.assertThrows(IllegalArgumentException.class, () -> ListenerEventType.of(-1));
}
}

3
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
View File

@ -63,7 +63,8 @@ import com.google.common.collect.Lists;
@Slf4j
public class AlertDao {
private static final int QUERY_ALERT_THRESHOLD = 100;
@Value("${alert.query_alert_threshold:100}")
private Integer QUERY_ALERT_THRESHOLD;
@Value("${alert.alarm-suppression.crash:60}")
private Integer crashAlarmSuppression;

19
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/AlertPluginInstance.java
View File

@ -17,6 +17,9 @@
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import java.util.Date;
import lombok.Data;
@ -55,6 +58,18 @@ public class AlertPluginInstance {
@TableField("plugin_instance_params")
private String pluginInstanceParams;
/**
* instance_type. 0 normal, 1 global
*/
@TableField("instance_type")
private AlertPluginInstanceType instanceType;
/**
* warning_type
*/
@TableField("warning_type")
private WarningType warningType;
/**
* create_time
*/
@ -80,9 +95,11 @@ public class AlertPluginInstance {
this.instanceName = instanceName;
}
public AlertPluginInstance(int id, String pluginInstanceParams, String instanceName, Date updateDate) {
public AlertPluginInstance(int id, String pluginInstanceParams, String instanceName, WarningType warningType,
Date updateDate) {
this.id = id;
this.pluginInstanceParams = pluginInstanceParams;
this.warningType = warningType;
this.updateTime = updateDate;
this.instanceName = instanceName;
}

88
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/ListenerEvent.java Normal file
View File

@ -0,0 +1,88 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@TableName("t_ds_listener_event")
public class ListenerEvent {
/**
* primary key
*/
@TableId(value = "id", type = IdType.AUTO)
private Integer id;
/**
* content
*/
@TableField(value = "content")
private String content;
/**
* sign
*/
@TableField(value = "sign")
private String sign;
/**
* alert_status
*/
@TableField(value = "event_type")
private ListenerEventType eventType;
/**
* post_status
*/
@TableField("post_status")
private AlertStatus postStatus;
/**
* log
*/
@TableField("log")
private String log;
/**
* create_time
*/
@TableField("create_time")
private Date createTime;
/**
* update_time
*/
@TableField("update_time")
private Date updateTime;
}

27
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/AbstractListenerEvent.java Normal file
View File

@ -0,0 +1,27 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
public interface AbstractListenerEvent {
ListenerEventType getEventType();
String getTitle();
}

195
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/ProcessDefinitionCreatedListenerEvent.java Normal file
View File

@ -0,0 +1,195 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.enums.ProcessExecutionTypeEnum;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog;
import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import java.util.Date;
import java.util.List;
import java.util.Map;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ProcessDefinitionCreatedListenerEvent implements AbstractListenerEvent {
/**
* id
*/
private Integer id;
/**
* code
*/
private long code;
/**
* name
*/
private String name;
/**
* version
*/
private int version;
/**
* release state : online/offline
*/
private ReleaseState releaseState;
/**
* project code
*/
private long projectCode;
/**
* description
*/
private String description;
/**
* user defined parameters
*/
private String globalParams;
/**
* user defined parameter list
*/
private List<Property> globalParamList;
/**
* user define parameter map
*/
private Map<String, String> globalParamMap;
/**
* create time
*/
private Date createTime;
/**
* update time
*/
private Date updateTime;
/**
* process is valid: yes/no
*/
private Flag flag;
/**
* process user id
*/
private int userId;
/**
* create user name
*/
private String userName;
/**
* project name
*/
private String projectName;
/**
* locations array for web
*/
private String locations;
/**
* schedule release state : online/offline
*/
private ReleaseState scheduleReleaseState;
/**
* process warning time out. unit: minute
*/
private int timeout;
/**
* modify user name
*/
private String modifyBy;
/**
* warningGroupId
*/
private Integer warningGroupId;
/**
* execution type
*/
private ProcessExecutionTypeEnum executionType;
/**
* task definitions
*/
List<TaskDefinitionLog> taskDefinitionLogs;
/**
*
*/
List<ProcessTaskRelationLog> taskRelationList;
public ProcessDefinitionCreatedListenerEvent(ProcessDefinition processDefinition) {
this.setId(processDefinition.getId());
this.setCode(processDefinition.getCode());
this.setName(processDefinition.getName());
this.setVersion(processDefinition.getVersion());
this.setReleaseState(processDefinition.getReleaseState());
this.setProjectCode(processDefinition.getProjectCode());
this.setDescription(processDefinition.getDescription());
this.setGlobalParams(processDefinition.getGlobalParams());
this.setGlobalParamList(processDefinition.getGlobalParamList());
this.setGlobalParamMap(processDefinition.getGlobalParamMap());
this.setCreateTime(processDefinition.getCreateTime());
this.setUpdateTime(processDefinition.getUpdateTime());
this.setFlag(processDefinition.getFlag());
this.setUserId(processDefinition.getUserId());
this.setUserName(processDefinition.getUserName());
this.setProjectName(processDefinition.getProjectName());
this.setLocations(processDefinition.getLocations());
this.setScheduleReleaseState(processDefinition.getScheduleReleaseState());
this.setTimeout(processDefinition.getTimeout());
this.setModifyBy(processDefinition.getModifyBy());
this.setWarningGroupId(processDefinition.getWarningGroupId());
this.setExecutionType(processDefinition.getExecutionType());
}
@Override
public ListenerEventType getEventType() {
return ListenerEventType.PROCESS_DEFINITION_CREATED;
}
@Override
public String getTitle() {
return String.format("process definition created:%s", this.name);
}
}

52
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/ProcessDefinitionDeletedListenerEvent.java Normal file
View File

@ -0,0 +1,52 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ProcessDefinitionDeletedListenerEvent implements AbstractListenerEvent {
private Integer projectId;
private Long projectCode;
private String projectName;
private String owner;
private Integer id;
private Long code;
private String name;
private Integer userId;
private String modifiedBy;
private Date eventTime;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.PROCESS_DEFINITION_DELETED;
}
@Override
public String getTitle() {
return String.format("process definition deleted:%s", this.name);
}
}

195
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/ProcessDefinitionUpdatedListenerEvent.java Normal file
View File

@ -0,0 +1,195 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.enums.ProcessExecutionTypeEnum;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog;
import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import java.util.Date;
import java.util.List;
import java.util.Map;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ProcessDefinitionUpdatedListenerEvent implements AbstractListenerEvent {
/**
* id
*/
private Integer id;
/**
* code
*/
private long code;
/**
* name
*/
private String name;
/**
* version
*/
private int version;
/**
* release state : online/offline
*/
private ReleaseState releaseState;
/**
* project code
*/
private long projectCode;
/**
* description
*/
private String description;
/**
* user defined parameters
*/
private String globalParams;
/**
* user defined parameter list
*/
private List<Property> globalParamList;
/**
* user define parameter map
*/
private Map<String, String> globalParamMap;
/**
* create time
*/
private Date createTime;
/**
* update time
*/
private Date updateTime;
/**
* process is valid: yes/no
*/
private Flag flag;
/**
* process user id
*/
private int userId;
/**
* create user name
*/
private String userName;
/**
* project name
*/
private String projectName;
/**
* locations array for web
*/
private String locations;
/**
* schedule release state : online/offline
*/
private ReleaseState scheduleReleaseState;
/**
* process warning time out. unit: minute
*/
private int timeout;
/**
* modify user name
*/
private String modifyBy;
/**
* warningGroupId
*/
private Integer warningGroupId;
/**
* execution type
*/
private ProcessExecutionTypeEnum executionType;
/**
* task definitions
*/
List<TaskDefinitionLog> taskDefinitionLogs;
/**
*
*/
List<ProcessTaskRelationLog> taskRelationList;
public ProcessDefinitionUpdatedListenerEvent(ProcessDefinition processDefinition) {
this.setId(processDefinition.getId());
this.setCode(processDefinition.getCode());
this.setName(processDefinition.getName());
this.setVersion(processDefinition.getVersion());
this.setReleaseState(processDefinition.getReleaseState());
this.setProjectCode(processDefinition.getProjectCode());
this.setDescription(processDefinition.getDescription());
this.setGlobalParams(processDefinition.getGlobalParams());
this.setGlobalParamList(processDefinition.getGlobalParamList());
this.setGlobalParamMap(processDefinition.getGlobalParamMap());
this.setCreateTime(processDefinition.getCreateTime());
this.setUpdateTime(processDefinition.getUpdateTime());
this.setFlag(processDefinition.getFlag());
this.setUserId(processDefinition.getUserId());
this.setUserName(processDefinition.getUserName());
this.setProjectName(processDefinition.getProjectName());
this.setLocations(processDefinition.getLocations());
this.setScheduleReleaseState(processDefinition.getScheduleReleaseState());
this.setTimeout(processDefinition.getTimeout());
this.setModifyBy(processDefinition.getModifyBy());
this.setWarningGroupId(processDefinition.getWarningGroupId());
this.setExecutionType(processDefinition.getExecutionType());
}
@Override
public ListenerEventType getEventType() {
return ListenerEventType.PROCESS_DEFINITION_UPDATED;
}
@Override
public String getTitle() {
return String.format("process definition updated:%s", this.name);
}
}

59
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/ProcessEndListenerEvent.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ProcessEndListenerEvent implements AbstractListenerEvent {
private Long projectCode;
private String projectName;
private String owner;
private Integer processId;
private Long processDefinitionCode;
private String processName;
private CommandType processType;
private WorkflowExecutionStatus processState;
private Flag recovery;
private Integer runTimes;
private Date processStartTime;
private Date processEndTime;
private String processHost;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.PROCESS_END;
}
@Override
public String getTitle() {
return String.format("process end: %s", processName);
}
}

59
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/ProcessFailListenerEvent.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ProcessFailListenerEvent implements AbstractListenerEvent {
private Long projectCode;
private String projectName;
private String owner;
private Integer processId;
private Long processDefinitionCode;
private String processName;
private CommandType processType;
private WorkflowExecutionStatus processState;
private Flag recovery;
private Integer runTimes;
private Date processStartTime;
private Date processEndTime;
private String processHost;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.PROCESS_FAIL;
}
@Override
public String getTitle() {
return String.format("process fail: %s", processName);
}
}

58
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/ProcessStartListenerEvent.java Normal file
View File

@ -0,0 +1,58 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ProcessStartListenerEvent implements AbstractListenerEvent {
private Long projectCode;
private String projectName;
private String owner;
private Long processDefinitionCode;
private String processDefinitionName;
private Integer processId;
private String processName;
private CommandType processType;
private WorkflowExecutionStatus processState;
private Integer runTimes;
private Flag recovery;
private Date processStartTime;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.PROCESS_START;
}
@Override
public String getTitle() {
return String.format("process start: %s", processName);
}
}

45
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/ServerDownListenerEvent.java Normal file
View File

@ -0,0 +1,45 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ServerDownListenerEvent implements AbstractListenerEvent {
private String type;
private String host;
private Date eventTime;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.SERVER_DOWN;
}
@Override
public String getTitle() {
return String.format("%s server down: %s", type, host);
}
}

59
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/TaskEndListenerEvent.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class TaskEndListenerEvent implements AbstractListenerEvent {
private long projectCode;
private String projectName;
private String owner;
private long processId;
private long processDefinitionCode;
private String processName;
private int taskInstanceId;
private long taskCode;
private String taskName;
private String taskType;
private TaskExecutionStatus taskState;
private Date taskStartTime;
private Date taskEndTime;
private String taskHost;
private String logPath;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.TASK_END;
}
@Override
public String getTitle() {
return String.format("task end: %s", taskName);
}
}

59
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/TaskFailListenerEvent.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class TaskFailListenerEvent implements AbstractListenerEvent {
private long projectCode;
private String projectName;
private String owner;
private long processId;
private long processDefinitionCode;
private String processName;
private int taskInstanceId;
private long taskCode;
private String taskName;
private String taskType;
private TaskExecutionStatus taskState;
private Date taskStartTime;
private Date taskEndTime;
private String taskHost;
private String logPath;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.TASK_FAIL;
}
@Override
public String getTitle() {
return String.format("task fail: %s", taskName);
}
}

59
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/event/TaskStartListenerEvent.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity.event;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class TaskStartListenerEvent implements AbstractListenerEvent {
private long projectCode;
private String projectName;
private String owner;
private long processId;
private long processDefinitionCode;
private String processName;
private int taskInstanceId;
private long taskCode;
private String taskName;
private String taskType;
private TaskExecutionStatus taskState;
private Date taskStartTime;
private Date taskEndTime;
private String taskHost;
private String logPath;
@Override
public ListenerEventType getEventType() {
return ListenerEventType.TASK_START;
}
@Override
public String getTitle() {
return String.format("task start: %s", taskName);
}
}

7
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertPluginInstanceMapper.java
View File

@ -36,6 +36,13 @@ public interface AlertPluginInstanceMapper extends BaseMapper<AlertPluginInstanc
*/
List<AlertPluginInstance> queryAllAlertPluginInstanceList();
/**
* query all global alert plugin instance
*
* @return global AlertPluginInstance list
*/
List<AlertPluginInstance> queryAllGlobalAlertPluginInstanceList();
/**
* query by alert group id
*

42
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ListenerEventMapper.java Normal file
View File

@ -0,0 +1,42 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.dao.entity.ListenerEvent;
import org.apache.ibatis.annotations.Param;
import java.util.Date;
import java.util.List;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
public interface ListenerEventMapper extends BaseMapper<ListenerEvent> {
int batchInsert(@Param("events") List<ListenerEvent> events);
void insertServerDownEvent(@Param("event") ListenerEvent event,
@Param("crashAlarmSuppressionStartTime") Date crashAlarmSuppressionStartTime);
List<ListenerEvent> listingListenerEventByStatus(@Param("postStatus") AlertStatus postStatus,
@Param("limit") int limit);
void updateListenerEvent(@Param("eventId") int eventId, @Param("postStatus") AlertStatus postStatus,
@Param("log") String log, @Param("updateTime") Date updateTime);
}

10
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertPluginInstanceMapper.xml
View File

@ -20,7 +20,7 @@
<mapper namespace="org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper">
<sql id="baseSql">
id, plugin_define_id, plugin_instance_params, create_time, update_time, instance_name
id, plugin_define_id, plugin_instance_params, instance_type, warning_type, create_time, update_time, instance_name
</sql>
@ -32,6 +32,14 @@
where 1 = 1 order by update_time desc
</select>
<select id="queryAllGlobalAlertPluginInstanceList"
resultType="org.apache.dolphinscheduler.dao.entity.AlertPluginInstance">
select
<include refid="baseSql"/>
from t_ds_alert_plugin_instance
where instance_type = 1 order by update_time desc
</select>
<select id="queryByIds" resultType="org.apache.dolphinscheduler.dao.entity.AlertPluginInstance">
select
<include refid="baseSql"/>

66
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ListenerEventMapper.xml Normal file
View File

@ -0,0 +1,66 @@
<?xml version="1.0" encoding="UTF-8" ?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You under the Apache License, Version 2.0
~ (the "License"); you may not use this file except in compliance with
~ the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="org.apache.dolphinscheduler.dao.mapper.ListenerEventMapper">
<sql id="baseSql">
id
, content, sign, event_type, post_status, log, create_time, update_time
</sql>
<insert id="batchInsert">
insert into t_ds_listener_event ( content, sign, post_status, event_type, create_time, update_time)
values
<foreach collection='events' item='event' separator=','>
(#{event.content},#{event.sign},#{event.postStatus.code},
#{event.eventType.code},#{event.createTime},#{event.updateTime})
</foreach>
</insert>
<insert id="insertServerDownEvent">
insert into t_ds_listener_event (content, sign, post_status, event_type, create_time, update_time)
SELECT #{event.sign},
#{event.content},
#{event.postStatus.code},
#{event.eventType.code},
#{event.createTime},
#{event.updateTime}
from t_ds_listener_event
where create_time >= #{crashAlarmSuppressionStartTime}
and sign = #{event.sign}
and post_status = #{event.postStatus.code}
having count(*) = 0
</insert>
<update id="updateListenerEvent">
update t_ds_listener_event
set log = #{log},
post_status = #{postStatus.code},
update_time = #{updateTime}
where id = #{eventId}
</update>
<select id="listingListenerEventByStatus" resultType="org.apache.dolphinscheduler.dao.entity.ListenerEvent">
select
<include refid="baseSql"/>
from t_ds_listener_event
where post_status = #{postStatus.code}
limit #{limit}
</select>
</mapper>

25
dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql
View File

@ -1062,6 +1062,10 @@ INSERT INTO t_ds_alertgroup(alert_instance_ids, create_user_id, group_name, desc
VALUES (NULL, 1, 'default admin warning group', 'default admin warning group', '2018-11-29 10:20:39',
'2018-11-29 10:20:39');
INSERT INTO t_ds_alertgroup(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', '2018-11-29 10:20:39',
'2018-11-29 10:20:39');
-- ----------------------------
-- Records of t_ds_user
-- ----------------------------
@ -1097,6 +1101,8 @@ CREATE TABLE t_ds_alert_plugin_instance
create_time timestamp NULL DEFAULT CURRENT_TIMESTAMP,
update_time timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
instance_name varchar(200) DEFAULT NULL,
instance_type int NOT NULL default '0',
warning_type int NOT NULL default '3',
PRIMARY KEY (id)
);
@ -2141,3 +2147,22 @@ CREATE TABLE t_ds_relation_sub_workflow (
INDEX idx_parent_task_code (parent_task_code),
INDEX idx_sub_workflow_instance_id (sub_workflow_instance_id)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
-- ----------------------------
-- Table structure for t_ds_listener_event
-- ----------------------------
DROP TABLE IF EXISTS t_ds_listener_event;
CREATE TABLE t_ds_listener_event
(
id int NOT NULL AUTO_INCREMENT,
content text,
sign char(64) NOT NULL DEFAULT '',
post_status tinyint(4) NOT NULL DEFAULT '0',
event_type int(11),
log text,
create_time datetime DEFAULT NULL,
update_time datetime DEFAULT NULL,
PRIMARY KEY (id),
KEY idx_status (post_status),
KEY idx_event_sign (sign)
);

22
dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql
View File

@ -1051,6 +1051,8 @@ INSERT IGNORE INTO `t_ds_version` VALUES ('1', '3.3.0');
-- ----------------------------
INSERT IGNORE INTO `t_ds_alertgroup`(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'default admin warning group', 'default admin warning group', current_timestamp, current_timestamp);
INSERT IGNORE INTO `t_ds_alertgroup`(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', current_timestamp, current_timestamp);
-- ----------------------------
-- Records of t_ds_user
@ -1086,6 +1088,8 @@ CREATE TABLE `t_ds_alert_plugin_instance` (
`create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
`update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`instance_name` varchar(255) DEFAULT NULL COMMENT 'alert instance name',
`instance_type` int NOT NULL default '0',
`warning_type` int NOT NULL default '3',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE = utf8_bin;
@ -2114,3 +2118,21 @@ CREATE TABLE `t_ds_relation_sub_workflow` (
KEY `idx_parent_task_code` (`parent_task_code`),
KEY `idx_sub_workflow_instance_id` (`sub_workflow_instance_id`)
);
-- ----------------------------
-- Table structure for t_ds_listener_event
-- ----------------------------
DROP TABLE IF EXISTS `t_ds_listener_event`;
CREATE TABLE `t_ds_listener_event` (
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
`content` text COMMENT 'listener event json content',
`sign` char(64) NOT NULL DEFAULT '' COMMENT 'sign=sha1(content)',
`post_status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '0:wait running,1:success,2:failed,3:partial success',
`event_type` int(11) NOT NULL COMMENT 'listener event type',
`log` text COMMENT 'log',
`create_time` datetime DEFAULT NULL COMMENT 'create time',
`update_time` datetime DEFAULT NULL COMMENT 'update time',
PRIMARY KEY (`id`),
KEY `idx_status` (`post_status`) USING BTREE,
KEY `idx_sign` (`sign`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE = utf8_bin;

26
dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql
View File

@ -1035,6 +1035,8 @@ VALUES (-1, 'default', 'default tenant', '1', '2018-03-27 15:48:50', '2018-10-24
-- Records of t_ds_alertgroup, default admin warning group
INSERT INTO t_ds_alertgroup(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'default admin warning group', 'default admin warning group', '2018-11-29 10:20:39', '2018-11-29 10:20:39');
INSERT INTO t_ds_alertgroup(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', '2018-11-29 10:20:39', '2018-11-29 10:20:39');
-- Records of t_ds_queue,default queue name : default
INSERT INTO t_ds_queue(queue_name, queue, create_time, update_time)
@ -1066,11 +1068,13 @@ CREATE TABLE t_ds_plugin_define (
DROP TABLE IF EXISTS t_ds_alert_plugin_instance;
CREATE TABLE t_ds_alert_plugin_instance (
id serial NOT NULL,
plugin_define_id int4 NOT NULL,
plugin_define_id int NOT NULL,
plugin_instance_params text NULL,
create_time timestamp NULL,
update_time timestamp NULL,
instance_name varchar(255) NULL,
instance_type int NOT NULL default '0',
warning_type int NOT NULL default '3',
CONSTRAINT t_ds_alert_plugin_instance_pk PRIMARY KEY (id)
);
@ -2103,3 +2107,23 @@ CREATE INDEX idx_parent_workflow_instance_id ON t_ds_relation_sub_workflow (pare
CREATE INDEX idx_parent_task_code ON t_ds_relation_sub_workflow (parent_task_code);
CREATE INDEX idx_sub_workflow_instance_id ON t_ds_relation_sub_workflow (sub_workflow_instance_id);
--
-- Table structure for table t_ds_alert
--
DROP TABLE IF EXISTS t_ds_listener_event;
CREATE TABLE t_ds_listener_event(
id int NOT NULL,
content text,
sign varchar(64) NOT NULL DEFAULT '',
post_status int NOT NULL DEFAULT '0',
event_type int NOT NULL,
log text,
create_time timestamp DEFAULT NULL,
update_time timestamp DEFAULT NULL,
PRIMARY KEY (id)
);
comment on column t_ds_listener_event.sign is 'sign=sha1(content)';
create index idx_listener_event_post_status on t_ds_listener_event (post_status);
create index idx_listener_event_sign on t_ds_listener_event (sign);

7
dolphinscheduler-dao/src/main/resources/sql/upgrade/3.2.0_schema/mysql/dolphinscheduler_ddl.sql
View File

@ -447,6 +447,13 @@ BEGIN
THEN
ALTER TABLE `t_ds_fav_task` DROP COLUMN `task_name`;
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.COLUMNS
WHERE TABLE_NAME='t_ds_fav_task'
AND TABLE_SCHEMA=(SELECT DATABASE())
AND COLUMN_NAME ='task_type')
THEN
ALTER TABLE `t_ds_fav_task` ADD `task_type` varchar(64) NOT NULL COMMENT 'favorite task type name';
END IF;
END;
d//
delimiter ;

20
dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/mysql/dolphinscheduler_ddl.sql
View File

@ -14,3 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-- Modify "t_ds_alertgroup" table
ALTER TABLE `t_ds_alertgroup` AUTO_INCREMENT 3;
-- Modify "t_ds_alert_plugin_instance" table
ALTER TABLE `t_ds_alert_plugin_instance`
ADD COLUMN `instance_type` int NOT NULL DEFAULT 0, ADD COLUMN `warning_type` int NOT NULL DEFAULT 3;
-- Create "t_ds_listener_event" table
CREATE TABLE `t_ds_listener_event`
(
`id` int NOT NULL AUTO_INCREMENT COMMENT "key",
`content` text NULL COMMENT "listener event json content",
`sign` char(64) NOT NULL DEFAULT "" COMMENT "sign=sha1(content)",
`post_status` tinyint NOT NULL DEFAULT 0 COMMENT "0:wait running,1:success,2:failed,3:partial success",
`event_type` int NOT NULL COMMENT "listener event type",
`log` text NULL COMMENT "log",
`create_time` datetime NULL COMMENT "create time",
`update_time` datetime NULL COMMENT "update time",
PRIMARY KEY (`id`),
INDEX `idx_sign` (`sign`),
INDEX `idx_status` (`post_status`)
) CHARSET utf8 COLLATE utf8_bin;

2
dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/mysql/dolphinscheduler_dml.sql
View File

@ -14,3 +14,5 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
INSERT IGNORE INTO `t_ds_alertgroup`(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', current_timestamp, current_timestamp);

10
dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/postgresql/dolphinscheduler_ddl.sql
View File

@ -14,3 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-- Modify "t_ds_alert_plugin_instance" table
ALTER TABLE "t_ds_alert_plugin_instance" ADD COLUMN "instance_type" integer NOT NULL DEFAULT 0, ADD COLUMN "warning_type" integer NOT NULL DEFAULT 3;
-- Create "t_ds_listener_event" table
CREATE TABLE "t_ds_listener_event" ("id" integer NOT NULL, "content" text NULL, "sign" character varying(64) NOT NULL DEFAULT '', "post_status" integer NOT NULL DEFAULT 0, "event_type" integer NOT NULL, "log" text NULL, "create_time" timestamp NULL, "update_time" timestamp NULL, PRIMARY KEY ("id"));
-- Create index "idx_listener_event_post_status" to table: "t_ds_listener_event"
CREATE INDEX "idx_listener_event_post_status" ON "t_ds_listener_event" ("post_status");
-- Create index "idx_listener_event_sign" to table: "t_ds_listener_event"
CREATE INDEX "idx_listener_event_sign" ON "t_ds_listener_event" ("sign");
-- Set comment to column: "sign" on table: "t_ds_listener_event"
COMMENT ON COLUMN "t_ds_listener_event" ."sign" IS 'sign=sha1(content)';

2
dolphinscheduler-dao/src/main/resources/sql/upgrade/3.3.0_schema/postgresql/dolphinscheduler_dml.sql
View File

@ -14,3 +14,5 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
INSERT INTO t_ds_alertgroup(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', '2018-11-29 10:20:39', '2018-11-29 10:20:39');

48
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinitionCreatedListenerEventTest.java Normal file
View File

@ -0,0 +1,48 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionCreatedListenerEvent;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class ProcessDefinitionCreatedListenerEventTest {
@Test
public void testBuildProcessDefinitionUpdatedListenerEvent() {
int id = 1;
long code = 1L;
String name = "testName";
ReleaseState releaseState = ReleaseState.OFFLINE;
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setId(id);
processDefinition.setCode(code);
processDefinition.setName(name);
processDefinition.setReleaseState(releaseState);
ProcessDefinitionCreatedListenerEvent event = new ProcessDefinitionCreatedListenerEvent(processDefinition);
Assertions.assertEquals(event.getEventType(), ListenerEventType.PROCESS_DEFINITION_CREATED);
Assertions.assertEquals(event.getId(), id);
Assertions.assertEquals(event.getCode(), code);
Assertions.assertEquals(event.getName(), name);
Assertions.assertEquals(event.getReleaseState(), releaseState);
Assertions.assertEquals(String.format("process definition created:%s", name), event.getTitle());
}
}

48
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/entity/ProcessDefinitionUpdatedListenerEventTest.java Normal file
View File

@ -0,0 +1,48 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionUpdatedListenerEvent;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class ProcessDefinitionUpdatedListenerEventTest {
@Test
public void testBuildProcessDefinitionUpdatedListenerEvent() {
int id = 1;
long code = 1L;
String name = "testName";
ReleaseState releaseState = ReleaseState.OFFLINE;
ProcessDefinition processDefinition = new ProcessDefinition();
processDefinition.setId(id);
processDefinition.setCode(code);
processDefinition.setName(name);
processDefinition.setReleaseState(releaseState);
ProcessDefinitionUpdatedListenerEvent event = new ProcessDefinitionUpdatedListenerEvent(processDefinition);
Assertions.assertEquals(event.getEventType(), ListenerEventType.PROCESS_DEFINITION_UPDATED);
Assertions.assertEquals(event.getId(), id);
Assertions.assertEquals(event.getCode(), code);
Assertions.assertEquals(event.getName(), name);
Assertions.assertEquals(event.getReleaseState(), releaseState);
Assertions.assertEquals(String.format("process definition updated:%s", name), event.getTitle());
}
}

55
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/AlertPluginInstanceMapperTest.java
View File

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.common.enums.AlertPluginInstanceType;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
@ -41,6 +42,9 @@ public class AlertPluginInstanceMapperTest extends BaseDaoTest {
@Autowired
private PluginDefineMapper pluginDefineMapper;
@Autowired
private AlertGroupMapper alertGroupMapper;
/**
* Test function queryAllAlertPluginInstanceList behavior with different size.
*/
@ -49,13 +53,17 @@ public class AlertPluginInstanceMapperTest extends BaseDaoTest {
List<AlertPluginInstance> withoutSingleOne = alertPluginInstanceMapper.queryAllAlertPluginInstanceList();
Assertions.assertEquals(0, withoutSingleOne.size());
createAlertPluginInstance("test_instance_1");
createNormalAlertPluginInstance("test_instance_1");
List<AlertPluginInstance> withExactlyOne = alertPluginInstanceMapper.queryAllAlertPluginInstanceList();
Assertions.assertEquals(1, withExactlyOne.size());
createAlertPluginInstance("test_instance_2");
createNormalAlertPluginInstance("test_instance_2");
List<AlertPluginInstance> withExactlyTwo = alertPluginInstanceMapper.queryAllAlertPluginInstanceList();
Assertions.assertEquals(2, withExactlyTwo.size());
createGlobalAlertPluginInstance("test_global_instance_1");
List<AlertPluginInstance> withExactlyThree = alertPluginInstanceMapper.queryAllAlertPluginInstanceList();
Assertions.assertEquals(3, withExactlyThree.size());
}
/**
@ -65,7 +73,9 @@ public class AlertPluginInstanceMapperTest extends BaseDaoTest {
public void testExistInstanceName() {
String instanceName = "test_instance";
Assertions.assertNull(alertPluginInstanceMapper.existInstanceName(instanceName));
createAlertPluginInstance(instanceName);
createNormalAlertPluginInstance(instanceName);
Assertions.assertTrue(alertPluginInstanceMapper.existInstanceName(instanceName));
createGlobalAlertPluginInstance(instanceName);
Assertions.assertTrue(alertPluginInstanceMapper.existInstanceName(instanceName));
}
@ -74,8 +84,8 @@ public class AlertPluginInstanceMapperTest extends BaseDaoTest {
*/
@Test
public void testQueryByInstanceNamePage() {
createAlertPluginInstance("test_with_pattern_instance");
createAlertPluginInstance("test_no_instance");
createNormalAlertPluginInstance("test_with_pattern_instance");
createNormalAlertPluginInstance("test_no_instance");
Page<AlertPluginInstance> page = new Page<>(1, 10);
IPage<AlertPluginInstance> matchTwoRecord = alertPluginInstanceMapper.queryByInstanceNamePage(page, "test");
@ -86,14 +96,45 @@ public class AlertPluginInstanceMapperTest extends BaseDaoTest {
}
/**
* Create alert plugin instance according to given alter plugin name.
* Test function queryAllGlobalAlertPluginInstanceList returning with different search variables.
*/
private void createAlertPluginInstance(String alterPluginInsName) {
@Test
public void testQueryAllGlobalAlertPluginInstanceList() {
List<AlertPluginInstance> withoutSingleOne = alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList();
Assertions.assertEquals(0, withoutSingleOne.size());
createNormalAlertPluginInstance("test_normal_instance");
List<AlertPluginInstance> withExactlyOne = alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList();
Assertions.assertEquals(0, withExactlyOne.size());
createGlobalAlertPluginInstance("test_global_instance_1");
List<AlertPluginInstance> withExactlyTwo = alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList();
Assertions.assertEquals(1, withExactlyTwo.size());
createGlobalAlertPluginInstance("test_global_instance_2");
List<AlertPluginInstance> withExactlyThree = alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList();
Assertions.assertEquals(2, withExactlyThree.size());
}
/**
* Create normal alert plugin instance according to given alter plugin name.
*/
private void createNormalAlertPluginInstance(String alterPluginInsName) {
PluginDefine pluginDefine = makeSurePluginDefineExists();
AlertPluginInstance alertPluginInstance = new AlertPluginInstance(pluginDefine.getId(), "", alterPluginInsName);
alertPluginInstanceMapper.insert(alertPluginInstance);
}
/**
* Create global alert plugin instance according to given alter plugin name.
*/
private void createGlobalAlertPluginInstance(String alterPluginInsName) {
PluginDefine pluginDefine = makeSurePluginDefineExists();
AlertPluginInstance alertPluginInstance = new AlertPluginInstance(pluginDefine.getId(), "", alterPluginInsName);
alertPluginInstance.setInstanceType(AlertPluginInstanceType.GLOBAL);
alertPluginInstanceMapper.insert(alertPluginInstance);
}
/**
* Make sure plugin define exists.
* <p>

134
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ListenerEventMapperTest.java Normal file
View File

@ -0,0 +1,134 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.ListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ServerDownListenerEvent;
import org.apache.commons.codec.digest.DigestUtils;
import java.util.Date;
import java.util.List;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.common.collect.Lists;
/**
* AlertPluginInstanceMapper mapper test
*/
public class ListenerEventMapperTest extends BaseDaoTest {
@Autowired
private ListenerEventMapper listenerEventMapper;
/**
* test insert
*
* @return
*/
@Test
public void testInsert() {
ListenerEvent serverDownListenerEvent = generateServerDownListenerEvent("192.168.x.x");
listenerEventMapper.insert(serverDownListenerEvent);
Assertions.assertTrue(serverDownListenerEvent.getId() > 0);
}
/**
* test batch insert
*
* @return
*/
@Test
public void testBatchInsert() {
ListenerEvent event1 = generateServerDownListenerEvent("192.168.x.1");
ListenerEvent event2 = generateServerDownListenerEvent("192.168.x.2");
listenerEventMapper.batchInsert(Lists.newArrayList(event1, event2));
Assertions.assertEquals(listenerEventMapper.selectCount(new QueryWrapper<>()), 2L);
}
/**
* test list listener event by status
*
* @return
*/
@Test
public void testListingListenerEventByStatus() {
ListenerEvent event1 = generateServerDownListenerEvent("192.168.x.1");
ListenerEvent event2 = generateServerDownListenerEvent("192.168.x.2");
listenerEventMapper.batchInsert(Lists.newArrayList(event1, event2));
List<ListenerEvent> listenerEvents =
listenerEventMapper.listingListenerEventByStatus(AlertStatus.WAIT_EXECUTION, 50);
Assertions.assertEquals(listenerEvents.size(), 2);
}
/**
* test update server down event
*
* @return
*/
@Test
public void testUpdateListenerEvent() {
ListenerEvent event = generateServerDownListenerEvent("192.168.x.1");
listenerEventMapper.insert(event);
listenerEventMapper.updateListenerEvent(event.getId(), AlertStatus.EXECUTION_FAILURE, "fail", new Date());
ListenerEvent updatedEvent = listenerEventMapper.selectById(event.getId());
Assertions.assertEquals(updatedEvent.getPostStatus(), AlertStatus.EXECUTION_FAILURE);
Assertions.assertEquals(updatedEvent.getLog(), "fail");
}
/**
* test delete listener event
*/
@Test
public void testDeleteListenerEvent() {
ListenerEvent event = generateServerDownListenerEvent("192.168.x.1");
listenerEventMapper.insert(event);
listenerEventMapper.deleteById(event);
ListenerEvent actualAlert = listenerEventMapper.selectById(event.getId());
Assertions.assertNull(actualAlert);
}
/**
* create server down event
* @param host worker host
* @return listener event
*/
private ListenerEvent generateServerDownListenerEvent(String host) {
ServerDownListenerEvent event = new ServerDownListenerEvent();
event.setEventTime(new Date());
event.setHost(host);
event.setType("WORKER");
ListenerEvent listenerEvent = new ListenerEvent();
listenerEvent.setEventType(ListenerEventType.SERVER_DOWN);
listenerEvent.setContent(JSONUtils.toJsonString(event));
listenerEvent.setSign(DigestUtils.sha1Hex(listenerEvent.getContent()));
listenerEvent.setLog("success");
listenerEvent.setCreateTime(DateUtils.getCurrentDate());
listenerEvent.setUpdateTime(DateUtils.getCurrentDate());
listenerEvent.setPostStatus(AlertStatus.WAIT_EXECUTION);
return listenerEvent;
}
}

4
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/event/TaskStateEventHandler.java
View File

@ -54,6 +54,10 @@ public class TaskStateEventHandler implements StateEventHandler {
"Handle task instance state event, the current task instance state {} will be changed to {}",
task.getState().name(), taskStateEvent.getStatus().name());
if (taskStateEvent.getStatus().isRunning()) {
workflowExecuteRunnable.taskStart(task);
}
Set<Long> completeTaskSet = workflowExecuteRunnable.getCompleteTaskCodes();
if (task.getState().isFinished()
&& (taskStateEvent.getStatus() != null && taskStateEvent.getStatus().isRunning())) {

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java
View File

@ -32,6 +32,7 @@ import org.apache.dolphinscheduler.registry.api.SubscribeListener;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.WorkerGroupNotFoundException;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
@ -101,6 +102,9 @@ public class ServerNodeManager implements InitializingBean {
@Autowired
private MasterConfig masterConfig;
@Autowired
private ListenerEventAlertManager listenerEventAlertManager;
private final List<WorkerInfoChangeListener> workerInfoChangeListeners = new ArrayList<>();
private final List<MasterInfoChangeListener> masterInfoChangeListeners = new ArrayList<>();
@ -171,6 +175,7 @@ public class ServerNodeManager implements InitializingBean {
} else if (type == Type.REMOVE) {
log.info("Worker node : {} down.", path);
alertDao.sendServerStoppedAlert(1, path, "WORKER");
listenerEventAlertManager.publishServerDownListenerEvent(path, "WORKER");
} else if (type == Type.UPDATE) {
syncSingleWorkerNodeInfo(workerAddress, JSONUtils.parseObject(data, WorkerHeartBeat.class));
}
@ -203,6 +208,7 @@ public class ServerNodeManager implements InitializingBean {
} else if (type.equals(Type.REMOVE)) {
log.info("master node : {} down.", path);
alertDao.sendServerStoppedAlert(1, path, "MASTER");
listenerEventAlertManager.publishServerDownListenerEvent(path, "MASTER");
}
} catch (Exception ex) {
log.error("MasterNodeListener capture data change and get data failed.", ex);

28
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
View File

@ -85,6 +85,7 @@ import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecu
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.utils.TaskUtils;
import org.apache.dolphinscheduler.server.master.utils.WorkflowInstanceUtils;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
import org.apache.dolphinscheduler.service.command.CommandService;
import org.apache.dolphinscheduler.service.cron.CronUtils;
@ -225,6 +226,8 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
private final MasterConfig masterConfig;
private final ListenerEventAlertManager listenerEventAlertManager;
public WorkflowExecuteRunnable(
@NonNull IWorkflowExecuteContext workflowExecuteContext,
@NonNull CommandService commandService,
@ -235,7 +238,8 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
@NonNull StateWheelExecuteThread stateWheelExecuteThread,
@NonNull CuringParamsService curingParamsService,
@NonNull TaskInstanceDao taskInstanceDao,
@NonNull DefaultTaskExecuteRunnableFactory defaultTaskExecuteRunnableFactory) {
@NonNull DefaultTaskExecuteRunnableFactory defaultTaskExecuteRunnableFactory,
@NonNull ListenerEventAlertManager listenerEventAlertManager) {
this.processService = processService;
this.commandService = commandService;
this.processInstanceDao = processInstanceDao;
@ -246,6 +250,7 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
this.curingParamsService = curingParamsService;
this.taskInstanceDao = taskInstanceDao;
this.defaultTaskExecuteRunnableFactory = defaultTaskExecuteRunnableFactory;
this.listenerEventAlertManager = listenerEventAlertManager;
TaskMetrics.registerTaskPrepared(readyToSubmitTaskQueue::size);
}
@ -372,6 +377,17 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
return true;
}
}
public void processStart() {
ProcessInstance workflowInstance = workflowExecuteContext.getWorkflowInstance();
ProjectUser projectUser = processService.queryProjectWithUserByProcessInstanceId(workflowInstance.getId());
this.listenerEventAlertManager.publishProcessStartListenerEvent(workflowInstance, projectUser);
}
public void taskStart(TaskInstance taskInstance) {
ProcessInstance workflowInstance = workflowExecuteContext.getWorkflowInstance();
ProjectUser projectUser = processService.queryProjectWithUserByProcessInstanceId(workflowInstance.getId());
this.listenerEventAlertManager.publishTaskStartListenerEvent(workflowInstance, taskInstance, projectUser);
}
public void processTimeout() {
ProcessInstance workflowInstance = workflowExecuteContext.getWorkflowInstance();
@ -398,6 +414,9 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
completeTaskSet.add(taskInstance.getTaskCode());
mergeTaskInstanceVarPool(taskInstance);
processInstanceDao.upsertProcessInstance(workflowInstance);
ProjectUser projectUser =
processService.queryProjectWithUserByProcessInstanceId(workflowInstance.getId());
listenerEventAlertManager.publishTaskEndListenerEvent(workflowInstance, taskInstance, projectUser);
// save the cacheKey only if the task is defined as cache task and the task is success
if (taskInstance.getIsCache().equals(Flag.YES)) {
saveCacheTaskInstance(taskInstance);
@ -411,6 +430,9 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
retryTaskInstance(taskInstance);
} else if (taskInstance.getState().isFailure()) {
completeTaskSet.add(taskInstance.getTaskCode());
ProjectUser projectUser =
processService.queryProjectWithUserByProcessInstanceId(workflowInstance.getId());
listenerEventAlertManager.publishTaskFailListenerEvent(workflowInstance, taskInstance, projectUser);
// There are child nodes and the failure policy is: CONTINUE
if (workflowInstance.getFailureStrategy() == FailureStrategy.CONTINUE && DagHelper.haveAllNodeAfterNode(
taskInstance.getTaskCode(),
@ -725,6 +747,7 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
log.info("workflowStatue changed to :{}", workflowRunnableStatus);
}
if (workflowRunnableStatus == WorkflowRunnableStatus.INITIALIZE_QUEUE) {
processStart();
submitPostNode(null);
workflowRunnableStatus = WorkflowRunnableStatus.STARTED;
log.info("workflowStatue changed to :{}", workflowRunnableStatus);
@ -753,6 +776,9 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
processAlertManager.sendAlertProcessInstance(workflowInstance, getValidTaskList(), projectUser);
if (workflowInstance.getState().isSuccess()) {
processAlertManager.closeAlert(workflowInstance);
listenerEventAlertManager.publishProcessEndListenerEvent(workflowInstance, projectUser);
} else {
listenerEventAlertManager.publishProcessFailListenerEvent(workflowInstance, projectUser);
}
if (checkTaskQueue()) {
// release task group

7
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableFactory.java
View File

@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.exception.WorkflowCreateException;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
import org.apache.dolphinscheduler.service.command.CommandService;
import org.apache.dolphinscheduler.service.expand.CuringParamsService;
@ -69,6 +70,9 @@ public class WorkflowExecuteRunnableFactory {
@Autowired
private WorkflowExecuteContextFactory workflowExecuteContextFactory;
@Autowired
private ListenerEventAlertManager listenerEventAlertManager;
public Optional<WorkflowExecuteRunnable> createWorkflowExecuteRunnable(Command command) throws WorkflowCreateException {
try {
Optional<IWorkflowExecuteContext> workflowExecuteRunnableContextOptional =
@ -83,7 +87,8 @@ public class WorkflowExecuteRunnableFactory {
stateWheelExecuteThread,
curingGlobalParamsService,
taskInstanceDao,
defaultTaskExecuteRunnableFactory));
defaultTaskExecuteRunnableFactory,
listenerEventAlertManager));
} catch (Exception ex) {
throw new WorkflowCreateException("Create workflow execute runnable failed", ex);
}

7
dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnableTest.java
View File

@ -40,6 +40,7 @@ import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.graph.IWorkflowGraph;
import org.apache.dolphinscheduler.server.master.runner.execute.DefaultTaskExecuteRunnableFactory;
import org.apache.dolphinscheduler.server.master.runner.execute.TaskExecuteRunnable;
import org.apache.dolphinscheduler.service.alert.ListenerEventAlertManager;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.command.CommandService;
@ -102,6 +103,8 @@ public class WorkflowExecuteRunnableTest {
private WorkflowExecuteContextFactory workflowExecuteContextFactory;
private ListenerEventAlertManager listenerEventAlertManager;
@BeforeEach
public void init() throws Exception {
applicationContext = Mockito.mock(ApplicationContext.class);
@ -117,6 +120,7 @@ public class WorkflowExecuteRunnableTest {
taskDefinitionLogDao = Mockito.mock(TaskDefinitionLogDao.class);
defaultTaskExecuteRunnableFactory = Mockito.mock(DefaultTaskExecuteRunnableFactory.class);
workflowExecuteContextFactory = Mockito.mock(WorkflowExecuteContextFactory.class);
listenerEventAlertManager = Mockito.mock(ListenerEventAlertManager.class);
Map<String, String> cmdParam = new HashMap<>();
cmdParam.put(CMD_PARAM_COMPLEMENT_DATA_START_DATE, "2020-01-01 00:00:00");
@ -146,7 +150,8 @@ public class WorkflowExecuteRunnableTest {
stateWheelExecuteThread,
curingGlobalParamsService,
taskInstanceDao,
defaultTaskExecuteRunnableFactory));
defaultTaskExecuteRunnableFactory,
listenerEventAlertManager));
}
@Test

278
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/ListenerEventAlertManager.java Normal file
View File

@ -0,0 +1,278 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.service.alert;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.ListenerEventType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.ListenerEvent;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.entity.event.AbstractListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionCreatedListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionDeletedListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessDefinitionUpdatedListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessEndListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessFailListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ProcessStartListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.ServerDownListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.TaskEndListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.TaskFailListenerEvent;
import org.apache.dolphinscheduler.dao.entity.event.TaskStartListenerEvent;
import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ListenerEventMapper;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.collections4.CollectionUtils;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class ListenerEventAlertManager {
@Value("${alert.alarm-suppression.crash:60}")
private int crashAlarmSuppression;
@Autowired
private ListenerEventMapper listenerEventMapper;
@Autowired
private AlertPluginInstanceMapper alertPluginInstanceMapper;
public void publishServerDownListenerEvent(String host, String type) {
ServerDownListenerEvent event = new ServerDownListenerEvent();
event.setEventTime(new Date());
event.setHost(host);
event.setType(type);
this.saveEvent(event);
}
public void publishProcessDefinitionCreatedListenerEvent(User user,
ProcessDefinition processDefinition,
List<TaskDefinitionLog> taskDefinitionLogs,
List<ProcessTaskRelationLog> processTaskRelationLogs) {
ProcessDefinitionCreatedListenerEvent event = new ProcessDefinitionCreatedListenerEvent(processDefinition);
event.setUserName(user.getUserName());
event.setModifyBy(user.getUserName());
event.setTaskDefinitionLogs(taskDefinitionLogs);
event.setTaskRelationList(processTaskRelationLogs);
this.saveEvent(event);
}
public void publishProcessDefinitionUpdatedListenerEvent(User user, ProcessDefinition processDefinition,
List<TaskDefinitionLog> taskDefinitionLogs,
List<ProcessTaskRelationLog> processTaskRelationLogs) {
ProcessDefinitionUpdatedListenerEvent event = new ProcessDefinitionUpdatedListenerEvent(processDefinition);
event.setTaskDefinitionLogs(taskDefinitionLogs);
event.setTaskRelationList(processTaskRelationLogs);
event.setUserName(user.getUserName());
event.setModifyBy(user.getUserName());
this.saveEvent(event);
}
public void publishProcessDefinitionDeletedListenerEvent(User user, Project project,
ProcessDefinition processDefinition) {
ProcessDefinitionDeletedListenerEvent event = new ProcessDefinitionDeletedListenerEvent();
event.setProjectId(project.getId());
event.setProjectCode(project.getCode());
event.setProjectName(project.getName());
event.setOwner(processDefinition.getUserName());
event.setId(processDefinition.getId());
event.setCode(processDefinition.getCode());
event.setName(processDefinition.getName());
event.setEventTime(new Date());
event.setUserId(user.getId());
event.setModifiedBy(user.getUserName());
this.saveEvent(event);
}
public void publishProcessStartListenerEvent(ProcessInstance processInstance, ProjectUser projectUser) {
ProcessStartListenerEvent event = new ProcessStartListenerEvent();
event.setProjectCode(projectUser.getProjectCode());
event.setProjectName(projectUser.getProjectName());
event.setOwner(projectUser.getUserName());
event.setProcessId(processInstance.getId());
event.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
event.setProcessName(processInstance.getName());
event.setProcessType(processInstance.getCommandType());
event.setProcessState(processInstance.getState());
event.setRunTimes(processInstance.getRunTimes());
event.setRecovery(processInstance.getRecovery());
event.setProcessStartTime(processInstance.getStartTime());
this.saveEvent(event);
}
public void publishProcessEndListenerEvent(ProcessInstance processInstance, ProjectUser projectUser) {
ProcessEndListenerEvent event = new ProcessEndListenerEvent();
event.setProjectCode(projectUser.getProjectCode());
event.setProjectName(projectUser.getProjectName());
event.setOwner(projectUser.getUserName());
event.setProcessId(processInstance.getId());
event.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
event.setProcessName(processInstance.getName());
event.setProcessType(processInstance.getCommandType());
event.setProcessState(processInstance.getState());
event.setRecovery(processInstance.getRecovery());
event.setRunTimes(processInstance.getRunTimes());
event.setProcessStartTime(processInstance.getStartTime());
event.setProcessEndTime(processInstance.getEndTime());
event.setProcessHost(processInstance.getHost());
this.saveEvent(event);
}
public void publishProcessFailListenerEvent(ProcessInstance processInstance,
ProjectUser projectUser) {
ProcessFailListenerEvent event = new ProcessFailListenerEvent();
event.setProjectCode(projectUser.getProjectCode());
event.setProjectName(projectUser.getProjectName());
event.setOwner(projectUser.getUserName());
event.setProcessId(processInstance.getId());
event.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
event.setProcessName(processInstance.getName());
event.setProcessType(processInstance.getCommandType());
event.setProcessState(processInstance.getState());
event.setRecovery(processInstance.getRecovery());
event.setRunTimes(processInstance.getRunTimes());
event.setProcessStartTime(processInstance.getStartTime());
event.setProcessEndTime(processInstance.getEndTime());
event.setProcessHost(processInstance.getHost());
this.saveEvent(event);
}
public void publishTaskStartListenerEvent(ProcessInstance processInstance,
TaskInstance taskInstance,
ProjectUser projectUser) {
TaskStartListenerEvent event = new TaskStartListenerEvent();
event.setProjectCode(projectUser.getProjectCode());
event.setProjectName(projectUser.getProjectName());
event.setOwner(projectUser.getUserName());
event.setProcessId(processInstance.getId());
event.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
event.setProcessName(processInstance.getName());
event.setTaskCode(taskInstance.getTaskCode());
event.setTaskName(taskInstance.getName());
event.setTaskType(taskInstance.getTaskType());
event.setTaskState(taskInstance.getState());
event.setTaskStartTime(taskInstance.getStartTime());
event.setTaskEndTime(taskInstance.getEndTime());
event.setTaskHost(taskInstance.getHost());
event.setLogPath(taskInstance.getLogPath());
this.saveEvent(event);
}
public void publishTaskEndListenerEvent(ProcessInstance processInstance,
TaskInstance taskInstance,
ProjectUser projectUser) {
TaskEndListenerEvent event = new TaskEndListenerEvent();
event.setProjectCode(projectUser.getProjectCode());
event.setProjectName(projectUser.getProjectName());
event.setOwner(projectUser.getUserName());
event.setProcessId(processInstance.getId());
event.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
event.setProcessName(processInstance.getName());
event.setTaskCode(taskInstance.getTaskCode());
event.setTaskName(taskInstance.getName());
event.setTaskType(taskInstance.getTaskType());
event.setTaskState(taskInstance.getState());
event.setTaskStartTime(taskInstance.getStartTime());
event.setTaskEndTime(taskInstance.getEndTime());
event.setTaskHost(taskInstance.getHost());
event.setLogPath(taskInstance.getLogPath());
this.saveEvent(event);
}
public void publishTaskFailListenerEvent(ProcessInstance processInstance,
TaskInstance taskInstance,
ProjectUser projectUser) {
TaskFailListenerEvent event = new TaskFailListenerEvent();
event.setProjectCode(projectUser.getProjectCode());
event.setProjectName(projectUser.getProjectName());
event.setOwner(projectUser.getUserName());
event.setProcessId(processInstance.getId());
event.setProcessDefinitionCode(processInstance.getProcessDefinitionCode());
event.setProcessName(processInstance.getName());
event.setTaskCode(taskInstance.getTaskCode());
event.setTaskName(taskInstance.getName());
event.setTaskType(taskInstance.getTaskType());
event.setTaskState(taskInstance.getState());
event.setTaskStartTime(taskInstance.getStartTime());
event.setTaskEndTime(taskInstance.getEndTime());
event.setTaskHost(taskInstance.getHost());
event.setLogPath(taskInstance.getLogPath());
this.saveEvent(event);
}
private void saveEvent(AbstractListenerEvent event) {
if (!needSendGlobalListenerEvent()) {
return;
}
ListenerEvent listenerEvent = new ListenerEvent();
String content = JSONUtils.toJsonString(event);
listenerEvent.setContent(content);
listenerEvent.setPostStatus(AlertStatus.WAIT_EXECUTION);
listenerEvent.setSign(generateSign(content));
listenerEvent.setCreateTime(new Date());
listenerEvent.setUpdateTime(new Date());
listenerEvent.setEventType(event.getEventType());
if (event.getEventType() == ListenerEventType.SERVER_DOWN) {
saveServerDownEvent(listenerEvent);
} else {
saveNormalEvent(listenerEvent);
}
}
private void saveNormalEvent(ListenerEvent listenerEvent) {
int insert = listenerEventMapper.insert(listenerEvent);
if (insert < 1) {
log.error("insert listener event failed: {}", listenerEvent);
}
}
private void saveServerDownEvent(ListenerEvent listenerEvent) {
Date crashAlarmSuppressionStartTime = Date.from(
LocalDateTime.now().plusMinutes(-crashAlarmSuppression).atZone(ZoneId.systemDefault()).toInstant());
listenerEventMapper.insertServerDownEvent(listenerEvent, crashAlarmSuppressionStartTime);
}
private String generateSign(String content) {
return DigestUtils.sha256Hex(content).toLowerCase();
}
private boolean needSendGlobalListenerEvent() {
List<AlertPluginInstance> globalPluginInstanceList =
alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList();
return CollectionUtils.isNotEmpty(globalPluginInstanceList);
}
}

148
dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/ListenerEventAlertManagerTest.java Normal file
View File

@ -0,0 +1,148 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.service.alert;
import static org.mockito.ArgumentMatchers.any;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ListenerEventMapper;
import java.util.ArrayList;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* ProcessAlertManager Test
*/
@ExtendWith(MockitoExtension.class)
public class ListenerEventAlertManagerTest {
private static final Logger logger = LoggerFactory.getLogger(ListenerEventAlertManagerTest.class);
@InjectMocks
ListenerEventAlertManager listenerEventAlertManager;
@Mock
AlertPluginInstanceMapper alertPluginInstanceMapper;
@Mock
ListenerEventMapper listenerEventMapper;
@Test
public void sendServerDownListenerEventTest() {
String host = "127.0.0.1";
String type = "WORKER";
List<AlertPluginInstance> globalPluginInstanceList = new ArrayList<>();
AlertPluginInstance instance = new AlertPluginInstance(1, "instanceParams", "instanceName");
globalPluginInstanceList.add(instance);
Mockito.when(alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList())
.thenReturn(globalPluginInstanceList);
Mockito.doNothing().when(listenerEventMapper).insertServerDownEvent(any(), any());
listenerEventAlertManager.publishServerDownListenerEvent(host, type);
}
@Test
public void sendProcessDefinitionCreatedListenerEvent() {
User user = Mockito.mock(User.class);
ProcessDefinition processDefinition = Mockito.mock(ProcessDefinition.class);
List<TaskDefinitionLog> taskDefinitionLogs = new ArrayList<>();
List<ProcessTaskRelationLog> processTaskRelationLogs = new ArrayList<>();
AlertPluginInstance instance = new AlertPluginInstance(1, "instanceParams", "instanceName");
List<AlertPluginInstance> globalPluginInstanceList = new ArrayList<>();
globalPluginInstanceList.add(instance);
Mockito.when(alertPluginInstanceMapper.queryAllGlobalAlertPluginInstanceList())
.thenReturn(globalPluginInstanceList);
Mockito.when(listenerEventMapper.insert(any())).thenReturn(1);
listenerEventAlertManager.publishProcessDefinitionCreatedListenerEvent(user, processDefinition,
taskDefinitionLogs, processTaskRelationLogs);
}
@Test
public void sendProcessDefinitionUpdatedListenerEvent() {
User user = new User();
ProcessDefinition processDefinition = new ProcessDefinition();
List<TaskDefinitionLog> taskDefinitionLogs = new ArrayList<>();
List<ProcessTaskRelationLog> processTaskRelationLogs = new ArrayList<>();
listenerEventAlertManager.publishProcessDefinitionUpdatedListenerEvent(user, processDefinition,
taskDefinitionLogs, processTaskRelationLogs);
}
@Test
public void sendProcessDefinitionDeletedListenerEvent() {
User user = new User();
Project project = new Project();
ProcessDefinition processDefinition = new ProcessDefinition();
listenerEventAlertManager.publishProcessDefinitionDeletedListenerEvent(user, project, processDefinition);
}
@Test
public void sendProcessStartListenerEvent() {
ProcessInstance processInstance = new ProcessInstance();
ProjectUser projectUser = new ProjectUser();
listenerEventAlertManager.publishProcessStartListenerEvent(processInstance, projectUser);
}
@Test
public void sendProcessEndListenerEvent() {
ProcessInstance processInstance = new ProcessInstance();
ProjectUser projectUser = new ProjectUser();
listenerEventAlertManager.publishProcessEndListenerEvent(processInstance, projectUser);
}
@Test
public void sendProcessFailListenerEvent() {
ProcessInstance processInstance = new ProcessInstance();
ProjectUser projectUser = new ProjectUser();
listenerEventAlertManager.publishProcessFailListenerEvent(processInstance, projectUser);
}
@Test
public void sendTaskStartListenerEvent() {
ProcessInstance processInstance = Mockito.mock(ProcessInstance.class);
TaskInstance taskInstance = Mockito.mock(TaskInstance.class);
ProjectUser projectUser = Mockito.mock(ProjectUser.class);
listenerEventAlertManager.publishTaskStartListenerEvent(processInstance, taskInstance, projectUser);
}
@Test
public void sendTaskEndListenerEvent() {
ProcessInstance processInstance = Mockito.mock(ProcessInstance.class);
TaskInstance taskInstance = Mockito.mock(TaskInstance.class);
ProjectUser projectUser = Mockito.mock(ProjectUser.class);
listenerEventAlertManager.publishTaskEndListenerEvent(processInstance, taskInstance, projectUser);
}
@Test
public void sendTaskFailListenerEvent() {
ProcessInstance processInstance = Mockito.mock(ProcessInstance.class);
TaskInstance taskInstance = Mockito.mock(TaskInstance.class);
ProjectUser projectUser = Mockito.mock(ProjectUser.class);
listenerEventAlertManager.publishTaskFailListenerEvent(processInstance, taskInstance, projectUser);
}
}

1
dolphinscheduler-standalone-server/src/main/resources/application.yaml
View File

@ -223,6 +223,7 @@ alert:
# Define value is (0 = infinite), and alert server would be waiting alert result.
wait-timeout: 0
heartbeat-interval: 60s
query_alert_threshold: 100
api:
audit-enable: false

2
dolphinscheduler-ui/src/locales/en_US/security.ts
View File

@ -204,6 +204,8 @@ export default {
alarm_instance_name: 'Alarm instance name',
alarm_instance_name_tips: 'Please enter alarm plugin instance name',
alarm_plugin_name: 'Alarm plugin name',
alarm_instance_type: 'Alarm instance type',
is_global_instance: 'Is Global Instance',
create_time: 'Create Time',
update_time: 'Update Time',
operation: 'Operation',

2
dolphinscheduler-ui/src/locales/zh_CN/security.ts
View File

@ -201,6 +201,8 @@ export default {
alarm_instance_name: '告警实例名称',
alarm_instance_name_tips: '请输入告警实例名称',
alarm_plugin_name: '告警插件名称',
alarm_instance_type: '告警实例类型',
is_global_instance: '是否全局告警实例',
create_time: '创建时间',
update_time: '更新时间',
operation: '操作',

7
dolphinscheduler-ui/src/service/modules/alert-group/index.ts
View File

@ -41,6 +41,13 @@ export function listAlertGroupById(): any {
})
}
export function listNormalAlertGroupById(): any {
return axios({
url: '/alert-groups/normal-list',
method: 'get'
})
}
export function queryAlertGroupById(data: IdReq): any {
return axios({
url: '/alert-groups/query',

3
dolphinscheduler-ui/src/service/modules/alert-plugin/types.ts
View File

@ -24,6 +24,8 @@ interface ListReq {
interface PluginInstanceReq {
instanceName: string
pluginDefineId: number
instanceType: string
warningType: string
pluginInstanceParams: string
}
@ -34,6 +36,7 @@ interface InstanceNameReq {
interface UpdatePluginInstanceReq {
alertPluginInstanceId: number
instanceName: string
warningType: string
pluginInstanceParams: string
}

4
dolphinscheduler-ui/src/views/projects/preference/components/use-alert-group.ts
View File

@ -18,7 +18,7 @@
import { ref, onMounted } from 'vue'
import { useI18n } from 'vue-i18n'
import type { IJsonItem } from '../../task/components/node/types'
import { listAlertGroupById } from '@/service/modules/alert-group'
import { listNormalAlertGroupById } from '@/service/modules/alert-group'
export function useAlertGroup(): IJsonItem {
const { t } = useI18n()
@ -26,7 +26,7 @@ export function useAlertGroup(): IJsonItem {
const options = ref([] as { label: string; value: string }[])
const getAlertGroups = async () => {
const res = await listAlertGroupById()
const res = await listNormalAlertGroupById()
options.value = res.map((item: any) => ({
label: item.groupName,
value: item.id

4
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts
View File

@ -17,7 +17,7 @@
import { ref, onMounted, computed, h } from 'vue'
import { useI18n } from 'vue-i18n'
import { listAlertGroupById } from '@/service/modules/alert-group'
import { listNormalAlertGroupById } from '@/service/modules/alert-group'
import styles from '../index.module.scss'
import type { IJsonItem } from '../types'
@ -43,7 +43,7 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] {
const getGroups = async () => {
if (groupsLoading.value) return
groupsLoading.value = true
const res = await listAlertGroupById()
const res = await listNormalAlertGroupById()
groups.value = res.map((item: { id: number; groupName: string }) => ({
label: item.groupName,
value: item.id

4
dolphinscheduler-ui/src/views/projects/task/definition/components/use-start.tsx
View File

@ -24,7 +24,7 @@ import { queryProcessDefinitionByCode } from '@/service/modules/process-definiti
import { queryAllWorkerGroups } from '@/service/modules/worker-groups'
import { queryTenantList } from '@/service/modules/tenants'
import { queryAllEnvironmentList } from '@/service/modules/environment'
import { listAlertGroupById } from '@/service/modules/alert-group'
import { listNormalAlertGroupById } from '@/service/modules/alert-group'
import type { EnvironmentItem } from '@/service/modules/environment/types'
import type { IStartState } from '../types'
@ -90,7 +90,7 @@ export const useStart = (
}
const getAlertGroups = () => {
listAlertGroupById().then((res: any) => {
listNormalAlertGroupById().then((res: any) => {
variables.startState.alertGroups = res.map((item: any) => ({
label: item.groupName,
value: item.id

4
dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag-startup-param.tsx
View File

@ -18,7 +18,7 @@
import _ from 'lodash'
import { defineComponent, onMounted, PropType, ref, computed } from 'vue'
import { useI18n } from 'vue-i18n'
import { listAlertGroupById } from '@/service/modules/alert-group'
import { listNormalAlertGroupById } from '@/service/modules/alert-group'
import { queryAllWorkerGroups } from '@/service/modules/worker-groups'
import { runningType, warningTypeList } from '@/common/common'
import { IStartupParam } from './types'
@ -42,7 +42,7 @@ export default defineComponent({
const commandParam = JSON.parse(props.startupParam?.commandParam || '{}')
const getAlertGroupList = () => {
listAlertGroupById().then((res: any) => {
listNormalAlertGroupById().then((res: any) => {
alertGroupListRef.value = res
})
}

4
dolphinscheduler-ui/src/views/projects/workflow/definition/components/use-modal.ts
View File

@ -28,7 +28,7 @@ import {
} from '@/service/modules/process-definition'
import { queryAllWorkerGroups } from '@/service/modules/worker-groups'
import { queryAllEnvironmentList } from '@/service/modules/environment'
import { listAlertGroupById } from '@/service/modules/alert-group'
import { listNormalAlertGroupById } from '@/service/modules/alert-group'
import { startProcessInstance } from '@/service/modules/executors'
import {
createSchedule,
@ -256,7 +256,7 @@ export function useModal(
}
const getAlertGroups = () => {
listAlertGroupById().then((res: any) => {
listNormalAlertGroupById().then((res: any) => {
variables.alertGroups = res.map((item: any) => ({
label: item.groupName,
value: item.id

41
dolphinscheduler-ui/src/views/security/alarm-instance-manage/detail.tsx
View File

@ -23,7 +23,7 @@ import {
ref,
getCurrentInstance
} from 'vue'
import { NSelect, NInput } from 'naive-ui'
import { NSelect, NInput, NSwitch, NRadioGroup, NSpace, NRadio } from 'naive-ui'
import { isFunction } from 'lodash'
import { useI18n } from 'vue-i18n'
import { useForm } from './use-form'
@ -57,6 +57,7 @@ const DetailModal = defineComponent({
const rules = ref<IFormRules>({})
const elements = ref<IFormItem[]>([]) as IElements
const warningTypeSpan = ref(24)
const {
meta,
@ -95,6 +96,10 @@ const DetailModal = defineComponent({
props.show && props.currentRecord && setDetail(props.currentRecord)
}
)
watch(
() => state.detailForm.instanceType,
() => warningTypeSpan.value = state.detailForm.instanceType === 'GLOBAL' ? 0 : 24
)
watch(
() => state.json,
() => {
@ -121,6 +126,7 @@ const DetailModal = defineComponent({
...toRefs(state),
...toRefs(status),
meta,
warningTypeSpan,
rules,
elements,
onChangePlugin,
@ -134,6 +140,7 @@ const DetailModal = defineComponent({
show,
t,
meta,
warningTypeSpan,
rules,
elements,
detailForm,
@ -183,6 +190,38 @@ const DetailModal = defineComponent({
/>
)
},
{
path: 'instanceType',
label: t('security.alarm_instance.is_global_instance'),
widget: (
<NSwitch
checkedValue={'GLOBAL'}
uncheckedValue={'NORMAL'}
disabled={!!currentRecord?.id}
v-model:value={detailForm.instanceType}
/>
)
},
{
path: 'warningType',
label: t('security.alarm_instance.WarningType'),
span: warningTypeSpan,
widget: (
<NRadioGroup v-model:value={detailForm.warningType}>
<NSpace>
<NRadio value={'SUCCESS'}>
{"success"}
</NRadio>
<NRadio value={'FAILURE'} >
{"failure"}
</NRadio>
<NRadio value={'ALL'} >
{"all"}
</NRadio>
</NSpace>
</NRadioGroup>
)
},
{
path: 'pluginDefineId',
label: t('security.alarm_instance.select_plugin'),

2
dolphinscheduler-ui/src/views/security/alarm-instance-manage/types.ts
View File

@ -29,6 +29,8 @@ interface IRecord {
createTime?: string
id: number
instanceName: string
instanceType: string
warningType: string
pluginDefineId: number
pluginInstanceParams?: string
updateTime?: string

4
dolphinscheduler-ui/src/views/security/alarm-instance-manage/use-columns.ts
View File

@ -35,6 +35,10 @@ export function useColumns(onCallback: Function) {
title: t('security.alarm_instance.alarm_instance_name'),
key: 'instanceName'
},
{
title: t('security.alarm_instance.alarm_instance_type'),
key: 'instanceType'
},
{
title: t('security.alarm_instance.alarm_plugin_name'),
key: 'alertPluginName'

3
dolphinscheduler-ui/src/views/security/alarm-instance-manage/use-detail.ts
View File

@ -58,6 +58,7 @@ export function useDetail(getFormValues: Function) {
{
alertPluginInstanceId: values.pluginDefineId,
instanceName: values.instanceName,
warningType: values.warningType,
pluginInstanceParams: formatParams(json, values)
},
currentRecord.id
@ -65,6 +66,8 @@ export function useDetail(getFormValues: Function) {
: await createAlertPluginInstance({
instanceName: values.instanceName,
pluginDefineId: values.pluginDefineId,
instanceType: values.instanceType,
warningType: values.warningType,
pluginInstanceParams: formatParams(json, values)
})

8
dolphinscheduler-ui/src/views/security/alarm-instance-manage/use-form.ts
View File

@ -35,7 +35,9 @@ export function useForm() {
const initialValues = {
instanceName: '',
pluginDefineId: null
pluginDefineId: null,
instanceType: 'NORMAL',
warningType: 'ALL'
}
const state = reactive({
@ -44,7 +46,7 @@ export function useForm() {
uiPlugins: [],
pluginsLoading: false,
json: []
} as { detailFormRef: Ref; json: IJsonItem[]; detailForm: { instanceName: string; pluginDefineId: number | null }; pluginsLoading: boolean; uiPlugins: [] })
} as { detailFormRef: Ref; json: IJsonItem[]; detailForm: { instanceName: string; pluginDefineId: number | null; instanceType: string; warningType: string }; pluginsLoading: boolean; uiPlugins: [] })
const meta = {
model: state.detailForm,
@ -109,6 +111,8 @@ export function useForm() {
const setDetail = (record: IRecord) => {
state.detailForm.instanceName = record.instanceName
state.detailForm.pluginDefineId = record.pluginDefineId
state.detailForm.instanceType = record.instanceType
state.detailForm.warningType = record.warningType
if (record.pluginInstanceParams)
state.json = JSON.parse(record.pluginInstanceParams)
}