zhuangchong commented on a change in pull request #4645:
URL:
https://github.com/apache/incubator-dolphinscheduler/pull/4645#discussion_r571795840
##########
File path:
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/AlertClientService.java
##########
@@ -78,40 +72,46 @@ public void close() {
/**
* alert sync send data
- * @param groupId
- * @param title
- * @param content
- * @return
*/
- public AlertSendResponseCommand sendAlert(int groupId, String title,
String content) {
- return this.sendAlert(this.host,this.port,groupId,title,content);
+ public AlertSendResponseCommand sendAlert(int groupId, String title,
String content) {
+ return this.sendAlert(this.host, this.port, groupId, title, content);
}
/**
* alert sync send data
+ *
* @param host host
* @param port port
* @param groupId groupId
* @param title title
* @param content content
* @return AlertSendResponseCommand
*/
- public AlertSendResponseCommand sendAlert(String host, int port, int
groupId, String title, String content) {
+ public AlertSendResponseCommand sendAlert(String host, int port, int
groupId, String title, String content) {
logger.info("sync alert send, host : {}, port : {}, groupId : {},
title : {} ", host, port, groupId, title);
AlertSendRequestCommand request = new AlertSendRequestCommand(groupId,
title, content);
- final Host address = new Host(host, port);
- try {
- Command command = request.convert2Command();
- Command response = this.client.sendSync(address, command,
ALERT_REQUEST_TIMEOUT);
- if (response != null) {
- return JsonSerializer.deserialize(response.getBody(),
AlertSendResponseCommand.class);
+ AlertSendResponseCommand alertSendResponseCommand = null;
+ String[] hosts = host.split(",");
+ for (String hostName : hosts) {
+ final Host address = new Host(hostName, port);
+ try {
+ Command command = request.convert2Command();
+ Command response = this.client.sendSync(address, command,
ALERT_REQUEST_TIMEOUT);
+ if (response != null) {
+ alertSendResponseCommand =
JsonSerializer.deserialize(response.getBody(), AlertSendResponseCommand.class);
+ if (alertSendResponseCommand.getResStatus()) {
+ return alertSendResponseCommand;
+ }
Review comment:
If only alertSendResponseCommand getResStatus return
alertSendResponseCommand () = = true, then the problem comes, synchronous alarm
failure will have the news, so sqlTask will not receive the alarm failed
messages
---
如果只有alertSendResponseCommand.getResStatus()==true时返回alertSendResponseCommand，那么问题来了，同步告警失败也会有失败的消息，这样sqlTask就收不到告警失败的消息
##########
File path:
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/AlertClientService.java
##########
@@ -78,40 +72,46 @@ public void close() {
/**
* alert sync send data
- * @param groupId
- * @param title
- * @param content
- * @return
*/
- public AlertSendResponseCommand sendAlert(int groupId, String title,
String content) {
- return this.sendAlert(this.host,this.port,groupId,title,content);
+ public AlertSendResponseCommand sendAlert(int groupId, String title,
String content) {
+ return this.sendAlert(this.host, this.port, groupId, title, content);
}
/**
* alert sync send data
+ *
* @param host host
* @param port port
* @param groupId groupId
* @param title title
* @param content content
* @return AlertSendResponseCommand
*/
- public AlertSendResponseCommand sendAlert(String host, int port, int
groupId, String title, String content) {
+ public AlertSendResponseCommand sendAlert(String host, int port, int
groupId, String title, String content) {
logger.info("sync alert send, host : {}, port : {}, groupId : {},
title : {} ", host, port, groupId, title);
AlertSendRequestCommand request = new AlertSendRequestCommand(groupId,
title, content);
- final Host address = new Host(host, port);
- try {
- Command command = request.convert2Command();
- Command response = this.client.sendSync(address, command,
ALERT_REQUEST_TIMEOUT);
- if (response != null) {
- return JsonSerializer.deserialize(response.getBody(),
AlertSendResponseCommand.class);
+ AlertSendResponseCommand alertSendResponseCommand = null;
+ String[] hosts = host.split(",");
+ for (String hostName : hosts) {
+ final Host address = new Host(hostName, port);
Review comment:
1. Split the host by comma. If the alert. Listen. Host =IP1,IP2 in
Worker.properties, will the SQLTask alert be sent twice?
2.Alert-server has already used ZK, this time it is recommended that the
alert-server be registered in ZK, synchronous Alert get alert-server host from
ZK (like worker/master)
---
1.host按逗号分割，如果worker.properties内alert.listen.host=IP1,IP2,
那是不是sqlTask的告警会发送两遍？
2.Alert-server 已经使用ZK，此次建议alert-server在zk注册，同步告警从zk获取alert-server
host（像worker/master一样）
##########
File path:
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/ZookeeperClient.java
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.utils;
+
+import static
org.apache.dolphinscheduler.alert.utils.Constants.ZOOKEEPER_ABNORMAL_TOLERATING_NUMBER;
+import static
org.apache.dolphinscheduler.alert.utils.Constants.ZOOKEEPER_DOLPHINSCHEDULER_LOCK_ALERTS;
+import static
org.apache.dolphinscheduler.alert.utils.Constants.ZOOKEEPER_PROPERTIES_PATH;
+import static org.apache.dolphinscheduler.alert.utils.Constants.ZOOKEEPER_ROOT;
+
+import org.apache.dolphinscheduler.common.utils.IOUtils;
+
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.CuratorFrameworkState;
+import org.apache.curator.framework.recipes.locks.InterProcessMutex;
+import org.apache.curator.retry.RetryOneTime;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+public class ZookeeperClient {
+
+
+ private static final Properties properties = new Properties();
+
+ public static void concurrentOperation(LockCallBall callBall,String
zookeeperConnectorStr) throws Exception {
+
+ RetryPolicy retryPolicy = new RetryOneTime(1000);
+ CuratorFramework zkClient =
CuratorFrameworkFactory.newClient(zookeeperConnectorStr, retryPolicy);
+ zkClient.start();
+
+ String alertLockPath = PropertyUtils.getString(ZOOKEEPER_ROOT,
"/dolphinscheduler") + ZOOKEEPER_DOLPHINSCHEDULER_LOCK_ALERTS;
+
Review comment:
Personal idea: there are common methods of ZK in the service module, can
we implement the distributed lock in the service module?
---
个人想法：在service模块里面有zk的常用方法，是否可以把分布式锁的实现也在service模块下实现?
##########
File path:
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/AlertServer.java
##########
@@ -129,9 +132,30 @@ private void runSender() {
if (alertPluginManager == null ||
alertPluginManager.getAlertChannelMap().size() == 0) {
logger.warn("No Alert Plugin . Can not send alert info. ");
} else {
- List<Alert> alerts = alertDao.listWaitExecutionAlert();
- alertSender = new AlertSender(alerts, alertDao,
alertPluginManager);
- alertSender.run();
+ try {
+ Properties properties =
ZookeeperClient.getZookeeperProperties();
+ ZookeeperClient.concurrentOperation(new
ZookeeperClient.LockCallBall() {
+ @Override
+ public void handle() {
+ List<Alert> alerts =
alertDao.listWaitExecutionAlert();
+ alertSender = new AlertSender(alerts, alertDao,
alertPluginManager);
+ alertSender.run();
+ zookeeperStateAbnormalToleratingNumber = 0;
+ }
+ }, properties.getProperty(Constants.ZOOKEEPER_LIST));
+ } catch (Exception e) {
+ logger.error("alert server with error : ", e);
+ }
+
+ if (zookeeperStateAbnormalToleratingNumber >
ZookeeperClient.checkZkStateAbnormalToleratingNumber()) {
+ List<Alert> alerts = alertDao.listWaitExecutionAlert();
+ alertSender = new AlertSender(alerts, alertDao,
alertPluginManager);
+ alertSender.run();
+ zookeeperStateAbnormalToleratingNumber = 0;
Review comment:
Resolve code redundancy
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
