2.1 输入插件
在"hello World" 示例中,我们已经见到并介绍了Logstash 的运行流程和配置的基础语法。
请记住一个原则: Logstash 配置一定要有一个input和一个output
在演示过程中,如果没有写明input,默认就会使用 logstash-input-stdin
同理,没有写明的output 就是logstash-output-stdout
2.1.1 标准输入
[elk@Vsftp logstash]$ cat stdin.conf
input {
stdin {
add_field =>{"key11"=>"value22"}
codec=>"plain"
tags=>["add"]
type=>"std"
}
}
output {
stdout {
codec => rubydebug
}
}
[elk@Vsftp logstash]$ logstash -f stdin.conf
Settings: Default pipeline workers: 4
Pipeline main started
abc123
{
"message" => "abc123",
"@version" => "1",
"@timestamp" => "2017-02-08T02:14:53.476Z",
"type" => "std",
"key11" => "value22",
"tags" => [
[0] "add"
],
"host" => "Vsftp"
}
[elk@Vsftp logstash]$ cat stdin.conf
input {
stdin {
add_field =>{"key11"=>"value22"}
codec=>"plain"
tags=>["add","xxyy"]
type=>"std"
}
}
output {
stdout {
codec => rubydebug
}
}
[elk@Vsftp logstash]$ logstash -f stdin.conf
Settings: Default pipeline workers: 4
Pipeline main started
this is scan
{
"message" => "this is scan",
"@version" => "1",
"@timestamp" => "2017-02-08T02:15:39.183Z",
"type" => "std",
"key11" => "value22",
"tags" => [
[0] "add",
[1] "xxyy"
],
"host" => "Vsftp"
}
根据tags 判断:
[elk@Vsftp logstash]$ cat stdin.conf
input {
stdin {
add_field =>{"key11"=>"value22"}
codec=>"plain"
tags=>["add","xxyy"]
type=>"std"
}
}
output {
if "tttt" in [tags]{
stdout {
codec=>rubydebug{}
}
}
else if "add" in [tags]{
stdout {
codec=>json
}
}
}
[elk@Vsftp logstash]$ logstash -f stdin.conf
Settings: Default pipeline workers: 4
Pipeline main started
yyyyyjjjj
{"message":"yyyyyjjjj","@version":"1","@timestamp":"2017-02-08T02:20:42.833Z","type":"std","key11":"value22","tags":["add","xxyy"],"host":"Vsftp"}
2.1.2 文件输入:
logstash 使用一个名叫FileWatch的Ruby Gem库来监听文件变化。这个库支持glob展开文件路径,而且会记录一个叫.sincedb的数据库文件来跟踪被监听日志文件的当前读取位置
[elk@Vsftp logstash]$ cat log.conf
input {
file {
path =>["/var/log/*.log","/var/log/mm"]
type=>"system"
start_position =>"beginning"
}
}
output {
stdout {
codec => rubydebug
}
}
[elk@Vsftp logstash]$ logstash -f log.conf
Settings: Default pipeline workers: 4
Pipeline main started
{
"message" => "11111111111",
"@version" => "1",
LogStash::Inputs::File 只是在进程运行的注册阶段初始化一个FileWatch对象。
所以它不能支持类型fluentd 那样的path=>"
2.1.3 TCP 输入
未来你可能会用Redis 服务器或者其他的消息队列系统来作为Logstash Broker的角色。
不过Logstash 其实也有自己的TCP/UDP 插件,在临时任务的时候,也算能用,尤其是测试环境。
[elk@Vsftp logstash]$ cat tcp.conf
input {
tcp {
port =>8888
mode=>"server"
ssl_enable =>false
}
}
output {
stdout {
codec => rubydebug
}
}
[elk@Vsftp logstash]$ logstash -f tcp.conf
Settings: Default pipeline workers: 4
Pipeline main started
{
"message" => "9999999999",
"@version" => "1",
"@timestamp" => "2017-02-08T03:02:43.746Z",
"host" => "127.0.0.1",
"port" => 47187
}
{
"message" => "000000000",
"@version" => "1",
"@timestamp" => "2017-02-08T03:02:43.747Z",
"host" => "127.0.0.1",
"port" => 47187
}
Vsftp:/var/log# nc 127.0.0.1 8888 < mm
Vsftp:/var/log# cat mm
9999999999
000000000
2.1.4 syslog输入:
syslog 可能是运维领域最流行的数据传输协议了,当你想从设备上收集系统日志的时候,syslog 应该会是你第一选择。
尤其是网络设备
介绍 如何把Logstash 配置成一个syslog 服务器来接收数据。
2.2 编解码配置:
Codec 是Logstash 从1.3.0 开始引入的概念(Codec 来自Coder/decoder 两个单词的首字母缩写)
我们在第一个"Hello World" 用例就已经用过Codec了
rubydebug就是一种Codec 虽然它一般只会在stdout 插件中,作为配置测试或者调试的工具
2.2.1 JSON 编解码:
2.2.2 多行事件编码
有些时候,应用程序调试日志会包含非常丰富的内容,为一个事件打印出很多行内容。
这种日志通常都很难通过命令行解析的方式做分析
而Logstash 正为此准备好了 codec/multiline 插件！当然,multiline插件也可以用于其他类似的堆栈信息
Vsftp:/home/elk/logstash# cat multi.conf
input {
stdin {
codec=>multiline {
pattern =>"^["
negate =>true
what=>"previous"
}
}
}
output {
stdout {
codec => rubydebug
}
}
[elk@Vsftp logstash]$ logstash -f multi.conf
Settings: Default pipeline workers: 4
Pipeline main started
[Aug/02/03 11:45:27] aaaa
bbbb
cccc
[Aug/02/03 11:45:27] 998877
{
"@timestamp" => "2017-02-08T05:27:07.442Z",
"message" => "[Aug/02/03 11:45:27] aaaa
bbbb
cccc",
"@version" => "1",
"tags" => [
[0] "multiline"
],
"host" => "Vsftp"
}
其实这个插件的原理很简单,就是把当前行的数据添加到前面一行后面,直到新进的当前行匹配^[正则为止。
2.3 过滤器配置:
2.3.1 date时间处理
之前章节已经提过,logstash-filter-date插件可以用来转换你的日志记录中的时间字符串,变成LogStash::Timestamp 对象,然后转存到@timestamp 字段里
因为在稍后的logstash-outout-elasticsearch 中常用的%{+YYYY.MM.dd}这种写法必须读取@timestamp 数据
%{TIMESTAMP_ISO8601:time} 匹配如下时间格式:
2011-04-18 08:20:11
2011-04-18 08:20:11,108
[elk@Vsftp logstash]$ logstash -f stdin.conf
Settings: Default pipeline workers: 4
Pipeline main started
aaaaabbbbbccccc
{"message":"aaaaabbbbbccccc","@version":"1","@timestamp":"2017-02-08T05:44:44.165Z","type":"std","key11":"value22","tags":["add","xxyy"],"host":"Vsftp"}