Logstash收集日志

logstash是一种分布式日志收集框架&＃xff0c;开发语言是JRuby&＃xff0c;当然是为了与Java平台对接&＃xff0c;不过与Ruby语法兼容良好&＃xff0c;非常简洁强大&＃xff0c;经常与ElasticSearch&＃xff0c;Kibana配置&＃xff0c;组成著名的ELK技术栈&＃xff0c;非常适合用来做日志数据的分析。

当然它可以单独出现&＃xff0c;作为日志收集软件&＃xff0c;你可以收集日志到多种存储系统或临时中转系统&＃xff0c;如MySQL&＃xff0c;redis&＃xff0c;kakfa&＃xff0c;HDFS, lucene&＃xff0c;solr等并不一定是ElasticSearch。

官网下载地址&＃xff1a;https://www.elastic.co/downloads/logstash 

官网文档地址&＃xff1a;https://www.elastic.co/guide/en/logstash/current/index.html 

中文文档地址&＃xff1a;http://udn.yyuap.com/doc/logstash-best-practice-cn/index.html

Logstash 的安装非常简单&＃xff0c;直接官网下载对应的版本&＃xff08;需要和 Elasticsearch 的版本匹配&＃xff09;解压缩即可

# tar -zxvf logstash-2.3.2.tar.gz

测试启动

# cd logstash/
# cd bin/
# ./logstash -e &＃39;input { stdin { } } output { stdout {} }&＃39;

启动成功提示

Settings: Default pipeline workers: 4
Pipeline main started Logstash配置

  启动logstash收集需要创建自定义conf配置文件 

   通常 Logstash 的一个配置文件含 input{}&＃xff0c;filter{}&＃xff0c;output{} 三部分&＃xff0c;该三部分的作用也很好理解&＃xff0c;一个配置输入源&＃xff0c;一个配置过滤规则&＃xff0c;一个配置输出&＃xff1b;

   其每个部分又可以配置各种不同的插件&＃xff0c;所有插件的说明请参考官网文档的说明

   三部分我觉得最难的配置就是filter了&＃xff0c;需要配置grok插件

   在dir下创建一个文件&＃xff0c;文件名随意

   将dir加入grok路径&＃xff1a; patterns_dir &＃61;> "./dir"

   将想要增加的SYNTAX写入&＃xff1a; SYNTAX_NAME regexp

   使用方法和使用默认SYNTAX相同&＃xff1a;%{SYNTAX_NAME:SEMANTIC}

#log4j日志
# 2016-04-10 07:19:16-|INFO|-Root WebApplicationContext: initialization started
MYTIMESTAMP 20%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:%{MINUTE}:%{SECOND}
MYLOG %{MYTIMESTAMP:mytimestamp}-\|%{LOGLEVEL:level}\|-%{JAVALOGMESSAGE:logmsg}SERVER_LOG %{DATA:year}-%{DATA:month}-%{DATA:day}\ %{DATA:hour}\:%{DATA:min}\:%{DATA:sec}\ %{DATA:level}\ %{DATA:class} -{ip:%{DATA:ip},url:%{DATA:url},param:%{DATA:param},return:%{DATA:return},cost:%{BASE10NUM:cost}#tomcatAccess日志
ACCESSIP (?:[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})
ACCESSTIMESTAMP %{MONTHDAY}\/%{MONTH}\/20%{YEAR}:%{HOUR}:%{MINUTE}:%{SECOND} %{ISO8601_TIMEZONE}
HTTPMETHOD (GET|POST|PUT|DELETE)
PRJNAME ([^\s]*)
HTTPVERSION (https?\/[0-9]{1}\.[0-9]{1})
STATUSCODE ([0-9]{3})
# 192.168.1.101 - - [10/Apr/2016:08:31:34 &＃43;0800] "GET /spring-mvc-showcase HTTP/1.1" 302 -
ACCESSLOG %{ACCESSIP:accIP}\s-\s\-\s\[%{ACCESSTIMESTAMP:accstamp}\]\s"%{HTTPMETHOD:method}\s\/%{PRJNAME:prjName}\s%{JAVALOGMESSAGE:statusCode}

  创建.conf文件

input {file {type&＃61;>"xx_server_log"path&＃61;>"/opt/software/apache-tomcat-7.0.59/logs/catalina.out"}
}
filter {if [type] &＃61;&＃61; "xx_server_log" {grok {match &＃61;> [ "message","%{SERVER_LOG}"]patterns_dir &＃61;> ["/opt/conf/logstash"]remove_field &＃61;> ["message"]}}
}
output {redis{host &＃61;> &＃39;10.4.8.5&＃39; #redis服务器地址key &＃61;> &＃39;logstash:redis&＃39;datatype &＃61;> &＃39;list&＃39;}stdout{     #调试方便可以在输出到stdout,判断grok解析是否正确.如果tag中没有出现grok failure字样&＃xff0c;说明解析正确。codec&＃61;>json}
} Logstash启动

# ./bin/logstash -f ./conf/access-log.conf

  解释&＃xff1a; -f 后面指定的是配置文件

以上我是将采集到的日志先放在redis缓存&＃xff0c;再次通过logstash导入到es

input {redis{host &＃61;> &＃39;10.4.8.5&＃39;key &＃61;> &＃39;logstash:redis&＃39;datatype &＃61;> &＃39;list&＃39;}
}
#日志在上面已经结构化了&＃xff0c;可以不再解析。
#filter{
#}
output
{# stdout{}elasticsearch {host &＃61;> &＃39;10.4.8.5&＃39;protocol &＃61;> &＃39;http&＃39;index &＃61;> "logstash-%{type}"#设置在elasticsearch中的indexer名字&＃xff0c;默认为logstash-%{yyyy.MM.dd}}
}

博客地址&＃xff1a;http://my.oschina.net/wangnian