前言

在《使用EFK快速搭建安全可靠的日志服务》一文中，我们已经大致介绍了分布式日志服务的各个组件。但是对于很多实现细节，并未涵盖到。本文主要介绍logstash用到的插件，以及对插件的离线打包，避免每次手动安装插件。本文主要分为4部分：

1. 插件的离线打包
2. input插件
3. filter插件
4. output插件

在使用插件之前，我们先了解一个概念：事件。Logstash 每读取一次数据的行为叫做事件。

1.插件的离线打包

在生产环境中，logstash可能是部署到多台机器上，如果每次logstash的安装更新都需要手动更新一遍插件成本是非常高的。那我们可以对用到的插件进行offline打包。
制作logstash-offline-plugins-x.x.x.zip说明：

1. 需要在一台已经安装了logstash的机器上，先给logstash安装好所需要的插件
2. 然后再导出插件，具体命令：

bin/logstash-plugin update logstash-filter-mutate 
bin/logstash-plugin install logstash-filter-json_encode 
bin/logstash-plugin prepare-offline-pack logstash-filter-mutate logstash-filter-json_encode

参考elastic的官方文档：https://www.elastic.co/guide/...

在logstash安装时通过如下命令进行插件包的安装：

bin/logstash-plugin install file://$(pwd)/logstash-offline-plugins.zip

安装插件后要及时观察logstash的打印日志，避免因为插件版本冲突导致的logstash启动失败的情况。

2.input插件

输入插件允许一个特定的事件源可以读取到 Logstash 管道中，配置在 input {} 中，且可以设置多个。 在《使用EFK快速搭建安全可靠的日志服务》一文中我们是分别将input、filter和output三部分的配置放到了不同的配置文件中。input的配置在input.conf文件中。
由于我们使用的是EFK，因此需要beats插件。可以参考官网：https://www.elastic.co/guide/...

input {
    beats {
        port => 5044
        client_inactivity_timeout => 600
        ssl => true
        ssl_certificate_authorities => ["/home/work/certificate/chain-ca.pem"]
        ssl_certificate => "/home/work/certificate/server.crt.pem"
        ssl_key => "/home/work/certificate/server.key.pem"
        ssl_verify_mode => "force_peer"
    }
}

需要注意的是，在6.4版本以上的beats配置中，多了ssl_peer_metadata，可以拿到证书里的meta信息，可以便于我们后续在logstash上做一些鉴权校验。
除了beats插件，inputs类型插件还有很多种，主要列举如下：

• elasticsearch
• file
• log4j
• kafka

更多插件配置介绍参考官网：https://www.elastic.co/guide/...

3.filter插件

filter插件就比较多了。日志接入之后要进行日志的规则匹配、过滤等操作，配置filter是必不可少的。最长用到的filter插件有以下几种：

• grok
• date
• mutate
• geoip
• split
• uuid
• json
• jsonencode

3.1 grok插件

Grok是logstash最主要的过滤插件，grok是通过系统预定义的正则表达式或者通过自己定义正则表达式来匹配日志中的各个值，安装了logstash后默认会有一些常用软件的日志匹配正则，在使用时自己可以根据自己的日志格式或者对匹配正则进行调整或者直接调用。如果，自己想在其他目录定义正则规则匹配日志，在使用时需要指定正则的路径。
对于以下一行日志：

55.3.244.1 GET /index.html 15824 0.043

使用如下的规则：

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

grok配置示例如下：

filter {
  grok {
    match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
  }
}

那么这行日志经过grok之后会得到以下的field：

client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043

常用的配置选项：

• match：用来对字段的模式进行匹配
• patterns_dir：用来指定规则的匹配路劲，如果使用logstash自定义的规则时，不需要写此参数。Patterns_dir可以同时制定多个存放过滤规则的目录。语法格式：patterns_dir => [“/ opt / logstash / patterns”，“/ opt / logstash / extra_patterns”]
• remove_field：如果匹配到某个”日志字段，则将匹配的这个日志字段从这条日志中删除

Grok过滤正则规则可以自己根据自己的日志格式自行编写，在编写grok过滤规则时容易出错，此时可以使用grokdebug网站对自己的日志及编写的正则规则进行调试，grokdebug网址为（https://grokdebug.herokuapp.com/）

3.2 date插件

date用于解析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。
对于date插件的介绍，在https://segmentfault.com/a/11...（感觉是官网https://www.elastic.co/guide/... 的翻译版本）中有非常详细的介绍。我来列举一个自己在配置过程中遇到的示例：
对于给定的时间格式：

2019-01-10T18:11:28.699+08:00

需要进行配置的pattern应该如下：

date {
    match => ["[@metadata][log_timestamp]", "yyyy-MM-dd'T'HH:mm:ss.SSSZZ"]
    timezone => "Asia/Shanghai"
}

需要注意的是，对于filter.conf配置是可以热加载的，如果时间格式配置错误，会导致logstash进程挂掉，对于问题的排查，可以自行查询logstash的日志。

3.3 mutate插件

mutate 插件可以在字段上执行变换，包括重命名、删除、替换和修改。这个插件相当常用。
比如：
你已经根据 Grok 表达式将 Tomcat 日志的内容放到各个字段中，想把状态码、字节大小或是响应时间，转换成整型；
你已经根据正则表达式将日志内容放到各个字段中，但是字段的值，大小写都有，这对于 Elasticsearch 的全文检索来说，显然用处不大，那么可以用该插件，将字段内容全部转换成小写。
示例如下：

filter {
    mutate {
        split => ["hostname", "."]
        add_field => { "shortHostname" => "%{hostname[0]}" }
    }

    mutate {
        rename => ["shortHostname", "hostname" ]
    }
}

在笔者之前写的《使用EFK快速搭建安全可靠的日志服务》一文中，主要使用mutate插件来add_field, remove_field, convert，gsub等操作。其中covert可以将字符串类型转换为int/long/float等类型，便于进行聚合等操作。
例如：

mutate {
  convert => {
        "averageSliceQueryTime" => "integer"
        "maxSliceQueryTime" => "integer"
  }

关于更多filter类型插件可以参考：https://www.elastic.co/guide/...

4.output插件

和前面介绍的input插件一样，output插件也是最基础最简单的输出插件。在《使用EFK快速搭建安全可靠的日志服务》一文中我们使用的elasticsearch作为最终的输出存储，因此output插件使用的是elasticsearch。对于grok解析失败的文件会写入到本地文件中，因此用了file插件，通过分支进行判断。示例如下：

output {
    if "_grokparsefailure" in [tags] or "_jsonparsefailure" in [tags] {
        file {
            path => "/home/work/logstash/failure_output/failure-%{+YYYY-MM-dd}.log"
        }
    } else {
        elasticsearch {
            hosts => ["https://es-host1:9920","https://es-host2:9920","https://es-host3:9920"]
            index => "logstash-%{[@metadata][index_name]}-%{+YYYY.MM.dd}"
            document_id => "%{[@metadata][document_id]}"
            ssl => true
            ssl_certificate_verification => true
            truststore => "/home/work/certificate/truststore.jks"
            truststore_password => "adofkoe"
            user => "logstash"
            password => "dafodmfkamkefadfg"
        }
    }
}

除了elasticsearch，还支持以下输出：

• email
• file
• influxdb
• mongodb
• redis

更多可以参考官网：https://www.elastic.co/guide/...

总结

logstash的各种插件极大丰富了其功能。在实际使用中，我们可以根据自己的实际需求进行相关配置，来构建自己的日志服务系统。笔者此处主要是对自己使用过程中所用到内容的总结。遇到的很多实际问题，可以自行查询官网进行解决。