ELK统一日志管理平台第三篇-logstash grok插件的使用

1. ELK统一日志管理平台第三篇-logstash grok插件的使用

在本篇博文中，主要讲解如下几个知识点和实践经验，供大家参考：

1. 关于JAVA应用程序的日志内容标准规范：

2. 如何使用logstash的grok插件来完成message字段的拆分：

3. 定时删除Es的索引：

1. 关于JAVA应用程序的日志内容标准规范：

最近公司一直在主推ELK这个项目，而我是ELK这个项目的运维人员。所以针对ELK项目会有很多经验输出；由于我们公司的业务系统以JAVA语言开发为主，特别是Spring Cloud、Spring Boot等框架为主。如何将业务系统的日志标准化是研发架构师需要考虑的问题。目前我们的ELK日志规范定义如下：

<pattern>[%date{ISO8601}][%level] %logger{80} [%thread] Line:%-3L [%X{TRACE_ID}] ${dev-group-name}
${app-name} - %msg%n</pattern>

|时间|日志级别|类文件|线程数|代码发生行|全局流水号|开发团队|系统名称|日志信息

时间：记录日志产生时间;
日志级别：ERROR、WARN、INFO、DEBUG;
类文件：打印类文件名称；
线程名：执行操作线程名称;
代码发生行：日志事件发生在代码中位置;
全局流水号：贯穿一次业务流程的全局流水号;
开发团队： 系统开发的团队名称
系统名称：项目名称组建名
INFO: 记录详细日志信息

比如一个业务系统的日志输出标准格式如下：

[2019-06-2409:32:14,262] [ERROR] com.bqjr.cmm.aps.job.ApsAlarmJob [scheduling-1] []
tstteam tst Line:157 - ApsAlarmJob类execute方法，‘【测试系统预警】校验指标异常三次预警‘预警出错：nested
exception is org.apache.ibatis.exceptions.PersistenceException: ### Error
querying database. Cause: java.lang.NullPointerException ### Cause:
java.lang.NullPointerException org.mybatis.spring.MyBatisSystemException:
nested exception is

2. 如何使用logstash的grok插件来完成message字段的拆分：

现在我们的日志都按照标准字段输出了，但是在kibana界面还是一个message字段。现在就要实现将message分解为每一个字段，可以通过每一个字段进行搜索；

我们的ELK日志平台的架构是：所有业务系统安装filebeat日志收集软件，将日志原封不动的收集到KAFKA集群，由kafka集群再发送到logstash集群，由logstash集群再输出到ES集群，由ES集群再输出到kibana展示和搜索。中间为什么会用到logstash软件，主要是因为logstash软件有强大的文本处理功能，如grok插件。能够实现文本的格式化输出；

logstash软件已经内置了很多正则表达式模板，可以匹配常用的nginx、httpd、syslog等日志；

#logstash默认grok语法模板路径:
/usr/local/logstash-6.2.4/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns

#logstash自带的grok语法模板：
[[email protected] patterns]# ll
total 116
-rw-r--r-- 1 root root   271 Jun 24 16:05 application
-rw-r--r-- 1 root root  1831 Apr 13  2018 aws
-rw-r--r-- 1 root root  4831 Apr 13  2018 bacula
-rw-r--r-- 1 root root   260 Apr 13  2018 bind
-rw-r--r-- 1 root root  2154 Apr 13  2018 bro
-rw-r--r-- 1 root root   879 Apr 13  2018 exim
-rw-r--r-- 1 root root 10095 Apr 13  2018 firewalls
-rw-r--r-- 1 root root  5338 Apr 13  2018 grok-patterns
-rw-r--r-- 1 root root  3251 Apr 13  2018 haproxy
-rw-r--r-- 1 root root   987 Apr 13  2018 httpd
-rw-r--r-- 1 root root  1265 Apr 13  2018 java
-rw-r--r-- 1 root root  1087 Apr 13  2018 junos
-rw-r--r-- 1 root root  1037 Apr 13  2018 linux-syslog
-rw-r--r-- 1 root root    74 Apr 13  2018 maven
-rw-r--r-- 1 root root    49 Apr 13  2018 mcollective
-rw-r--r-- 1 root root   190 Apr 13  2018 mcollective-patterns
-rw-r--r-- 1 root root   614 Apr 13  2018 mongodb
-rw-r--r-- 1 root root  9597 Apr 13  2018 nagios
-rw-r--r-- 1 root root   142 Apr 13  2018 postgresql
-rw-r--r-- 1 root root   845 Apr 13  2018 rails
-rw-r--r-- 1 root root   224 Apr 13  2018 redis
-rw-r--r-- 1 root root   188 Apr 13  2018 ruby
-rw-r--r-- 1 root root   404 Apr 13  2018 squid

#其中有一个java的模板，已经内置了很多java类、时间戳等
[[email protected] patterns]# cat java
JAVACLASS (?:[a-zA-Z$_][a-zA-Z$_0-9]*\.)*[a-zA-Z$_][a-zA-Z$_0-9]*
#Space is an allowed character to match special cases like ‘Native Method‘ or ‘Unknown Source‘
JAVAFILE (?:[A-Za-z0-9_. -]+)
#Allow special <init>, <clinit> methods
JAVAMETHOD (?:(<(?:cl)?init>)|[a-zA-Z$_][a-zA-Z$_0-9]*)
#Line number is optional in special cases ‘Native method‘ or ‘Unknown source‘
JAVASTACKTRACEPART %{SPACE}at %{JAVACLASS:class}\.%{JAVAMETHOD:method}\(%{JAVAFILE:file}(?::%{NUMBER:line})?\)
# Java Logs
JAVATHREAD (?:[A-Z]{2}-Processor[\d]+)
JAVACLASS (?:[a-zA-Z0-9-]+\.)+[A-Za-z0-9$]+
JAVAFILE (?:[A-Za-z0-9_.-]+)
JAVALOGMESSAGE (.*)
# MMM dd, yyyy HH:mm:ss eg: Jan 9, 2014 7:13:13 AM
CATALINA_DATESTAMP %{MONTH} %{MONTHDAY}, 20%{YEAR} %{HOUR}:?%{MINUTE}(?::?%{SECOND}) (?:AM|PM)
# yyyy-MM-dd HH:mm:ss,SSS ZZZ eg: 2014-01-09 17:32:25,527 -0800
TOMCAT_DATESTAMP 20%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:?%{MINUTE}(?::?%{SECOND}) %{ISO8601_TIMEZONE}
CATALINALOG %{CATALINA_DATESTAMP:timestamp} %{JAVACLASS:class} %{JAVALOGMESSAGE:logmessage}
# 2014-01-09 20:03:28,269 -0800 | ERROR | com.example.service.ExampleService - something compeletely unexpected happened...
TOMCATLOG %{TOMCAT_DATESTAMP:timestamp} \| %{LOGLEVEL:level} \| %{JAVACLASS:class} - %{JAVALOGMESSAGE:logmessage}
[[email protected] patterns]#

#但是仅靠默认的这个模板还是不能匹配我们公司自定义的日志内容，所以我还自己写了一个
[[email protected] patterns]# cat application
APP_DATESTAMP 20%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:?%{MINUTE}(?::?%{SECOND})
THREADS_NUMBER (?:[a-zA-Z0-9-]+)
GLOBAL_PIPELINE_NUMBER (?:[a-zA-Z0-9-]+)
DEV_TEAM (?:[a-zA-Z0-9-]+)
SYSTEM_NAME (?:[a-zA-Z0-9-]+)
LINE_NUMBER (Line:[0-9]+)
JAVALOGMESSAGE (.*)
APPLOG \[%{APP_DATESTAMP:timestamp}\] \[%{LOGLEVEL:loglevel}\] %{JAVACLASS:class} \[%{THREADS_NUMBER:threads_number}\] \[%{GLOBAL_PIPELINE_NUMBER:global_pipeline_number}\] %{DEV_TEAM:team} %{SYSTEM_NAME:system_name} %{LINE_NUMBER:linenumber} %{JAVALOGMESSAGE:logmessage}

# 然后就是配置logstash

[[email protected] patterns]# cat /usr/local/logstash/config/yunyan.conf
input {
  kafka {
    bootstrap_servers => "192.168.1.12:9092,192.168.1.14:9092,192.168.1.15:9092"
    topics_pattern => "elk-tst-tst-info.*"
    group_id => "test-consumer-group"
    codec => json
    consumer_threads => 3
    decorate_events => true
    auto_offset_reset => "latest"
  }
}

filter {
    grok {
             match => {"message" => ["%{APPLOG}","%{JAVALOGMESSAGE:message}"]}  #注意这里的APPLOG就是我上面自定义的名字
             overwrite => ["message"]

}
}

output {
  elasticsearch {
     hosts => ["192.168.1.19:9200","192.168.1.24:9200"]
     user => "elastic"
     password => "111111"
     index => "%{[@metadata][kafka][topic]}-%{+YYYY-MM-dd}"
     workers => 1
  }
}

#output {
#   stdout{
#      codec => "rubydebug"
#  }
#}

#一般建议调试的时候，先输出到stdout标准输出，不要直接输出到es.等在标准输出确认已经OK了，所有的格式化字段都可以分别输出了，再去输出到ES；
# 如何将grok的正则表达式写好，有一个在线的grok表达式测试地址:  http://grokdebug.herokuapp.com/

输出了标准的日志内容之后，就可以根据key:value的格式查询搜索了，比如可以在搜索栏输入: loglevel:ERROR 只搜索级别为ERROR的日志内容；

3. 定时删除Es的索引：

索引的定义是根据logstash里面的output插件配置，比如按天的索引，就是索引名后面跟-%{+YYYY-MM-dd},如果想改成按照月的索引，那就是-%{+YYYY-MM}。不同内容的索引，定义的方式应该也是不同的。比如操作系统类的日志，每天的变化不多，就可以按照月来划分索引。但是业务系统本身的程序日志，由于每天产生的日志比较多，就比较适合使用按天的索引。因为对于elasticsearch，索引太大的话也会影响性能，索引太多的话也会影响性能。elasticsearch的主要性能瓶颈在CPU
我在运维ELK这个项目的过程中，就发现了因为索引文件太大，索引数量太多，但是我们的es data节点cpu配置太低，引起ES集群崩溃。解决此问题有几种途径，第一就是定时删除没有用的索引，第二就是优化ES的索引参数。第二点我还没有实践，后续实践了再总结文档出来；先把定时删除索引和手工删除索引的方法写出来。

#/bin/bash
#指定日期(7天前)
DATA=`date -d "1 week ago" +%Y-%m-%d`

#当前日期
time=`date`

#删除7天前的日志
curl -u elastic:654321 -XGET "http://192.168.1.19:9200/_cat/indices/?v"|grep $DATA
if [ $? == 0 ];then
  curl -u elastic:654321 -XDELETE "http://127.0.0.1:9200/*-${DATA}"
  echo "于 $time 清理 $DATA 索引!"
fi

curl -u elastic:654321 \-XGET "http://192.168.1.19:9200/_cat/indices/?v"|awk ‘{print $3}‘|grep elk >> /tmp/es.txt
#手工删除索引，将索引名输出到一个文本文件，然后通过循环删除的方法
for i in `cat /tmp/es.txt`;do curl -u elastic:654321 -X DELETE "192.168.1.19:9200/$i";done

好的，今天暂时就写到这里了。最近工作特别忙，很难抽出时间来更新技术博客。基本都是晚上加班很晚或者早上起来很早把博客更新一下，工作时间任务很多真的很难抽出时间更新博文了。还是感谢大家的持续关注。

博文的更详细内容请关注我的个人微信公众号 “云时代IT运维”,本公众号旨在共享互联网运维新技术，新趋势；包括IT运维行业的咨询，运维技术文档分享。重点关注devops、jenkins、zabbix监控、kubernetes、ELK、各种中间件的使用，比如redis、MQ等；shell和python等运维编程语言；本人从事IT运维相关的工作有十多年。2008年开始专职从事Linux/Unix系统运维工作；对运维相关技术有一定程度的理解。本公众号所有博文均是我的实际工作经验总结，基本都是原创博文。我很乐意将我积累的经验、心得、技术与大家分享交流！希望和大家在IT运维职业道路上一起成长和进步；

原文地址：https://blog.51cto.com/zgui2000/2413917

时间： 2024-11-14 21:09:31

ELK统一日志管理平台第三篇-logstash grok插件的使用

1. 关于JAVA应用程序的日志内容标准规范：

2. 如何使用logstash的grok插件来完成message字段的拆分：

3. 定时删除Es的索引：

ELK统一日志管理平台第三篇-logstash grok插件的使用的相关文章

七牛云智能日志管理平台正式发布

Spring MVC 中使用AOP 进行统一日志管理--XML配置实现

性能优化分析Spring Cloud ELK+kafka日志分析平台

ELK日志管理平台部署简介

centos7搭建ELK Cluster集群日志分析平台（三）

centos7搭建ELK Cluster日志分析平台（一）

Centos7下ELK+Redis日志分析平台的集群环境部署记录

使用Spring进行统一日志管理 + 统一异常管理

NC外部统一流程管理平台方案