This is an automated email from the ASF dual-hosted git repository. journey pushed a commit to branch dev in repository https://gitbox.apache.org/repos/asf/incubator-dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push: new 2f01025 [Fix-3299][dao&server] Fix 3299,when Json string parsing problem caused by non-standard json format. (#3552) 2f01025 is described below commit 2f0102580268ff045e6042c3a691f1dee4c49962 Author: felix.wang <59079269+felix-thinkingd...@users.noreply.github.com> AuthorDate: Wed Aug 19 16:27:39 2020 +0800 [Fix-3299][dao&server] Fix 3299,when Json string parsing problem caused by non-standard json format. (#3552) * #3299 Json string parsing problem caused by non-standard json format. * #3299 Json string parsing problem caused by non-standard json format. * #3299 Json string parsing problem caused by non-standard json format. fix code style * #3299 Json string parsing problem caused by non-standard json format. fix code style Co-authored-by: wangjianda <fe...@thinkingdata.com> --- .../org/apache/dolphinscheduler/dao/AlertDao.java | 81 ++++++++++----- .../server/utils/AlertManager.java | 109 +++++++++------------ 2 files changed, 104 insertions(+), 86 deletions(-) diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java index 49b8c01..685d72c 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java @@ -14,15 +14,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dolphinscheduler.dao; +package org.apache.dolphinscheduler.dao; -import org.apache.dolphinscheduler.common.utils.StringUtils; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Component; import org.apache.dolphinscheduler.common.enums.AlertStatus; import org.apache.dolphinscheduler.common.enums.AlertType; import org.apache.dolphinscheduler.common.enums.ShowType; +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.common.utils.StringUtils; import org.apache.dolphinscheduler.dao.datasource.ConnectionFactory; import org.apache.dolphinscheduler.dao.entity.Alert; import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; @@ -30,13 +29,17 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance; import org.apache.dolphinscheduler.dao.entity.User; import org.apache.dolphinscheduler.dao.mapper.AlertMapper; import org.apache.dolphinscheduler.dao.mapper.UserAlertGroupMapper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - +import java.util.ArrayList; import java.util.Date; +import java.util.LinkedHashMap; import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + @Component public class AlertDao extends AbstractBaseDao { @@ -56,21 +59,23 @@ public class AlertDao extends AbstractBaseDao { /** * insert alert + * * @param alert alert * @return add alert result */ - public int addAlert(Alert alert){ + public int addAlert(Alert alert) { return alertMapper.insert(alert); } /** * update alert + * * @param alertStatus alertStatus * @param log log * @param id id * @return update alert result */ - public int updateAlert(AlertStatus alertStatus,String log,int id){ + public int updateAlert(AlertStatus alertStatus, String log, int id) { Alert alert = alertMapper.selectById(id); alert.setAlertStatus(alertStatus); alert.setUpdateTime(new Date()); @@ -80,46 +85,61 @@ public class AlertDao extends AbstractBaseDao { /** * query user list by alert group id + * * @param alerGroupId alerGroupId * @return user list */ - public List<User> queryUserByAlertGroupId(int alerGroupId){ + public List<User> queryUserByAlertGroupId(int alerGroupId) { return userAlertGroupMapper.listUserByAlertgroupId(alerGroupId); } /** * MasterServer or WorkerServer stoped + * * @param alertgroupId alertgroupId * @param host host * @param serverType serverType */ - public void sendServerStopedAlert(int alertgroupId,String host,String serverType){ + public void sendServerStopedAlert(int alertgroupId, String host, String serverType) { Alert alert = new Alert(); - String content = String.format("[{'type':'%s','host':'%s','event':'server down','warning level':'serious'}]", - serverType, host); + List<LinkedHashMap> serverStopList = new ArrayList<>(1); + LinkedHashMap<String, String> serverStopedMap = new LinkedHashMap(); + serverStopedMap.put("type", serverType); + serverStopedMap.put("host", host); + serverStopedMap.put("event", "server down"); + serverStopedMap.put("warning level", "serious"); + serverStopList.add(serverStopedMap); + String content = JSONUtils.toJsonString(serverStopList); alert.setTitle("Fault tolerance warning"); saveTaskTimeoutAlert(alert, content, alertgroupId, null, null); } /** * process time out alert + * * @param processInstance processInstance * @param processDefinition processDefinition */ - public void sendProcessTimeoutAlert(ProcessInstance processInstance, ProcessDefinition processDefinition){ + public void sendProcessTimeoutAlert(ProcessInstance processInstance, ProcessDefinition processDefinition) { int alertgroupId = processInstance.getWarningGroupId(); String receivers = processDefinition.getReceivers(); String receiversCc = processDefinition.getReceiversCc(); Alert alert = new Alert(); - String content = String.format("[{'id':'%d','name':'%s','event':'timeout','warnLevel':'middle'}]", - processInstance.getId(), processInstance.getName()); + List<LinkedHashMap> processTimeoutList = new ArrayList<>(1); + LinkedHashMap<String, String> processTimeoutMap = new LinkedHashMap(); + processTimeoutMap.put("id", String.valueOf(processInstance.getId())); + processTimeoutMap.put("name", processInstance.getName()); + processTimeoutMap.put("event", "timeout"); + processTimeoutMap.put("warnLevel", "middle"); + processTimeoutList.add(processTimeoutMap); + String content = JSONUtils.toJsonString(processTimeoutList); alert.setTitle("Process Timeout Warn"); saveTaskTimeoutAlert(alert, content, alertgroupId, receivers, receiversCc); } - private void saveTaskTimeoutAlert(Alert alert, String content, int alertgroupId, - String receivers, String receiversCc){ + private void saveTaskTimeoutAlert(Alert alert, String content, int alertgroupId, + String receivers, String receiversCc) { alert.setShowType(ShowType.TABLE); alert.setContent(content); alert.setAlertType(AlertType.EMAIL); @@ -135,9 +155,9 @@ public class AlertDao extends AbstractBaseDao { alertMapper.insert(alert); } - /** * task timeout warn + * * @param alertgroupId alertgroupId * @param receivers receivers * @param receiversCc receiversCc @@ -146,34 +166,45 @@ public class AlertDao extends AbstractBaseDao { * @param taskId taskId * @param taskName taskName */ - public void sendTaskTimeoutAlert(int alertgroupId,String receivers,String receiversCc, int processInstanceId, - String processInstanceName, int taskId,String taskName){ + public void sendTaskTimeoutAlert(int alertgroupId, String receivers, String receiversCc, int processInstanceId, + String processInstanceName, int taskId, String taskName) { Alert alert = new Alert(); - String content = String.format("[{'process instance id':'%d','task name':'%s','task id':'%d','task name':'%s'," + - "'event':'timeout','warnLevel':'middle'}]", processInstanceId, processInstanceName, taskId, taskName); + List<LinkedHashMap> taskTimeoutList = new ArrayList<>(1); + LinkedHashMap<String, String> taskTimeoutMap = new LinkedHashMap(); + taskTimeoutMap.put("process instance id", String.valueOf(processInstanceId)); + taskTimeoutMap.put("process name", processInstanceName); + taskTimeoutMap.put("task id", String.valueOf(taskId)); + taskTimeoutMap.put("task name", taskName); + taskTimeoutMap.put("event", "timeout"); + taskTimeoutMap.put("warnLevel", "middle"); + taskTimeoutList.add(taskTimeoutMap); + String content = JSONUtils.toJsonString(taskTimeoutList); alert.setTitle("Task Timeout Warn"); saveTaskTimeoutAlert(alert, content, alertgroupId, receivers, receiversCc); } /** * list the alert information of waiting to be executed + * * @return alert list */ - public List<Alert> listWaitExecutionAlert(){ + public List<Alert> listWaitExecutionAlert() { return alertMapper.listAlertByStatus(AlertStatus.WAIT_EXECUTION); } /** * list user information by alert group id + * * @param alertgroupId alertgroupId * @return user list */ - public List<User> listUserByAlertgroupId(int alertgroupId){ + public List<User> listUserByAlertgroupId(int alertgroupId) { return userAlertGroupMapper.listUserByAlertgroupId(alertgroupId); } /** * for test + * * @return AlertMapper */ public AlertMapper getAlertMapper() { diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java index 49ec9d3..08c6022 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/AlertManager.java @@ -14,29 +14,30 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dolphinscheduler.server.utils; +package org.apache.dolphinscheduler.server.utils; import org.apache.dolphinscheduler.common.enums.AlertType; import org.apache.dolphinscheduler.common.enums.CommandType; import org.apache.dolphinscheduler.common.enums.ShowType; import org.apache.dolphinscheduler.common.enums.WarningType; import org.apache.dolphinscheduler.common.utils.DateUtils; -import org.apache.dolphinscheduler.common.utils.*; +import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.dao.AlertDao; import org.apache.dolphinscheduler.dao.DaoFactory; import org.apache.dolphinscheduler.dao.entity.Alert; import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; import org.apache.dolphinscheduler.dao.entity.ProcessInstance; import org.apache.dolphinscheduler.dao.entity.TaskInstance; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Date; import java.util.LinkedHashMap; import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + /** * alert manager */ @@ -50,8 +51,7 @@ public class AlertManager { /** * alert dao */ - private AlertDao alertDao = DaoFactory.getDaoInstance(AlertDao.class); - + private final AlertDao alertDao = DaoFactory.getDaoInstance(AlertDao.class); /** * command type convert chinese @@ -87,49 +87,36 @@ public class AlertManager { } /** - * process instance format - */ - private static final String PROCESS_INSTANCE_FORMAT = - "\"id:%d\"," + - "\"name:%s\"," + - "\"job type: %s\"," + - "\"state: %s\"," + - "\"recovery:%s\"," + - "\"run time: %d\"," + - "\"start time: %s\"," + - "\"end time: %s\"," + - "\"host: %s\"" ; - - /** * get process instance content - * @param processInstance process instance - * @param taskInstances task instance list + * + * @param processInstance process instance + * @param taskInstances task instance list * @return process instance format content */ public String getContentProcessInstance(ProcessInstance processInstance, - List<TaskInstance> taskInstances){ + List<TaskInstance> taskInstances) { String res = ""; - if(processInstance.getState().typeIsSuccess()){ - res = String.format(PROCESS_INSTANCE_FORMAT, - processInstance.getId(), - processInstance.getName(), - getCommandCnName(processInstance.getCommandType()), - processInstance.getState().toString(), - processInstance.getRecovery().toString(), - processInstance.getRunTimes(), - DateUtils.dateToString(processInstance.getStartTime()), - DateUtils.dateToString(processInstance.getEndTime()), - processInstance.getHost() - - ); - res = "[" + res + "]"; - }else if(processInstance.getState().typeIsFailure()){ + if (processInstance.getState().typeIsSuccess()) { + List<LinkedHashMap> successTaskList = new ArrayList<>(1); + LinkedHashMap<String, String> successTaskMap = new LinkedHashMap(); + successTaskMap.put("id", String.valueOf(processInstance.getId())); + successTaskMap.put("name", processInstance.getName()); + successTaskMap.put("job type", getCommandCnName(processInstance.getCommandType())); + successTaskMap.put("state", processInstance.getState().toString()); + successTaskMap.put("recovery", processInstance.getRecovery().toString()); + successTaskMap.put("run time", String.valueOf(processInstance.getRunTimes())); + successTaskMap.put("start time", DateUtils.dateToString(processInstance.getStartTime())); + successTaskMap.put("end time", DateUtils.dateToString(processInstance.getEndTime())); + successTaskMap.put("host", processInstance.getHost()); + successTaskList.add(successTaskMap); + res = JSONUtils.toJsonString(successTaskList); + } else if (processInstance.getState().typeIsFailure()) { List<LinkedHashMap> failedTaskList = new ArrayList<>(); - for(TaskInstance task : taskInstances){ - if(task.getState().typeIsSuccess()){ + for (TaskInstance task : taskInstances) { + if (task.getState().typeIsSuccess()) { continue; } LinkedHashMap<String, String> failedTaskMap = new LinkedHashMap(); @@ -154,15 +141,15 @@ public class AlertManager { /** * getting worker fault tolerant content * - * @param processInstance process instance + * @param processInstance process instance * @param toleranceTaskList tolerance task list * @return worker tolerance content */ - private String getWorkerToleranceContent(ProcessInstance processInstance, List<TaskInstance> toleranceTaskList){ + private String getWorkerToleranceContent(ProcessInstance processInstance, List<TaskInstance> toleranceTaskList) { - List<LinkedHashMap<String, String>> toleranceTaskInstanceList = new ArrayList<>(); + List<LinkedHashMap<String, String>> toleranceTaskInstanceList = new ArrayList<>(); - for(TaskInstance taskInstance: toleranceTaskList){ + for (TaskInstance taskInstance : toleranceTaskList) { LinkedHashMap<String, String> toleranceWorkerContentMap = new LinkedHashMap(); toleranceWorkerContentMap.put("process name", processInstance.getName()); toleranceWorkerContentMap.put("task name", taskInstance.getName()); @@ -176,11 +163,11 @@ public class AlertManager { /** * send worker alert fault tolerance * - * @param processInstance process instance + * @param processInstance process instance * @param toleranceTaskList tolerance task list */ - public void sendAlertWorkerToleranceFault(ProcessInstance processInstance, List<TaskInstance> toleranceTaskList){ - try{ + public void sendAlertWorkerToleranceFault(ProcessInstance processInstance, List<TaskInstance> toleranceTaskList) { + try { Alert alert = new Alert(); alert.setTitle("worker fault tolerance"); alert.setShowType(ShowType.TABLE); @@ -188,13 +175,13 @@ public class AlertManager { alert.setContent(content); alert.setAlertType(AlertType.EMAIL); alert.setCreateTime(new Date()); - alert.setAlertGroupId(processInstance.getWarningGroupId() == null ? 1:processInstance.getWarningGroupId()); + alert.setAlertGroupId(processInstance.getWarningGroupId() == null ? 1 : processInstance.getWarningGroupId()); alert.setReceivers(processInstance.getProcessDefinition().getReceivers()); alert.setReceiversCc(processInstance.getProcessDefinition().getReceiversCc()); alertDao.addAlert(alert); logger.info("add alert to db , alert : {}", alert.toString()); - }catch (Exception e){ + } catch (Exception e) { logger.error("send alert failed:{} ", e.getMessage()); } @@ -202,40 +189,40 @@ public class AlertManager { /** * send process instance alert - * @param processInstance process instance - * @param taskInstances task instance list + * + * @param processInstance process instance + * @param taskInstances task instance list */ public void sendAlertProcessInstance(ProcessInstance processInstance, - List<TaskInstance> taskInstances){ + List<TaskInstance> taskInstances) { boolean sendWarnning = false; WarningType warningType = processInstance.getWarningType(); - switch (warningType){ + switch (warningType) { case ALL: - if(processInstance.getState().typeIsFinished()){ + if (processInstance.getState().typeIsFinished()) { sendWarnning = true; } break; case SUCCESS: - if(processInstance.getState().typeIsSuccess()){ + if (processInstance.getState().typeIsSuccess()) { sendWarnning = true; } break; case FAILURE: - if(processInstance.getState().typeIsFailure()){ + if (processInstance.getState().typeIsFailure()) { sendWarnning = true; } break; - default: + default: } - if(!sendWarnning){ + if (!sendWarnning) { return; } Alert alert = new Alert(); - String cmdName = getCommandCnName(processInstance.getCommandType()); - String success = processInstance.getState().typeIsSuccess() ? "success" :"failed"; + String success = processInstance.getState().typeIsSuccess() ? "success" : "failed"; alert.setTitle(cmdName + " " + success); ShowType showType = processInstance.getState().typeIsSuccess() ? ShowType.TEXT : ShowType.TABLE; alert.setShowType(showType); @@ -254,7 +241,7 @@ public class AlertManager { /** * send process timeout alert * - * @param processInstance process instance + * @param processInstance process instance * @param processDefinition process definition */ public void sendProcessTimeoutAlert(ProcessInstance processInstance, ProcessDefinition processDefinition) {