Elasticsearch+Kibana+Logstash 搭建日志平台

阅读量：5804 次

发布时间：2019-06-18

本文共 7423 字，大约阅读时间需要 24 分钟。

使用elk方式对日志系统的解析

Java 环境部署

网上很多教程,此处只做出测试

java -versionjava version "1.7.0_45"Java(TM) SE Runtime Environment (build 1.7.0_45-b18)Java HotSpot(TM) 64-Bit Server VM (build 24.45-b08, mixed mode)

Elasticsearch 搭建

curl -O https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-1.5.1.tar.gztar zxvf elasticsearch-1.5.1.tar.gzcd elasticsearch-1.5.1/./bin/elasticsearchbin/elasticsearch -d #(后台运行)

es在此处不需要设置多少东西,基本上默认的就可以满足我们的要求了...

Logstash 搭建

初步搭建

curl -O http://download.elastic.co/logstash/logstash/logstash-1.5.1.tar.gz

现在你应该有了一个叫logstash-1.5.2.tar.gz的文件了。我们把它解压一下

tar zxvf logstash-1.4.2.tar.gz

cd logstash-1.5.1

现在我们来运行一下：

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

我们现在可以在命令行下输入一些字符，然后我们将看到logstash的输出内容：

hello world

2015-06-17T01:22:14.405+1000 0.0.0.0 hello world

Ok,还挺有意思的吧... 以上例子我们在运行logstash中，定义了一个叫"stdin"的input还有一个"stdout"的output，无论我们输入什么字符，Logstash都会按照某种格式来返回我们输入的字符。这里注意我们在命令行中使用了-e参数，该参数允许Logstash直接通过命令行接受设置。这点尤其快速的帮助我们反复的测试配置是否正确而不用写配置文件。

让我们再试个更有意思的例子。首先我们在命令行下使用CTRL-C命令退出之前运行的Logstash。现在我们重新运行Logstash使用下面的命令：

bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'

我们再输入一些字符，这次我们输入"goodnight moon" 将出现:

goodnight moon{  "message" => "goodnight moon",  "@timestamp" => "2013-11-20T23:48:05.335Z",  "@version" => "1",  "host" => "my-laptop"}

以上示例通过重新设置了叫"stdout"的output(添加了"codec"参数)，我们就可以改变Logstash的输出表现。类似的我们可以通过在你的配置文件中添加或者修改inputs、outputs、filters，就可以使随意的格式化日志数据成为可能，从而订制更合理的存储格式为查询提供便利。

集成Elasticsearch插入数据

以上的步骤已经成功的搭建了logstash,接下来增加logstash的配置文件,使其配置文件启动,将数据存入ES中,显示

1、在/root/config/目录下面增加logs.conf

input{    file{        type => "all"        path => "/root/tomcat7/logs/catalina.out"    }    file{        type => "access"        path => "/root/tomcat7/logs/access.log"    }}filter {    multiline {      pattern => "^[^\[]"      what => "previous"    }    if [type] == "access" {      grok {        pattern => "(?
    
     {.*}$)"      }      json {        source => request_info      }      geoip {        source => "client_ip"        fields => ["country_name", "region_name", "city_name", "real_region_name", "latitude", "longitude"]        remove_field => [ "[geoip][longitude]", "[geoip][latitude]","location","region_name" ]      }      useragent {          source => "user_agent"          prefix => "useragent_"          remove_field => [ "useragent_device", "useragent_major", "useragent_minor" ,"useragent_patch","useragent_os","useragent_os_major","useragent_os_minor"]      }    } else if [type] == 'all' {      grok {	pattern => "\[(?
     
      \w*).*\] (?
      
       \d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d{3})\s"      }    }    mutate {      remove_field => [ "request_info", "@version", "tags" ]      remove_tag => [ "_grokparsefailure" ]      replace => [ "host", "gd1_prd_yowoo_tomcat4" ]    }}output {  stdout { codec => rubydebug }  elasticsearch {    host => "localhost"    index => "logstash-%{type}-%{+YYYY.MM.dd}"    index_type => "%{type}"  }}

2、启动logstash(配置文件启动)

sh logstash -f /root/config/logs.conf

3、在上述的配置文件中,有指定了tomcat的日志,all是tomcat的日志,access是我们自己在程序中写的的日志,在log4j.xml中有:

/>

在log4j.xml中配置的是日滚的日志文件,logstash指向了生成日志文件的地址,进行监听,日志不会的,我在博客中有一个分类叫日志,进去看哇...

Kibana 搭建

下载Kibana

wget https://download.elastic.co/kibana/kibana/kibana-4.1.0-linux-x64.tar.gz

解压出来就行了...

配置使其读取ES的数据展示

进入Kibana目录/config中,修改kibana.yml文件,指定ES访问地址(ps:以前的版本是修改conf.js,不要让被人误导你...)

# Kibana is served by a back end server. This controls which port to use.port: 5601# The host to bind the server to.host: "0.0.0.0"# The Elasticsearch instance to use for all your queries.elasticsearch_url: "http://localhost:9200"

最后进入web页面中查看

下面我们讲解一下logstash中config配置参数

首先我们能看到的是三个input,filter和output,分别是读取日志,过滤分割和输出日志,重点在filter上:

[03-Jun-2013 13:15:29] PHP Fatal error: Uncaught exception 'Leb_Exception' in /data1/www/bbs.xman.com/htdocs/framework/xbox/ufo.php:68

Stack trace:

#0 /data/www/bbs.xman.com/htdocs/framework/dao/abstract.php(299): Leb_Dao_Pdo->connect(Array, 'read')

#1 /data/www/bbs.xman.com/htdocs/framework/dao/pdo.php(108): Leb_Dao_Abstract->initConnect(false)

#2 /data/www/bbs.xman.com/htdocs/framework/dao/abstract.php(1123): Leb_Dao_Pdo->query('SELECT * FROM `...')

#3 /data/www/bbs.xman.com/htdocs/framework/dao/abstract.php(1217): Leb_Dao_Abstract->select(Array)

#4 /data/www/bbs.xman.com/htdocs/framework/model.php(735): Leb_Dao_Abstract->daoSelect(Array, false)

#5 /data/www/bbs.xman.com/htdocs/app/configure/model/configure.php(40): Leb_Model->find()

#6 /data/www/bbs.xman.com/htdocs/app/search/default.php(131): Configure->get_configure_by_type('news')

#7 /data/www/bbs.xman.com/htdocs/framework/dispatcher.php(291): defaultController->indexAction()

#8 /data/www/bbs.xman.com/htdocs/framework/dispatcher.php(222): Leb_Di in /data1/www/bbs.xman.com/htdocs/framework/dao/pdo.php on line 68

这个时候 logstash一般会只记录上面一行，所以这类的日志就看不全了。怎么办呢？logstash提供了一个功能解决了这个问题就是"multiline"

这个filter的功能顾名思义就是对多行的日志进行处理这个是官网上的说明

multiline filter

This filter will collapse multiline messages into a single event.

The multiline filter is for combining multiple events from a single source into the same event.

下面看下格式

filter {

multiline {

type => "type" #类型,不多说

pattern => "pattern, a regexp" #参数,也可以认为是字符,有点像grep ,如果符合什么字符就交给下面的 what 去处理

negate => boolean

what => "previous" or "next" #这个是符合上面 pattern 的要求后具体怎么处理,处理方法有两种,合并到上面一条日志或者下面的日志

}

The 'negate' can be "true" or "false" (defaults false). If true, a message not matching the pattern will constitute a match of the multiline filter and the what will be applied. (vice-versa is also true)

这个 negate 有两种 true 或者 false，默认是 true,如果选了false 的话估计就是取反的意思。

看看例子

filter {

multiline {

pattern => "^[^\[]"

what => "previous"

}

这个例子是针对我上面的php日志写的，意思就是如果不是以 "["开头的日志都跟上一个日志合并在一起。以此类推遇到其他的多行日志也可以按照这个方法来做合并。

2、logstash 根据@message内容来触发命令"exec"

我当初的想法是这样的，如果php日志文件中出现 "PHP Fatal error"的时候将相关的错误日志发给相关开发的负责人。一开始想到了 output 的 email 功能，但我尝试

了很多次这个email功能不太稳定，有时候能发出来邮件有的时候却发不出来。不知道是邮件服务器的问题还是 logstash本身的问题,具体配置如下

output {

email {

match => [ "@message", "aaaaa" ]

to => "storyskya@gmail.com"

from => "monitor@mib.com.cn"

options => [ "smtpIporHost", "smtp.mibnet.com",

"port", "25",

"userName", "monitor@mib.com.cn",

"starttls", "true",

"password", "opmonitor",

"authenticationType", "login"

]

subject => "123"

body => '123'

via => smtp

}

下面我们介绍grop插件,对日志文件的解析,

grok {	pattern => "\[(?
    
     \w*).*\] (?
     
      \d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d{3})\s"}

上面的这个配置,就是将日志文件进行了解析,其中的level/datetime等字段会被抽取出来作为es的搜索键存在,但是整个日志信息是存在于message这个字段中的,其余的键都是为了找寻这条message,但是如果我们想要改变这个message的信息,怎么办呢?这时候就需要用到match这个参数,他可以对匹配成功的正则日志,进行相应的处理,例如

日志文件信息: 55.3.244.1 GET /index.html 15824 0.043

grok {

match => [ "message", "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" ]

}

对其日志进行匹配成功后呢,message信息就变成了这样的,也就是说,对其日志进行了match后呢,只要是满足正则的,那么都会被存在于message中

client: 55.3.244.1

method: GET

request: /index.html

bytes: 15824

duration: 0.043

在上面的正则中我们看到了有IP,WORD这样的字段,这些是logstash内置的正则表达式,在 https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns 这里都是可以找到的,

我给出一个grop的正则在线测试的网址 http://grokdebug.herokuapp.com/