ELK处理Nginx日志

本文主要内容是分析 nginx 日志然后进行可视化图形展示。

一、日志收集配置

1、定义nginx日志格式

修改nginx配置文件 nginx.conf:

1
2
3
log_format  main  '$remote_addr | $time_local | $request | $uri | '
'$status | $body_bytes_sent | $bytes_sent | $gzip_ratio | $http_referer | '
'"$http_user_agent" | $http_x_forwarded_for | $upstream_addr | $upstream_response_time | $upstream_status | $request_time';

定义以上格式后,nginx输出的log就是按照上面格式来进行的:

1
27.19.249.25 | 06/Sep/2018:15:37:50 +0800 | POST /yiqun/order/getEOrderDetails/18926?cursor=0&eorderState=1&inviteType=1&keyWord=&offset=25&userId=18926 HTTP/1.1 | /yiqun/order/getEOrderDetails/18926 | 200 | 33 | 427 | - | https://www.5ecat.com/bg/ | "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36" | - | 172.17.140.115:8080 | 0.012 | 200 | 0.012

当然上面的格式可以按照自己习惯自定义配置。

2、grok正则解析log文件

我们的目的是将nginx的日志文件存入 elastic ,在 Logstash 中我们使用 grok 数据结构化转换工具来进行匹配解析,解析出nginx日志中的各个有用字段。
这里可以借助 grop官网debuggerpatterns来快速帮助我们写正则表达式。
这就是我自定义的 grop 正则表达式:

1
"%{IPORHOST:clientip} \| %{HTTPDATE:timestamp} \| (?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:http_version})?|-) \| %{URIPATH:uripath} \| %{NUMBER:response} \| (?:%{NUMBER:body_bytes_sent}|-) \| (?:%{NUMBER:bytes_sent}|-) \| (?:%{NOTSPACE:gzip_ratio}|-) \| (?:%{NOTSPACE:http_referer}|-) \| (?:%{QS:agent}|-) \| (?:%{QS:http_x_forwarded_for}|-) \| (%{URIHOST:upstream_addr}|-) \| (%{BASE16FLOAT:upstream_response_time}) \| %{NUMBER:upstream_status} \| (%{BASE16FLOAT:request_time})"

grop官网的debugger解析结果如下解析结果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
{
"clientip": [
[
"27.19.249.25"
]
],
"HOSTNAME": [
[
"27.19.249.25",
"172.17.140.115"
]
],
"IP": [
[
null,
null
]
],
"IPV6": [
[
null,
null
]
],
"IPV4": [
[
null,
null
]
],
"timestamp": [
[
"06/Sep/2018:15:37:50 +0800"
]
],
"MONTHDAY": [
[
"06"
]
],
"MONTH": [
[
"Sep"
]
],
"YEAR": [
[
"2018"
]
],
"TIME": [
[
"15:37:50"
]
],
"HOUR": [
[
"15"
]
],
"MINUTE": [
[
"37"
]
],
"SECOND": [
[
"50"
]
],
"INT": [
[
"+0800"
]
],
"verb": [
[
"POST"
]
],
"request": [
[
"/yiqun/order/getEOrderDetails/18926?cursor=0&eorderState=1&inviteType=1&keyWord=&offset=25&userId=18926"
]
],
"http_version": [
[
"1.1"
]
],
"BASE10NUM": [
[
"1.1",
"200",
"33",
"427",
"200"
]
],
"uripath": [
[
"/yiqun/order/getEOrderDetails/18926"
]
],
"response": [
[
"200"
]
],
"body_bytes_sent": [
[
"33"
]
],
"bytes_sent": [
[
"427"
]
],
"gzip_ratio": [
[
"-"
]
],
"http_referer": [
[
"https://www.5ecat.com/bg/"
]
],
"agent": [
[
""Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36""
]
],
"QUOTEDSTRING": [
[
""Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36"",
null
]
],
"http_x_forwarded_for": [
[
null
]
],
"upstream_addr": [
[
"172.17.140.115:8080"
]
],
"IPORHOST": [
[
"172.17.140.115"
]
],
"port": [
[
"8080"
]
],
"upstream_response_time": [
[
"0.012"
]
],
"upstream_status": [
[
"200"
]
],
"request_time": [
[
"0.012"
]
]
}

3、地理坐标分析-geoip

logstash 中配置的 GeoIP 的数据库解析ip了,这里是用了开源的ip数据源,用来分析客户端的ip归属地。官网在这里:MAXMIND,下载 GeoLite City 库。
logstashfilter 中配置:

1
2
3
4
5
geoip {
source => "clientip"
##这里指定好解压后GeoIP数据库文件的位置
database => "/usr/local/src/logstash-5.5.3/GeoLite2-City.mmdb"
}

4、 logstash完整配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
input {
file {
path => [ "/root/yimiao_access.log" ]
# 从文件的开始读取事件。另外还有end参数
start_position => "beginning"
# 忽略早于24小时(默认值86400)的日志,设为0,即关闭该功能,以防止文件中的事件由于是早期的被logstash所忽略。
ignore_older => 0
}
}

filter {
# 数据结构化转换工具
grok {
# match:匹配条件格式
match => [
"message" => "%{IPORHOST:clientip} \| %{HTTPDATE:timestamp} \| (?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:http_version})?|-) \| %{URIPATH:uripath} \| %{NUMBER:response} \| (?:%{NUMBER:body_bytes_sent}|-) \| (?:%{NUMBER:bytes_sent}|-) \| (?:%{NOTSPACE:gzip_ratio}|-) \| (?:%{NOTSPACE:http_referer}|-) \| (?:%{QS:agent}|-) \| (?:%{QS:http_x_forwarded_for}|-) \| (%{URIHOST:upstream_addr}|-) \| (%{BASE16FLOAT:upstream_response_time}) \| %{NUMBER:upstream_status} \| (%{BASE16FLOAT:request_time})"
]

}
# geoip该过滤器从geoip中匹配ip字段,显示该ip的地理位置
geoip {
# source:ip来源字段,这里我们选择的是日志文件中的第一个字段
source => "clientip"
# target:指定插入的logstash字断目标存储为geoip
target => "geoip"
# database:geoip数据库的存放路径
database => "/usr/local/src/logstash-5.5.3/GeoLite2-City.mmdb"
# add_field: 增加的字段,坐标经度
add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
# add_field: 增加的字段,坐标纬度
add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}

# mutate: 数据的修改、删除、类型转换
mutate {
# convert: 将坐标转为float类型
convert => [ "[geoip][coordinates]", "float" ]
# convert: http的响应代码字段转换成 int
convert => [ "response","integer" ]
# convert: http的传输字节转换成int
convert => [ "body_bytes_sent","integer" ]
convert => [ "bytes_sent","integer" ]
# replace: 替换一个字段
replace => { "type" => "nginx_access" }
# remove_field: 移除message 的内容,因为数据已经过滤了一份,这里不必在用到该字段了。不然会相当于存两份
remove_field => "message"
}

# date: 时间处理,该插件很实用,主要是用你日志文件中事件的事件来对timestamp进行转换,导入老的数据必备!
date {
# match:匹配到timestamp字段后,修改格式为dd/MMM/yyyy:HH:mm:ss Z
match => [ "timestamp","dd/MMM/yyyy:HH:mm:ss Z"]

}

# mutate:数据修改
mutate {
# remove_field: 移除timestamp字段。
remove_field => "timestamp"

}
}

output {
# elasticsearch:输出到es中
elasticsearch {
# host: es的主机ip+端口或者es 的FQDN+端口
hosts => ["192.168.1.27:9200"]
# index: 为日志创建索引logstash-nginx-access,这里也就是kibana那里添加索引时的名称
index => "logstash-nginx-access"
}
stdout {codec => rubydebug}
}

logstash 配置完成后通过命令导入log文件到Elastic中。

二、Kibana分析Nginx日志并生成图表

上一步数据已经导入到了Elastic ,我们需要明白每个字段的含义,然后在 Kibana 中根据需求自定义生成报表。
这里不做过多说明,根据需求生成。