ELK對nginx日志進行流量監控

本文轉載自查看原文 2019-11-12 08:11 803 Linux 總結

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  }