ELK对nginx日志进行流量监控

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主配置文件中加入：

log_format json ‘{"@timestamp":"$time_iso8601",‘
  ‘"host":"$server_addr",‘
  ‘"clientip":"$remote_addr",‘
  ‘"size":$body_bytes_sent,‘
  ‘"responsetime":$request_time,‘
  ‘"upstreamtime":$upstream_response_time,‘
  ‘"upstreamhost":"$upstream_addr",‘
  ‘"http_host":"$host",‘
  ‘"url":"$uri",‘
  ‘"xff":"$http_x_forwarded_for",‘
  ‘"referer":"$http_referer",‘
  ‘"agent":"$http_user_agent",‘
  ‘"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文件
添加要读取的日志路径

ELK对nginx日志进行流量监控

（b）配置filebeat输出到logstash，

注释输出到elasticsearch的配置：

ELK对nginx日志进行流量监控

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

ELK对nginx日志进行流量监控

（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

input {
         beats {
             port => 4568
             client_inactivity_timeout => 600  #600秒之后关闭空闲的连接
        }
}

filter {
      mutate {
          gsub => ["message", "\\x", "\\\x"]
          gsub => ["message", ":-,", ":0,"]
      }

      json {
            source => "message"
            remove_field => "message"
            remove_field => "[beat][hostname]"
            remove_field => "[beat][name]"
            remove_field => "[beat][version]"
            remove_field => "@version"
            remove_field => "offset"
            remove_field => "input_type"
            remove_field => "tags"
            remove_field => "type"
            remove_field => "host"
      }
      mutate {
          convert => ["status", "integer"]
          convert => ["size","integer"]
          convert => ["upstreamtime", "float"]
          convert => ["responsetime", "float"]
      }

       geoip {
            source => "clientip"
            database => "/home/elk/elk5.2/logstash-5.2.1/config/GeoLite2-City.mmdb"
            fields => ["city_name", "country_code2", "country_name", "latitude", "longitude", "region_name"] 
            add_field => [ "[geoip][location]", "%{[geoip][longitude]}" ]
            add_field => [ "[geoip][location]", "%{[geoip][latitude]}" ]
            target => "geoip"
        }
        mutate {
           convert => [ "[geoip][location]", "float" ]
         }
        if "_geoip_lookup_failure" in [tags] { drop { } }
}

output {
    elasticsearch {
        hosts => ["127.0.0.1:9500"]
        index => "logstash-nginx-xtw"
    }
 }

说明：

（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