码迷,mamicode.com
首页 > 其他好文 > 详细

ELK对nginx日志进行流量监控

时间:2019-11-12 09:24:53      阅读:137      评论:0      收藏:0      [点我收藏+]

标签:cli   实现   包含   日志采集   市级   off   级别   搜索框   href   

ELK对nginx日志进行流量监控

一、前言

     线上有一套ELK单机版,版本为5.2.1。现在想把nginx访问日志接入到elk里,进行各个域名使用流量带宽的统计分析。要把nginx日志传输到elk上,可以在存有nginx日志的服务器上使用logstash或者filebeat。但是因为logstash是jvm跑的,资源消耗比较大,启动一个logstash就需要消耗500M左右的内存(这就是为什么logstash启动特别慢的原因),而filebeat只需要10来M内存资源,所以最终决定使用filebeat。

     filebeat是轻量级日志采集器,能输送至 Elasticsearch 或 Logstash,在 Kibana 中实现可视化,架构图如下:

    技术图片

 

二、具体部署过程

1、设置nginx日志输出为json格式

ps:如果不先把nginx日志设为json格式,而是沿用默认的日志输出格式,则需要利用logstash grok插件对日志字段进行正则匹配。grok表达式的语法来源于Ruby正则,它的主要作用是将文本格式的字符串,转换为具体的结构化数据,配合正则表达式使用。但是grok是非常耗费CPU资源的,所以最终还是按简单方式来,把nginx日志设置为json输出(表示不懂Ruby正则 = =)

     在nginx主配置文件中加入:

 1     log_format json {"@timestamp":"$time_iso8601",
 2   "host":"$server_addr",
 3   "clientip":"$remote_addr",
 4   "size":$body_bytes_sent,
 5   "responsetime":$request_time,
 6   "upstreamtime":$upstream_response_time,
 7   "upstreamhost":"$upstream_addr",
 8   "http_host":"$host",
 9   "url":"$uri",
10   "xff":"$http_x_forwarded_for",
11   "referer":"$http_referer",
12   "agent":"$http_user_agent",
13   "status":"$status"};

 

2、日志采集客户端安装filebeat

(1)下载filebeat-5.2.2的rpm包并安装

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-5.2.2-x86_64.rpm 

sudo rpm -vi filebeat-5.2.2-x86_64.rpm

官方安装文档:https://www.elastic.co/guide/en/beats/filebeat/5.2/filebeat-installation.html

 

(2)修改配置文件

vim /etc/filebeat/filebeat.yml

(a)指定paths列表,配置要读取的nginx日志路径

注释默认的paths列表:“- /var/log/*.log”,默认的/var/log/*.log表示将会抓取/var/log子目录目录下所有.log文件
添加要读取的日志路径

技术图片

 

b) 配置filebeat输出到logstash,

注释输出到elasticsearch的配置:

技术图片

 打开logstash的输出,其中hosts设置为logstash服务器的ip,端口要确保logstash服务器是没有被占用的

技术图片

 

(3)启动filebeat

nohup sh /usr/share/filebeat/bin/filebeat -e -c /etc/filebeat/filebeat.yml >> /tmp/startfilebeat.log 2>/tmp/startServerError.log &

 

3、配置日志收集引擎logstash

注意:确保已经安装以下插件(查看:bin/logstash-plugin list)

(1)logstash-input-beats:Beats插件用于建立监听服务,接收Filebeat或者其他beat发送的Events

(2)logstash-filter-mutate:允许在字段上执行常规突变,包括重命名,删除,替换和修改事件中的字段

(3)logstash-filter-json:JSON解析过滤器,它接受一个包含JSON的现有字段,并将其扩展为Logstash事件中的实际数据结构。

(4)logstash-filter-geoip:转换ip为详细地址(经纬度,具体所属地区等)

 

(1)编辑logstash配置文件

vim /home/elk/elk5.2/logstash-5.2.1/config/logstash-nginx.conf

 1 input {
 2          beats {
 3              port => 4568
 4              client_inactivity_timeout => 600  #600秒之后关闭空闲的连接
 5         }
 6 }
 7 
 8 filter {
 9       mutate {
10           gsub => ["message", "\\x", "\\\x"]
11           gsub => ["message", ":-,", ":0,"]
12       }
13 
14       json {
15             source => "message"
16             remove_field => "message"
17             remove_field => "[beat][hostname]"
18             remove_field => "[beat][name]"
19             remove_field => "[beat][version]"
20             remove_field => "@version"
21             remove_field => "offset"
22             remove_field => "input_type"
23             remove_field => "tags"
24             remove_field => "type"
25             remove_field => "host"
26       }
27       mutate {
28           convert => ["status", "integer"]
29           convert => ["size","integer"]
30           convert => ["upstreamtime", "float"]
31           convert => ["responsetime", "float"]
32       }
33 
34        geoip {
35             source => "clientip"
36             database => "/home/elk/elk5.2/logstash-5.2.1/config/GeoLite2-City.mmdb"
37             fields => ["city_name", "country_code2", "country_name", "latitude", "longitude", "region_name"] 
38             add_field => [ "[geoip][location]", "%{[geoip][longitude]}" ]
39             add_field => [ "[geoip][location]", "%{[geoip][latitude]}" ]
40             target => "geoip"
41         }
42         mutate {
43            convert => [ "[geoip][location]", "float" ]
44          }
45         if "_geoip_lookup_failure" in [tags] { drop { } }
46 }
47 
48 output {
49     elasticsearch {
50         hosts => ["127.0.0.1:9500"]
51         index => "logstash-nginx-xtw"
52     }
53  }

 

说明:

(a)gsub => ["message", "\\x", "\\\x"]

用于兼容中文路径,数据替换

(b)gsub => ["message", ":-,", ":0,"]

Nginx日志有些字段输出为空时用“-”表示,json会解析不到,所以转换为0

(c)source => "message"

从filebeat数据中取出message

(d)remove_field 

删除kibana页面展示的多余字段

(e)mutate convert

用于转换数据类型,方便后续数据统计

(f)geoip

需要下载全球IP库:GeoLite2-City.mmdb,它可以解决通过IP对城市级别的定位问题。

(g)output输出设置

logstash 使用 Elasticsearch 输出插件,把数据发送到 Elasticsearch 进行存储和搜索

 

(2)启动logstash

/home/elk/elk5.2/logstash-5.2.1/bin/logstash -f /home/elk/elk5.2/logstash-5.2.1/config/logstash-nginx.conf

 

三、最终效果图及流量统计

技术图片

 

技术图片

 热力图:

技术图片

 

 

针对具体域名进行流量统计

思路:主要是对nginx日志字段 $body_bytes_sent (kibana显示为size字段)进行流量统计的

注意:因为nginx字段中 $body_bytes_sent ,单位是 byte 传输的字节,但是在画图统计后,显示的不太直观,所以需要转换成KB/MB/GB

 

1、格式化size字段类型

选中左边菜单栏的“Management”,找到对应的索引,修改size字段,Format改成“Bytes”类型,保存即可,这样数据显示都是经过单位换算的,如B/KB/MB/GB等

技术图片

 

 2、绘制线条图查看流量使用情况

找到左边菜单栏的“Visualize”,选择线条图Line chart

技术图片

 

切换到Data窗口

(1)metrics

Aggregation选择“Sum”,代表求和;

Field选择“size”,代表针对size字段

(2)buckets

Aggregation选择“Date Histogram”,代表按照日期自动划分

Field选择“timestamp”,代表时间戳

技术图片

 

比如统计某个域名在过去一个小时里使用的流量,可以在搜索框输入域名

技术图片

 

四、注意及总结

 1、logstash配置文件中的output输出的索引要以“logstash”作为前缀命名,经纬度写入到[geoip][location]上,否则Tile map成不了图,除非自定义索引模板,logstash* 有系统自带的索引模板,默认类型是geo_point,能被地图识别

 2、一开始使用nginx日志中的$http_x_forwarded_for作为客户端的ip,发现它有好几种值:空值(用“-”表示),1个或多个。

可能是下载的全球IP库解析不出部分ip地址,因为$http_x_forwarded_fo很多时候是代理ip地址,内网ip等是解析不出具体方位的。

最终换成$remote_addr作为客户端真实访问ip,没有出现ip解析错误的问题

技术图片

 

 

ELK对nginx日志进行流量监控

标签:cli   实现   包含   日志采集   市级   off   级别   搜索框   href   

原文地址:https://www.cnblogs.com/windysai/p/11832843.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!