ELK 架构之 Logstash 和 Filebeat 配置使用(采集过滤)
相关文章:
- ELK 架构之 Elasticsearch 和 Kibana 安装配置
- ELK 架构之 Logstash 和 Filebeat 安装配置
ELK 使用步骤:Spring Boot 日志输出到指定目录,Filebeat 进行采集,Logstash 进行过滤,Elasticsearch 进行存储,Kibana 进行展示。
Filebeat 示例配置(vi /etc/filebeat/filebeat.yml
):
filebeat.prospectors: - input_type: log paths: - /var/log/spring-boot-log4j2/*.log document_type: "spring-boot-log4j2" # 定义写入 ES 时的 _type 值 multiline: #pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})' # 指定匹配的表达式(匹配以 2017-11-15 08:04:23:889 时间格式开头的字符串) pattern: '^\s*("{)' # 指定匹配的表达式(匹配以 "{ 开头的字符串) negate: true # 是否匹配到 match: after # 合并到上一行的末尾 max_lines: 1000 # 最大的行数 timeout: 30s # 如果在规定的时候没有新的日志事件就不等待后面的日志 fields: logsource: node1 logtype: spring-boot-log4j2 - input_type: log paths: - /var/log/messages #- /var/log/*.log document_type: "syslog" # 定义写入 ES 时的 _type 值 fields: logsource: node1 logtype: syslog #output.elasticsearch: #hosts: ["node1:9200"] output.logstash: hosts: ["node1:10515"]
上面的配置需要注意几点:
pattern
:配置的正则表达式,是为了合并异常信息(而不是单行显示),匹配以"{
开头的字符串(判断是否 Json 格式),如果匹配不到的话,就进行合并行。document_type
:配置的是 Elasticsearch 的 Type 值,方便 Elasticsearch 对日志数据的归类。logtype
:新增的字段,用于 Filebeat 和 Logstash 之间传递参数,进行过滤的判断逻辑。
Logstash 示例配置(vi /etc/logstash/conf.d/logstash.conf
):
input { beats { port => 10515 } } filter { if [fields][logtype] == "syslog" { grok { match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" } add_field => [ "received_at", "%{@timestamp}" ] add_field => [ "received_from", "%{host}" ] } syslog_pri { } date { match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ] } } if [fields][logtype] == "spring-boot-log4j2" { json { source => "message" target => "data" } } } output { if [fields][logtype] == "spring-boot-log4j2"{ elasticsearch { hosts => ["127.0.0.1:9200"] index => "spring-boot-log4j2-%{+YYYY.MM.dd}" } } if [fields][logtype] == "syslog"{ elasticsearch { hosts => ["127.0.0.1:9200"] index => "filebeat-%{+YYYY.MM.dd}" } } }
上面的配置需要注意几点:
- logstash.conf:配置文件可以配置多个,
input
、filter
和output
可以单独文件配置。 - fields logtype:就是上面 Filebeat 配置的字段,这边用来判断服务来源,然后进行单独的处理。
- filter:过滤器做了两件事,一个是使用
grok
插件,匹配数据和增加字段值,另一个就是使用json
插件,将字符串转换成 Json 对象(会创建data
层级结构,如果不想新建层级的话,删掉target
配置即可)。 - output:根据
logtype
判断,输出到指定的 Elasticsearch 地址,以及创建指定的索引。
简单总结下, Filebeat 是客户端,一般部署在 Service 所在服务器(有多少服务器,就有多少 Filebeat),不同 Service 配置不同的input_type
(也可以配置一个),采集的数据源可以配置多个,然后 Filebeat 将采集的日志数据,传输到指定的 Logstash 进行过滤,最后将处理好的日志数据,存储到指定的 Elasticsearch。
好了,下面我们测试下上面的配置,是否可行。
Logstash 和 Filebeat 配置好之后,重启一下:
$ systemctl restart logstash && systemctl restart filebeat
Spring Boot 中log4j2.xml
中的配置(参考文章):
<?xml version="1.0" encoding="UTF-8"?> <Configuration status="OFF" monitorInterval="30"> <Properties> <Property name="LOG_PATTERN">%m%n%ex</Property> <Property name="LOG_FILE_PATH">/Users/xishuai/Downloads/logs</Property> </Properties> <Appenders> <Console name="ConsoleAppender" target="SYSTEM_OUT" follow="true"> <PatternLayout pattern="${LOG_PATTERN}"/> </Console> <RollingFile name="FileAppender" fileName="${LOG_FILE_PATH}/spring-boot-log4j2-demo.log" filePattern="${LOG_FILE_PATH}/spring-boot-log4j2-demo-%d{yyyy-MM-dd}-%i.log"> <PatternLayout> <Pattern>${LOG_PATTERN}</Pattern> </PatternLayout> <Filters> <ThresholdFilter level="ERROR" onMatch="ACCEPT" onMismatch="DENY" /> </Filters> <Policies> <SizeBasedTriggeringPolicy size="10MB" /> <TimeBasedTriggeringPolicy interval="1" /> </Policies> <DefaultRolloverStrategy max="10"/> </RollingFile> </Appenders> <Loggers> <Root level="ERROR"> <AppenderRef ref="ConsoleAppender" /> <AppenderRef ref="FileAppender"/> </Root> </Loggers> </Configuration>
配置说明可以参考之前的文章,这边的LOG_PATTERN
配置改为了%m%n%ex
,直接输出日志信息或异常信息。
测试代码:
@Log4j2 @RestController @EnableDiscoveryClient @SpringBootApplication public class SpringBootLog4j2Application implements ApplicationRunner { public static void main(String[] args) { SpringApplication.run(SpringBootLog4j2Application.class, args); } @Override public void run(ApplicationArguments applicationArguments) throws Exception { logger.debug("Debugging log"); logger.info("Info log"); logger.warn("Hey, This is a warning!"); logger.error("jack! We have an Error. OK"); logger.fatal("xishuai! Fatal error. Please fix me."); } @RequestMapping("/log") public String log() { log.error("{\"msg\":\"出现一个异常错误:请求连接失败\",\"level\":\"ERROR\",\"createTime\":\"2018-5-21 20:22:22\",\"provider\":\"xishuai\",\"ip\":\"192.168.1.11\",\"stackTrace\":\"java.lang.Exception\\n\\tat com.example.log_demo.LogDemoTests.logCustomField(LogDemoTests.java:33)\\n\\tat org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:191)\\n\\tat org.junit.runner.JUnitCore.run(JUnitCore.java:137)\\n\\tat com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)\\n\\tat com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)\\n\\tat com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)\\n\\tat com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)\\n\",\"tag\":\"\",\"url\":\"\"}"); return "Hello World ----spring-boot-log4j2"; } }
启动服务,然后访问http://localhost:8280/log
,手动产生一条日志数据(Json 格式)。
查看下 Elasticsearch 中产生的索引:
[root@node1 ~]# curl http://node1:9200/_cat/indices?v health status index uuid pri rep docs.count docs.deleted store.size pri.store.size yellow open .kibana _8JIwyIZTJ-y9p3_o04tdA 1 1 2 0 28.3kb 28.3kb yellow open spring-boot-log4j2-2018.05.21 IZuVdDfjRIGXR5Rei7-gFQ 5 1 2 0 29.1kb 29.1kb
可以看到,有一条新的索引spring-boot-log4j2-2018.05.21
产生。
接着,我们使用 Kibana,创建一个索引模版(``spring-boot-log4j2-*):
然后,我们就可以看到日志信息了:
第一条日志数据中,红框里面是我们输出的日志信息,测试代码中总共输出了 5 种日志级别的信息,因为配置文件中设置的日志级别是ERROR
,所以这边ERROR
以下的日志不会输出,另外,因为设置了日志匹配规则,两条数据都不是以"{
开头,这边就将两条日志数据,合并成一条了。
我们再看一下自定义输出的日志信息(Json 格式):
测试代码中输出的是 Json 字符串,经过 Logstash 过滤处理之后,就转换成 Json对象了。
另外,我们还可以data.level:ERROR
这样格式进行搜索,或者data.msg:(错误)
格式进行模糊搜索。
还需要注意的是,上面多行合并的配置是在 Filebeat 中,如果每个服务都是一样的规则,那么每台服务器都需要配置,如果规则更改了,这样每台服务器的 Filebat 配置都需要更改,就比较不方便。
Logstash 也提供了多行合并的配置功能,我们只需要这样配置(使用codec/multiline
):
input { beats { port => 10515 codec => multiline { pattern => '^\s*({")' negate => true what => "previous" } } }
效果和 Filebeat 配置是一样的。
如果日志组件使用 Log4j/Log4j2,Logstash 还提供了另一种处理 Log4j 的方式:input/log4j。与codec/multiline
不同,这个插件是直接调用了org.apache.log4j.spi.LoggingEvent
处理 TCP 端口接收的数据。
参考资料:
- ELK 使用 - 多个日志源 filebeat 配置详解
- filebeat和logstash收集处理java多行日志
- JSON 编解码(官网推荐)
- 合并多行数据(Multiline)(官网推荐)