Logstash配置详解

https://blog.csdn.net/hushukang/article/details/

Logstash配置文件位于Logstash安装目录下bin/logstash.conf

启动命令: logstash -f logstash.conf

1. Input Plugin

1.1 从文件输入

从文件读取数据，如常见的日志文件。文件读取通常要解决几个问题：

1.1.1 logstash-input-file配置

配置：

1.1.2 path

例如：

• /var/log/*.log：匹配/var/log目录下以.log结尾的所有文件
• /var/log//*.log：匹配/var/log所有子目录下以.log结尾的文件
• /var/log/{app1,app2,app3}/*.log：匹配/var/log目录下app1，app2，app3子目录中以.log结尾的文件

1.1.3 sincedb

通过logstash-input-file插件导入了一些本地日志文件时，logstash会通过一个名为sincedb的独立文件中来跟踪记录每个文件中的当前位置。这使得停止和重新启动Logstash成为可能，并让它在不丢失在停止Logstashwas时添加到文件中的行数的情况下继续运行。sincedb默认保存在当前用户的Home目录下。

在调试的时候，我们可能希望取消sincedb的记录功能，使文件每次都能从头开始读取。此时，我们可以这样来做：

input {    file {        path => ["文件路径"]        start_position => "beginning"        sincedb_path => "/dev/null"   } } 

1.2 从HTTP输入

input {  http {    port => 端口号 } } 12345

1.3 从TCP输入

input {  tcp {    mode => "server"    host => "0.0.0.0"    port => 端口号    codec => json_lines } } 

2. Codec Plugin

Codec Plugin作用于Input和Output Plugin，负责将数据在原始与Logstash Event之间转换，常见的codec有：

• plain：读取原始内容
• dots：将内容简化为点进行输出
• rubydebug：将Logstash Events按照ruby格式输出，方便调试
• line：处理带有换行符的内容
• json：处理json格式的内容
• multiline：处理多行数据的内容

2.1 multiline

当一个Event的message由多行组成时，需要使用multiline，常见的情况是处理日志信息，如下所示：

Exception in thread "main" java.lang.NullPointerException at com.example.myproject.Book.getTitle(Book.java:16) at com.example.myproject.Author.getBookTitles(Author.java:25) at com.example.myproject.Bootstrap.main(Bootstrap.java:14) 1234

主要设置参数如下：

例如，匹配日志文件：

input { file { path => ["文件路径"]       start_position => "beginning" codec => multiline { pattern => "^\s" #匹配以空格开头的行 what => "previous" } } } 

3. Filter Plugin

Filter是Logstash功能强大的主要原因，它可以对Logstash Event进行丰富的处理，比如解析数据、删除字段、类型转换等等，常见的有如下几个：

3.1 date插件

date插件可以将日期字符串解析为日期类型，然后替换@timestamp字段或者指定其他字段：

filter { date { match => ["logdate", "MM dd yyy HH:mm:ss"] } } 12345

通过以上配置，我们可以将

{ "logdate": "Jan 01 2018 12:02:2018" "target": "@timestamp" } 1234

转换为：

{ "@version" => "1" "host" => "..." "logdate" => Jan01 2018 12:02:08 "@timestamp" => "2018-01-01T04:02:08.000Z" } 

3.2 grok插件

grok插件用于解析日志内容，例如我们要将下面这一条日志解析成json数组。

83.149.9.216 [17/May/2015:10:05:03 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200  "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36" 1

那么我们就可以这样来配置grok：

%{IPORHOST:clientip} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response:int} (?:-|%{NUMBER:bytes:int}) %{QS:referrer} %{QS:agent} 1

grok语法：

例子：%{SYNTAX:SEMANTIC} SYNTAX为grok pattern的名称，SEMANTIC为赋值字段名称。%{NUMBER:duration}可以匹配数值类型，但是grok匹配出的内容都是字符串类型，可以通过在最后指定为int或者float来强转类型：%{NUMBER:duration:int}

grok内置pattern：logstash-patterns-core

自定义grok pattern 我们通过pattern_definitions参数，以键值对的方式定义pattern名称和内容。也可以通过pattern_dir参数，以文件的形式读取pattern。

filter { grok { match => { "message" => "%{SERVICE:service}" } pattern_definitions => { "SERVICE" => "[a-z0-9]{10,11}" } } } 

3.3 dissect插件

基于分隔符原理解析数据，解决grok解析时消耗过多cpu资源的问题。

同样的日志信息，dissect可以这样来写：

83.149.9.216 [17/May/2015:10:05:03 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200  "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36" 1

dissect配置：

%{clientip} [%{timestamp}] "%{request}" %{response} %{bytes} "%{referrer}" "%{agent}" 1

但是，正因为dissect语法简单，因此它能处理的场景比较有限。它只能处理格式相似，且有分隔符的字符串。它的语法如下：

• %{}里面是字段
• 两个%{}之间是分隔符。

例子1： 有以下日志：

Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool 1

我想要把前面的日期和时间解析到同一个字段中，那么就可以这样来做：

filter { dissect { mapping => { "message" => "%{ts} %{+ts} %{+ts} %{src} %{prog}[%{pid}]: %{msg}" } } } 

例子2： 如果有以下日志：

name=hushukang&age=28 1

我想要把它解析成：

{ "name": "hushukang", "age": 28 } 1234

那么，可以这样来写dissect:

filter { dissect { mapping => { "message" => "%{?key1}=%{&key1}&%{?key2}=%{&key2}" } convert_datatype => { age => "int" } } } 

%{?}代表忽略匹配值，但是赋予字段名，用于后续匹配用。%{&}代表将匹配值赋到指定字段中。convert_datatype可以将指定字符串转为int或者float类型。

3.4 mutate插件

mutate是使用最频繁的插件，可以对字段进行各种操作，比如重命名、删除、替换、更新等，主要操作如下：

• convert类型转换
• gsub字符串替换
• split、join、merge字符串切割、数组合并为字符串、数组合并为数组
• rename字段重命名
• update、replace字段内容更新或替换。它们都可以更新字段的内容，区别在于update只在字段存在时生效，而replace在字段不存在时会执行新增字段的操作
• remove_field删除字段

mutate { convert => {"age" => "integer"} gsub => [ "birthday", "-", "/", "image", "\/", "_" ] split => {"hobities" => ","} rename => {"hobities" => "hobityList"} replace => {"icon" => "%{image}"} remove_field => ["message"] } 11

3.5 json插件

将字段内容为json格式的数据解析出来。 语法如下：

filter { json { source => "message" target => "msg_json" } } 

如果不指定target的话，那么filter会把解析出来的json数据直接放到根级别。

3.6 geoip插件

常用的插件，根据ip地址提供对应的地域信息，比如经纬度、城市名等，方便进行地理数据分析。

语法如下：

filter { geoip { source => "clientIp" } } 12345

4. Output Plugin

4.1 输出到控制台

多用于调试

output { stdout { codec => rubydebug } } 12345

4.2 输出到文件

实现将分散在多地的文件统一到一处的需求，比如将所有web机器的web日志收集到1个文件中，从而方便查阅信息

output { file { path => "文件路径" codec => line {format => %{message}} } } 

4.3 输出到elasticsearch

output {  elasticsearch {    hosts => ["http://192.168.3.12:9200"]    index => "logstash-%{+YYYY.MM.dd}"    document_type => "_doc"    user => "用户名"    password => "密码" } }