diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/procesor/AlarmMessageProcessor.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/procesor/AlarmMessageProcessor.java new file mode 100644 index 000000000..668cc1e75 --- /dev/null +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/procesor/AlarmMessageProcessor.java @@ -0,0 +1,157 @@ +package com.ai.cloud.skywalking.alarm.procesor; + +import com.ai.cloud.skywalking.alarm.model.AlarmRule; +import com.ai.cloud.skywalking.alarm.model.ApplicationInfo; +import com.ai.cloud.skywalking.alarm.model.UserInfo; +import com.ai.cloud.skywalking.alarm.model.parameter.Application; +import com.ai.cloud.skywalking.alarm.util.MailUtil; +import com.ai.cloud.skywalking.alarm.util.RedisUtil; +import freemarker.template.Configuration; +import freemarker.template.Template; +import freemarker.template.TemplateException; +import freemarker.template.Version; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import redis.clients.jedis.Jedis; + +import java.io.IOException; +import java.io.StringReader; +import java.io.StringWriter; +import java.text.SimpleDateFormat; +import java.util.*; + +public class AlarmMessageProcessor { + + private static Logger logger = LogManager.getLogger(AlarmMessageProcessor.class); + + public static void process(UserInfo userInfo, AlarmRule rule) { + Set sentData = new HashSet(); + List applications = new ArrayList(); + Set toBeSendData; + Application temApplication; + long currentFireTimeM = System.currentTimeMillis() / (10000 * 6); + // 获取待发送数据 + if (checkerProcessInterval(rule, currentFireTimeM)) { + for (ApplicationInfo applicationInfo : rule.getApplicationInfos()) { + toBeSendData = new HashSet(); + + for (int i = 0; i < currentFireTimeM - rule.getPreviousFireTimeM(); i++) { + toBeSendData.addAll(getAlarmMessage(generateAlarmKey(userInfo.getUserId(), + applicationInfo.getAppCode(), i))); + + toBeSendData.removeAll(sentData); + sentData.addAll(toBeSendData); + } + + temApplication = new Application(applicationInfo.getAppId()); + temApplication.setTraceIds(toBeSendData); + applications.add(temApplication); + } + + // 没有数据需要发送 + if (sentData.size() <= 0) { + // 清理数据 + for (ApplicationInfo applicationInfo : rule.getApplicationInfos()) { + for (int i = 0; i < currentFireTimeM - rule.getPreviousFireTimeM(); i++) { + expiredAlarmMessage(generateAlarmKey(userInfo.getUserId(), + applicationInfo.getAppCode(), i)); + } + } + // 修改-保存上次处理时间 + rule.setPreviousFireTimeM(currentFireTimeM); + savePreviousFireTime(userInfo.getUserId(), rule.getRuleId(), currentFireTimeM); + } else { + if ("0".equals(rule.getTodoType())) { + // 发送邮件 + String subjects = generateSubjects(sentData.size(), + rule.getPreviousFireTimeM(), currentFireTimeM); + Map parameter = new HashMap(); + parameter.put("applications", applications); + parameter.put("name", userInfo.getUserId()); + MailUtil.sendMail(rule.getConfigArgsDescriber().getMailInfo().getMailTo(), + rule.getConfigArgsDescriber().getMailInfo().getMailCc(), + generateContent(rule.getConfigArgsDescriber().getMailInfo() + .getMailTemp(), parameter), + subjects); + } + // 清理数据 + for (ApplicationInfo applicationInfo : rule.getApplicationInfos()) { + for (int i = 0; i < currentFireTimeM - rule.getPreviousFireTimeM(); i++) { + expiredAlarmMessage(generateAlarmKey(userInfo.getUserId(), + applicationInfo.getAppCode(), i)); + } + } + // 修改-保存上次处理时间 + rule.setPreviousFireTimeM(currentFireTimeM); + savePreviousFireTime(userInfo.getUserId(), rule.getRuleId(), currentFireTimeM); + } + } + + + } + + private static String generateSubjects(int count, long startTime, long endTime) { + String title = "[Warning] There were " + count + " alarm information between " + + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(startTime)) + + " to " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(endTime)); + + return title; + } + + private static boolean checkerProcessInterval(AlarmRule rule, long currentFireTimeM) { + return currentFireTimeM - rule.getPreviousFireTimeM() >= rule.getConfigArgsDescriber().getPeriod(); + } + + private static void expiredAlarmMessage(String key) { + Jedis client = RedisUtil.getRedisClient(); + client.expire(key, 0); + if (client != null) { + client.close(); + } + } + + + private static void savePreviousFireTime(String userId, String ruleId, long currentFireTimeM) { + Jedis client = RedisUtil.getRedisClient(); + client.hset(userId, ruleId, String.valueOf(currentFireTimeM)); + if (client != null) { + client.close(); + } + } + + + private static String generateAlarmKey(String userId, String appCode, int period) { + return userId + "-" + appCode + "-" + ((System.currentTimeMillis() / (10000 * 6)) + - period); + } + + private static Collection getAlarmMessage(String key) { + Jedis client = RedisUtil.getRedisClient(); + Map result = client.hgetAll(key); + if (result == null) { + return new ArrayList(); + } + + client.close(); + + return result.values(); + } + + private static String generateContent(String templateStr, Map parameter) { + Configuration cfg = new Configuration(new Version("2.3.23")); + cfg.setDefaultEncoding("UTF-8"); + Template t = null; + try { + t = new Template(null, new StringReader(templateStr), cfg); + StringWriter out = new StringWriter(); + t.process(parameter, out); + return out.getBuffer().toString(); + } catch (IOException e) { + logger.error("Template illegal.", e); + } catch (TemplateException e) { + logger.error("Failed to generate content.", e); + } + + return ""; + } +} diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/MailUtil.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/MailUtil.java new file mode 100644 index 000000000..6611d3390 --- /dev/null +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/MailUtil.java @@ -0,0 +1,65 @@ +package com.ai.cloud.skywalking.alarm.util; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import javax.mail.*; +import javax.mail.internet.AddressException; +import javax.mail.internet.InternetAddress; +import javax.mail.internet.MimeMessage; +import java.util.Properties; + +public class MailUtil { + + private static Logger logger = LogManager.getLogger(MailUtil.class); + + private static Session session; + private static String sendAccount; + private static Transport ts; + + static { + try { + Properties prop = new Properties(); + prop.load(MailUtil.class.getResourceAsStream("/mail/mail.config")); + session = Session.getInstance(prop); + ts = session.getTransport(); + ts.connect(prop.getProperty("mail.host"), prop.getProperty("mail.username"), prop.getProperty("mail.password")); + sendAccount = prop.getProperty("mail.username") + prop.getProperty("mail.account.prefix"); + } catch (Exception e) { + logger.error("Failed to connect the mail System.", e); + System.exit(-1); + } + } + + + public static void sendMail(String[] recipientAccounts, String[] ccList, String context, String title) { + try { + MimeMessage message = new MimeMessage(session); + message.setFrom(new InternetAddress(sendAccount)); + InternetAddress[] recipientAccountArray = new InternetAddress[recipientAccounts.length]; + for (int i = 0; i < recipientAccounts.length; i++) { + recipientAccountArray[i] = new InternetAddress(recipientAccounts[i]); + } + message.addRecipients(Message.RecipientType.TO, recipientAccountArray); + + InternetAddress[] ccAccountArray = new InternetAddress[ccList.length]; + for (int i = 0; i < recipientAccounts.length; i++) { + ccAccountArray[i] = new InternetAddress(ccList[i]); + } + message.addRecipients(Message.RecipientType.CC, ccAccountArray); + + message.setSubject(title); + message.setContent(context, "text/html;charset=UTF-8"); + + ts.sendMessage(message, message.getAllRecipients()); + } catch (AddressException e) { + logger.error("Recipient Account is not correct.", e); + } catch (NoSuchProviderException e) { + logger.error("Failed to send mail.", e); + } catch (MessagingException e) { + logger.error("Failed to send mail.", e); + } + } + + +} diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/RedisUtil.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/RedisUtil.java new file mode 100644 index 000000000..1f829923c --- /dev/null +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/RedisUtil.java @@ -0,0 +1,46 @@ +package com.ai.cloud.skywalking.alarm.util; + +import com.ai.cloud.skywalking.alarm.conf.Config; +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; + +public class RedisUtil { + + private static Logger logger = LogManager.getLogger(RedisUtil.class); + private static JedisPool jedisPool; + private static String[] config; + + static { + GenericObjectPoolConfig genericObjectPoolConfig = buildGenericObjectPoolConfig(); + String redisServerConfig = Config.Alarm.REDIS_SERVER; + if (redisServerConfig == null || redisServerConfig.length() <= 0) { + logger.error("Redis server is not setting. Switch off alarm module. "); + } else { + config = redisServerConfig.split(":"); + if (config.length != 2) { + logger.error("Redis server address is illegal setting, need to be 'ip:port'. Switch off alarm module. "); + Config.Alarm.ALARM_OFF_FLAG = true; + } else { + jedisPool = new JedisPool(genericObjectPoolConfig, config[0], + Integer.valueOf(config[1])); + } + } + } + + private static GenericObjectPoolConfig buildGenericObjectPoolConfig() { + GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig(); + genericObjectPoolConfig.setTestOnBorrow(true); + genericObjectPoolConfig.setMaxIdle(Config.Alarm.REDIS_MAX_IDLE); + genericObjectPoolConfig.setMinIdle(Config.Alarm.REDIS_MIN_IDLE); + genericObjectPoolConfig.setMaxTotal(Config.Alarm.REDIS_MAX_TOTAL); + return genericObjectPoolConfig; + } + + public static Jedis getRedisClient(){ + return jedisPool.getResource(); + } + +}