基于SpringBoot轻量非侵入式数据库数据告警器

Posted 山河已无恙

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了基于SpringBoot轻量非侵入式数据库数据告警器相关的知识,希望对你有一定的参考价值。

基于SpringBoot的轻量、非侵入式数据库数据告警器


我的需求:

需要写一个数据库数据监控的告警小工具,要求:

  • 非侵入式的,对监控的数据只有查询权限,没有写权限
  • 可以对数据表的部分数据状态,数据数量进行监控告警
  • 监控数据,告警条件等是可配置的

我需要解决的问题:

  • 抽象告警行为,解耦告警流程构建过程
  • 告警命中之后如何避免重复告警
  • 可配置的部分如何从流程代码中解耦为配置
  • 如何动态配置告警扫描计划

我是这样做的:


整体来讲,逻辑很简单,没啥技术难点,属于重复造轮子,考虑到需要解析配置文件、多数据源配置,定时任务等,所以使用SpringBoot,利用其自动化配置,类型安全配置属性,集成简单的任务调度等优点,可以方便地的配置不同的数据源,同时将复杂配置文件中的数据注入Bean中,动态配置定时计划

关于多数据源配置和类型安全配置属性等不是本文重点,这里不多讲。

编码思路:

  • 一是解耦告警器类的构建和构建步骤
  • 二是解耦告警流程,涉及的单个行为从流程解耦,对于行为可变的部分从代码解耦为配置文件。
  • 三是对于告警缓存的处理,非侵入式需要解决重复告警,当前集成了H2,但是没有使用,感觉有点重,所以利用WeakHashMap构建了一个弱键的缓存工具类来实现。

解耦告警器类的构建和构建步骤

对于告警器类的构建,涉及初始化告警规则生成两部分,初始化负责告警配置文件加载解析校验,告警规则生成负责告警流程的建立。

这里可以使用默认的初始化规则,和告警解析规程,也可以使用自定义的规则。整体上编码基于构建者设计模式,类似于Spring Security配置对象的构建

可以使用默认的告警解析流程,调用方式

alarms.alarmStart()

或者

alarms.alarmsInit(null).alarmsRun(null);

也可以通过自定义告警解析流程,这里采用函数式编程的思想,通过行为参数化的方式,可以动态编写告警解析流程。

// 告警器初始化
 alarms.alarmsInit(alarmsInit -> 
                logger.info("告警器扫描时间周期cron:" + alarmsInit.getMinute());
                alarmsInit.getAlarms().forEach((alarm -> 
                    logger.info("加载的告警器名称:" + alarm.getItemsName());
                    logger.info("触发器:" + alarm.getTrigger());
                    logger.info("动作:" + Arrays.toString(alarm.getActions()));
                    logger.info("告警媒介:" + Arrays.toString(alarm.getMediaType()));
                    logger.info("告警内容:" + alarm.getMedia());
                    logger.info("告警短信插表SQL:" + alarm.getMediaSql());
                ));
                return alarmsInit;
            // 告警规则生成
            ).alarmsRun(alarmsRun -> 
                logger.info("告警器扫描......");
                alarmsRun.getAlarms().forEach(alarm -> 
                    Boolean boo = Long.class.cast(jdbcTemplateOne.queryForList(alarm.getTrigger()).get(0).get("isAlarms")) == 1L ? Boolean.TRUE : Boolean.FALSE;
                    if (boo) 
                        logger.info("告警规则命中......" + alarm.getTrigger());

                        Arrays.stream(alarm.getActions()).forEach(sql -> 
                            List<Map<String, Object>> list = jdbcTemplateOne.queryForList(sql);
                            Object[] codes = list.stream().map((code) -> code.get("code").toString()).toArray();
                            Arrays.stream(alarm.getMediaType()).forEach(phone -> 

                                String msg = String.format(alarm.getMedia(), Arrays.toString(codes));

                                logger.info("告警消息生产......》》》》 content:" + msg + " phone:" + phone);

                                Object oldTime = cache.get(msg + phone);

                                if (Objects.isNull(oldTime)) 
                                    cache.put(msg + phone, System.currentTimeMillis());
                                    jdbcTemplateTow.execute(String.format(alarm.getMediaSql(), msg, phone, phone));
                                 else 
                                    if (System.currentTimeMillis() - Long.class.cast(oldTime) > 7200000L) 
                                        jdbcTemplateTow.execute(String.format(alarm.getMediaSql(), msg, phone, phone));
                                        cache.put(msg + phone, System.currentTimeMillis());
                                     else 
                                        logger.info("2小时内重复告警消息....不发送");
                                    
                                
                            );
                        );
                    
                );
                return alarmsRun;
            );

解耦告警行为和流程

关于告警流程,这里结合zabbix监控告警的配置方式,抽象出触发器,动作,告警媒介,告警消息模板,插表sql等行为,整个告警流程行为通过配置文件配置,在上面告警器构建中告警规则生成中整合行为成完成流程。

  • 触发器(trigger):这里的触发器是一个返回0/1布尔值的SQL,当为true时人为告警被触发,会执行动作。
  • 动作(actions[]):动作在这里是一组返回触发告警标识内容的SQL,用于描述告警触发后的行为,返回触发告警的数据
  • 告警媒介(mediaType[]): 当前告警通过短信的方式,所以这里是一组电话号码,要给哪些用户发生告警消息
  • 告警消息模板(media):不多讲,结合上面动作获取的告警数据,生成完整告警消息
  • 插表sql(mediaSql): 当前发送短信的方式通过插表的方式,如何通过邮件或则短信发生调API的方式,就需要自定义告警规则

我们通配置文件看几个具体的场景

活动监控场景:适用一些批量处理任务的数据,通过where条件判断是否有不符合预期状态的数据,有则获取这部分数据的唯一标识,生成告警消息发送。

空表校验场景: 适用一些账期表,在某些时间会数据落表,通过where条件判断是否存在数据,没有则通过select 'XXX 表数据为空' as code 的方式构建告警消息,发生告警讯息

大表监控场景: 适用部分大表在数据量达到某个峰值的时候,会影响系统性能、SQL超时甚至部分持久化数据丢失,需要对冗余数据进行备份清理。需要提前告警.

alarms:
  minute: "*/5 * * * * ?"
  alarms:
      # 告警策略名称
    - itemsName: "活动监控 "
      #  SQL return isAlarms is Boolean 触发器
      trigger: "select count(*) > 0 as isAlarms from 活动表  a WHERE a.CREATED_DATE >  SUBDATE(NOW(),interval 1 day)  AND a.CREATED_DATE < SUBDATE(NOW(), interval 1 hour)  AND a.STATE ='A' AND a.SP_ID=0"
      # SQL return code is String[], 动作
      actions: [ "select  ACTIVITY_CODE as code from 活动表  a WHERE a.CREATED_DATE >  SUBDATE(NOW(),interval 1 day)  AND a.CREATED_DATE < SUBDATE(NOW(), interval 1 hour)  AND a.STATE ='A' AND a.SP_ID=0" ]
      #  MMS is String[] 告警媒介
      mediaType: [ "18147405370","13147405370","12147405370"]
      # MMS content 告警消息
      media: "%s 活动发生异常,请排查"
      #  MMS install SQL 插表sql
      mediaSql: "INSERT INTO 短信表 (MT_SEQ,  MT_SERV_TYPE, SEND_PRIORITY,  MSG_CONTENT, DEST_TERMID, SRC_TERMID, FEE_USER_TERMID,  FEE_TYPE,   MAKE_TIME,  REQUIRE_APPLY ) VALUES (sms.ACCT_SENDSMS_SEQ.NEXTVAL,'123',2,'%s', '%s','1183111', '%s', '00',SYSDATE,0)"
   - itemsName: "XXX 空表校验"
     trigger: "select count(*) = 0 as isAlarms from XXX"
     actions: ["select 'XXX 表数据为空' as code "]
     mediaType: ['123123','3123']
     media: "%s 异常信息,请排查"
     mediaSql: "insert into sms(phone,content) values('%s','%s')"
   - itemsName: "XXX 大表监控"
     trigger: "select count(*) > 40000000 as isAlarms from XXX"
     actions: ["select 'XXX 表数据量超过峰值' as code "]
     mediaType: ['123123','3123']
     media: "%s 异常信息,请排查"
     mediaSql: "insert into sms(phone,content) values('%s','%s')"  

重复告警的问题

关于重复告警的问题,集成了H2,但是目前告警数据量小,所以没有使用,对于重复告警使用了WeakHashMap构建了一个弱键的缓存工具类实现。

第一告警触发后,存到缓存里,之后2小时内触发告警不发送告警消息,2小时候在发送一次

  Object oldTime = cache.get(msg + phone);

  if (Objects.isNull(oldTime)) 
      cache.put(msg + phone, System.currentTimeMillis());
      jdbcTemplateTow.execute(String.format(alarm.getMediaSql(), msg, phone,phone));
   else 
      if (System.currentTimeMillis() -Long.class.cast(oldTime) > 7200000L)
          jdbcTemplateTow.execute(String.format(alarm.getMediaSql(), msg, phone,phone));
          cache.put(msg + phone, System.currentTimeMillis());
      else 
          logger.info("2小时内重复告警消息....不发送");
      
  

动态定时任务配置

通过SchedulingConfigurer配置类实现动态配置,重配置文件获取cron表达式

DynamicCronSchedule.java

package com.example.alarms.alert;

import com.example.alarms.alert.dto.Alarms;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.logging.Logger;

/**
 * @author LiRuilong
 * @Classname DynamicCronSchedule
 * @Description TODO
 * @Date 2022/5/16 11:31
 */
@Component
@EnableScheduling
public class DynamicCronSchedule implements SchedulingConfigurer 
    private static Logger logger = Logger.getLogger("com.example.alarms.alert.DynamicCronSchedule");

    @Autowired
    private Alarms alarms;


    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) 

        alarms.alarmsInit(null);

        taskRegistrar.addTriggerTask(() -> 

            // 自定义逻辑:
            //alarms.alarmsRun(alarmsRun -> alarmsRun);

            // 默认逻辑:
            alarms.alarmsRun(null);

        , (triggerContext) -> 
            String cron = alarms.getMinute();

            logger.fine("cron expression is " + cron);
            logger.fine("trigger list size is " + taskRegistrar.getTriggerTaskList().size());


            CronTrigger cronTrigger = new CronTrigger(cron);
            Date nextExecTime = cronTrigger.nextExecutionTime(triggerContext);
            return nextExecTime;
        );


    



全部代码

配置相关
---
server:
  port: 30036
  tomcat:
    uri-encoding: utf-8

spring:
  datasource:
    one:
      type: com.alibaba.druid.pool.DruidDataSource
      driver-class-name: com.mysql.jdbc.Driver
      username: 
      password: 
      url: 
      test-while-idle: true

    tow:
      driver-class-name: oracle.jdbc.OracleDriver
      username: 
      password: 
      url: 
      test-while-idle: true
    h2db:
      type: com.alibaba.druid.pool.DruidDataSource
      driver-class-name: org.h2.Driver
      schema: classpath:db/schema.sql
      username: sa
      password: sa
      url: jdbc:h2:mem:alarms


  #http://本地端口/h2-console 通过项目来进行访问数据库
  h2:
    console:
      enabled: true


alarms:
  minute: "*/5 * * * * ?"
  alarms:
      # 告警策略名称
    - itemsName: "活动监控 "
      #  SQL return isAlarms is Boolean 触发器
      trigger: "select count(*) > 0 as isAlarms from 活动表  a WHERE a.CREATED_DATE >  SUBDATE(NOW(),interval 1 day)  AND a.CREATED_DATE < SUBDATE(NOW(), interval 1 hour)  AND a.STATE ='A' AND a.SP_ID=0"
      # SQL return code is String[], 动作
      actions: [ "select  ACTIVITY_CODE as code from 活动表  a WHERE a.CREATED_DATE >  SUBDATE(NOW(),interval 1 day)  AND a.CREATED_DATE < SUBDATE(NOW(), interval 1 hour)  AND a.STATE ='A' AND a.SP_ID=0" ]
      #  MMS is String[] 告警媒介
      mediaType: [ "18147405370","13147405370","12147405370"]
      # MMS content 告警消息
      media: "%s 活动发生异常,请排查"
      #  MMS install SQL 插表sql
      mediaSql: "INSERT INTO 短信表 (MT_SEQ,  MT_SERV_TYPE, SEND_PRIORITY,  MSG_CONTENT, DEST_TERMID, SRC_TERMID, FEE_USER_TERMID,  FEE_TYPE,   MAKE_TIME,  REQUIRE_APPLY ) VALUES (sms.ACCT_SENDSMS_SEQ.NEXTVAL,'123',2,'%s', '%s','1183111', '%s', '00',SYSDATE,0)"
   - itemsName: "XXX 空表校验"
     trigger: "select count(*) = 0 as isAlarms from XXX"
     actions: ["select 'XXX 表数据为空' as code "]
     mediaType: ['123123','3123']
     media: "%s 异常信息,请排查"
     mediaSql: "insert into sms(phone,content) values('%s','%s')"
   - itemsName: "XXX 大表监控"
     trigger: "select count(*) > 40000000 as isAlarms from XXX"
     actions: ["select 'XXX 表数据量超过峰值' as code "]
     mediaType: ['123123','3123']
     media: "%s 异常信息,请排查"
     mediaSql: "insert into sms(phone,content) values('%s','%s')" 




package com.example.alarms.alert.config;

import com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceBuilder;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.sql.DataSource;

/**
 * @author LiRuilong
 * @Classname DataSourceConfig
 * @Description TODO
 * @Date 2022/5/10 9:59
 */

@Configuration
public class DataSourceConfig 


    @Bean
    @ConfigurationProperties("spring.datasource.one")
    DataSource dsOne() 
        return DruidDataSourceBuilder.create().build();
    

    @Bean
    @ConfigurationProperties("spring.datasource.tow")
    DataSource dsTow() 
        return DruidDataSourceBuilder.create().build();
    

    @Bean
    @ConfigurationProperties("spring.datasource.h2db")
    DataSource dsH2db() 
        return DruidDataSourceBuilder.create().build();
    



package com.example.alarms.alert.config;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;

import javax.sql.DataSource;

/**
 * @author LiRuilong
 * @Classname JdbcTemplateConfig
 * @Description TODO
 * @Date 2022/5/10 10:01
 */

@Configuration
public class JdbcTemplateConfig 


    @Bean
    JdbcTemplate jdbcTemplateOne(@Qualifier("dsOne") DataSource ds) 
        return new JdbcTemplate(ds);
    

    @Bean
    JdbcTemplate jdbcTemplateTow(@Qualifier("dsTow") DataSource ds) 
        return new JdbcTemplate(ds);
    

    @Bean
    JdbcTemplate jdbcTemplateH2db(@Qualifier("dsH2db") DataSource ds) 
        return new JdbcTemplate(ds);
    




package com.example.alarms.alert;

import com.example.alarms.alert.dto.Alarms;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework

以上是关于基于SpringBoot轻量非侵入式数据库数据告警器的主要内容,如果未能解决你的问题,请参考以下文章

BlockCanary 一个轻量的,非侵入式的性能监控组件(阿里)

Android基于AOP的非侵入式监控之——AspectJ实战

获取非侵入式升压序列化 C++ 的私有数据成员

Vue数据响应Object.defineProperty

Vue数据响应Object.defineProperty

SpringBoot简介和Banner