1.logstash的概念及特点。 概念:logstash是一个数据采集、加工处理以及传输(输出)的工具。 特点: - 所有类型的数据集中处理 - 不同模式和格式数据的正常化 - 自定义日志格式的迅速扩展 - 为自定义数据源轻松添加插件
2.logstash安装配置。 ①.下载安装 [root@node1 ~]# wget https://download./logstash/logstash/packages/centos/logstash-2.3.4-1.noarch.rpm [root@node1 ~]# rpm -ivh logstash-2.3.4-1.noarch.rpm ②.简单测试logstash,启动后输入"hello,xkops"。 [root@node1 ~]# /opt/logstash/bin/logstash -e 'input{ stdin{} } output{ stdout{} }' *提示:如果输出"hello,xkops",则证明logstash启动成功。 ③.以服务启动方式。 [root@node1 ~]# service logstash start 总结:logstash三种启动方式,-e sting类型启动,-f 指定配置文件启动,服务启动。
3.logstash配置语句详解。 logstash配置文件包含三个配置部分,分别为:input{}、filter{}、output{}。 {} 定义区域,区域内可以定义一个或多个插件,通过插件对数据进行收集,加工处理,输出。
数据类型: 布尔值类型: ssl_enable => true 字节类型: bytes => "1MiB" 字符串类型: name => "xkops" 数值类型: port => 22 数组: match => ["datetime","UNIX"] 哈希: options => {key1 => "value1",key2 => "value2"} 编码解码: codec => "json" 路径: file_path => "/tmp/filename" 注释: #
条件判断: 等于: == 不等于: != 小于: < 大于: > 小于等于: <= 大于等于: >= 匹配正则: =~ 不匹配正则: !~ 包含: in 不包含: not in 与: and 或: or 非与: nand 非或: xor 复合表达式: () 取反符合: !()
4.logstash常用插件。 类型: input类:也就是在input区域内定义使用的插件。 codec类:定于处理数据格式,如plain,json,json_lines等格式。这个可以定义在input、output区域中。 filter类:也就是在filter区域内定义使用的插件。 output类:也就是在output区域内定义使用的插件。 各类插件地址:https://github.com/logstash-plugins
----------------------input类插件------------------------------- input类插件,常用的插件:file、tcp、udp、syslog,beats等。 ①.file插件: file插件字段解释: codec => #可选项,默认是plain,可设置其他编码方式。 discover_interval => #可选项,logstash多久检查一下path下有新文件,默认15s。 exclude => #可选项,排除path下不想监听的文件。 sincedb_path => #可选项,记录文件以及文件读取信息位置的数据文件。~/.sincedb_xxxx sincedb_write_interval => #可选项,logstash多久写一次sincedb文件,默认15s. stat_interval => #可选项,logstash多久检查一次被监听文件的变化,默认1s。 start_position => #可选项,logstash从哪个位置读取文件数据,默认从尾部,值为:end。初次导入,设置为:beginning。 path => #必选项,配置文件路径,可定义多个。 tags => #可选项,在数据处理过程中,由具体的插件来添加或者删除的标记。 type => #可选项,自定义处理时间类型。比如nginxlog。
实例: [root@node1 conf.d]# cat /tmp/file{1,2}.log This is test file-plugin in file1.log This is test file-plugin in file2.log [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat file.conf
input{
file{
start_position => "beginning"
path => ["/tmp/file1.log","/tmp/file2.log"]
type => 'filelog'
}
}
output{
stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f file.conf -t Configuration OK [root@node1 conf.d]# /opt/logstash/bin/logstash -f file.conf Settings: Default pipeline workers: 1 Pipeline main started 2016-07-16T02:50:07.410Z node1 This is test file-plugin in file1.log 2016-07-16T02:50:07.428Z node1 This is test file-plugin in file2.log *提示:能够输出内容则证明file插件正常工作。
②.tcp插件: tcp插件字段解释: add_field => #可选项,默认{}。 codec => #可选项,默认plain。 data_timeout => #可选项,默认-1。 host => #可选项,默认0.0.0.0。 mode => #可选项,值为["server","client"]之一,默认为server。 port => #必选,端口。 ssl_cacert => #可选项,定义相关路径。 ssl_cert => #可选项,定义相关路径。 ssl_enable => #可选项,默认为false。 ssl_key => #可选项,定义相关路径。 ssl_key_passphrase => #可选项,默认nil ssl_verify => #可选项,默认false。 tags => #可选项 type => #可选项
实例: [root@node1 conf.d]# cat /tmp/tcp.log this is test tcp-plugin in tcp.log send tcplog data output tcplog data [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat tcp.conf
input{
tcp{
host => "127.0.0.1"
port => 8888
type => "tcplog"
}
}
output{
stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f tcp.conf 此时开启另个终端,使用nc开启一个tcp端口8888,并将数据推送到8888端口。 [root@node1 conf.d]# nc 127.0.0.1 8888 < /tmp/tcp.log *提示:前一个终端如果出现数据,则证明tcp插件工作正常。
③udp插件: udp插件字段解释: add_field => #可选项,默认{}。 host => #可选项,默认0.0.0.0。 queue_size => #默认2000 tags => #可选项 type => #可选项 workers => #默认为2
实例: [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat udp.conf
input{
udp{
host => "127.0.0.1"
port => 9999
}
}
output{
stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f udp.conf 打开另一终端执行如下脚本,并输入内容:"hello,udplog"。 [root@node1 conf.d]# cat /tmp/udpclient.py
#/usr/bin/env python
import socket
host = "127.0.0.1"
port = 9999
file_input = raw_input("Please input udp log: ")
s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
s.sendto(file_input,(host,port))
*提示:如果前一个终端收到日志,则证明udp插件工作正常。
④.syslog插件: 实例: [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat syslog.conf
input{
syslog{
host => "127.0.0.1"
port => 518
type => "syslog"
}
}
output{
stdout{}
}
[root@node1 conf.d]# echo '*.* @@127.0.0.1:518' >> /etc/rsyslog.conf [root@node1 conf.d]# /etc/init.d/rsyslog restart 关闭系统日志记录器: [确定] 启动系统日志记录器: [确定] [root@node1 conf.d]# /opt/logstash/bin/logstash -f syslog.conf 使用logger命令向系统写入日志: [root@node1 conf.d]# logger
*提示:此处随便输入内容,查看前一个终端是否会有内容输出,如果输出则证明syslog插件工作正常。
--------------------------codec类插件------------------------------------ codec类插件,常用的插件:plain、json、json_lines、rubydebug、multiline等。 ①.plain插件: 实例: [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat plain.conf
input{
stdin{
codec => "plain"
}
}
output{
stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f plain.conf 输入信息查看输出。
②.json插件: 实例: [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat json.conf
input{
stdin{}
}
output{
stdout{
codec => "json"
}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f json.conf 输入信息查看输出。
③.json_lines插件:(json文本过长时使用) 实例: [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat jsonlines.conf
input{
tcp{
host => "127.0.0.1"
port => 8888
codec => "json_lines"
}
}
output{
stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f jsonlines.conf 启动一个新的终端,执行如下命令。 [root@node1 conf.d]# cat /tmp/jsonlines.txt You run a price alerting platform which allows price-savvy customers to specify a rule like "I am interested in buying a specific electronic gadget and I want to be notified if the price of gadget falls below $X from any vendor within the next month". In this case you can scrape vendor prices, push them into Elasticsearch and use its reverse-search (Percolator) capability to match price movements against customer queries and eventually push the alerts out to the customer once matches are found. [root@node1 conf.d]# nc 127.0.0.1 8888 < /tmp/jsonlines.txt *提示:观察前一个终端的输出,如果正常输出,则json_lines插件工作正常。
④.rubedebug插件: 实例: [root@node1 conf.d]# cat rubydebug.conf
input{
stdin{
codec => "json"
}
}
output{
stdout{
codec => "rubydebug"
}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f rubydebug.conf 输入json串查看输出效果。 json串:{"name":"xkops","age":"25"}
⑤.multiline插件:(处理错误日志) multiline插件字段: charset => #字符编码,可选 max_bytes => #bytes类型,设置最大的字节数,可选 max_lines => #number类型,设置最大的行数,默认是500行,可选 multiline_tag => #string类型,设置一个事件标签,默认是"multiline" ,可选 pattern => #string 类型,设置匹配的正则表达式 ,必选 patterns_dir => #array类型,可以设置多个正则表达式,可选 negate => #boolean类型,设置正向匹配还是反向匹配,默认是false,可选 what => #设置未匹配的内容是向前合并还是向后合并,previous, next 两个值选择,必选
错误日志: [16-07-2016 22:54:01] PHP warning: unknown exception in /xxx/test/index.php:99 111111111111111111 222222222222222222 [16-07-2016 23:19:43] PHP warning: unknown exception in /xxx/test/index.php:93 [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat codecmultilines.conf
input{
stdin{
codec => multiline{
pattern => "^\["
negate => true
what => "previous"
}
}
}
output{
stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f codecmultilines.conf *提示:输入上述错误日志查看输出。
---------------------filter类插件--------------------------------------------- filter插件,常用的filter插件:json、grok等。 ①.json插件: add_field => #hash(可选项),默认{} add_tag => #array(可选项),默认[] remove_field => #array(可选项),默认[] remove_tag => #array(可选项),默认[] source => #string(必选项) target => #string(可选项)
实例: [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat filterjson.conf
input{
stdin{}
}
filter{
json{
source => "message"
#target => "content"
}
}
output{
stdout{
codec => "rubydebug"
}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f filterjson.conf 输入两个串,查看输出。 {"name":"xkops","age":"25"} name xkops
②.grok插件:解析各种非结构化的日志数据插件。 grok有丰富的patterns,查看方式: cat /opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns/grok-patterns https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns 实例: 输入的元数据内容: 20.3.1.3 GET /xkops/index.html 8838 1.323 [root@node1 conf.d]# pwd /etc/logstash/conf.d [root@node1 conf.d]# cat filtergrok.conf
input{
stdin{}
}
filter{
grok{
match => ["message","%{IP:ip} %{WORD:method} %{URIPATH:uri} %{NUMBER:bytes} %{NUMBER:duration}"]
}
}
output{
stdout{
codec => "rubydebug"
}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f filtergrok.conf *提示:输入上述元数据查看输出。 *提示:grok在线工具使用:https://grokdebug./
③.kv插件: 解析处理key-value键值对数据
--------------------output类插件----------------------------- output插件: ①.file插件 ②.tcp/udp插件 ③.redis/kfaka ④.elasticsearch
附录: redis配置:
input{
redis{
host => 'redis-server'
port => '6379'
data_type => 'list'
key => 'lb'
codec => 'json'
}
}
|