diff --git a/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMService.java b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMService.java
index cf1bca8dc..6d7e6915d 100644
--- a/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMService.java
+++ b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMService.java
@@ -26,7 +26,9 @@
*/
public interface IMService {
- /** Send post request alert message by dingding. */
+ void sendPostMsgByLark(String data, String url);
+
+ /** Send post request alert message by dingding. */
public void sendPostMsgByDingDing(String data, String url);
/** Send alert message by wechat. */
diff --git a/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMServiceImpl.java b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMServiceImpl.java
index 0703c3681..6b943cb6b 100644
--- a/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMServiceImpl.java
+++ b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/IMServiceImpl.java
@@ -17,14 +17,15 @@
*/
package org.smartloli.kafka.eagle.api.im;
-import java.util.Date;
-
import org.smartloli.kafka.eagle.api.im.queue.DingDingJob;
+import org.smartloli.kafka.eagle.api.im.queue.LarkJob;
import org.smartloli.kafka.eagle.api.im.queue.MailJob;
import org.smartloli.kafka.eagle.api.im.queue.WeChatJob;
import org.smartloli.kafka.eagle.common.protocol.alarm.queue.BaseJobContext;
import org.smartloli.kafka.eagle.common.util.QuartzManagerUtils;
+import java.util.Date;
+
/**
* Implements IMService all method.
*
@@ -36,6 +37,15 @@ public class IMServiceImpl implements IMService {
private static final String KE_JOB_ID = "ke_job_id_";
+ /** Send Json msg by Lark. */
+ @Override
+ public void sendPostMsgByLark(String data, String url) {
+ BaseJobContext jobContext = new BaseJobContext();
+ jobContext.setData(data);
+ jobContext.setUrl(url);
+ QuartzManagerUtils.addJob(jobContext, KE_JOB_ID + new Date().getTime(), LarkJob.class, QuartzManagerUtils.getCron(new Date(), 5));
+ }
+
/** Send Json msg by dingding. */
@Override
public void sendPostMsgByDingDing(String data, String url) {
diff --git a/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/queue/LarkJob.java b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/queue/LarkJob.java
new file mode 100644
index 000000000..2970dd6d5
--- /dev/null
+++ b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/queue/LarkJob.java
@@ -0,0 +1,73 @@
+/**
+ * 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.smartloli.kafka.eagle.api.im.queue;
+
+import com.alibaba.fastjson.JSONObject;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.smartloli.kafka.eagle.common.protocol.alarm.queue.BaseJobContext;
+import org.smartloli.kafka.eagle.common.util.HttpClientUtils;
+import org.smartloli.kafka.eagle.common.util.KConstants.AlarmQueue;
+import org.smartloli.kafka.eagle.common.util.LoggerUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Add alarm message to lark job queue.
+ *
+ * @author Baiye.
+ *
+ * Created by Aug 10, 2022
+ */
+public class LarkJob implements Job {
+
+ /**
+ * Send alarm information by Lark.
+ */
+ public void execute(JobExecutionContext jobContext) throws JobExecutionException {
+ BaseJobContext bjc = (BaseJobContext) jobContext.getJobDetail().getJobDataMap().get(AlarmQueue.JOB_PARAMS);
+ sendMsg(bjc.getData(), bjc.getUrl());
+ }
+
+ private int sendMsg(String data, String url) {
+ try {
+ Map larkMessage = getLarkMessage(data);
+ String result = HttpClientUtils.doPostJson(url, JSONObject.toJSONString(larkMessage));
+ LoggerUtils.print(this.getClass()).info("Lark SendMsg Result: " + result);
+ } catch (Exception e) {
+ LoggerUtils.print(this.getClass()).error("Send alarm message has error by Lark, msg is ", e);
+ return 0;
+ }
+ return 1;
+ }
+
+ private static Map getLarkMessage(String text) {
+ Map map = new HashMap<>();
+ map.put("msg_type", "text");
+
+ Map textContainer = new HashMap<>();
+ textContainer.put("text", text);
+ map.put("content", textContainer);
+
+
+ return map;
+ }
+
+}
diff --git a/efak-api/src/main/java/org/smartloli/kafka/eagle/api/util/AlertUtils.java b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/util/AlertUtils.java
index 34b62ec28..fe4d9cde7 100644
--- a/efak-api/src/main/java/org/smartloli/kafka/eagle/api/util/AlertUtils.java
+++ b/efak-api/src/main/java/org/smartloli/kafka/eagle/api/util/AlertUtils.java
@@ -43,6 +43,11 @@ private AlertUtils() {
}
+ public static String sendTestMsgByLark(String url, String data) {
+ Map larkMessage = getLarkMessage(data);
+ return HttpClientUtils.doPostJson(url, JSONObject.toJSONString(larkMessage));
+ }
+
/**
* Send Json msg by wechat.
*/
@@ -51,6 +56,18 @@ public static String sendTestMsgByWeChat(String url, String data) {
return HttpClientUtils.doPostJson(url, JSONObject.toJSONString(wechatMarkdownMessage));
}
+ private static Map getLarkMessage(String text) {
+ Map map = new HashMap<>();
+ map.put("msg_type", "text");
+
+ Map textContainer = new HashMap<>();
+ textContainer.put("text", text);
+ map.put("content", textContainer);
+
+
+ return map;
+ }
+
private static Map getWeChatMarkdownMessage(String text) {
Map map = new HashMap<>();
map.put("msgtype", MARKDOWN);
diff --git a/efak-api/src/test/java/org/smartloli/kafka/eagle/api/sms/TestIM.java b/efak-api/src/test/java/org/smartloli/kafka/eagle/api/sms/TestIM.java
index 37a406988..1f02cc9f3 100644
--- a/efak-api/src/test/java/org/smartloli/kafka/eagle/api/sms/TestIM.java
+++ b/efak-api/src/test/java/org/smartloli/kafka/eagle/api/sms/TestIM.java
@@ -31,12 +31,29 @@
*/
public class TestIM {
public static void main(String[] args) {
- testAlarmClusterByDingDingMarkDownNewStyle();
+ // testAlarmClusterByDingDingMarkDownNewStyle();
// testAlarmClusterByDingDingMarkDown();
-// testAlarmClusterByWeChatMarkDown();
+ testAlarmClusterByLark();
}
+ private static void testAlarmClusterByLark() {
+ AlarmMessageInfo alarmMsg = new AlarmMessageInfo();
+ // FF0000 (red), 008000(green), FFA500(yellow)
+ alarmMsg.setTitle("`Kafka Eagle Alarm Notice`\n");
+ alarmMsg.setAlarmContent("node.shutdown [ localhost:9092 ]");
+ // alarmMsg.setAlarmContent("node.alive [
+ // localhost:9092 ]");
+ alarmMsg.setAlarmDate("2019-10-07 21:43:22");
+ alarmMsg.setAlarmLevel("P0");
+ alarmMsg.setAlarmProject("Kafka");
+ alarmMsg.setAlarmStatus("PROBLEM");
+ // alarmMsg.setAlarmStatus("NORMAL");
+ alarmMsg.setAlarmTimes("current(1), max(7)");
+
+ IMServiceImpl im = new IMServiceImpl();
+ im.sendPostMsgByLark(alarmMsg.toMail(), "xxx");
+ }
/**
* New alarm im api.
*/
diff --git a/efak-common/src/main/java/org/smartloli/kafka/eagle/common/util/KConstants.java b/efak-common/src/main/java/org/smartloli/kafka/eagle/common/util/KConstants.java
index 165432706..1ab0ff025 100644
--- a/efak-common/src/main/java/org/smartloli/kafka/eagle/common/util/KConstants.java
+++ b/efak-common/src/main/java/org/smartloli/kafka/eagle/common/util/KConstants.java
@@ -345,7 +345,7 @@ private Component() {
}
public interface AlarmType {
- public static String[] TYPE = new String[]{"DingDing", "WeChat", "Email"};
+ public static String[] TYPE = new String[]{"DingDing", "WeChat", "Email","Lark"};
public static String[] CLUSTER = new String[]{"Kafka", "Zookeeper", "Topic", "Producer"};
public static String[] LEVEL = new String[]{"P0", "P1", "P2", "P3"};
public static int[] MAXTIMES = new int[]{-1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
@@ -353,6 +353,7 @@ public interface AlarmType {
public static String EMAIL_TEST_TITLE = "Kafka Eagle Send Test MSG";
public static String DingDing = "DingDing";
public static String WeChat = "WeChat";
+ public static String LARK = "Lark";
public static String HTTP_GET = "get";
public static String HTTP_POST = "post";
public static String DISABLE = "N";
diff --git a/efak-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AlarmController.java b/efak-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AlarmController.java
index e87f6a784..013954e89 100644
--- a/efak-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AlarmController.java
+++ b/efak-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AlarmController.java
@@ -985,6 +985,8 @@ public void sendTestMsgAlarmConfig(@RequestBody AlarmEmailMockInfo emailMock, Ht
result = AlertUtils.sendTestMsgByDingDing(url, msg);
} else if (AlarmType.WeChat.equals(type)) {
result = AlertUtils.sendTestMsgByWeChat(url, msg);
+ } else if (AlarmType.LARK.equals(type)) {
+ result = AlertUtils.sendTestMsgByLark(url, msg);
}
byte[] output = result.getBytes();
BaseController.response(output, response);
diff --git a/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/alert/ClusterStrategyContext.java b/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/alert/ClusterStrategyContext.java
index 4bc15e479..0ee9903a3 100644
--- a/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/alert/ClusterStrategyContext.java
+++ b/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/alert/ClusterStrategyContext.java
@@ -178,6 +178,33 @@ public static void sendAlarmClusterError(AlarmConfigInfo alarmConfing, AlarmClus
alarmMsg.setAlarmTimes("current(" + cluster.getAlarmTimes() + "), max(" + cluster.getAlarmMaxTimes() + ")");
IMServiceImpl im = new IMServiceImpl();
im.sendPostMsgByWeChat(alarmMsg.toWeChatMarkDown(), alarmConfing.getAlarmUrl());
+ } else if (alarmConfing.getAlarmType().equals(KConstants.AlarmType.LARK)) {
+ AlarmMessageInfo alarmMsg = new AlarmMessageInfo();
+ alarmMsg.setAlarmId(cluster.getId());
+ alarmMsg.setAlarmCluster(alarmConfing.getCluster());
+ alarmMsg.setTitle("`EFAK - Alert Cluster Error`\n");
+ if (KConstants.AlarmType.TOPIC.equals(cluster.getType())) {
+ JSONObject alarmTopicMsg = JSON.parseObject(server);
+ String topic = alarmTopicMsg.getString("topic");
+ long alarmCapacity = alarmTopicMsg.getLong("alarmCapacity");
+ long realCapacity = alarmTopicMsg.getLong("realCapacity");
+ alarmMsg.setAlarmContent("topic.capacity.overflow [topic(" + topic + "), real.capacity(" + StrUtils.stringify(realCapacity) + "), alarm.capacity(" + StrUtils.stringify(alarmCapacity) + ")]");
+ } else if (KConstants.AlarmType.PRODUCER.equals(cluster.getType())) {
+ JSONObject alarmTopicMsg = JSON.parseObject(server);
+ String topic = alarmTopicMsg.getString("topic");
+ String alarmSpeeds = alarmTopicMsg.getString("alarmSpeeds");
+ long realSpeeds = alarmTopicMsg.getLong("realSpeeds");
+ alarmMsg.setAlarmContent("producer.speed.overflow [topic(" + topic + "), real.speeds(" + realSpeeds + "), alarm.speeds.range(" + alarmSpeeds + ")]");
+ } else {
+ alarmMsg.setAlarmContent("node.shutdown [ " + server + " ]");
+ }
+ alarmMsg.setAlarmDate(CalendarUtils.getDate());
+ alarmMsg.setAlarmLevel(cluster.getAlarmLevel());
+ alarmMsg.setAlarmProject(cluster.getType());
+ alarmMsg.setAlarmStatus("PROBLEM");
+ alarmMsg.setAlarmTimes("current(" + cluster.getAlarmTimes() + 1 + "), max(" + cluster.getAlarmMaxTimes() + ")");
+ IMServiceImpl im = new IMServiceImpl();
+ im.sendPostMsgByLark(alarmMsg.toMail(), alarmConfing.getAlarmUrl());
}
}
@@ -271,6 +298,33 @@ public static void sendAlarmClusterNormal(AlarmConfigInfo alarmConfing, AlarmClu
alarmMsg.setAlarmTimes("current(" + cluster.getAlarmTimes() + "), max(" + cluster.getAlarmMaxTimes() + ")");
IMServiceImpl im = new IMServiceImpl();
im.sendPostMsgByWeChat(alarmMsg.toWeChatMarkDown(), alarmConfing.getAlarmUrl());
+ } else if (alarmConfing.getAlarmType().equals(KConstants.AlarmType.LARK)) {
+ AlarmMessageInfo alarmMsg = new AlarmMessageInfo();
+ alarmMsg.setAlarmId(cluster.getId());
+ alarmMsg.setAlarmCluster(alarmConfing.getCluster());
+ alarmMsg.setTitle("`EFAK - Alert Cluster Error`\n");
+ if (KConstants.AlarmType.TOPIC.equals(cluster.getType())) {
+ JSONObject alarmTopicMsg = JSON.parseObject(server);
+ String topic = alarmTopicMsg.getString("topic");
+ long alarmCapacity = alarmTopicMsg.getLong("alarmCapacity");
+ long realCapacity = alarmTopicMsg.getLong("realCapacity");
+ alarmMsg.setAlarmContent("topic.capacity.overflow [topic(" + topic + "), real.capacity(" + StrUtils.stringify(realCapacity) + "), alarm.capacity(" + StrUtils.stringify(alarmCapacity) + ")]");
+ } else if (KConstants.AlarmType.PRODUCER.equals(cluster.getType())) {
+ JSONObject alarmTopicMsg = JSON.parseObject(server);
+ String topic = alarmTopicMsg.getString("topic");
+ String alarmSpeeds = alarmTopicMsg.getString("alarmSpeeds");
+ long realSpeeds = alarmTopicMsg.getLong("realSpeeds");
+ alarmMsg.setAlarmContent("producer.speed.overflow [topic(" + topic + "), real.speeds(" + realSpeeds + "), alarm.speeds.range(" + alarmSpeeds + ")]");
+ } else {
+ alarmMsg.setAlarmContent("node.shutdown [ " + server + " ]");
+ }
+ alarmMsg.setAlarmDate(CalendarUtils.getDate());
+ alarmMsg.setAlarmLevel(cluster.getAlarmLevel());
+ alarmMsg.setAlarmProject(cluster.getType());
+ alarmMsg.setAlarmStatus("PROBLEM");
+ alarmMsg.setAlarmTimes("current(" + cluster.getAlarmTimes() + 1 + "), max(" + cluster.getAlarmMaxTimes() + ")");
+ IMServiceImpl im = new IMServiceImpl();
+ im.sendPostMsgByLark(alarmMsg.toMail(), alarmConfing.getAlarmUrl());
}
}
}
diff --git a/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/sub/AlertConsumerSubTask.java b/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/sub/AlertConsumerSubTask.java
index a4931e95e..e254d5827 100644
--- a/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/sub/AlertConsumerSubTask.java
+++ b/efak-web/src/main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/sub/AlertConsumerSubTask.java
@@ -146,6 +146,12 @@ private void sendAlarmConsumerError(AlarmConfigInfo alarmConfing, AlarmConsumerI
alarmMsg.setAlarmStatus("PROBLEM");
IMServiceImpl im = new IMServiceImpl();
im.sendPostMsgByWeChat(alarmMsg.toWeChatMarkDown(), alarmConfing.getAlarmUrl());
+ } else if (alarmConfing.getAlarmType().equals(KConstants.AlarmType.LARK)) {
+ alarmMsg.setTitle("`EFAK - Alert Consumer Notice`\n");
+ alarmMsg.setAlarmContent("lag.overflow [ cluster(" + alarmConsumer.getCluster() + "), group(" + alarmConsumer.getGroup() + "), topic(" + alarmConsumer.getTopic() + "), current(" + lag + "), max(" + alarmConsumer.getLag() + ") ]");
+ alarmMsg.setAlarmStatus("PROBLEM");
+ IMServiceImpl im = new IMServiceImpl();
+ im.sendPostMsgByLark(alarmMsg.toMail(), alarmConfing.getAlarmUrl());
}
}
@@ -176,6 +182,12 @@ private void sendAlarmConsumerNormal(AlarmConfigInfo alarmConfing, AlarmConsumer
alarmMsg.setAlarmStatus("NORMAL");
IMServiceImpl im = new IMServiceImpl();
im.sendPostMsgByWeChat(alarmMsg.toWeChatMarkDown(), alarmConfing.getAlarmUrl());
+ } else if (alarmConfing.getAlarmType().equals(KConstants.AlarmType.LARK)) {
+ alarmMsg.setTitle("`EFAK - Alert Consumer Notice`\n");
+ alarmMsg.setAlarmContent("lag.normal [ cluster(" + alarmConsumer.getCluster() + "), group(" + alarmConsumer.getGroup() + "), topic(" + alarmConsumer.getTopic() + "), current(" + lag + "), max(" + alarmConsumer.getLag() + ") ]");
+ alarmMsg.setAlarmStatus("NORMAL");
+ IMServiceImpl im = new IMServiceImpl();
+ im.sendPostMsgByLark(alarmMsg.toMail(), alarmConfing.getAlarmUrl());
}
}
}
diff --git a/pom.xml b/pom.xml
index 62cf1b8c9..05ee77015 100644
--- a/pom.xml
+++ b/pom.xml
@@ -13,6 +13,8 @@
3.0.0
${project.build.directory}/endorsed
UTF-8
+ 1.8
+ 1.8