LogStash的安装部署与应用

介绍

1、Logstash是一个接收，处理，转发日志的工具;
2、Logstash支持网络日志、系统日志、应用日志、apache日志等等，总之可以处理所有日志类型;
3、典型应用场景ELK：logstash负责采集、解析日志，elasticsearch负责数据存储，kibana负责前端报表展示。

下载

https://www.elastic.co/cn/downloads/logstash

安装部署

上传服务器、解压即可使用。

配置

根据需要修改配置文件config/logstash.yml，默认不修改即可运行

节点名称：node.name
日志级别：log.level（默认debug，如果想要看详细日志改为trace）

根据需要调整jvm.options配置文件：

-Xms256m
-Xmx1g

测试安装是否成功:
快速启动，标准输入输出作为input和output

./bin/logstash -e 'input { stdin {} } output { stdout {} }'

使用

主要组件

Input组件：负责采集日志数据，包括文件、syslog、collectd、kafka、redis等等；
Filter：负责解析日志数据，包括解析、加工、转换数据等；
Output：负责输出日志数据，对接到redis、kafka、elasticsearch、hdfs等存储组件；

常用启动参数

-e   立即执行，使用命令行里的配置参数启动实例 ./bin/logstash -e ‘input {stdin {}} output {stdout {}}'
-f  指定启动实例的配置文件 ./bin/logstash -f config/test.conf
-t  测试配置文件的正确性  ./bin/logstash-f config/test.conf -t
-l  指定日志文件名称    ./bin/logstash-f config/test.conf -l logs/test.log
-w  指定filter线程数量，默认线程数是5 ./bin/logstash-f config/test.conf -w

常用input配置

File

文件读取插件主要用来抓取文件的变化信息，将变化信息封装成Event进程处理或者传递。

inputfile {#监听文件的路径path => ["E:/software/logstash-1.5.4/logstash-1.5.4/data/*","F:/test.txt"]#排除不想监听的文件exclude => "1.log"#添加自定义的字段add_field => {"test"=>"test"}#增加标签tags => "tag1"#设置新事件的标志delimiter => "\n"#设置多长时间扫描目录，发现新文件discover_interval => 15#设置多长时间检测文件是否修改stat_interval => 1#监听文件的起始位置，默认是endstart_position => beginning#监听文件读取信息记录的位置sincedb_path => "E:/software/logstash-1.5.4/logstash-1.5.4/test.txt"#设置多长时间会写入读取的位置信息sincedb_write_interval => 15}
}

kafka

inputkafka {#集群地址bootstrap_servers => ["10.142.134.179:9092"]#消费组idgroup_id => "ete_serv_common"#数组类型，可配置多个topic topics_pattern => ["BO_TOPO_\d{1,}$"]正则匹配topics => ["logq","loge"]#从最早的偏移量开始消费latest从最新的开始消费auto_offset_reset => "earliest"#起多少个input线程数consumer_threads => 5#此属性会将当前topic、offset、group、partition等信息也带到message中decorate_events => true}
}

Beats

Beats插件用于建立监听服务，接收Filebeat或者其他beat发送的Events；

配置示例

input {beats {port => 5044}
}

TCP

TCP插件有两种工作模式，“Client"和"Server”，分别用于发送网络数据和监听网络数据。

配置示例

tcp {port => 41414
}

Redis

input {redis {host => "127.0.0.1"port => 6379data_type => "list"key => "logstash-list"}
}

常用的Filter配置

丰富的过滤器插件的是 logstash威力如此强大的重要因素，过滤器插件主要处理流经当前Logstash的事件信息，可以添加字段、移除字段、转换字段类型，通过正则表达式切分数据等，也可以根据条件判断来进行不同的数据处理方式。

grok 过滤器

grok 是Logstash中将非结构化数据解析成结构化数据以便于查询的最好工具，非常适合解析syslog logs，apache log， mysql log，以及一些其他的web log

Logstash提供120个常用正则表达式可供安装使用，安装之后你可以通过名称调用它们
语法如下：%{SYNTAX:SEMANTIC}

SYNTAX：表示已经安装的正则表达式的名称

SEMANTIC：表示从Event中匹配到的内容的名称

例如：Event的内容为"[debug] 127.0.0.1 - test log content"，匹配%{IP:client}将获得"client: 127.0.0.1"的结果，前提安装了IP表达式；

通过配置grok可以把 [debug] 127.0.0.1 - test log content 这样的非结构化数据转为：
“cllient”:“127.0.0.1”.

一个完整的例子

 日志文件http.log内容：192.168.1.11 GET /index.html 15994 0.053表达式：%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}配置文件内容：input {file {path => "/var/log/http.log"}
}
filter {grok {match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"}}
}输出结果：
client: 192.168.1.11
method: GET
request: /index.html
bytes: 15994
duration: 0.053

自定义表达式
与预定义表达式相同，你也可以将自定义的表达式配置到Logstash中，然后就可以像于定义的表达式一样使用；

语法：(?<field_name>the pattern here)举例：捕获10或11和长度的十六进制queue_id可以使用表达式(?<queue_id>[0-9A-F]{10,11})安装自定义表达式1、在Logstash根目录下创建文件夹"patterns"，在"patterns"文件夹中创建文件"extra"（文件名称无所谓，可自己选择有意义的文件名称）；2、在文件"extra"中添加表达式，格式：patternName regexp，名称与表达式之间用空格隔开即可，如下：# contents of ./patterns/postfix:POSTFIX_QUEUEID [0-9A-F]{10,11}3、使用自定义的表达式时需要指定"patterns_dir"变量，变量内容指向表达式文件所在的目录举例如下：## 日志内容
Jan  1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=20130101142543.5828399CCAF@mailserver14.example.com>## Logstash配置filter {grok {patterns_dir => ["./patterns"]match => { "message" => "%{SYSLOGBASE} %{POSTFIX_QUEUEID:queue_id}: %{GREEDYDATA:syslog_message}" }}}## 运行结果 ##timestamp: Jan 1 06:25:43logsource: mailserver14program: postfix/cleanuppid: 21403queue_id: BEF25A72965-Grok表达式在线debug地址：http://grokdebug.herokuapp.com
-预定义正则表达式参考地址：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns

date 时间处理过滤器

该插件用于时间字段的格式转换，比如将"Apr 17 09:32:01"（MMM dd HH:mm:ss）转换为"MM-dd HH:mm:ss"。
而且通常情况下，Logstash会为自动给Event打上时间戳，但是这个时间戳是Event的处理时间（主要是input接收数据的时间），和日志记录时间会存在偏差（主要原因是buffer），我们可以使用此插件用日志发生时间替换掉默认是时间戳的值。

#日志内容
{"agent":"Windows 7","client_time":"2017-11-20 12:00:00","client_ip":"123.10.91.106"
}# Filter 配置
filter {date {match => ["client_time", "yyyy-MM-dd HH:mm:ss"] #这里是如果client_time跟后面的格式匹配上了就会去替换，替换什么呢?#target默认指的就是@timestamp,所以就是以client_time的时间更新@timestamp的时间}
}

mutate数据修改过滤器

mutate 插件是 Logstash另一个重要插件。它提供了丰富的基础类型数据处理能力。
可以重命名，删除，替换和修改事件中的字段。

重命名 – rename 对于已经存在的字段，重命名其字段名称

filter {mutate {rename => ["syslog_host", "host"]}
}

更新字段内容 – update 更新字段内容，如果字段不存在，不会新建

filter {mutate {update => { "sample" => "My new message" }}
}

替换字段内容 – replace 与 update 功能相同，区别在于如果字段不存在则会新建字段

filter {mutate {replace => { "message" => "%{source_host}: My new message" }}
}

数据类型转换 – convert

filter {mutate {convert => ["request_time", "float"]}
}

JSON过滤器

JSON插件用于解码JSON格式的字符串，一般是一堆日志信息中，部分是JSON格式，部分不是的情况下

配置示例json {source => ...
}## 示例配置，message是JSON格式的字符串："{\"uid\":3081609001,\"type\":\"signal\"}" ##filter {json {source => "message"target => "jsoncontent"}
}## 输出结果 ##
{"@version": "1","@timestamp": "2014-11-18T08:11:33.000Z","host": "web121.mweibo.tc.sinanode.com","message": "{\"uid\":333333333,\"type\":\"signal\"}","jsoncontent": {"uid": 333333333,"type": "signal"}
}## 如果从示例配置中删除`target`，输出结果如下 ##
{"@version": "1","@timestamp": "2014-11-18T08:11:33.000Z","host": "web121.mweibo.tc.sinanode.com","message": "{\"uid\":333333333,\"type\":\"signal\"}","uid": 333333333,"type": "signal"
}

Output

ElasticSearch输出插件

用于将事件信息写入到Elasticsearch中，官方推荐插件，ELK必备插件

配置事例

output {elasticsearch {hosts => ["127.0.0.1:9200"]index => "filebeat-%{type}-%{+yyyy.MM.dd}"user => "upuptop" password => "upuptop"template => "/dd/c/caaaa.json"template_name => "index_scene_control"template_overwrite => true}
}

File输出插件

用于将Event输出到文件内

配置事例

output {file {path => "/home/upuptop/test.txt "}
}