ELK收集Nginx自定义日志格式输出

1.ELK收集日志的有两种常用的方式:

1.1:不修改源日志格式,简单的说就是在logstash中转通过 grok方式进行过滤处理,将原始无规则的日志转换为规则日志(Logstash自定义日志格式)

1.2:修改 源日志格式,将需要的日志格式进行规则输出,logstash只负责日志的收集和传输,不对日志做任何过滤处理(filebeat生产者自定义日志格式)

优缺点:

首先我们来看下不修改源日志格式,这样Logstash会通过grok来处理分析,对线上业务无任何影响;但是在高压环境下,Logstash中的grok会成为性能瓶颈,最终会阻塞正常的日志输出,所以,在Logsatsh中,尽量不要使用grok过滤功能

第二种是修改 源日志格式,也就是在收集生产日志的过程中,自定义日志格式,虽然有一定的工作量,但是优势很明显,因为是实现定义好了日志输出格式,logstash那就只负责收集和传输了,这样大大减轻了logstash负担,可以更高效的收集和传输日志;是企业首选方案

下图是引用了网上与之本次实验效果图

最前方是一台Apache服务器用于生产日志,filebeat收集web服务产生的日志,将收集到的日志推送到Kafka集群中,完成日志的收集工作

接着Logstash去kafka集群中拉取日志并进行过滤分析之后转发到Elasticsearch集群中进行索引和存储,最后由Kibana完成日志的可视化查询展示

ELK收集Nginx自定义日志格式输出

ELK收集Nginx自定义日志格式输出

Nginx支持自定义输出日志格式;先来了解一下关于多层代理获取用户真实IP的几个概念

#remote_addr:    客户端地址,如果没有使用代理,默认就是客户端真实IP,如果使用了代理,这个就是上层代理IP

#X-Forwarded-For: 简称XFF,一个http扩展头,格式为X-Forwarded-For:client,proxy1,proxy2,如果请求经过三个代理层(proxy1,proxy2.proxy3),用户IP为IP0,那么标准的XFF,服务器最终只会收到X-Forwarded-For:IP0,IP1,IP2

Ps:经过了三层代理。IP3这个地址X-Fowarded-For并未获取到,而remote_addr刚好获取的就是IP3的地址

下面红色标记的字段是自定义nginx日志输出格式:其中调用了nginx中map指令,通过map定义了一个变量$clientRealIp,这个就是获取客户端真实IP的变量,map指令由ngx_http_map_module模块提供并默认加载

map首选定义了一个$clientRealIp变量,如果$HTTP_x_forwarded_for为空的话“(""为空),则remote_addr的变量值则赋予给clientRealIP,如果不为空,则通过正则表达式取出第一个IP赋值给firstAddr,最后由firstAddr赋值给clientRealIP;

接着,通过log_format指令自定义nginx日志格式定义了13个字段,access_log指令指定了日志文件存放路径

user nginx;
worker_processes auto;
error_log /var/log/nginx/error.log;
pid /run/nginx.pid;
include /usr/share/nginx/modules/*.conf;
events {
worker_connections 1024;
}
http {
map $http_x_forwarded_for $clientRealIp {
"" $remote_addr;
~^(?P<firstAddr>[0-9\.]+),?.*$ $firstAddr;
}
log_format nginx_log_json '{"accessip_list":"$proxy_add_x_forwarded_for","client_ip":"$clientRealIp","http_host":"$host","@timestamp":"$time_iso8601","method":"$request_method","url":"$request_uri","status":"$status","http_referer":"$http_referer","body_bytes_sent":"$body_bytes_sent","request_time":"$request_time","http_user_agent":"$http_user_agent","total_bytes_sent":"$bytes_sent","server_ip":"$server_addr"}';
access_log /var/log/nginx/access.log nginx_log_json;

sendfile on;
tcp_nopush on;
tcp_nodelay on;
keepalive_timeout 65;
types_hash_max_size 2048;
include /etc/nginx/mime.types;
default_type application/octet-stream;
include /etc/nginx/conf.d/*.conf;
server {
listen 80 default_server;
listen [::]:80 default_server;
server_name _;
root /usr/share/nginx/html;
include /etc/nginx/default.d/*.conf;
location / {
}
error_page 404 /404.html;
location = /40x.html {
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
}
}
}

输出的日志为

#tail /var/log/nginx/access.log

{"accessip_list":"124.207.82.22","client_ip":"124.207.82.22","http_host":"203.195.163.239","@timestamp":"2018-09-03T19:47:42+08:00","method":"GET","url":"/","status":"304","http_r
eferer":"-","body_bytes_sent":"0","request_time":"0.000","http_user_agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 S
afari/537.36","total_bytes_sent":"180","server_ip":"10.104.137.230"}

ELK收集Nginx自定义日志格式输出

分析详解:

accessip_list:输出时代理叠加而成的IP地址列表

client_ip:客户端访问真实IP

http_host:客户端请求的地址,也就是浏览器输入的IP或者域名

@timestamp:时间戳,表示请求的时间

method:表示HTTP请求方法,通常为“GET”或者“POST”

url:表示客户端请求参数的原始URL

status:表示请求状态

http_reserer:表示来源页面,即从哪个页面请求过来的,专业名称叫referer

body_bytes_sent:表示发送客户端的字节数,不包括响应头的大小

request_time:表示请求处理时间,单位为秒,精度毫秒

http_user_agent:表示用户浏览器信息,例如浏览器版本,类型等

total_bytes_sent:表示传输给客户端字节数

server_ip:表示本地服务器的IP地址信息

【filebeat配置】

ELK收集Nginx自定义日志格式输出

ELK收集Nginx自定义日志格式输出

【Logstash事件配置 】

input {
kafka {
bootstrap_servers => "192.168.37.134:9092,192.168.37.135:9092,192.168.37.136:9092"
topics => "nginx_log" #指定输入源中需要从哪个topic中读取日志数据
group_id => "logstash"
codec => json {
charset => "UTF-8" #将输入的json日志输出进行utf8格式编码
}
add_field => { "[@metadata][myid]" => "nginxaccess-log" } #增加一个字段,用于标识和判断,与下方output输出相对应
}
}
filter {
if [@metadata][myid] == "nginxaccess-log" {
mutate {
gsub => ["message", "\\x", "\\x"] #message字段,也就是日志的输出内容,插件作用就是讲message字段内容中UTF8单字节编码做替换处理,目的就是应对URL出现中文情况,防止乱码的出现
}
if ( 'method":"HEAD' in [message] ) { #如果message字段中有HEAD请求,就会删除
drop {}
}
json { #启动json解码插件,因为输入的数据是复合的数据结构,只是一部分记录的是json格式
source => "message" #指定json格式字段,也就是message字段
remove_field => "prospector"
remove_field => "beat"
remove_field => "source"
remove_field => "input"
remove_field => "offset"
remove_field => "fields"
remove_field => "host"
remove_field => "@version"
remove_field => "message" #因为json格式中已经定义好了每个字段,所以输出也就是按照每个字段输出,不需要message字段了,这里直接移除
}
}
}
output {
if [@metadata][myid]=="nginxaccess-log" { #当有多个输入源的时候,可根据不同的标识,指定不同的输出地址
elasticsearch {
hosts =>["192.168.37.134:9200","192.168.37.135:9200","192.168.37.136:9200"]
index => "webnginx_log-%{+YYYY-MM-dd}" #自定义指定索引名称
}
}

[root@localhost etc]# nohup /usr/local/logstash/bin/logstash -f nginxlog.conf &    #后台运行Logstash事件

http://192.168.37.134:9100/   #访问Elasticsearch集群ip,验证es是否将nginx日志事件存储为索引;

ok!  Elasticsearch成功存储索引名称

ELK收集Nginx自定义日志格式输出

http://192.168.37.136:5601   接下来在kibana创建索引,如下图所示

ELK收集Nginx自定义日志格式输出

ELK收集Nginx自定义日志格式输出

ELK收集Nginx自定义日志格式输出

ELK收集Nginx自定义日志格式输出

ELK收集Nginx自定义日志格式输出

最后我们可以看到。从最开始的自定义日志格式,到索引存储,最终到Kibana展示,过程一点毛病都没有,展示的日志都 是我们自定义的样子,依旧是13个字段,很有条理性,我们还可以在左侧进行删选操作,这样更加的高效~ok,打完收工~

ELK收集Nginx自定义日志格式输出

ELK收集Nginx自定义日志格式输出

上一篇:nginx日志格式字段


下一篇:ELF Executable Reconstruction From A Core Image