如何通过图文讲解掌握Logstash实战在Elasticsearch中的应用？

前言 在上一篇中我们介绍了Logstash快速入门，本文主要介绍的是ELK日志系统中的Logstash的实战使用。实战使用我打算从以下的几个场景来进行讲解。 时区问题解决方案 在我们使用logstash将采集的数据传输到ES中的时候，会发现采集的时间@timestamp的时间和我们本地的不一致，这个主要是因为时区的问题导致的，我们在计算时间的时候需要将这个时间增加8小时，但是这样会很不方便。为了永久解决这个问题，我们可以在logstash中的filter中对该字段进行转换，增加8小时。 添加的配置如下: ruby { code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)" } ruby { code => "event.set('@timestamp',event.get('timestamp'))" } mutate { remove_field => ["timestamp"] } 原本示例图: 添加配置之后的示例图: 可以看到添加配置之后@timestamp时间已经和本地时间基本一致了。 日志内容切分 我们在进行采集日志到ES中的时候，有时需要对日志内容进行切割。比如得到日志内容的时间以及日志级别等等。这时我们就可以通过grok来对日志内容进行切分，比如将制定好的日志内容切割为日志时间、线程名称、日志级别、类名以及详细内容等等。我们只需要在logstash的filter中使用grok语法即可完成日志内容切割。 这里我们使用JAVA的Logback来制定日志输出格式，然后通过日志的格式编写grok语法，最后将grok配置添加到logstash的filter中。 Logback输出配置: |%d{yyyy-MM-dd HH:mm:ss.SSS}|[%thread]|%-5level|%logger{50}|-%msg%n 日志样例数据: |2020-07-24 17:08:33.159|[Thread-5]|INFO|com.pancm.Application|-测试示例三: All things in their being are good for something. 天生我才必有用3 grok模式: |%{DATA:log_time}|%{DATA:thread}|%{DATA:log_level}|%{DATA:class_name}|-% 使用grok分析 可以看到以及分析匹配成功了。 然后我们在filter中添加如下配置: grok { match => { "message" =>"|%{DATA:log_time}|%{DATA:thread}|%{DATA:log_level}|%{DATA:class_name}|-%{GREEDYDATA:content}" } } 最终输出的日志到ES的示例图: 自定义模板 我们在使用Logstash采集日志的时候，如果没有指定索引库或模板，则会使用ElasticSearch默认自带的名字为”logstash”的模板，默认应用于Logstash写入数据到ElasticSearch使用。但是我们希望使用自定义的索引模板，将采集的日志按照我们自身的想法来写入，此时我们就需要用到自定义模板了。 主要有两种方式，一种是在logstash的output插件中使用template指定本机器上的一个模板json路径， 例如 template => "/home/logstash.json"，json里面的内容为我们自定的索引mapping，虽然这种方式简单，但是分散在Logstash机器上，维护起来比较麻烦。还有一种是在elasticsearc服务端自定义配置模板，事先将模板设置好，然后在logstash的output输出中指定该模板即可，这种方式比较灵活方便，可动态更改，全局生效。 这里我们还是通过一个示例来进行说明，我们首先创建一个template_mylog的模板，配置这几个字段: log_time、thread、log_level、class_name、content。