简易日志告警系统
1、需求分析
随着公司业务发展,支撑公司业务的各种系统越来越多,为了保证公司的业务正常发展,急需要对这些线上系统的运行进行监控,做到问题的及时发现和处理,最大程度减少对业务的影响。
目前系统分类有:
- 有基于Tomcat的web应用
- 有独立的Java Application应用
- 有运行在linux上的脚本程序
- 有大规模的集群框架(zookeeper、Hadoop、Storm、SRP……)
- 有操作系统的运行日志
主要功能需求分为:监控系统日志中的内容,按照一定规则进行过滤,发现问题之后通过短信和邮件进行告警。
2、功能分析
- 数据输入:使用flume客户端获取个系统的数据; 用户通过页面输入系统名称、负责人触发规则等信息。
- 数据存储:使用flume采集数据并存放在kafka集群中。
- 数据计算:使用storm编写程序对日志进行过滤,将满足过滤规则的信息,通过邮件短信告警并保存到数据库中。
- 数据展示:管理页面可以查看触发规则的信息,系统负责人,联系方式,触发信息明细等。
3、架构设计
3.1、整体架构设计
主要架构为应用+flume+kafka+storm+mysql+Java web。数据流程如下:
- 应用程序使用log4j产生日志
- 部署flume客户端监控应用程序产生的日志信息,并发送到kafka集群中
- storm spout拉去kafka的数据进行消费,逐条过滤每条日志的进行规则判断,对符合规则的日志进行邮件告警。
- 最后将告警的信息保存到mysql数据库中,用来进行管理。
3.2、Flume设计
使用 Flume EXEC执行一个linux命令来生成数据源。例如,可以用tail命令监控一个文件,那么,只要文件增加内容,EXEC就可以将增加的内容作为数据源发送出去。
使用 org.apache.flume.plugins.KafkaSink,将Flume EXEC产生的数据源发送到Kafka中。
3.3、Kafka设计
部署kafka集群,在集群中添加一个Topic:monitor_realtime_javaxy
3.4、Storm设计
- KafkaSpout读取数据,需要配置Topic:monitor_realtime_javaxy
- FilterBolt判断规则
- NotifyBolt用来发送邮件或短信息
- Save2DB用来将告警信息写入mysql数据库
3.5、数据模型设计
3.5.1、用户表
用来保存用户的信息,包括账号、手机号码、邮箱、是否有效等信息。
3.5.2、应用表
用来保存应用的信息,包括应用名称、应用描述、应用是否在线等信息。
3.5.3、应用类型表
用来保存应用的类型等信息。
3.5.4、规则表
用来保存规则的信息,包括规则名称,规则描述,规则关键词等信息。
3.5.5、规则记录表
用来保存触发规则后的记录,包括告警编号、是否短信告知、是否邮件告知、告警明细等信息。
4、代码开发
4.1、整体结构
4.2、LogMonitorDao
LogMonitorDao用于操作数据库,用来操作数据模型。
public class LogMonitorDao {private JdbcTemplate jdbcTemplate=new JdbcTemplate(DataSourceUtil.getDataSource());// 查询所有规则信息public List<Rule> getRuleList() {String sql = "SELECT `id`,`name`,`keyword`,`isValid`,`appId` FROM `log_monitor`.`log_monitor_rule` WHERE isValid =1";return jdbcTemplate.query(sql, new BeanPropertyRowMapper<Rule>(Rule.class));}//查询所有应用的信息public List<App> getAppList() {String sql = "SELECT `id`,`name`,`isOnline`,`typeId`,`userId` FROM `log_monitor`.`log_monitor_app` WHERE isOnline =1";return jdbcTemplate.query(sql, new BeanPropertyRowMapper<App>(App.class));}//插叙所有用户的信息public List<User> getUserList() {String sql = "SELECT `id`,`name`,`mobile`,`email`,`isValid` FROM `log_monitor`.`log_monitor_user` WHERE isValid =1";return jdbcTemplate.query(sql, new BeanPropertyRowMapper<User>(User.class));}//插入触发规则的信息public void saveRecord(Record record) {String sql = "INSERT INTO `log_monitor`.`log_monitor_rule_record` (`appId`,`ruleId`,`isEmail`,`isPhone`,`isColse`,`noticeInfo`,`updataDate`) VALUES ( ?,?,?,?,?,?,?)";jdbcTemplate.update(sql, record.getAppId(), record.getRuleId(), record.getIsEmail(), record.getIsPhone(), 0, record.getLine(), new Date());}
}
4.3、数据库模型类
public class App {private int id;//应用编号private String name;//应用名称private int isOnline;//应用是否在线private int typeId;//应用所属类别private String userId;//应用的负责人,多个用户用逗号分开
}
public class Message implements Serializable {private String appId;//消息所属服务器编号private String line;//消息内容private String ruleId;//规则编号private String keyword;//规则中的关键词private int isEmail;//是否已发送邮件private int isPhone;//是否已发送短信private String appName;//应用的名称
}
//触发报警之后的记录
public class Record {private int id;//告警信息编号private int appId;//告警信息所属的应用private int ruleId;//告警信息所属的规则private int isEmail;//告警信息是否通过邮件告警private int isPhone;//告警信息是否通过短信告警private int isColse;//告警信息是否处理完毕private String line;//原始日志信息
}
public class Rule {private int id;//规则编号private String name;//规则名称private String keyword;//规则过滤的关键字private int isValid;//规则是否可用private int appId;//规则所属的应用
}
public class User {private int id;//用户编号private String name;//用户名称private String mobile;//用户手机private String email;//用户邮箱private int isValid;//用户是否可用
}
4.4、邮件发送
public class MessageSender {private static final Logger logger = Logger.getLogger(MessageSender.class);//发送邮件-邮件内容为文本格式public static boolean sendMail(MailInfo mailInfo) {try {Message mailMessage = generateBaseInfo(mailInfo);mailMessage.setText(mailInfo.getMailContent());// 设置邮件消息的主要内容Transport.send(mailMessage); // 发送邮件logger.info("【 TEXT 邮件发送完毕,成功时间: " + System.currentTimeMillis() + " 】");return true;} catch (MessagingException ex) {ex.printStackTrace();} catch (UnsupportedEncodingException e) {e.printStackTrace();}return false;}//邮件内容为html格式public static boolean sendHtmlMail(MailInfo mailInfo) {try {Message mailMessage = generateBaseInfo(mailInfo);Multipart mainPart = new MimeMultipart();// MiniMultipart类是一个容器类,包含MimeBodyPart类型的对象BodyPart html = new MimeBodyPart();// 创建一个包含HTML内容的MimeBodyParthtml.setContent(mailInfo.getMailContent(), "text/html; charset=utf-8");// 设置HTML内容mainPart.addBodyPart(html);mailMessage.setContent(mainPart);// 将MiniMultipart对象设置为邮件内容Transport.send(mailMessage);// 发送邮件logger.info("【 HTML 邮件发送完毕,成功时间: " + System.currentTimeMillis() + " 】");return true;} catch (MessagingException e) {e.printStackTrace();} catch (UnsupportedEncodingException e) {e.printStackTrace();}return false;}public static Message generateBaseInfo(MailInfo mailInfo) throws UnsupportedEncodingException, MessagingException {// 判断是否需要身份认证MailAuthenticator authenticator = null;Properties pro = mailInfo.getProperties();// 如果需要身份认证,则创建一个密码验证器if (mailInfo.isAuthValidate()) {authenticator = new MailAuthenticator(mailInfo.getUserName(),mailInfo.getUserPassword());}// 根据邮件会话属性和密码验证器构造一个发送邮件的sessionSession sendMailSession = Session.getDefaultInstance(pro, authenticator);Message mailMessage = null;mailMessage = new MimeMessage(sendMailSession); // 根据session创建一个邮件消息Address from = new InternetAddress(mailInfo.getFromAddress(), mailInfo.getFromUserName()); // 创建邮件发送者地址mailMessage.setFrom(from); // 设置邮件消息的发送者if (mailInfo.getToAddress() != null && mailInfo.getToAddress().contains(",")) {mailMessage.setRecipients(Message.RecipientType.TO, InternetAddress.parse(mailInfo.getToAddress()));// Message.RecipientType.TO属性表示接收者的类型为TO} else {Address to = new InternetAddress(mailInfo.getToAddress()); // 创建邮件的接收者地址,并设置到邮件消息中mailMessage.setRecipient(Message.RecipientType.TO, to);// Message.RecipientType.TO属性表示接收者的类型为TO}if (StringUtils.isNotBlank(mailInfo.getCcAddress())) {if (mailInfo.getCcAddress().contains(",")) {mailMessage.setRecipients(Message.RecipientType.CC, InternetAddress.parse(mailInfo.getCcAddress())); // Message.RecipientType.CC属性表示接收者的类型为CC} else {Address cc = new InternetAddress(mailInfo.getCcAddress()); // 创建邮件的抄送者地址,并设置到邮件消息中mailMessage.setRecipient(Message.RecipientType.CC, cc); // Message.RecipientType.CC属性表示接收者的类型为CC}}mailMessage.setSubject(mailInfo.getMailSubject());// 设置邮件消息的主题mailMessage.setSentDate(new Date());// 设置邮件消息发送的时间return mailMessage;}
}
public class MailInfo {private String mailServerHost; // 发送邮件的服务器的IPprivate String mailServerPort = "25"; // 发送邮件的服务器端口private String userName; // 登陆邮件发送服务器的用户名private String userPassword; // 登陆邮件发送服务器的密码private String fromAddress; // 邮件发送者的地址private String toAddress; // 邮件接收者的地址private String ccAddress; // 邮件抄送者的地址private String fromUserName = "日志监控平台"; // 邮件发送者的名称,显示在他人邮件的发件人private String mailSubject; // 邮件主题private String mailContent; // 邮件的文本内容private boolean authValidate = true; // 是否需要身份验证private Properties properties; // 邮件会话属性
}
4.5、日志核心处理
public class MonitorHandler {private static Logger logger = Logger.getLogger(MonitorHandler.class);//定义一个map,其中appId为Key,以该appId下的所有rule为Valueprivate static Map<String, List<Rule>> ruleMap;//定义一个map,其中appId为Key,以该appId下的所有user为Valueprivate static Map<String, List<User>> userMap;//定义一个list,用来封装所有的应用信息private static List<App> applist;//定义一个list,用来封装所有的用户信息private static List<User> userList;//定时加载配置文件的标识private static boolean reloaded = false;//定时加载配置文件的标识private static long nextReload = 0l;static {load();}/*** 解析输入的日志,将数据按照一定的规则进行分割。* 判断日志是否合法,主要校验日志所属应用的appId是否存在** @param line 一条日志*/public static Message parser(String line) {//日志内容分为两个部分:由5个$$$$$符号作为分隔符,第一部分为appid,第二部分为日志内容。String[] messageArr = line.split("\\$\\$\\$\\$\\$");//对日志进行校验if (messageArr.length != 2) {return null;}if (StringUtils.isBlank(messageArr[0]) || StringUtils.isBlank(messageArr[1])) {return null;}//检验当前日志所属的appid是否是经过授权的。if (apppIdisValid(messageArr[0].trim())) {Message message = new Message();message.setAppId(messageArr[0].trim());message.setLine(messageArr[1]);return message;}return null;}/*** 验证appid是否经过授权*/private static boolean apppIdisValid(String appId) {try {for (App app : applist) {if (app.getId() == Integer.parseInt(appId)) {return true;}}} catch (Exception e) {return false;}return false;}/*** 对日志进行规制判定,看看是否触发规则* @param message* @return*/public static boolean trigger(Message message) {//如果规则模型为空,需要初始化加载规则模型if (ruleMap == null) {load();}//从规则模型中获取当前appid配置的规则List<Rule> keywordByAppIdList = ruleMap.get(message.getAppId());for (Rule rule : keywordByAppIdList) {//如果日志中包含过滤过的关键词,即为匹配成功if (message.getLine().contains(rule.getKeyword())) {message.setRuleId(rule.getId() + "");message.setKeyword(rule.getKeyword());return true;}}return false;}/*** 加载数据模型,主要是用户列表、应用管理表、组合规则模型、组合用户模型。*/public static synchronized void load() {if (userList == null) {userList = loadUserList();}if (applist == null) {applist = loadAppList();}if (ruleMap == null) {ruleMap = loadRuleMap();}if (userMap == null) {userMap = loadUserMap();}}/*** 访问数据库获取所有有效的app列表* @return*/private static List<App> loadAppList() {return new LogMonitorDao().getAppList();}/*** 访问数据库获取所有有效用户的列表*/private static List<User> loadUserList() {return new LogMonitorDao().getUserList();}/*** 封装应用与用户对应的map*/private static Map<String, List<User>> loadUserMap() {//以应用的appId为key,以应用的所有负责人的userList对象为value。//HashMap<String, List<User>>HashMap<String, List<User>> map = new HashMap<String, List<User>>();for (App app : applist) {String userIds = app.getUserId();List<User> userListInApp = map.get(app.getId());if (userListInApp == null) {userListInApp = new ArrayList<User>();map.put(app.getId() + "", userListInApp);}String[] userIdArr = userIds.split(",");for (String userId : userIdArr) {userListInApp.add(queryUserById(userId));}map.put(app.getId() + "", userListInApp);}return map;}/*** 封装应用与规则的map*/private static Map<String, List<Rule>> loadRuleMap() {Map<String, List<Rule>> map = new HashMap<String, List<Rule>>();LogMonitorDao logMonitorDao = new LogMonitorDao();List<Rule> ruleList = logMonitorDao.getRuleList();//将代表rule的list转化成一个map,转化的逻辑是,// 从rule.getAppId作为map的key,然后将rule对象作为value传入map//Map<appId,ruleList> 一个appid的规则信息,保存在一个list中。for (Rule rule : ruleList) {List<Rule> ruleListByAppId = map.get(rule.getAppId()+"");if (ruleListByAppId == null) {ruleListByAppId = new ArrayList<Rule>();map.put(rule.getAppId() + "", ruleListByAppId);}ruleListByAppId.add(rule);map.put(rule.getAppId() + "", ruleListByAppId);}return map;}/*** 通过用户编号获取用户的JavaBean*/private static User queryUserById(String userId) {for (User user : userList) {if (user.getId() == Integer.parseInt(userId)) {return user;}}return null;}/*** 通过app编号,获取当前app的所有负责人列表* @param appId* @return*/public static List<User> getUserIdsByAppId(String appId) {return userMap.get(appId);}/*** 告警模块,用来发送邮件和短信* 短信功能由于短信资源匮乏,目前默认返回已发送。*/public static void notifly(String appId, Message message) {//通过appId获取应用负责人的对象List<User> users = getUserIdsByAppId(appId);//发送邮件if (sendMail(appId, users, message)) {message.setIsEmail(1);}//发送短信if (sendSMS(appId, users, message)) {message.setIsPhone(1);}}/*** 发送短信的模块* 由于短信资源匮乏,目前该功能不开启,默认true,即短信发送成功。* 目前发送短信功能使用的是外部接口,外面接口的并发性没法保证,会影响storm程序运行的效率。* 后期可以改造为将短信数据发送到外部的消息队里中,然后创建一个worker去发送短信。*/private static boolean sendSMS(String appId, List<User> users, Message message) {List<String> mobileList = new ArrayList<String>();for (User user : users) {mobileList.add(user.getMobile());}for (App app : applist) {if (app.getId() == Integer.parseInt(appId.trim())) {message.setAppName(app.getName());break;}}String content = "系统【" + message.getAppName() + "】在 " + DateUtils.getDateTime() + " 触发规则 " + message.getRuleId() + ",关键字:" + message.getKeyword();return SMSBase.sendSms(listToStringFormat(mobileList), content);}/*** 发送邮件* 后期可以改造为将邮件数据发送到外部的消息队里中,然后创建一个worker去发送短信。*/private static boolean sendMail(String appId, List<User> userList, Message message) {List<String> receiver = new ArrayList<String>();for (User user : userList) {receiver.add(user.getEmail());}for (App app : applist) {if (app.getId() == Integer.parseInt(appId.trim())) {message.setAppName(app.getName());break;}}if (receiver.size() >= 1) {String date = DateUtils.getDateTime();String content = "系统【" + message.getAppName() + "】在 " + date + " 触发规则 " + message.getRuleId() + " ,过滤关键字为:" + message.getKeyword() + " 错误内容:" + message.getLine();MailInfo mailInfo = new MailInfo("系统运行日志监控", content, receiver, null);return MessageSender.sendMail(mailInfo);}return false;}/*** 保存触发规则的信息,将触发信息写入到mysql数据库中。*/public static void save(Record record) {new LogMonitorDao().saveRecord(record);}/*** 将list转换为String* @param list*/private static String listToStringFormat(List<String> list) {StringBuilder stringBuilder = new StringBuilder();for (int i = 0; i < list.size(); i++) {if (i == list.size() - 1) {stringBuilder.append(list.get(i));} else {stringBuilder.append(list.get(i)).append(",");}}return stringBuilder.toString();}/*** 配置scheduleLoad重新加载底层数据模型。*/public static synchronized void reloadDataModel() {
// * thread 1 reloaded = true ----> reloaded = false
// * thread 2 reloaded = false
// * thread 3 reloaded = false
// * thread 4 reloaded = falseif (reloaded) {long start = System.currentTimeMillis();userList = loadUserList();applist = loadAppList();ruleMap = loadRuleMap();userMap = loadUserMap();reloaded = false;nextReload = 0l;logger.info("配置文件reload完成,时间:"+DateUtils.getDateTime()+" 耗时:"+ (System.currentTimeMillis()-start));}}/*** 定时加载配置信息* 配合reloadDataModel模块一起使用。* 主要实现原理如下:* 1,获取分钟的数据值,当分钟数据是10的倍数,就会触发reloadDataModel方法,简称reload时间。* 2,reloadDataModel方式是线程安全的,在当前worker中只有一个线程能够操作。* 3,为了保证当前线程操作完毕之后,其他线程不再重复操作,设置了一个标识符reloaded。* 在非reload时间段时,reloaded一直被置为true;* 在reload时间段时,第一个线程进入reloadDataModel后,加载完毕之后会将reloaded置为false。*/public static void scheduleLoad() {String date = DateUtils.getDateTime();int now = Integer.parseInt(date.split(":")[1]);//每10分钟加载一次if (now % 10 == 0) {//1,2,3,4,5,6reloadDataModel();}else {reloaded = true;}if (System.currentTimeMillis()==nextReload){//thread 1,2,3,reloadDataModel();}}
}
4.6、Storm Bolt
/*** Describe: 过滤规则信息* /BaseRichBolt 需要手动调ack方法,BaseBasicBolt由storm框架自动调ack方法*/
public class FilterBolt extends BaseBasicBolt {private static Logger logger = Logger.getLogger(FilterBolt.class);@Overridepublic void prepare(Map stormConf, TopologyContext context) {super.prepare(stormConf, context);}public void execute(Tuple input, BasicOutputCollector collector) {//获取kafka发送的数据,是一个byte数组byte[] value = (byte[]) input.getValue(0);//将数组转化成字符串String line = new String(value);//对数据进行解析Message message = MonitorHandler.parser(line);if (message == null) {return;}if (MonitorHandler.trigger(message)) {collector.emit(new Values(message.getAppId(), message));}//定时更新规则信息MonitorHandler.scheduleLoad();}public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("appId", "message"));}
}
/*** Describe: 将触发信息保存到mysql数据库中*/
public class PrepareRecordBolt extends BaseBasicBolt {public void execute(Tuple input, BasicOutputCollector collector) {Message message = (Message) input.getValueByField("message");String appId = input.getStringByField("appId");//将触发规则的信息进行通知MonitorHandler.notifly(appId, message);Record record = new Record();try {BeanUtils.copyProperties(record, message);collector.emit(new Values(record));} catch (Exception e) {}}public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("record"));}}
public class SaveMessage2MySql extends BaseBasicBolt {private static Logger logger = Logger.getLogger(SaveMessage2MySql.class);public void execute(Tuple input, BasicOutputCollector collector) {Record record = (Record) input.getValueByField("record");MonitorHandler.save(record);}public void declareOutputFields(OutputFieldsDeclarer declarer) {}
}
4.7、Storm驱动类
public class LogMonitorTopologyMain {public static void main(String[] args) throws Exception{// 使用TopologyBuilder进行构建驱动类TopologyBuilder builder = new TopologyBuilder();// 设置kafka的zookeeper集群BrokerHosts hosts = new ZkHosts("zk01:2181,zk02:2181,zk03:2181");// 初始化配置信息SpoutConfig spoutConfig = new SpoutConfig(hosts, "logmonitor", "/aaa", "log_monitor");// 在topology中设置spoutbuilder.setSpout("kafka-spout", new KafkaSpout(spoutConfig),3);builder.setBolt("filter-bolt",new FilterBolt(),3).shuffleGrouping("kafka-spout");builder.setBolt("prepareRecord-bolt",new PrepareRecordBolt(),2).fieldsGrouping("filter-bolt", new Fields("appId"));builder.setBolt("saveMessage-bolt",new SaveMessage2MySql(),2).shuffleGrouping("prepareRecord-bolt");//启动topology的配置信息Config topologConf = new Config();//TOPOLOGY_DEBUG(setDebug), 当它被设置成true的话, storm会记录下每个组件所发射的每条消息。//这在本地环境调试topology很有用, 但是在线上这么做的话会影响性能的。topologConf.setDebug(true);//storm的运行有两种模式: 本地模式和分布式模式.if (args != null && args.length > 0) {//定义你希望集群分配多少个工作进程给你来执行这个topologytopologConf.setNumWorkers(2);//向集群提交topologyStormSubmitter.submitTopologyWithProgressBar(args[0], topologConf, builder.createTopology());} else {topologConf.setMaxTaskParallelism(3);LocalCluster cluster = new LocalCluster();cluster.submitTopology("word-count", topologConf, builder.createTopology());Utils.sleep(10000000);cluster.shutdown();}}
}