From f8ecb536b71d6f33b71c73930832b62890b84ea1 Mon Sep 17 00:00:00 2001 From: ruanwenjun <861923274@qq.com> Date: Tue, 1 Jun 2021 10:21:46 +0800 Subject: [PATCH] [Improvement][Alert] server down will send repetitive message #5525 (#5529) * [Improvement][Alert] server down will send repetitive message #5525 * add ut --- .../apache/dolphinscheduler/dao/AlertDao.java | 22 ++++++++++++++----- .../dao/mapper/AlertMapper.java | 11 +++++++++- .../dao/mapper/AlertMapper.xml | 9 ++++++++ .../dolphinscheduler/dao/AlertDaoTest.java | 17 ++++++++++++++ 4 files changed, 52 insertions(+), 7 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 5b66932a36..6787b8c386 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 @@ -44,6 +44,8 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import com.google.common.collect.Lists; + @Component public class AlertDao extends AbstractBaseDao { @@ -99,15 +101,23 @@ public class AlertDao extends AbstractBaseDao { * @param serverType serverType */ public void sendServerStopedAlert(int alertGroupId, String host, String serverType) { - Alert alert = new Alert(); - List serverAlertContents = new ArrayList<>(1); ServerAlertContent serverStopAlertContent = ServerAlertContent.newBuilder(). - type(serverType).host(host).event(AlertEvent.SERVER_DOWN).warningLevel(AlertWarnLevel.SERIOUS). + type(serverType) + .host(host) + .event(AlertEvent.SERVER_DOWN) + .warningLevel(AlertWarnLevel.SERIOUS). build(); - serverAlertContents.add(serverStopAlertContent); - String content = JSONUtils.toJsonString(serverAlertContents); + String content = JSONUtils.toJsonString(Lists.newArrayList(serverStopAlertContent)); + + Alert alert = new Alert(); alert.setTitle("Fault tolerance warning"); - saveTaskTimeoutAlert(alert, content, alertGroupId); + alert.setAlertStatus(AlertStatus.WAIT_EXECUTION); + alert.setContent(content); + alert.setAlertGroupId(alertGroupId); + alert.setCreateTime(new Date()); + alert.setUpdateTime(new Date()); + // we use this method to avoid insert duplicate alert(issue #5525) + alertMapper.insertAlertWhenServerCrash(alert); } /** diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java index d97e16d19d..77786c5a1e 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java @@ -14,15 +14,18 @@ * 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.Alert; -import com.baomidou.mybatisplus.core.mapper.BaseMapper; + import org.apache.ibatis.annotations.Param; import java.util.List; +import com.baomidou.mybatisplus.core.mapper.BaseMapper; + /** * alert mapper interface */ @@ -35,4 +38,10 @@ public interface AlertMapper extends BaseMapper { */ List listAlertByStatus(@Param("alertStatus") AlertStatus alertStatus); + /** + * Insert server crash alert + *

This method will ensure that there is at most one unsent alert which has the same content in the database. + */ + void insertAlertWhenServerCrash(@Param("alert") Alert alert); + } diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml index 9be5c7c784..40f538339d 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml @@ -29,4 +29,13 @@ from t_ds_alert where alert_status = #{alertStatus} + + + insert into t_ds_alert(title, content, alert_status, log, alertgroup_id, create_time, update_time) + SELECT #{alert.title}, #{alert.content}, #{alert.alertStatus.code}, #{alert.log}, #{alert.alertGroupId}, + #{alert.createTime}, #{alert.updateTime} + from t_ds_alert + where content = #{alert.content} and alert_status = #{alert.alertStatus.code} + having count(*) = 0 + diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java index 0137bd5639..7b9e8c6f35 100644 --- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java +++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/AlertDaoTest.java @@ -24,7 +24,9 @@ import java.util.List; import org.junit.Assert; import org.junit.Test; +import org.springframework.transaction.annotation.Transactional; +@Transactional public class AlertDaoTest { @Test @@ -42,4 +44,19 @@ public class AlertDaoTest { Assert.assertNotNull(alerts); Assert.assertNotEquals(0, alerts.size()); } + + @Test + public void testSendServerStopedAlert() { + AlertDao alertDao = DaoFactory.getDaoInstance(AlertDao.class); + int alertGroupId = 1; + String host = "127.0.0.998165432"; + String serverType = "Master"; + alertDao.sendServerStopedAlert(alertGroupId, host, serverType); + alertDao.sendServerStopedAlert(alertGroupId, host, serverType); + long count = alertDao.listWaitExecutionAlert() + .stream() + .filter(alert -> alert.getContent().contains(host)) + .count(); + Assert.assertEquals(1L, count); + } }