共计 4303 个字符，预计需要花费 11 分钟才能阅读完成。
前言
在《使用 EFK 快速搭建安全可靠的日志服务》一文中，我们已经大致介绍了分布式日志服务的各个组件。但是对于很多实现细节，并未涵盖到。本文主要介绍 logstash 用到的插件，以及对插件的离线打包，避免每次手动安装插件。本文主要分为 4 部分：
插件的离线打包
input 插件
filter 插件
output 插件
在使用插件之前，我们先了解一个概念：事件。Logstash 每读取一次数据的行为叫做事件。
1. 插件的离线打包
在生产环境中，logstash 可能是部署到多台机器上，如果每次 logstash 的安装更新都需要手动更新一遍插件成本是非常高的。那我们可以对用到的插件进行 offline 打包。制作 logstash-offline-plugins-x.x.x.zip 说明：
需要在一台已经安装了 logstash 的机器上，先给 logstash 安装好所需要的插件
然后再导出插件，具体命令：
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：
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 的各种插件极大丰富了其功能。在实际使用中，我们可以根据自己的实际需求进行相关配置，来构建自己的日志服务系统。笔者此处主要是对自己使用过程中所用到内容的总结。遇到的很多实际问题，可以自行查询官网进行解决。