logstash的处理过程

logstash在处理日志的整个过程是一个流的形式&＃xff0c;按照 input -> filter-> output 这样的顺序进行。
(严格的说法是input -> decode -> filter -> encode -> output 这样的一个流&＃xff0c;这里为了便于说明&＃xff0c;简略下)
如图&＃xff1a;

input&＃xff1a;负责日志的接收&＃xff0c;服务端角色。比如收集各服务器的nginx日志&＃xff0c;MySQL日志&＃xff0c;系统日志&＃xff0c;php慢日志等。

filter: 对日志进行预处理等&＃xff0c;后面会着重说下。

output: 负责日志的输出&＃xff0c;比如储存到哪个地方或者执行某些动作。

input配置

可以通过如下方式来接收日志&＃xff1a;

file&＃xff1a;顾名思义&＃xff0c;直接读文件
stdin: 标准输入&＃xff0c;调试配置的时候玩玩
syslog: syslog协议的日志格式&＃xff0c;比如linux的rsyslog
tcp/udp&＃xff1a;使用tcp或udp传输过来的日志

看一个file的配置

input {file {path &＃61;> ["/var/log/*.log", "/var/log/message"]type &＃61;> "system"start_position &＃61;> "beginning"codec &＃61;> "json"}
}

这些参数用途如下&＃xff1a;

path: 日志文件或目录的绝对路径&＃xff0c;也可以是通配符的。
type: 类型&＃xff0c;自定义
start_position: logstash 从什么位置开始读取文件数据&＃xff0c;默认是结束位置&＃xff0c;也就是说 logstash 进程会以类似 tail -F 的形式运行。如果你是要导入原有数据&＃xff0c;把这个设定改成 "beginning"&＃xff0c;logstash 进程就从头开始读取&＃xff0c;类似 less &＃43;F 的形式运行。
codec: codec配置&＃xff0c;通过它可以更好更方便的与其他有自定义数据格式的运维产品共存&＃xff0c;比如 graphite、fluent、netflow、collectd&＃xff0c;以及使用 msgpack、json、edn 等通用数据格式的其他产品等。

再看一个tcp的配置

input {tcp {port &＃61;> 8888mode &＃61;> "server"ssl_enable &＃61;> false}tcp {port &＃61;> 9999mode &＃61;> "server"ssl_enable &＃61;> false}
}

这里可以看到它支持ssl加密&＃xff0c;传输更安全。
更多input的插件请参考: Logstash Input

filter配置

这是今天的主菜&＃xff1a;过滤器。
logstash收集到日志后&＃xff0c;这些日志是原始的&＃xff0c;但需求是多变的&＃xff0c;比如日志中的有些内容要拆分成不同的字段&＃xff0c;或者要把多种日志格式(比如有nginx日志&＃xff0c;mysql慢日志等)统一成一种数据格式(比如json)等等&＃xff0c;这些都通过filter来实现。

同input一样&＃xff0c;filter也有各种各样的插件来处理日志&＃xff0c;常见的有grok,ruby,kv,date等。这里主要介绍grok和ruby&＃xff0c;详细参考Logstash Filter Plugin

如果你的日志在生成阶段就已经处理好了&＃xff0c;不需要额外的处理时&＃xff0c;可以不用filter&＃xff0c;logstash可这样配置&＃xff1a;

input {file {path &＃61;> "/opt/logstash/log"codec &＃61;> "json"}
}
output{stdout{codec&＃61;>rubydebug}
}

grok

grok类似于grep命令&＃xff0c;是一个正则表达式的插件&＃xff0c;通过正则匹配出我们需要的内容。
比如nginx的日志如下&＃xff1a;

172.16.91.200 - - [19/Jan/2017:17:20:17 &＃43;0800] "GET /favicon.ico HTTP/1.1" 200 0 "http://172.16.93.237:9881/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.95 Safari/537.36"

我想把里面的ip 172.16.91.200存储到clientip中&＃xff0c;之后我在kibana中查看时&＃xff0c;通过clientip就能查到ip了。
看一下grok是怎么匹配的

%{IPORHOST:clientip} - - \[%{HTTPDATE:request_time}\] \"(?:%{WORD:method} %{URIPATH:url}(?:%{URIPARAM:params})?(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})\" %{NUMBER:status} (?:%{NUMBER:bytes:int}|-) \"%{DATA:referrer}\" \"%{DATA:agent}\"

可以发现&＃xff0c;它的形式并不像我们平常写的正则表达式。
看下grok语法&＃xff1a;

%{PATTERN_NAME:capture_name:data_type}

这里有三部分PATTERN_NAME&＃xff0c;capture_name&＃xff0c;data_type。

1. PATTERN_NAM
正则变量&＃xff0c;指向一个正则表达式&＃xff0c;可以自定义&＃xff0c;如

USERNAME [a-zA-Z0-9._-]&＃43; #定义个正则表达式的变量
USER %{USERNAME} #使用这个正则表达式

logstash默认提供了很多的正则表达式&＃xff0c;具体可参考&＃xff1a;Logstash Grok Patterns。
在调试grok时&＃xff0c;可以借助下Grok Debugger。

回到刚才的grok&＃xff0c;匹配客户端ip的部分是:

%{IPORHOST:clientip}

这里的正则用了IPORHOST&＃xff0c;它实际内容如下&＃xff1a;

IPORHOST (?:%{HOSTNAME}|%{IP})

可以看到它引用了两个正则变量HOSTNAME和IP&＃xff0c;这两个的实际内容如下:

HOSTNAME \b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b)
IP (?:%{IPV6}|%{IPV4})

IP又引用了两个正则变量IPV4和IPV6。

2. capture_name
可以理解为把匹配的值存储到哪个field中。比如这里的ip匹配&＃xff0c;存储为clientip。

3. data_type
数据类型&＃xff0c;不是必填项。默认是字符串&＃xff0c;其他类型还有float&＃xff0c;int等。

了解了这三部分内容后&＃xff0c;再看grok的配置就明了了。
看一个完整的配置&＃xff1a;

input {file {path &＃61;> "/opt/logstash/log"}
}filter {grok {match &＃61;> {"message" &＃61;> "%{IPORHOST:clientip} - - \[%{HTTPDATE:request_time}\] \"(?:%{WORD:method} %{URIPATH:url}(?:%{URIPARAM:params})?(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})\" %{NUMBER:status} (?:%{NUMBER:bytes:int}|-) \"%{DATA:referrer}\" \"%{DATA:agent}\""}}
}output{stdout{codec&＃61;>rubydebug}
}

ruby

通过filters/ruby插件&＃xff0c;可以在filter中使用ruby&＃xff0c;极大地方便了日志处理。
看一个官方的示例&＃xff1a;

filter {ruby {init &＃61;> "&＃64;kname &＃61; [&＃39;client&＃39;,&＃39;servername&＃39;,&＃39;url&＃39;,&＃39;status&＃39;,&＃39;time&＃39;,&＃39;size&＃39;,&＃39;upstream&＃39;,&＃39;upstreamstatus&＃39;,&＃39;upstreamtime&＃39;,&＃39;referer&＃39;,&＃39;xff&＃39;,&＃39;useragent&＃39;]"code &＃61;> "new_event &＃61; LogStash::Event.new(Hash[&＃64;kname.zip(event.get(&＃39;message&＃39;).split(&＃39;|&＃39;))])new_event.remove(&＃39;&＃64;timestamp&＃39;)event.append(new_event)"}
}

参数如下&＃xff1a;

init:用来预定义参数。
code:要运行的ruby语句。

比如我想对nginx日志进行一个简单的归类&＃xff0c;区分下动态和静态资源。把css&＃xff0c;图片&＃xff0c;字体归为静态资源&＃xff0c;其他的划为动态。配置示例如下&＃xff1a;

input {file {path &＃61;> "/opt/logstash/log"codec &＃61;> "json"}
}filter {if [url] {ruby {code &＃61;> "url_match &＃61; /(.*).(css|js|png|html|gif|png|woff)/.match(event.get(&＃39;url&＃39;))if ( url_match )url_type &＃61; &＃39;static&＃39;elseurl_type &＃61; &＃39;dynamic&＃39;endevent.set(&＃39;url_type&＃39;,url_type)"}}
}output{stdout{codec&＃61;>rubydebug}
}

用logstash运行测试下&＃xff0c;可以看到多了个值url_type

/opt/logstash/bin/logstash -f /etc/logstash/conf.d/test.conf

output配置

output负责把处理好日志输出到指定的地方&＃xff0c;和input一样&＃xff0c;output也有丰富的插件&＃xff1a;

elasticsearch&＃xff1a;可以通过http等方式存入elasticsearch 中
email&＃xff1a;通过邮件发送出去
file: 存到文件中
nagios&＃xff1a;发送到nagios中
exce: 执行某个程序或命令
statsd:输出到statsd中
stdout:有标准输入&＃xff0c;那就有标准输出
tcp/udp:通过tcp/udp输出
HDFS:输出到hadoop中&＃xff0c;搞大数据:)

这里主要看下输出到elasticsearch的配置。

output {elasticsearch {hosts &＃61;> ["192.168.0.2:9200"]index &＃61;> "logstash-%{type}-%{&＃43;YYYY.MM.dd}"document_type &＃61;> "%{type}"flush_size &＃61;> 20000idle_flush_time &＃61;> 10sniffing &＃61;> truetemplate_overwrite &＃61;> true}
}

主要参数如下&＃xff1a;

host: es的主机和端口
index:写入es的索引名称
document_type&＃xff1a;es的document_type
flush_size:指定数据达到多少条时再发送
idle_flush_time&＃xff1a;结合flush_size使用&＃xff0c;指在这个时间内即使没攒够flush_size数&＃xff0c;也发送。比如flush_size设置1000条&＃xff0c;idle_flush_time设置为5秒&＃xff0c;则在这5秒中&＃xff0c;即使数目没达到1000条也会发送。而如果到3秒时就有1000条了&＃xff0c;则会立即发送。

结语

logstash的input&＃xff0c;filter&＃xff0c;output三个阶段都有很丰富的插件&＃xff0c;可根据自己的需求来搭配使用。
每部分可配置多个不同的内容&＃xff0c;比如input可以同时配置file和tcp&＃xff0c;并且配置多个tcp。

参考

Logstash 到底该怎么用
logstash使用指南