Logstash收集日志

王念博客

发布于 2019-07-25 17:46:22

1.7K0

发布于 2019-07-25 17:46:22

文章被收录于专栏：王念博客

Logstash 的介绍

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

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

官网下载地址：https://www.elastic.co/downloads/logstash

官网文档地址：https://www.elastic.co/guide/en/logstash/current/index.html

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

Logstash 的安装

Logstash 的安装非常简单，直接官网下载对应的版本（需要和 Elasticsearch 的版本匹配）解压缩即可

# tar -zxvf logstash-2.3.2.tar.gz

测试启动

# cd logstash/
# cd bin/
# ./logstash -e 'input { stdin { } } output { stdout {} }'

启动成功提示

Settings: Default pipeline workers: 4
Pipeline main started

Logstash配置

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

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

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

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

在dir下创建一个文件，文件名随意

将dir加入grok路径： patterns_dir => "./dir"

将想要增加的SYNTAX写入： SYNTAX_NAME regexp

使用方法和使用默认SYNTAX相同：%{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 +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=>"xx_server_log"
   path=>"/opt/software/apache-tomcat-7.0.59/logs/catalina.out"
 }
}
filter {
        if [type] == "xx_server_log" {
           grok {
                 match => [ "message","%{SERVER_LOG}"]
                 patterns_dir => ["/opt/conf/logstash"]
                 remove_field => ["message"]
          }
        }
}
output {
   redis{
   host => '10.4.8.5' #redis服务器地址
   key => 'logstash:redis'
   datatype => 'list'
  }
  stdout{     #调试方便可以在输出到stdout,判断grok解析是否正确.如果tag中没有出现grok failure字样，说明解析正确。
        codec=>json
  }
}

Logstash启动

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

解释： -f 后面指定的是配置文件

导入到es

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

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

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

Elasticsearch Service