logstash经常使用插件介绍

时间 2019-11-16

原文原文链接

前言

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

插件的离线打包
input插件
filter插件
output插件

在使用插件以前，咱们先了解一个概念：事件。Logstash 每读取一次数据的行为叫作事件。正则表达式

1.插件的离线打包

在生产环境中，logstash多是部署到多台机器上，若是每次logstash的安装更新都须要手动更新一遍插件成本是很是高的。那咱们能够对用到的插件进行offline打包。
制做logstash-offline-plugins-x.x.x.zip说明：redis

须要在一台已经安装了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/...mongodb

在logstash安装时经过以下命令进行插件包的安装：json

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

安装插件后要及时观察logstash的打印日志，避免由于插件版本冲突致使的logstash启动失败的状况。segmentfault

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类型插件还有不少种，主要列举以下：app

elasticsearch
file
log4j
kafka

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

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的各类插件极大丰富了其功能。在实际使用中，咱们能够根据本身的实际需求进行相关配置，来构建本身的日志服务系统。笔者此处主要是对本身使用过程当中所用到内容的总结。遇到的不少实际问题，能够自行查询官网进行解决。