首页
登录 | 注册

初探ELK-logstash使用小结

初探ELK-logstash使用小结

2016/9/22


注:由于在实践过程中,会不断学习到新的东西,文章会不定时保持更新,如果遇到问题,请重新刷新一下文章查看是否有更新。


【写在前言】

说起处理日志的手段,大家或多或少都听说过ELK(elasticsearch+logstash+kibana),怎么入门呢?咱们从一个小小的目标开始。

目标:收集nginx日志,集中展示。

不少人对 ELK 的第一印象,容易觉得它这个工具组合似乎挺难上手的,错!只需动手试试就知道啦!

目标分解:

1)熟悉 logstash 的安装和基本操作

2)熟悉 elasticsearch 的安装和基本操作,然后结合 logstash 使用

3)熟悉 kibana 的安装和基本操作,然后结合 elasticsearch 使用


本文主要是带你进入 logstash 的世界,其余内容请参考相关文章(elasticsearch使用小结,kibana使用小结,filebeat使用小结)。



一、安装
1、jdk 和 环境变量
支持jdk-1.7以上,推荐jdk-1.8
在环境变量配置:JAVA_HOME

2、安装
有2种方式下载,推荐缓存rpm包到本地yum源
1)直接使用rpm
wget https://download.elastic.co/logstash/logstash/packages/centos/logstash-2.4.0.noarch.rpm

2)使用yum源
[[email protected] ~]# rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch
[[email protected] ~]# vim /etc/yum.repos.d/logstash.repo
[logstash-2.4]
name=Logstash repository for 2.4.x packages
baseurl=https://packages.elastic.co/logstash/2.4/centos
gpgcheck=1
gpgkey=https://packages.elastic.co/GPG-KEY-elasticsearch
enabled=1

[[email protected] ~]# yum install logstash
[[email protected] ~]# whereis logstash
logstash: /etc/logstash /opt/logstash/bin/logstash /opt/logstash/bin/logstash.bat


二、使用
1、命令行测试
1)输出到 stdout
[[email protected] ~]# /opt/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'
hi, let us go(输入)
Settings: Default pipeline workers: 4
Pipeline main started
2016-09-12T02:42:59.110Z 0.0.0.0 hi, let us go(输出)
why not TRY IT OUT(输入)
2016-09-12T02:43:11.904Z 0.0.0.0 why not TRY IT OUT(输出)

(CTRL-D 退出)
Pipeline main has been shutdown
stopping pipeline {:id=>"main"}

2)改变输出到 stdout 的格式
[[email protected] ~]# /opt/logstash/bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'
aaaa
Settings: Default pipeline workers: 4
Pipeline main started
{
       "message" => "aaaa",
      "@version" => "1",
    "@timestamp" => "2016-09-18T02:11:07.109Z",
          "host" => "0.0.0.0"
}
bbbb
{
       "message" => "bbbb",
      "@version" => "1",
    "@timestamp" => "2016-09-18T02:11:11.864Z",
          "host" => "0.0.0.0"
}
Pipeline main has been shutdown
stopping pipeline {:id=>"main"}

3)上述输出发现一个小问题,得到的 host 为 0.0.0.0,而不是期望中的当前 hostname
检查一下 hostname
[[email protected] ~]# hostname
vm49.localdomain
[[email protected] ~]# echo '10.50.200.49 vm49.localdomain' >>/etc/hosts
再次测试:
[[email protected] ~]# /opt/logstash/bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'
aaaaa
Settings: Default pipeline workers: 4
Pipeline main started
{
       "message" => "aaaaa",
      "@version" => "1",
    "@timestamp" => "2016-09-18T02:22:12.180Z",
          "host" => "vm49.localdomain"
}
Pipeline main has been shutdown
stopping pipeline {:id=>"main"}

符合预期。

注1:还有一种场景(hostname使用本地名称,而不是采取域名的命名方式)需要注意
/etc/hosts 中包括下述内容
------------------------
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
127.0.0.1 local-vm50
10.50.200.50 local-vm50
------------------------
期望的结果是:
"host" => "local-vm50"
实际得到的结果是:
"host" => "localhost"

此时,需要删除:
127.0.0.1 local-vm50
然后才能得到期望的结果。

注2:如果是在 logstash 服务启动后才去调整了 hostname 相关设置,则需要重启 logstash 服务,否则依然是旧的结果。


2、使用配置文件
目的:从日志文件中读取数据,输出到另一个文件中来查看。
前提:已经配置了一个nginx服务,生成了对应的日志文件。

首先,因为这里要读取nginx日志,请将 logstash 加入 nginx 的用户组。
示例:
[[email protected] ~]# ls -l /var/log/nginx/access_*.log
-rw-r----- 1 nginx adm 608083 Sep 13 18:18 /var/log/nginx/access_www.test.com_80.log
-rw-r----- 1 nginx adm 597070 Sep 13 18:18 /var/log/nginx/access_www.work.com_80.log

nginx属于 adm 组,使用 logstash 读取日志,可能产生权限异常:
failed to open /var/log/nginx/access.ecshop.test.d3dstore.com.log: Permission denied
因此,要加入对应的组:
[[email protected] ~]# usermod -G adm logstash


接着,我们尝试这样配置 logstash 来收集日志:
[[email protected] ~]# cat /etc/logstash/conf.d/nginx.conf 
input {
    file {
        path => "/var/log/nginx/access_*.log"
        start_position => beginning 
        ignore_older => 0 
    }
}

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}

output {
    file {
        path => "/tmp/test.log"
    }
}

上面使用到以下插件:
file:日志数据的输入和输出
grok:来匹配标准的apache日志格式

【细节延伸】
显然,在3个环节中,都有改进和调整的地方。
input:使用 filebeat
filter:使用其他插件和规则
output:使用ES,redis等
具体请参考:
https://www.elastic.co/guide/en/logstash/current/pipeline.html

3、测试配置文件:
[[email protected] ~]# service logstash configtest
Configuration OK

4、启动服务:
[[email protected] ~]# service logstash start
[[email protected] ~]# chkconfig logstash on

5、测试请求nginx服务,然后观察输出的内容:
[[email protected] ~]# cat /tmp/test.log

符合预期。

6、比较
去掉 filter 这一节,我们来对比一下 /tmp/test.log 收集到的内容的差异

【使用了 filter 的结果a】
{"message":"10.50.200.219 - - [12/Sep/2016:13:00:03 +0800] \"GET / HTTP/1.1\" 200 13 \"-\" \"curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.14.0.0 zlib/1.2.3 libidn/1.18 libssh2/1.4.2\" \"-\" 0.000 \"-\" \"-\"","@version":"1","@timestamp":"2016-09-12T05:00:04.140Z","path":"/var/log/nginx/access_www.test.com_80.log","host":"vm49.localdomain","clientip":"10.50.200.219","ident":"-","auth":"-","timestamp":"12/Sep/2016:13:00:03 +0800","verb":"GET","request":"/","httpversion":"1.1","response":"200","bytes":"13","referrer":"\"-\"","agent":"\"curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.14.0.0 zlib/1.2.3 libidn/1.18 libssh2/1.4.2\""}

【未使用 filter 的结果b】
{"message":"10.50.200.219 - - [12/Sep/2016:13:07:49 +0800] \"GET / HTTP/1.1\" 200 13 \"-\" \"curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.14.0.0 zlib/1.2.3 libidn/1.18 libssh2/1.4.2\" \"-\" 0.000 \"-\" \"-\"","@version":"1","@timestamp":"2016-09-12T05:07:49.917Z","path":"/var/log/nginx/access_www.test.com_80.log","host":"vm49.localdomain"}


a的内容中,多出来的地方,正是使用了 grok 分析和结构化数据
---------------------------------------------------
Information	              Field Name
-----------               ----------
IP Address                clientip
User ID                   ident
User Authentication	      auth
timestamp	      	      timestamp
HTTP Verb	      	      verb
Request body	      	  request
HTTP Version	          httpversion
HTTP Status Code	      response
Bytes served	          bytes
Referrer URL	          referrer
User agent	              agent
---------------------------------------------------

注:也可以将 filter 调整到服务端,让客户端仅仅是收集日志即可,这个取舍需要自行对比。

7、改进
Logstash 默认自带了 apache 标准日志的 grok 正则:
如何使用自定义的日志格式呢?
例如,默认的 nginx 日志是:
    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

改成自定义的日志格式:
                      
    log_format  online '$remote_addr [$time_local] "$request" '
                   '"$http_content_type" "$request_body" "$http_referer" '
                   '$status $request_time $body_bytes_sent';
                   
对应的数据:
【GET】# curl -H "Content-Type: text/html; charset=UTF-8" --referer 'www.abc.com/this_is_a_referer' http://www.test.com/a/b/c.html?key1=value1
【结果】10.50.200.219 [12/Sep/2016:15:11:04 +0800] "GET /a/b/c.html?key1=value1 HTTP/1.1" "text/html; charset=UTF-8" "-" "www.abc.com/this_is_a_referer" 404 0.000 168

【POST】# curl -H "Content-Type: application/xml" -d "{"name": "Mark Lee" }" "http://www.test.com/start"
【结果】10.50.200.218 [12/Sep/2016:15:02:07 +0800] "POST /start HTTP/1.1" "application/xml" "-" "-" 404 0.000 168

尝试一下:
[[email protected] ~]# mkdir -p /etc/logstash/patterns.d
[[email protected] ~]# vim /etc/logstash/patterns.d/extra_patterns
NGINXACCESS %{IPORHOST:clientip} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" (?:%{QS:content_type}|-) (?:%{QS:request_body}|-) (?:"(?:%{URI:referrer}|-)"|%{QS:referrer}) %{NUMBER:response} %{BASE16FLOAT:request_time} (?:%{NUMBER:bytes}|-)


调整配置为:
[[email protected] ~]# cat /etc/logstash/conf.d/nginx.conf 
input {
    file {
        path => "/var/log/nginx/access_*.log"
        start_position => beginning
        ignore_older => 0
    }
}

filter {
    grok {
        patterns_dir => ["/etc/logstash/patterns.d"]
        match => {
            "message" => "%{NGINXACCESS}"
        }
    }
}

output {
    file {
        path => "/tmp/test.log"
    }
}


[[email protected] ~]# service logstash restart
结果:
{"message":"10.50.200.218 [12/Sep/2016:15:28:23 +0800] \"POST /start HTTP/1.1\" \"application/xml\" \"-\" \"-\" 404 0.000 168","@version":"1","@timestamp":"2016-09-12T07:28:24.007Z","path":"/var/log/nginx/access_www.test.com_80.log","host":"vm49.localdomain","clientip":"10.50.200.218","timestamp":"12/Sep/2016:15:28:23 +0800","verb":"POST","request":"/start","httpversion":"1.1","content_type":"\"application/xml\"","request_body":"\"-\"","response":"404","request_time":"0.000","bytes":"168"}

{"message":"10.50.200.219 [12/Sep/2016:15:28:24 +0800] \"GET /a/b/c.html?key1=value1 HTTP/1.1\" \"text/html; charset=UTF-8\" \"-\" \"www.abc.com/this_is_a_referer\" 404 0.000 168","@version":"1","@timestamp":"2016-09-12T07:28:25.019Z","path":"/var/log/nginx/access_www.test.com_80.log","host":"vm49.localdomain","clientip":"10.50.200.219","timestamp":"12/Sep/2016:15:28:24 +0800","verb":"GET","request":"/a/b/c.html?key1=value1","httpversion":"1.1","content_type":"\"text/html; charset=UTF-8\"","request_body":"\"-\"","referrer":"\"www.abc.com/this_is_a_referer\"","response":"404","request_time":"0.000","bytes":"168"}

符合预期。


三、输出到 redis+elasticsearch+kibana
1、测试环境(已经部署了服务)
【客户端】10.50.200.49: logstash, nginx(www.test.com, www.work.com)
【服务端】10.50.200.220: logstash, redis, elasticsearch, kibana
【测试端】10.50.200.218, 10.50.200.219: curl 请求 nginx
1)测试脚本
--------------------------------------------------------
[[email protected] bin]# cat test_www_by_curl.sh
#!/bin/bash
#
#2016/9/20
# crontab:
# */2 * * * * /usr/local/bin/test_www_by_curl.sh >>/tmp/test_www_by_curl.log 2>&1 &
echo "[S] at: `date`"
tag="$(echo `hostname` |awk -F'-' '{print $NF}')"
do_curl(){
    www=$1
    rnd=$2
    #PUT
    curl -H "Content-Type: application/xml" -d "{"name": "Jack A"}" "http://${www}/from_${tag}"
    curl -H "Content-Type: application/json" -d "{"name": "Jack B"}" "http://${www}/from_${tag}"
    curl -H "Content-Type: application/png" -d "{"img_path": "p_w_picpaths/$(date +%F)/${rnd}.png"}" "http://${www}/from_${tag}"
    #GET
    curl -H "Content-Type: text/html; charset=UTF-8" --referer "www.${tag}.com/this_is_referer" "http://${www}/"
    curl -H "Content-Type: text/html; charset=UTF-8" --referer "www.${tag}.com/this_is_referer" "http://${www}/a/b/c.html?key=${tag}"
    curl -H "Content-Type: text/html; charset=GBK" --referer "www.${tag}.com/this_is_referer" "http://${www}/action.do?event=login&user=${tag}&sid=${rnd}"
    #
}
for i in `seq 1 2000`; do
    do_curl www.work.com "`od /dev/urandom -w12 -tx4 -An |sed -e 's/ //g' |head -n 1`"
    do_curl www.test.com "`od /dev/urandom -w12 -tx4 -An |sed -e 's/ //g' |head -n 1`"
done >/dev/null 2>&1
echo "[D] at: `date`"
--------------------------------------------------------


2)hosts 写入域名
10.50.200.49 www.test.com
10.50.200.49 www.work.com


2、场景1:只有1个域名/模糊匹配N个域名
目的:将匹配的 access 日志收集起来集中展示。
【客户端】
输入:file
输出:redis

[[email protected] ~]# cat /etc/logstash/conf.d/nginx.conf 
input {
    file {
        type => "NginxAccess"
        path => "/var/log/nginx/access_*.log"
        start_position => beginning
        ignore_older => 0
    }
}

filter {
    if[type] == "NginxAccess" {
        grok {
            patterns_dir => ["/etc/logstash/patterns.d"]
            match => {
                "message" => "%{NGINXACCESS}"
            }
        }
        date {
            match => [ "timestamp", "dd/MMM/YYYY:HH:mm:ss Z" ]
            remove_field => [ "timestamp" ]
        }
    }
}

output {
    if[type] == "NginxAccess" {
        redis {
            host => "10.50.200.220"
            data_type => "list"
            key => "logstash:nginxaccess"
        }
    }
}
[[email protected] ~]# service logstash restart


【服务端】
输入:redis
输出:elasticsearch

[[email protected] ~]# vim /etc/logstash/conf.d/redis.conf
input {
    redis {
        host => '127.0.0.1'
        data_type => 'list'
        port => "6379"
        key => 'logstash:nginxaccess'
        type => 'redis-input'
    }
}

output {
    if[type] == "NginxAccess" {
        elasticsearch {
            hosts => "127.0.0.1:9200"
            index => "logstash-nginxaccess-%{+YYYY.MM.dd}"
        }
    }
}
[[email protected] ~]# service logstash restart


可以通过命令行去观察 redis 的状态:
[[email protected] ~]# redis-cli monitor


指定kibana的索引名称,例如:
logstash-nginxaccess-*

结果:符合预期。



3、场景2:N个域名分开收集
目的:将 www.test.com 和 www.work.com 的 access 日志收集起来使用独立的索引名来分开展示
【客户端】
输入:file
输出:redis

[[email protected] ~]# cat /etc/logstash/conf.d/nginx.conf 
input {
    file {
        type => "NginxAccess-www.test.com"
        path => "/var/log/nginx/access_www.test.com*.log"
        start_position => beginning
        ignore_older => 0
    }
    file {
        type => "NginxAccess-www.work.com"
        path => "/var/log/nginx/access_www.work.com*.log"
        start_position => beginning
        ignore_older => 0
    }
}

filter {
    if[type] =~ "NginxAccess-" {
        grok {
            patterns_dir => ["/etc/logstash/patterns.d"]
            match => {
                "message" => "%{NGINXACCESS}"
            }
        }
        date {
            match => [ "timestamp", "dd/MMM/YYYY:HH:mm:ss Z" ]
            remove_field => [ "timestamp" ]
        }
    }
}

output {
    if[type] =~ "NginxAccess-" {
        redis {
            host => "10.50.200.220"
            data_type => "list"
            key => "logstash:nginxaccess"
        }
    }
}
[[email protected] ~]# service logstash restart


【服务端】
输入:redis
输出:elasticsearch

[[email protected] ~]# cat /etc/logstash/conf.d/redis.conf    
input {
    redis {
        host => '127.0.0.1'
        data_type => 'list'
        port => "6379"
        key => 'logstash:nginxaccess'
        type => 'redis-input'
    }
}

output {
    if[type] == "NginxAccess-www.test.com" {
        elasticsearch {
            hosts => "127.0.0.1:9200"
            index => "logstash-nginxaccess-www.test.com-%{+YYYY.MM.dd}"
        }
    }
    if[type] == "NginxAccess-www.work.com" {
        elasticsearch {
            hosts => "127.0.0.1:9200"
            index => "logstash-nginxaccess-www.work.com-%{+YYYY.MM.dd}"
        }
    }
}
[[email protected] ~]# service logstash restart

当然了,要调整kibana的索引名称,指定更具体的名称,例如:
logstash-nginxaccess-www.test.com-*
logstash-nginxaccess-www.work.com-*

结果:符合预期。


四、小结FAQ
1、数据流向
------------------------------------------------------------------
log_files  ->  logstash  ->  redis  ->  elasticsearch  ->  kibana
------------------------------------------------------------------


2、@timestamp 和 message 中记录的时间字段(timestamp)不一致
message 中记录的是原始时间,而 @timestamp 中记录的是事件采集到的时候记录的时间,两者可能差距有1s及以上。
可以通过 date 插件来调整记录到的时间,以日志中记录的原始时间为准,可以用于读取过去某段时间的日志文件,这样一来,收集日志时记录的时间不会错乱。
例如:
filter {
    date {
        match => [ "timestamp", "dd/MMM/YYYY:HH:mm:ss Z" ]
        remove_field => [ "timestamp" ]
    }
}

将匹配 field "timestamp" 格式为 "dd/MMM/YYYY:HH:mm:ss Z"
实际匹配到:
18/Sep/2016:16:43:15 +0800

默认更新 @timestamp 为:
September 18th 2016, 16:43:15.000

特别说明:
1)timestamp 是我们在原始日志中收集的时间字段,不一定是这个名称(这意味是任意的自定义的字段名称),因为只是我们在前述操作中使用 grok 插件中在 patterns 中定义了 "%{NGINXACCESS}" 这个变量来解析原始字段,里边定义了这个 "timestamp" 字段。
a、grok插件的定义
        grok {
            patterns_dir => ["/etc/logstash/patterns.d"]
            match => {
                "message" => "%{NGINXACCESS}"
            }
        }
b、grok插件指明的 patterns 的定义
[[email protected] ~]# cat /etc/logstash/patterns.d/extra_patterns
NGINXACCESS %{IPORHOST:clientip} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" (?:%{QS:content_type}|-) (?:%{QS:request_body}|-) (?:"(?:%{URI:referrer}|-)"|%{QS:referrer}) %{NUMBER:response} %{BASE16FLOAT:request_time} (?:%{NUMBER:bytes}|-)

2)为何要使用 remove_field 功能?
a、重复数据
我们已经有了一个时间字段 "@timestamp" 因此可以丢掉这个重复的 "timestamp" 字段。
注:由此联想一下,生产应用中,我们是不是应该好好利用下 "remove_field" 功能去筛选一下字段呢?主动的去丢弃一些字段,只保留特别重要的指标发送给 output 即可。


b、导入到 elasticsearch 时,模版 mapping type 时异常
以收集 nginx error log 为例:
error log 的时间字段格式是这样的:2016/09/21 10:34:41
对应的 date 插件是这样写的:
    date {
        match => [ "timestamp", "YYYY/MM/dd HH:mm:ss" ]
        remove_field => [ "timestamp" ]
    }
如果不删除字段 "timestamp",当传递到 elasticsearch 时,模版将试图解析这字段的格式,这个时候会遇到解析异常,从而影响到 index 的建立。
在 elasticsearch 的日志中将产生如下错误:
MapperParsingException[failed to parse [timestamp]]; nested: IllegalArgumentException[Invalid format: "2016/09/21 10:34:41" is malformed at "/09/21 10:34:41"];





3、如何使用 grok 插件来匹配 nginx error log
1)示例日志
2016/09/21 10:34:41 [error] 11990#0: *6718852 no live upstreams while connecting to upstream, client: 10.50.200.219, server: c.test.com, request: "GET http://www.work.com/cgi-bin/common/attr?id=260714&r=0.1011513018987016 HTTP/1.1", upstream: "http://upstream_backend/cgi-bin/common/attr?id=260714&r=0.1011513018987016", host: "www.work.com"
2016/09/21 11:26:41 [error] 1614#0: *25584123 open() "/data/www/www.test.com/404.html" failed (2: No such file or directory), client: 10.50.200.218, server: www.test.com, request: "GET /action.do?event=login&user=vm218&sid=15120ae826ca2290514bcd5b HTTP/1.1", host: "www.test.com", referrer: "www.vm218.com/this_is_referer"


2)示例匹配规则
NGINXERROR_DATESTAMP %{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME}
NGINXERROR_PID (?:[0-9]+#[0-9]+\:)
NGINXERROR_TID (?:\*[0-9]+)
NGINXERROR %{NGINXERROR_DATESTAMP:timestamp} \[%{LOGLEVEL:loglevel}\] %{NGINXERROR_PID:pid} %{NGINXERROR_TID:tid} %{GREEDYDATA:errormsg}, client: %{IPORHOST:clientip}, server: %{HOSTNAME:server}, request: %{QS:request}(?:, upstream: %{QS:upstream})?, host: \"%{HOSTNAME:hostname}\"(?:, referrer: (?:"(?:%{URI:referrer}|-)"|%{QS:referrer}))?

注:特别说明,“%{NGINXERROR_PID:pid} %{NGINXERROR_TID:tid}” 这一段只是为了匹配“11990#0: *6718852”并打算丢弃掉,并非实际需要这2个字段。

3)使用工具测试
http://grokconstructor.appspot.com/do/match

4)结果
------------------------------------------------------------------
2016/09/21 10:34:41 [error] 11990#0: *6718852 no live upstreams while connecting to upstream, client: 10.50.200.219, server: c.test.com, request: "GET http://www.work.com/cgi-bin/common/attr?id=260714&r=0.1011513018987016 HTTP/1.1", upstream: "http://upstream_backend/cgi-bin/common/attr?id=260714&r=0.1011513018987016", host: "www.work.com"
【MATCHED】
request 	"GET·http://www.work.com/cgi-bin/common/attr?id=260714&r=0.1011513018987016·HTTP/1.1"
server 	c.test.com
hostname 	www.work.com
timestamp 	2016/09/21·10:34:41
clientip 	10.50.200.219
port 	
pid 	11990#0:
upstream 	"http://upstream_backend/cgi-bin/common/attr?id=260714&r=0.1011513018987016"
referrer 	
tid 	*6718852
errormsg 	no·live·upstreams·while·connecting·to·upstream
loglevel 	error
==================================================================
2016/09/21 11:26:41 [error] 1614#0: *25584123 open() "/data/www/www.test.com/404.html" failed (2: No such file or directory), client: 10.50.200.218, server: www.test.com, request: "GET /action.do?event=login&user=vm218&sid=15120ae826ca2290514bcd5b HTTP/1.1", host: "www.test.com", referrer: "www.vm218.com/this_is_referer"
【MATCHED】
request 	"GET·/action.do?event=login&user=vm218&sid=15120ae826ca2290514bcd5b·HTTP/1.1"
server 	www.test.com
hostname 	www.test.com
timestamp 	2016/09/21·11:26:41
clientip 	10.50.200.218
port 	
pid 	1614#0:
upstream 	
referrer 	"www.vm218.com/this_is_referer"
tid 	*25584123
errormsg 	open()·"/data/www/www.test.com/404.html"·failed·(2:·No·such·file·or·directory)
loglevel 	error 
------------------------------------------------------------------



5)应用
【服务端】
a、增加上述匹配规则到配置文件
[[email protected] ~]# cat /etc/logstash/patterns.d/extra_patterns
NGINXERROR_DATESTAMP %{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME}
NGINXERROR_PID (?:[0-9]+#[0-9]+\:)
NGINXERROR_TID (?:\*[0-9]+)
NGINXERROR %{NGINXERROR_DATESTAMP:timestamp} \[%{LOGLEVEL:loglevel}\] %{NGINXERROR_PID:pid} %{NGINXERROR_TID:tid} %{GREEDYDATA:errormsg}, client: %{IPORHOST:clientip}, server: %{HOSTNAME:server}, request: %{QS:request}(?:, upstream: %{QS:upstream})?, host: \"%{HOSTNAME:hostname}\"(?:, referrer: (?:"(?:%{URI:referrer}|-)"|%{QS:referrer}))?

b、调整对应的 logstash 配置
[[email protected] ~]# vim /etc/logstash/conf.d/filebeat.conf 
在 filter 这一段增加:
    if[type] =~ "NginxError-" {
        grok {
            patterns_dir => ["/etc/logstash/patterns.d"]
            match => {
                "message" => "%{NGINXERROR}"
            }
        }
        date {
            match => [ "timestamp", "YYYY/MM/dd HH:mm:ss" ]
            remove_field => [ "timestamp" ]
        }
    }

在 output 这一段增加:
    if[type] == "NginxError-www.test.com" {
        elasticsearch {
            hosts => ["10.50.200.218:9200", "10.50.200.219:9200", "10.50.200.220:9200"]
            manage_template => false
            index => "%{[@metadata][beat]}-nginxerror-www.test.com-%{+YYYY.MM.dd}"
            document_type => "%{[@metadata][type]}"
        }
    }
    if[type] == "NginxError-www.work.com" {
        elasticsearch {
            hosts => ["10.50.200.218:9200", "10.50.200.219:9200", "10.50.200.220:9200"]
            manage_template => false
            index => "%{[@metadata][beat]}-nginxerror-www.work.com-%{+YYYY.MM.dd}"
            document_type => "%{[@metadata][type]}"
        }
    }


c、重启 logstash 服务
[[email protected] ~]# service logstash restart

【客户端】
d、调整对应的 filebeat 配置
[[email protected] ~]# vim /etc/filebeat/filebeat.yml
(增加对应的 prospectors 来收集日志,打上 type 标记,其他配置略)
  prospectors:
    -
      paths:
        - /var/log/nginx/error_www.test.com*.log
      input_type: log
      document_type: NginxError-www.test.com
    -
      paths:
        - /var/log/nginx/error_www.work.com*.log
      input_type: log
      document_type: NginxError-www.work.com


e、重启 filebeat 服务
[[email protected] ~]# service filebeat restart



4、怎么调试问题
直接改变 output 到 stdout 即可查看输出是否符合需求。
output {
    if[type] =~ "NginxAccess-" {
        stdout { codec => rubydebug }
    }
    if[type] =~ "NginxError-" {
        stdout { codec => rubydebug }
    }
}





ZYXW、参考
1、官网
https://www.elastic.co/guide/en/logstash/current/introduction.html
https://www.elastic.co/guide/en/logstash/current/getting-started-with-logstash.html
https://www.elastic.co/guide/en/logstash/current/installing-logstash.html
https://www.elastic.co/guide/en/logstash/current/first-event.html
https://www.elastic.co/guide/en/logstash/current/advanced-pipeline.html
https://www.elastic.co/guide/en/logstash/current/pipeline.html
https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html
https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html
2、ELK中文
http://kibana.logstash.es/content/
http://kibana.logstash.es/content/logstash/plugins/filter/grok.html
http://kibana.logstash.es/content/beats/file.html
3、用ELK搭建简单的日志收集分析系统
http://blog.csdn.net/lzw_2006/article/details/51280058
4、logstash host field is 0.0.0.0
http://www.v1en.com/2016/03/31/logstash-host-field-is-0-0-0-0/
5、grok
https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns
http://grokconstructor.appspot.com/do/match

相关文章

  • ELK——Logstash
    Logstash 是开源的服务器端数据处理管道,能够同时从多个来源采集数据,转换数据,然后将数据发送到您最喜欢的 "存储库" 中. 官网介绍:https://www.elastic.co/cn/products/logs ...
  • 资源路径JAVAhttp://www.oracle.com/technetwork/java/javase/downloads/jre8-downloads-2133155.html ELKhttps://www.elastic.co/ 安 ...
  • ELK - 简谈 logstash flume
    现主流的日志分析系统有 logstash 和 flume,结合好多网上前辈的说法,汇总了一下,希望和大家分享与探讨,有不同的想法欢迎留言.Flume  Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集.聚合和传输的系统:支 ...
  • ELK中如何搭建客户端客户端如何将日志发送给服务端的logstash
    [背景]我们要统一收集日志,统一分析,统一在一个平台上搜索过滤日志!在上一篇中已经完成了ELK的搭建,那么该如何将每台客户端的日志集中到ELK平台中呢?[本系统介绍]ELK -- 192.168.100.10  (这个地方要有FQDN 创建 ...
  • elk安装整理 kibana安装配置 rpm 安装 vim vim /etc/kibana/kibana.yml修改访问端口地址server.port: 5601 server.host: "ip" 修改elastics ...
  • ELK日志平台搭建
    ELK日志平台搭建 整体架构 整体架构主要分为5个模块,分别提供不同的功能: Filebeat:轻量级数据收集引擎.基于原先 Logstash-fowarder 的源码改造出来.是 ELK Stack 在 Agent 的第一选择.<b ...

2019 jeepyurongfu.net webmaster#jeepyurongfu.net
12 q. 0.023 s.
京ICP备10005923号