须知：本文将说明如何将一个SpringBoot项目的日志通过logstash的过滤，将error等级的日志推送到钉钉进行报警，不涉及到elk的具体搭建以及配置。
要将日志推送到钉钉需要使用钉钉自带的报警机器人，本文使用的为webhook机器人，可通过rpc请求推送消息到钉钉群内，具体钉钉机器人的使用不再多说，官方文档介绍的较为全面，读者可阅读文档（我是链接）或者自行阅读其他博客。
logstash配置文件中使用到了tcp input、Grok、throttle filter和http output几种插件。官方文档可查看上述链接，后续会介绍下本文用到的几种配置。
logstash部分
新建logstash配置文件logstash.conf如下：
input {# 开启logstash的tcp输入端口，端口号5000tcp {port => 5000mode => "server"ssl_enable => falsetags => ["hb"]}
}
filter {# grok过滤器grok {match => { "message" => "%{TIME:timestamp}%{SPACE}%{LOGLEVEL:level}%{SPACE}%{JAVAFILE:log_file}%{SPACE}%{NUMBER:log_num}%{SPACE}%{GREEDYDATA:content}" }}# throttle过滤器if [level] == "ERROR" {throttle {before_count => 3after_count => 5period => 1800max_age => 3600key => "%{[content]}"add_tag => "hb-throttle"}}
}
output {# 输出控制台方便调试stdout { codec => rubydebug }# elasticsearch {# hosts=> ["http://127.0.0.1:9200"]# index => "hb-%{+YYYY.MM.dd}"# user => elastic# password => elastic# }# 判断tag和日志等级if [level] == "ERROR" and "hb" in [tags] and "hb-throttle" not in [tags] {# 发送报警日志到钉钉http {url => "https://oapi.dingtalk.com/robot/send?access_token={
{token}}"http_method => "post"content_type => "application/json; charset=utf-8"format => "message"message => '{"msgtype":"text","text":{"content":"[hb] %{[content]}"}}'}}
}这里将配置文件分为三部分讲解：input、filter、output
input
使用logstash自带的TCP/UDP 插件来作为数据入口，暴露端口号为5000。
filter
logstash过滤器，首先使用grok插件进行数据捕获，其本质为正则捕获，但是grok提供了非常多的内置默认类型，可以在此处查看所有的默认类型。当然你也可以自己编写正则表达式来匹配自定义的格式。这里用到的几种默认类型分别为，TIME:时间匹配，SPACE:空格匹配，LOGLEVEL:日志等级匹配，JAVAFILE:java的文件名匹配，NUMBER:数字类型匹配，GREEDYDATA:字符串。（需要和后续的logback中内容匹配，读者自己调试时可以一个一个字段匹配。）通过这里的过滤后logstash会将这些都作为字段输出。
使用throttle filter过滤器来添加计数器，如果日志等级为error等级，则添加特殊tag，添加该tag的日志会推送到钉钉报警机器人。解释一下使用到的几个字段，before_count 出现的最小数量，配置为-1则忽略该配置，after_count 出现的最大数量，配置为-1则忽略该配置，period 两次发生时间事件之间的间隙，max_age 一个时隙的最大年龄，key 需要记录的内容，add_tag 新增的tag。上述配置的结果就是在30分钟内日志类型为error类型的日志出现一次就会添加上hb-throttle的tag，并且只会添加一次。其中before_count和after_count两个配置可能不是很好理解，通俗的讲就是当两个配置为3和5的时候，在max_age这个时间段内，会给3次之前和5次及5次之后的key的日志内容打上add_tag配置的tag内容。
output
首先通过stdout { codec => rubydebug }输出到控制台日志方便调试，其次判断日志的tag内容来判断是否需要通过钉钉机器人报警。
java部分
首先引入maven依赖：
<dependency><groupId>net.logstash.logback</groupId><artifactId>logstash-logback-encoder</artifactId><version>4.9</version></dependency>java项目中配置在logback的xml配置文件中，新增如下的配置：
<appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender"><destination>logstash地址:5000</destination><encoder><!--配置发送的格式 时间点 日志等级 文件名 行数 具体日志内容 --><pattern>%d{HH:mm:ss,GMT+8} %-5level %logger{36} %line %msg%n</pattern><charset>UTF-8</charset></encoder></appender><springProfile name="uat"><root level="info"><!--除了发送到logstash之外还打印到控制台 --><appender-ref ref="CONSOLE"/><appender-ref ref="LOGSTASH" /></root><logger name="org.apache.http" level="info"/></springProfile>其中encoder的内容需要和logstash配置中grok解析格式相同。
启动项目后通过lombok的@Slf4j来进行日志输出：
@Slf4j
@SpringBootApplication
public class TestApplication {public static void main(String[] args) {SpringApplication.run(TestApplication.class, args);log.warn("this is a warnMsg");log.info("this is a infoMsg");log.error("this is a error msg???");log.error("this is a error msg???");log.error("this is a error msg???");log.error("this is a error msg???");log.error("this is a error msg???");log.error("this is a error msg???");}
}
logstash日志打印如下：
钉钉收到三条报警信息：
成功。
参考链接