From b44ccf20aa01c19905eea37916a17d1cbe533a9b Mon Sep 17 00:00:00 2001 From: SheldonZheng Date: Thu, 11 Aug 2022 11:24:33 +0800 Subject: [PATCH] add lark bot support --- .../kafka/eagle/api/im/IMService.java | 4 +- .../kafka/eagle/api/im/IMServiceImpl.java | 14 +++- .../kafka/eagle/api/im/queue/LarkJob.java | 73 +++++++++++++++++++ .../kafka/eagle/api/util/AlertUtils.java | 17 +++++ .../smartloli/kafka/eagle/api/sms/TestIM.java | 21 +++++- .../kafka/eagle/common/util/KConstants.java | 3 +- .../eagle/web/controller/AlarmController.java | 2 + .../task/alert/ClusterStrategyContext.java | 54 ++++++++++++++ .../shard/task/sub/AlertConsumerSubTask.java | 12 +++ pom.xml | 2 + 10 files changed, 196 insertions(+), 6 deletions(-) create mode 100644 efak-api/src/main/java/org/smartloli/kafka/eagle/api/im/queue/LarkJob.java 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