Browse Source

Merge pull request #4616 from CalvinKirs/alert_param_reduce

[Improvement][alert-spi]plugin instance only saves the main information
pull/3/MERGE
gaojun2048 3 years ago committed by GitHub
parent
commit
947dcb90cc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 7
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-dingtalk/src/main/java/org/apache/dolphinscheduler/plugin/alert/dingtalk/DingTalkAlertChannel.java
  2. 7
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-email/src/main/java/org/apache/dolphinscheduler/plugin/alert/email/EmailAlertChannel.java
  3. 23
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-email/src/test/java/org/apache/dolphinscheduler/plugin/alert/email/EmailAlertChannelFactoryTest.java
  4. 6
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-email/src/test/java/org/apache/dolphinscheduler/plugin/alert/email/EmailAlertChannelTest.java
  5. 7
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-feishu/src/main/java/org/apache/dolphinscheduler/plugin/alert/feishu/FeiShuAlertChannel.java
  6. 7
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-http/src/main/java/org/apache/dolphinscheduler/plugin/alert/http/HttpAlertChannel.java
  7. 7
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-http/src/test/java/org/apache/dolphinscheduler/plugin/alert/http/HttpAlertChannelTest.java
  8. 7
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-script/src/main/java/org/apache/dolphinscheduler/plugin/alert/script/ScriptAlertChannel.java
  9. 2
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-script/src/test/java/org/apache/dolphinscheduler/plugin/alert/script/ScriptAlertChannelFactoryTest.java
  10. 7
      dolphinscheduler-alert-plugin/dolphinscheduler-alert-wechat/src/main/java/org/apache/dolphinscheduler/plugin/alert/wechat/WeChatAlertChannel.java
  11. 7
      dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/AlertServer.java
  12. 55
      dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/manager/EmailManager.java
  13. 40
      dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/manager/MsgManager.java
  14. 13
      dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/plugin/AlertPluginManager.java
  15. 9
      dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessor.java
  16. 26
      dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/runner/AlertSender.java
  17. 9
      dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/AlertServerTest.java
  18. 2
      dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/plugin/EmailAlertPluginTest.java
  19. 11
      dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessorTest.java
  20. 36
      dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java
  21. 65
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java
  22. 109
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertPluginInstanceServiceTest.java
  23. 18
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/PluginDao.java
  24. 8
      dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/alert/AlertInfo.java
  25. 8
      dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/alert/AlertResult.java
  26. 4
      dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/params/base/PluginParams.java

7
dolphinscheduler-alert-plugin/dolphinscheduler-alert-dingtalk/src/main/java/org/apache/dolphinscheduler/plugin/alert/dingtalk/DingTalkAlertChannel.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import java.util.Map;
@ -34,8 +33,10 @@ public class DingTalkAlertChannel implements AlertChannel {
public AlertResult process(AlertInfo alertInfo) {
AlertData alertData = alertInfo.getAlertData();
String alertParams = alertInfo.getAlertParams();
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(alertParams);
Map<String, String> paramsMap = alertInfo.getAlertParams();
if (null == paramsMap) {
return new AlertResult("false", "ding talk params is null");
}
return new DingTalkSender(paramsMap).sendDingTalkMsg(alertData.getTitle(), alertData.getContent());
}
}

7
dolphinscheduler-alert-plugin/dolphinscheduler-alert-email/src/main/java/org/apache/dolphinscheduler/plugin/alert/email/EmailAlertChannel.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import java.util.Map;
@ -38,8 +37,10 @@ public class EmailAlertChannel implements AlertChannel {
public AlertResult process(AlertInfo info) {
AlertData alert = info.getAlertData();
String alertParams = info.getAlertParams();
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(alertParams);
Map<String, String> paramsMap = info.getAlertParams();
if (null == paramsMap) {
return new AlertResult("false", "mail params is null");
}
MailSender mailSender = new MailSender(paramsMap);
AlertResult alertResult = mailSender.sendMails(alert.getTitle(), alert.getContent());

23
dolphinscheduler-alert-plugin/dolphinscheduler-alert-email/src/test/java/org/apache/dolphinscheduler/plugin/alert/email/EmailAlertChannelFactoryTest.java

@ -19,13 +19,10 @@ package org.apache.dolphinscheduler.plugin.alert.email;
import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import java.util.List;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
/**
@ -36,29 +33,13 @@ import org.junit.Test;
*/
public class EmailAlertChannelFactoryTest {
@Before
public void before() throws Exception {
}
@After
public void after() throws Exception {
}
/**
* Method: getName()
*/
@Test
public void testGetName() throws Exception {
}
/**
* Method: getParams()
*/
@Test
public void testGetParams() throws Exception {
public void testGetParams() {
EmailAlertChannelFactory emailAlertChannelFactory = new EmailAlertChannelFactory();
List<PluginParams> params = emailAlertChannelFactory.getParams();
System.out.println(JSONUtils.toJsonString(params));
Assert.assertEquals(12, params.size());
}
@ -66,7 +47,7 @@ public class EmailAlertChannelFactoryTest {
* Method: create()
*/
@Test
public void testCreate() throws Exception {
public void testCreate() {
EmailAlertChannelFactory emailAlertChannelFactory = new EmailAlertChannelFactory();
AlertChannel alertChannel = emailAlertChannelFactory.create();
Assert.assertNotNull(alertChannel);

6
dolphinscheduler-alert-plugin/dolphinscheduler-alert-email/src/test/java/org/apache/dolphinscheduler/plugin/alert/email/EmailAlertChannelTest.java

@ -24,6 +24,7 @@ import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.alert.ShowType;
import org.apache.dolphinscheduler.spi.params.InputParam;
import org.apache.dolphinscheduler.spi.params.PasswordParam;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import org.apache.dolphinscheduler.spi.params.RadioParam;
import org.apache.dolphinscheduler.spi.params.base.DataType;
import org.apache.dolphinscheduler.spi.params.base.ParamsOptions;
@ -34,6 +35,7 @@ import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
@ -66,7 +68,9 @@ public class EmailAlertChannelTest {
.setTitle("test");
AlertInfo alertInfo = new AlertInfo();
alertInfo.setAlertData(alertData);
alertInfo.setAlertParams(getEmailAlertParams());
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(getEmailAlertParams());
alertInfo.setAlertParams(paramsMap);
AlertResult alertResult = emailAlertChannel.process(alertInfo);
Assert.assertNotNull(alertResult);
Assert.assertEquals("false", alertResult.getStatus());

7
dolphinscheduler-alert-plugin/dolphinscheduler-alert-feishu/src/main/java/org/apache/dolphinscheduler/plugin/alert/feishu/FeiShuAlertChannel.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import java.util.Map;
@ -30,8 +29,10 @@ public class FeiShuAlertChannel implements AlertChannel {
public AlertResult process(AlertInfo alertInfo) {
AlertData alertData = alertInfo.getAlertData();
String alertParams = alertInfo.getAlertParams();
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(alertParams);
Map<String, String> paramsMap = alertInfo.getAlertParams();
if (null == paramsMap) {
return new AlertResult("false", "fei shu params is null");
}
return new FeiShuSender(paramsMap).sendFeiShuMsg(alertData);
}
}

7
dolphinscheduler-alert-plugin/dolphinscheduler-alert-http/src/main/java/org/apache/dolphinscheduler/plugin/alert/http/HttpAlertChannel.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import java.util.Map;
@ -33,8 +32,10 @@ public class HttpAlertChannel implements AlertChannel {
public AlertResult process(AlertInfo alertInfo) {
AlertData alertData = alertInfo.getAlertData();
String alertParams = alertInfo.getAlertParams();
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(alertParams);
Map<String, String> paramsMap = alertInfo.getAlertParams();
if (null == paramsMap) {
return new AlertResult("false", "http params is null");
}
return new HttpSender(paramsMap).send(alertData.getContent());
}

7
dolphinscheduler-alert-plugin/dolphinscheduler-alert-http/src/test/java/org/apache/dolphinscheduler/plugin/alert/http/HttpAlertChannelTest.java

@ -21,12 +21,14 @@ import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.params.InputParam;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import org.apache.dolphinscheduler.spi.params.base.Validate;
import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
@ -45,7 +47,7 @@ public class HttpAlertChannelTest {
alertData.setContent("Fault tolerance warning");
alertInfo.setAlertData(alertData);
AlertResult alertResult = alertChannel.process(alertInfo);
Assert.assertEquals("Request types are not supported", alertResult.getMessage());
Assert.assertEquals("http params is null", alertResult.getMessage());
}
@Test
@ -56,7 +58,8 @@ public class HttpAlertChannelTest {
AlertData alertData = new AlertData();
alertData.setContent("Fault tolerance warning");
alertInfo.setAlertData(alertData);
alertInfo.setAlertParams(getParams());
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(getParams());
alertInfo.setAlertParams(paramsMap);
AlertResult alertResult = alertChannel.process(alertInfo);
Assert.assertEquals("true", alertResult.getStatus());
}

7
dolphinscheduler-alert-plugin/dolphinscheduler-alert-script/src/main/java/org/apache/dolphinscheduler/plugin/alert/script/ScriptAlertChannel.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import java.util.Map;
@ -33,8 +32,10 @@ public class ScriptAlertChannel implements AlertChannel {
@Override
public AlertResult process(AlertInfo alertinfo) {
AlertData alertData = alertinfo.getAlertData();
String alertParams = alertinfo.getAlertParams();
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(alertParams);
Map<String, String> paramsMap = alertinfo.getAlertParams();
if (null == paramsMap) {
return new AlertResult("false", "ding talk params is null");
}
return new ScriptSender(paramsMap).sendScriptAlert(alertData.getTitle());
}
}

2
dolphinscheduler-alert-plugin/dolphinscheduler-alert-script/src/test/java/org/apache/dolphinscheduler/plugin/alert/script/ScriptAlertChannelFactoryTest.java

@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.plugin.alert.script;
import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import java.util.List;
@ -35,7 +34,6 @@ public class ScriptAlertChannelFactoryTest {
public void testGetParams() {
ScriptAlertChannelFactory scriptAlertChannelFactory = new ScriptAlertChannelFactory();
List<PluginParams> params = scriptAlertChannelFactory.getParams();
JSONUtils.toJsonString(params);
Assert.assertEquals(3, params.size());
}

7
dolphinscheduler-alert-plugin/dolphinscheduler-alert-wechat/src/main/java/org/apache/dolphinscheduler/plugin/alert/wechat/WeChatAlertChannel.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.spi.alert.AlertChannel;
import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import java.util.Map;
@ -33,8 +32,10 @@ public class WeChatAlertChannel implements AlertChannel {
@Override
public AlertResult process(AlertInfo info) {
AlertData alertData = info.getAlertData();
String alertParams = info.getAlertParams();
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(alertParams);
Map<String, String> paramsMap = info.getAlertParams();
if (null == paramsMap) {
return new AlertResult("false", "we chat params is null");
}
return new WeChatSender(paramsMap).sendEnterpriseWeChat(alertData.getTitle(), alertData.getContent());
}

7
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/AlertServer.java

@ -29,7 +29,6 @@ import org.apache.dolphinscheduler.alert.utils.PropertyUtils;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.DaoFactory;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.remote.NettyRemotingServer;
import org.apache.dolphinscheduler.remote.command.CommandType;
@ -53,8 +52,6 @@ public class AlertServer {
*/
private AlertDao alertDao = DaoFactory.getDaoInstance(AlertDao.class);
private PluginDao pluginDao = DaoFactory.getDaoInstance(PluginDao.class);
private AlertSender alertSender;
private static AlertServer instance;
@ -114,7 +111,7 @@ public class AlertServer {
NettyServerConfig serverConfig = new NettyServerConfig();
serverConfig.setListenPort(ALERT_RPC_PORT);
this.server = new NettyRemotingServer(serverConfig);
this.server.registerProcessor(CommandType.ALERT_SEND_REQUEST, new AlertRequestProcessor(alertDao, alertPluginManager, pluginDao));
this.server.registerProcessor(CommandType.ALERT_SEND_REQUEST, new AlertRequestProcessor(alertDao, alertPluginManager));
this.server.start();
}
@ -133,7 +130,7 @@ public class AlertServer {
logger.warn("No Alert Plugin . Can not send alert info. ");
} else {
List<Alert> alerts = alertDao.listWaitExecutionAlert();
alertSender = new AlertSender(alerts, alertDao, alertPluginManager, pluginDao);
alertSender = new AlertSender(alerts, alertDao, alertPluginManager);
alertSender.run();
}
}

55
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/manager/EmailManager.java

@ -1,55 +0,0 @@
///*
// * 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.manager;
//
//import org.apache.dolphinscheduler.alert.utils.MailUtils;
//
//import java.util.List;
//import java.util.Map;
//
///**
// * email send manager
// */
//public class EmailManager {
// /**
// * email send
// * @param receiversList the receiver list
// * @param receiversCcList the cc List
// * @param title the title
// * @param content the content
// * @param showType the showType
// * @return the send result
// */
// public Map<String,Object> send(List<String> receiversList,List<String> receiversCcList,String title,String content,String showType){
//
// return MailUtils.sendMails(receiversList, receiversCcList, title, content, showType);
// }
//
// /**
// * msg send
// * @param receiversList the receiver list
// * @param title the title
// * @param content the content
// * @param showType the showType
// * @return the send result
// */
// public Map<String,Object> send(List<String> receiversList,String title,String content,String showType){
//
// return MailUtils.sendMails(receiversList,title, content, showType);
// }
//
//}

40
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/manager/MsgManager.java

@ -1,40 +0,0 @@
/*
* 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.manager;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* SMS send manager
*/
public class MsgManager {
private static final Logger logger = LoggerFactory.getLogger(MsgManager.class);
/**
* SMS send
*
* @param alert the alert
*/
public void send(Alert alert) {
logger.info("send message {}", alert);
}
}

13
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/plugin/AlertPluginManager.java

@ -31,6 +31,7 @@ import org.apache.dolphinscheduler.spi.classloader.ThreadContextClassLoader;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ -47,6 +48,11 @@ public class AlertPluginManager extends AbstractDolphinPluginManager {
private final Map<String, AlertChannelFactory> alertChannelFactoryMap = new ConcurrentHashMap<>();
private final Map<String, AlertChannel> alertChannelMap = new ConcurrentHashMap<>();
/**
* k->pluginDefineId v->pluginDefineName
*/
private final Map<Integer, String> pluginDefineMap = new HashMap<>();
public void addAlertChannelFactory(AlertChannelFactory alertChannelFactory) {
requireNonNull(alertChannelFactory, "alertChannelFactory is null");
@ -83,6 +89,10 @@ public class AlertPluginManager extends AbstractDolphinPluginManager {
return alertChannelMap;
}
public String getPluginNameById(int id) {
return pluginDefineMap.get(id);
}
@Override
public void installPlugin(DolphinSchedulerPlugin dolphinSchedulerPlugin) {
for (AlertChannelFactory alertChannelFactory : dolphinSchedulerPlugin.getAlertChannelFactorys()) {
@ -93,7 +103,8 @@ public class AlertPluginManager extends AbstractDolphinPluginManager {
String paramsJson = PluginParamsTransfer.transferParamsToJson(params);
PluginDefine pluginDefine = new PluginDefine(nameEn, PluginType.ALERT.getDesc(), paramsJson);
pluginDao.addOrUpdatePluginDefine(pluginDefine);
int id = pluginDao.addOrUpdatePluginDefine(pluginDefine);
pluginDefineMap.put(id, pluginDefine.getPluginName());
}
}
}

9
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessor.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
import org.apache.dolphinscheduler.alert.runner.AlertSender;
import org.apache.dolphinscheduler.common.utils.Preconditions;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendRequestCommand;
@ -35,18 +34,16 @@ import org.slf4j.LoggerFactory;
import io.netty.channel.Channel;
/**
* alert request processor
* alert request processor
*/
public class AlertRequestProcessor implements NettyRequestProcessor {
private final Logger logger = LoggerFactory.getLogger(AlertRequestProcessor.class);
private AlertDao alertDao;
private PluginDao pluginDao;
private AlertPluginManager alertPluginManager;
public AlertRequestProcessor(AlertDao alertDao, AlertPluginManager alertPluginManager, PluginDao pluginDao) {
public AlertRequestProcessor(AlertDao alertDao, AlertPluginManager alertPluginManager) {
this.alertDao = alertDao;
this.pluginDao = pluginDao;
this.alertPluginManager = alertPluginManager;
}
@ -59,7 +56,7 @@ public class AlertRequestProcessor implements NettyRequestProcessor {
command.getBody(), AlertSendRequestCommand.class);
logger.info("received command : {}", alertSendRequestCommand);
AlertSender alertSender = new AlertSender(alertDao, alertPluginManager, pluginDao);
AlertSender alertSender = new AlertSender(alertDao, alertPluginManager);
AlertSendResponseCommand alertSendResponseCommand = alertSender.syncHandler(alertSendRequestCommand.getGroupId(), alertSendRequestCommand.getTitle(), alertSendRequestCommand.getContent());
channel.writeAndFlush(alertSendResponseCommand.convert2Command(command.getOpaque()));

26
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/runner/AlertSender.java

@ -20,8 +20,8 @@ package org.apache.dolphinscheduler.alert.runner;
import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendResponseCommand;
@ -33,6 +33,7 @@ import org.apache.dolphinscheduler.spi.alert.AlertResult;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -46,25 +47,22 @@ public class AlertSender {
private List<Alert> alertList;
private AlertDao alertDao;
private PluginDao pluginDao;
private AlertPluginManager alertPluginManager;
public AlertSender(AlertPluginManager alertPluginManager) {
this.alertPluginManager = alertPluginManager;
}
public AlertSender(AlertDao alertDao, AlertPluginManager alertPluginManager, PluginDao pluginDao) {
public AlertSender(AlertDao alertDao, AlertPluginManager alertPluginManager) {
super();
this.alertDao = alertDao;
this.pluginDao = pluginDao;
this.alertPluginManager = alertPluginManager;
}
public AlertSender(List<Alert> alertList, AlertDao alertDao, AlertPluginManager alertPluginManager, PluginDao pluginDao) {
public AlertSender(List<Alert> alertList, AlertDao alertDao, AlertPluginManager alertPluginManager) {
super();
this.alertList = alertList;
this.alertDao = alertDao;
this.pluginDao = pluginDao;
this.alertPluginManager = alertPluginManager;
}
@ -108,7 +106,7 @@ public class AlertSender {
List<AlertPluginInstance> alertInstanceList = alertDao.listInstanceByAlertGroupId(alertGroupId);
AlertData alertData = new AlertData();
alertData.setContent(content)
.setTitle(title);
.setTitle(title);
boolean sendResponseStatus = true;
List<AlertSendResponseResult> sendResponseResults = new ArrayList<>();
@ -143,7 +141,7 @@ public class AlertSender {
* @return AlertResult
*/
private AlertResult alertResultHandler(AlertPluginInstance instance, AlertData alertData) {
String pluginName = pluginDao.getPluginDefineById(instance.getPluginDefineId()).getPluginName();
String pluginName = alertPluginManager.getPluginNameById(instance.getPluginDefineId());
AlertChannel alertChannel = alertPluginManager.getAlertChannelMap().get(pluginName);
AlertResult alertResultExtend = new AlertResult();
String pluginInstanceName = instance.getInstanceName();
@ -157,8 +155,16 @@ public class AlertSender {
AlertInfo alertInfo = new AlertInfo();
alertInfo.setAlertData(alertData);
alertInfo.setAlertParams(instance.getPluginInstanceParams());
AlertResult alertResult = alertChannel.process(alertInfo);
Map<String, String> paramsMap = JSONUtils.toMap(instance.getPluginInstanceParams());
alertInfo.setAlertParams(paramsMap);
AlertResult alertResult;
try {
alertResult = alertChannel.process(alertInfo);
} catch (Exception e) {
alertResult = new AlertResult("false", e.getMessage());
logger.error("send alert error alert data id :{},", alertData.getId(), e);
}
if (alertResult == null) {
String message = String.format("Alert Plugin %s send error : return alertResult value is null", pluginInstanceName);

9
dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/AlertServerTest.java

@ -34,12 +34,13 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
@RunWith(PowerMockRunner.class)
@PrepareForTest({AlertServer.class,DaoFactory.class})
@PrepareForTest({AlertServer.class, DaoFactory.class})
public class AlertServerTest {
@Before
@ -61,7 +62,8 @@ public class AlertServerTest {
AlertPluginManager alertPluginManager = PowerMockito.mock(AlertPluginManager.class);
PowerMockito.whenNew(AlertPluginManager.class).withNoArguments().thenReturn(alertPluginManager);
ConcurrentHashMap alertChannelMap = new ConcurrentHashMap<>();
alertChannelMap.put("pluginName",alertChannelMock);
alertChannelMap.put("pluginName", alertChannelMock);
PowerMockito.when(alertPluginManager.getPluginNameById(Mockito.anyInt())).thenReturn("pluginName");
PowerMockito.when(alertPluginManager.getAlertChannelMap()).thenReturn(alertChannelMap);
DolphinPluginManagerConfig alertPluginManagerConfig = PowerMockito.mock(DolphinPluginManagerConfig.class);
@ -79,7 +81,8 @@ public class AlertServerTest {
Assert.assertNotNull(alertServer);
new Thread(() -> {
alertServer.start(); })
alertServer.start();
})
.start();
Thread.sleep(5 * Constants.ALERT_SCAN_INTERVAL);

2
dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/plugin/EmailAlertPluginTest.java

@ -136,7 +136,7 @@ public class EmailAlertPluginTest {
alertPluginInstance.setPluginInstanceParams(getEmailAlertParams());
alertDao.getAlertPluginInstanceMapper().insert(alertPluginInstance);
AlertSender alertSender = new AlertSender(alertList, alertDao, alertPluginManager, pluginDao);
AlertSender alertSender = new AlertSender(alertList, alertDao, alertPluginManager);
alertSender.run();
Alert alertResult = alertDao.getAlertMapper().selectById(alert1.getId());

11
dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/processor/AlertRequestProcessorTest.java

@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.alert.processor;
import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendRequestCommand;
@ -37,7 +36,6 @@ import io.netty.channel.Channel;
public class AlertRequestProcessorTest {
private AlertDao alertDao;
private PluginDao pluginDao;
private AlertPluginManager alertPluginManager;
private AlertRequestProcessor alertRequestProcessor;
@ -45,17 +43,16 @@ public class AlertRequestProcessorTest {
@Before
public void before() {
alertDao = PowerMockito.mock(AlertDao.class);
pluginDao = PowerMockito.mock(PluginDao.class);
alertPluginManager = PowerMockito.mock(AlertPluginManager.class);
alertRequestProcessor = new AlertRequestProcessor(alertDao,alertPluginManager,pluginDao);
alertRequestProcessor = new AlertRequestProcessor(alertDao, alertPluginManager);
}
@Test
public void testProcess() {
Channel channel = PowerMockito.mock(Channel.class);
AlertSendRequestCommand alertSendRequestCommand = new AlertSendRequestCommand(1,"title","content");
AlertSendRequestCommand alertSendRequestCommand = new AlertSendRequestCommand(1, "title", "content");
Command reqCommand = alertSendRequestCommand.convert2Command();
Assert.assertEquals(CommandType.ALERT_SEND_REQUEST,reqCommand.getType());
alertRequestProcessor.process(channel,reqCommand);
Assert.assertEquals(CommandType.ALERT_SEND_REQUEST, reqCommand.getType());
alertRequestProcessor.process(channel, reqCommand);
}
}

36
dolphinscheduler-alert/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java

@ -67,7 +67,7 @@ public class AlertSenderTest {
int alertGroupId = 1;
String title = "alert mail test title";
String content = "alert mail test content";
alertSender = new AlertSender(alertDao,alertPluginManager,pluginDao);
alertSender = new AlertSender(alertDao, alertPluginManager);
//1.alert instance does not exist
PowerMockito.when(alertDao.listInstanceByAlertGroupId(alertGroupId)).thenReturn(null);
@ -75,7 +75,7 @@ public class AlertSenderTest {
AlertSendResponseCommand alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
//2.alert plugin does not exist
int pluginDefineId = 1;
@ -83,30 +83,31 @@ public class AlertSenderTest {
String pluginInstanceName = "alert-instance-mail";
List<AlertPluginInstance> alertInstanceList = new ArrayList<>();
AlertPluginInstance alertPluginInstance = new AlertPluginInstance(
pluginDefineId,pluginInstanceParams,pluginInstanceName);
pluginDefineId, pluginInstanceParams, pluginInstanceName);
alertInstanceList.add(alertPluginInstance);
PowerMockito.when(alertDao.listInstanceByAlertGroupId(1)).thenReturn(alertInstanceList);
String pluginName = "alert-plugin-mail";
PluginDefine pluginDefine = new PluginDefine(pluginName,"1",null);
PluginDefine pluginDefine = new PluginDefine(pluginName, "1", null);
PowerMockito.when(pluginDao.getPluginDefineById(pluginDefineId)).thenReturn(pluginDefine);
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
//3.alert result value is null
AlertChannel alertChannelMock = PowerMockito.mock(AlertChannel.class);
PowerMockito.when(alertChannelMock.process(Mockito.any())).thenReturn(null);
Map<String, AlertChannel> alertChannelMap = new ConcurrentHashMap<>();
alertChannelMap.put(pluginName,alertChannelMock);
alertChannelMap.put(pluginName, alertChannelMock);
PowerMockito.when(alertPluginManager.getAlertChannelMap()).thenReturn(alertChannelMap);
PowerMockito.when(alertPluginManager.getPluginNameById(Mockito.anyInt())).thenReturn("alert-plugin-mail");
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
//4.abnormal information inside the alert plug-in code
AlertResult alertResult = new AlertResult();
@ -114,27 +115,27 @@ public class AlertSenderTest {
alertResult.setMessage("Abnormal information inside the alert plug-in code");
PowerMockito.when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
alertChannelMap = new ConcurrentHashMap<>();
alertChannelMap.put(pluginName,alertChannelMock);
alertChannelMap.put(pluginName, alertChannelMock);
PowerMockito.when(alertPluginManager.getAlertChannelMap()).thenReturn(alertChannelMap);
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
//5.alert plugin send success
alertResult = new AlertResult();
alertResult.setStatus(String.valueOf(true));
alertResult.setMessage(String.format("Alert Plugin %s send success",pluginInstanceName));
alertResult.setMessage(String.format("Alert Plugin %s send success", pluginInstanceName));
PowerMockito.when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
alertChannelMap = new ConcurrentHashMap<>();
alertChannelMap.put(pluginName,alertChannelMock);
alertChannelMap.put(pluginName, alertChannelMock);
PowerMockito.when(alertPluginManager.getAlertChannelMap()).thenReturn(alertChannelMap);
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content);
Assert.assertTrue(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}",result.getStatus(),result.getMessage()));
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
}
@ -150,28 +151,29 @@ public class AlertSenderTest {
alert.setContent(content);
alertList.add(alert);
alertSender = new AlertSender(alertList,alertDao,alertPluginManager,pluginDao);
alertSender = new AlertSender(alertList, alertDao, alertPluginManager);
int pluginDefineId = 1;
String pluginInstanceParams = "alert-instance-mail-params";
String pluginInstanceName = "alert-instance-mail";
List<AlertPluginInstance> alertInstanceList = new ArrayList<>();
AlertPluginInstance alertPluginInstance = new AlertPluginInstance(
pluginDefineId,pluginInstanceParams,pluginInstanceName);
pluginDefineId, pluginInstanceParams, pluginInstanceName);
alertInstanceList.add(alertPluginInstance);
PowerMockito.when(alertDao.listInstanceByAlertGroupId(alertGroupId)).thenReturn(alertInstanceList);
String pluginName = "alert-plugin-mail";
PluginDefine pluginDefine = new PluginDefine(pluginName,"1",null);
PluginDefine pluginDefine = new PluginDefine(pluginName, "1", null);
PowerMockito.when(pluginDao.getPluginDefineById(pluginDefineId)).thenReturn(pluginDefine);
PowerMockito.when(alertPluginManager.getPluginNameById(1)).thenReturn("alert-instance-mail");
AlertResult alertResult = new AlertResult();
alertResult.setStatus(String.valueOf(true));
alertResult.setMessage(String.format("Alert Plugin %s send success",pluginInstanceName));
alertResult.setMessage(String.format("Alert Plugin %s send success", pluginInstanceName));
AlertChannel alertChannelMock = PowerMockito.mock(AlertChannel.class);
PowerMockito.when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
ConcurrentHashMap alertChannelMap = new ConcurrentHashMap<>();
alertChannelMap.put(pluginName,alertChannelMock);
alertChannelMap.put(pluginName, alertChannelMock);
PowerMockito.when(alertPluginManager.getAlertChannelMap()).thenReturn(alertChannelMap);
Assert.assertTrue(Boolean.parseBoolean(alertResult.getStatus()));
alertSender.run();

65
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java

@ -24,12 +24,17 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.vo.AlertPluginInstanceVO;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.AlertGroupMapper;
import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import org.apache.commons.collections4.MapUtils;
import java.util.ArrayList;
import java.util.Arrays;
@ -37,6 +42,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.springframework.beans.factory.annotation.Autowired;
@ -73,7 +79,8 @@ public class AlertPluginInstanceServiceImpl extends BaseService implements Alert
@Override
public Map<String, Object> create(User loginUser, int pluginDefineId, String instanceName, String pluginInstanceParams) {
AlertPluginInstance alertPluginInstance = new AlertPluginInstance();
alertPluginInstance.setPluginInstanceParams(pluginInstanceParams);
String paramsMapJson = parsePluginParamsMap(pluginInstanceParams);
alertPluginInstance.setPluginInstanceParams(paramsMapJson);
alertPluginInstance.setInstanceName(instanceName);
alertPluginInstance.setPluginDefineId(pluginDefineId);
@ -88,7 +95,9 @@ public class AlertPluginInstanceServiceImpl extends BaseService implements Alert
if (i > 0) {
putMsg(result, Status.SUCCESS);
return result;
}
putMsg(result, Status.SAVE_ERROR);
return result;
}
@ -104,7 +113,8 @@ public class AlertPluginInstanceServiceImpl extends BaseService implements Alert
public Map<String, Object> update(User loginUser, int pluginInstanceId, String instanceName, String pluginInstanceParams) {
AlertPluginInstance alertPluginInstance = new AlertPluginInstance();
alertPluginInstance.setPluginInstanceParams(pluginInstanceParams);
String paramsMapJson = parsePluginParamsMap(pluginInstanceParams);
alertPluginInstance.setPluginInstanceParams(paramsMapJson);
alertPluginInstance.setInstanceName(instanceName);
alertPluginInstance.setId(pluginInstanceId);
Map<String, Object> result = new HashMap<>();
@ -112,8 +122,9 @@ public class AlertPluginInstanceServiceImpl extends BaseService implements Alert
if (i > 0) {
putMsg(result, Status.SUCCESS);
return result;
}
putMsg(result, Status.SAVE_ERROR);
return result;
}
@ -201,24 +212,66 @@ public class AlertPluginInstanceServiceImpl extends BaseService implements Alert
if (CollectionUtils.isEmpty(pluginDefineList)) {
return null;
}
Map<Integer, String> pluginDefineMap = pluginDefineList.stream().collect(Collectors.toMap(PluginDefine::getId, PluginDefine::getPluginName));
Map<Integer, PluginDefine> pluginDefineMap = pluginDefineList.stream().collect(Collectors.toMap(PluginDefine::getId, Function.identity()));
List<AlertPluginInstanceVO> alertPluginInstanceVOS = new ArrayList<>();
alertPluginInstances.forEach(alertPluginInstance -> {
AlertPluginInstanceVO alertPluginInstanceVO = new AlertPluginInstanceVO();
alertPluginInstanceVO.setAlertPluginName(pluginDefineMap.get(alertPluginInstance.getPluginDefineId()));
alertPluginInstanceVO.setCreateTime(alertPluginInstance.getCreateTime());
alertPluginInstanceVO.setUpdateTime(alertPluginInstance.getUpdateTime());
alertPluginInstanceVO.setPluginDefineId(alertPluginInstance.getPluginDefineId());
alertPluginInstanceVO.setInstanceName(alertPluginInstance.getInstanceName());
alertPluginInstanceVO.setId(alertPluginInstance.getId());
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?
if (null == pluginDefine) {
return;
}
alertPluginInstanceVO.setAlertPluginName(pluginDefine.getPluginName());
//todo List pages do not recommend returning this parameter
alertPluginInstanceVO.setPluginInstanceParams(alertPluginInstance.getPluginInstanceParams());
String pluginParamsMapString = alertPluginInstance.getPluginInstanceParams();
String uiPluginParams = parseToPluginUiParams(pluginParamsMapString, pluginDefine.getPluginParams());
alertPluginInstanceVO.setPluginInstanceParams(uiPluginParams);
alertPluginInstanceVOS.add(alertPluginInstanceVO);
});
return alertPluginInstanceVOS;
}
/**
* Get the parameters actually needed by the plugin
*
* @param pluginParams Complete parameters(include ui)
* @return k, v(json string)
*/
private String parsePluginParamsMap(String pluginParams) {
Map<String, String> paramsMap = PluginParamsTransfer.getPluginParamsMap(pluginParams);
return JSONUtils.toJsonString(paramsMap);
}
/**
* parseToPluginUiParams
*
* @param pluginParamsMapString k-v data
* @param pluginUiParams Complete parameters(include ui)
* @return Complete parameters list(include ui)
*/
private String parseToPluginUiParams(String pluginParamsMapString, String pluginUiParams) {
Map<String, String> paramsMap = JSONUtils.toMap(pluginParamsMapString);
if (MapUtils.isEmpty(paramsMap)) {
return null;
}
List<PluginParams> pluginParamsList = JSONUtils.toList(pluginUiParams, PluginParams.class);
List<PluginParams> newPluginParamsList = new ArrayList<>(pluginParamsList.size());
pluginParamsList.forEach(pluginParams -> {
pluginParams.setValue(paramsMap.get(pluginParams.getName()));
newPluginParamsList.add(pluginParams);
});
return JSONUtils.toJsonString(newPluginParamsList);
}
private boolean checkHasAssociatedAlertGroup(String id) {
List<String> idsList = alertGroupMapper.queryInstanceIdsList();
if (CollectionUtils.isEmpty(idsList)) {

109
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertPluginInstanceServiceTest.java

@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.api.service.impl.AlertPluginInstanceServiceIm
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.AlertGroupMapper;
import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
@ -29,6 +30,7 @@ import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@ -60,6 +62,82 @@ public class AlertPluginInstanceServiceTest {
private User user;
private String uiParams = "[\n"
+ " {\n"
+ " \"field\":\"userParams\",\n"
+ " \"name\":\"user.params\",\n"
+ " \"props\":{\n"
+ " \"placeholder\":\"please enter your custom parameters, which will be passed to you when calling your script\",\n"
+ " \"size\":\"small\"\n"
+ " },\n"
+ " \"type\":\"input\",\n"
+ " \"title\":\"user.params\",\n"
+ " \"value\":\"userParams\",\n"
+ " \"validate\":[\n"
+ " {\n"
+ " \"required\":false,\n"
+ " \"message\":null,\n"
+ " \"type\":\"string\",\n"
+ " \"trigger\":\"blur\",\n"
+ " \"min\":null,\n"
+ " \"max\":null\n"
+ " }\n"
+ " ]\n"
+ " },\n"
+ " {\n"
+ " \"field\":\"path\",\n"
+ " \"name\":\"path\",\n"
+ " \"props\":{\n"
+ " \"placeholder\":\"please upload the file to the disk directory of the alert server, and ensure that the path is absolute and has the corresponding access rights\",\n"
+ " \"size\":\"small\"\n"
+ " },\n"
+ " \"type\":\"input\",\n"
+ " \"title\":\"path\",\n"
+ " \"value\":\"/kris/script/path\",\n"
+ " \"validate\":[\n"
+ " {\n"
+ " \"required\":true,\n"
+ " \"message\":null,\n"
+ " \"type\":\"string\",\n"
+ " \"trigger\":\"blur\",\n"
+ " \"min\":null,\n"
+ " \"max\":null\n"
+ " }\n"
+ " ]\n"
+ " },\n"
+ " {\n"
+ " \"field\":\"type\",\n"
+ " \"name\":\"type\",\n"
+ " \"props\":{\n"
+ " \"placeholder\":null,\n"
+ " \"size\":\"small\"\n"
+ " },\n"
+ " \"type\":\"radio\",\n"
+ " \"title\":\"type\",\n"
+ " \"value\":0,\n"
+ " \"validate\":[\n"
+ " {\n"
+ " \"required\":true,\n"
+ " \"message\":null,\n"
+ " \"type\":\"string\",\n"
+ " \"trigger\":\"blur\",\n"
+ " \"min\":null,\n"
+ " \"max\":null\n"
+ " }\n"
+ " ],\n"
+ " \"options\":[\n"
+ " {\n"
+ " \"label\":\"SHELL\",\n"
+ " \"value\":0,\n"
+ " \"disabled\":false\n"
+ " }\n"
+ " ]\n"
+ " }\n"
+ "]\n"
+ "\n";
private String paramsMap = "{\"path\":\"/kris/script/path\",\"userParams\":\"userParams\",\"type\":\"0\"}";
@Before
public void before() {
user = new User();
@ -77,10 +155,10 @@ public class AlertPluginInstanceServiceTest {
@Test
public void testCreate() {
Mockito.when(alertPluginInstanceMapper.queryByInstanceName("test")).thenReturn(alertPluginInstances);
Map<String, Object> result = alertPluginInstanceService.create(user, 1, "test", "test params");
Map<String, Object> result = alertPluginInstanceService.create(user, 1, "test", uiParams);
Assert.assertEquals(Status.PLUGIN_INSTANCE_ALREADY_EXIT, result.get(Constants.STATUS));
Mockito.when(alertPluginInstanceMapper.insert(Mockito.any())).thenReturn(1);
result = alertPluginInstanceService.create(user, 1, "test1", "test params");
result = alertPluginInstanceService.create(user, 1, "test1", uiParams);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@ -96,4 +174,31 @@ public class AlertPluginInstanceServiceTest {
}
@Test
public void testUpdate() {
Mockito.when(alertPluginInstanceMapper.updateById(Mockito.any())).thenReturn(0);
Map<String, Object> result = alertPluginInstanceService.update(user, 1, "testUpdate", uiParams);
Assert.assertEquals(Status.SAVE_ERROR, result.get(Constants.STATUS));
Mockito.when(alertPluginInstanceMapper.updateById(Mockito.any())).thenReturn(1);
result = alertPluginInstanceService.update(user, 1, "testUpdate", uiParams);
Assert.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");
PluginDefine pluginDefine = new PluginDefine("script", "script", uiParams);
pluginDefine.setId(1);
List<PluginDefine> pluginDefines = Collections.singletonList(pluginDefine);
List<AlertPluginInstance> pluginInstanceList = Collections.singletonList(alertPluginInstance);
Mockito.when(alertPluginInstanceMapper.queryAllAlertPluginInstanceList()).thenReturn(pluginInstanceList);
Mockito.when(pluginDefineMapper.queryAllPluginDefineList()).thenReturn(pluginDefines);
Map<String, Object> result = alertPluginInstanceService.queryAll();
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
}

18
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/PluginDao.java

@ -58,22 +58,22 @@ public class PluginDao extends AbstractBaseDao {
*
* @param pluginDefine new pluginDefine
*/
public void addOrUpdatePluginDefine(PluginDefine pluginDefine) {
public int addOrUpdatePluginDefine(PluginDefine pluginDefine) {
requireNonNull(pluginDefine, "pluginDefine is null");
requireNonNull(pluginDefine.getPluginName(), "pluginName is null");
requireNonNull(pluginDefine.getPluginType(), "pluginType is null");
List<PluginDefine> pluginDefineList = pluginDefineMapper.queryByNameAndType(pluginDefine.getPluginName(), pluginDefine.getPluginType());
if (pluginDefineList == null || pluginDefineList.size() == 0) {
pluginDefineMapper.insert(pluginDefine);
} else {
PluginDefine currPluginDefine = pluginDefineList.get(0);
if (!currPluginDefine.getPluginParams().equals(pluginDefine.getPluginParams())) {
currPluginDefine.setUpdateTime(pluginDefine.getUpdateTime());
currPluginDefine.setPluginParams(pluginDefine.getPluginParams());
pluginDefineMapper.updateById(currPluginDefine);
}
return pluginDefineMapper.insert(pluginDefine);
}
PluginDefine currPluginDefine = pluginDefineList.get(0);
if (!currPluginDefine.getPluginParams().equals(pluginDefine.getPluginParams())) {
currPluginDefine.setUpdateTime(pluginDefine.getUpdateTime());
currPluginDefine.setPluginParams(pluginDefine.getPluginParams());
pluginDefineMapper.updateById(currPluginDefine);
}
return currPluginDefine.getId();
}
/**

8
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/alert/AlertInfo.java

@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.spi.alert;
import java.util.Map;
/**
* AlertInfo
*/
@ -25,18 +27,18 @@ public class AlertInfo {
/**
* all params this plugin need is in alertProps
*/
private String alertParams;
private Map<String,String> alertParams;
/**
* the alert content
*/
private AlertData alertData;
public String getAlertParams() {
public Map<String, String> getAlertParams() {
return alertParams;
}
public void setAlertParams(String alertParams) {
public void setAlertParams(Map<String, String> alertParams) {
this.alertParams = alertParams;
}

8
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/alert/AlertResult.java

@ -38,4 +38,12 @@ public class AlertResult {
public void setMessage(String message) {
this.message = message;
}
public AlertResult(String status, String message) {
this.status = status;
this.message = message;
}
public AlertResult() {
}
}

4
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/params/base/PluginParams.java

@ -165,6 +165,10 @@ public class PluginParams {
public List<Validate> getValidateList() {
return validateList;
}
public void setValue(Object value) {
this.value = value;
}
}

Loading…
Cancel
Save