logstash 输入插件

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"}