基於ELK Nginx日志分析

配置Nginx 日志

Nginx 默認的access 日志為log格式，需要logstash 進行正則匹配和清洗處理，從而極大的增加了logstash的壓力 所以我們Nginx 的日志修改為json 格式 。
Nginx access 日志和 Nginx error 日志

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;
    log_format  json  '{"@timestamp":"$time_iso8601",'
                      '"server_addr":"$server_addr",'
                      '"hostname":"$hostname",'
                      '"remote_add":"$remote_addr",'
                      '"request_method":"$request_method",'
                      '"scheme":"$scheme",'
                      '"server_name":"$server_name",'
                      '"http_referer":"$http_referer",'
                      '"request_uri":"$request_uri",'
                      '"args":"$args",'
                      '"body_bytes_sent":$body_bytes_sent,'
                      '"status": $status,'
                      '"request_time":$request_time,'
                      '"upstream_response_time":"$upstream_response_time",'
                      '"upstream_addr":"$upstream_addr",'
                      '"http_user_agent":"$http_user_agent",'
                      '"https":"$https"'
                      '}';
    access_log  /var/log/nginx/access.log json;

針對不同的虛擬主機配置Nginx日志

access_log  /var/log/nginx/80.access.log json;
error_log  /var/log/nginx/80.error.log error;
access_log  /var/log/nginx/8001.access.log json;
error_log  /var/log/nginx/8001.error.log error;

Nginx error_log 類型

[ debug | info | notice | warn | error | crit ] 

例如：error_log /var/log/nginx/8001.error.log crit;
解釋：日志文件存儲在/var/log/nginx/8001.error.log 文件中，錯誤類型為 crit ，也就是記錄最少錯誤信息（debug最詳細 crit最少）；

filebeat 配置

針對*.access.log 和 *.error.log 的日志進行不同的標簽封裝

[root@elk-node1 nginx]# egrep -v "*#|^$" /etc/filebeat/filebeat.yml 
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/*.access.log
  tags: ["nginx.access"]
- type: log
  paths:
    - /var/log/nginx/*.error.log
  tags: ["nginx.error"]
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 3
setup.kibana:
output.logstash:
  hosts: ["192.168.99.186:6044"]
processors:
  - add_host_metadata: ~
  - add_cloud_metadata: ~

logstash 配置

查看logstash 安裝已經安裝插件

/usr/share/logstash/bin/logstash-plugin list 
[root@elk-node2 ~]#/usr/share/logstash/bin/logstash-plugin list |grep geoip
logstash-filter-geoip
/usr/share/logstash/bin/logstash-plugin    install logstash-filter-geoip

Nginx 日志清洗規則

[root@elk-node2 ~]# cat /etc/logstash/conf.d/nginx.conf 
input {
  beats {
    port => 6044
  }

}

filter {
    if "nginx.access" in [tags] {
        json {
            source => "message"
            remove_field => "message"
        }
        date {
            match => ["timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]
        }
        useragent {
            target => "agent"
            source => "http_user_agent"
        }     
        geoip {
             #target => "geoip"
             source => "remote_add"
             fields => ["city_name", "country_code2", "country_name", "region_name","longitude","latitude","ip"]
             add_field => ["[geoip][coordinates]","%{[geoip][longitude]}"]
             add_field => ["[geoip][coordinates]","%{[geoip][latitude]}"]
        }
        mutate {
            convert => ["[geoip][coordinates]","float"] 
        }
    }
   else if "nginx.error" in [tags] {
        mutate {
            remove_field => ["@timestamp"]
        }
        grok {
            match => {"message" => "(?<datetime>%{YEAR}[./-]%{MONTHNUM}[./-]%{MONTHDAY}[- ]%{TIME}) \[%{LOGLEVEL:severity}\] %{POSINT:pid}#%{NUMBER}: %{GREEDYDATA:errormessage}(?:, client: (?<real_ip>%{IP}|%{HOSTNAME}))(?:, server: %{IPORHOST:domain}?)(?:, request: %{QS:request})?(?:, upstream: (?<upstream>\"%{URI}\"|%{QS}))?(?:, host: %{QS:request_host})?(?:, referrer: \"%{URI:referrer}\")?"}
        }   
        date {
            match => ["datetime", "yyyy/MM/dd HH:mm:ss"]
            target => "@timestamp"
        }
        mutate {
            remove_field => ["message"]
        }
   }
}

output{
    stdout{codec => rubydebug}
    
    if "nginx.access" in [tags]{
        elasticsearch{
            index => "logstash-nginx.access-%{+YYYY.MM.dd}"
            hosts => ["192.168.99.186:9200"]
        }
    }
    else if "nginx.error" in [tags]{
        elasticsearch {
            index => "nginx.error-%{+YYYY.MM.dd}"
            hosts => ["192.168.99.186:9200"]
        }
    }
}

注意：source 可以是任意處理后的字段，需要注意的是 IP 必須是公網 IP，否則logstash 的返回的geoip字段為空

Logstash解析

Logstash 分為 Input、Output、Filter、Codec 等多種plugins。

• Input：數據的輸入源也支持多種插件，如elk官網的beats、file、graphite、http、kafka、redis、exec等等。
• Output：數據的輸出目的也支持多種插件，如本文的elasticsearch，當然這可能也是最常用的一種輸出。以及exec、stdout終端、graphite、http、zabbix、nagios、redmine等等。
• Filter：使用過濾器根據日志事件的特征，對數據事件進行處理過濾后，在輸出。支持grok、date、geoip、mutate、ruby、json、kv、csv、checksum、dns、drop、xml等等。
• Codec：編碼插件，改變事件數據的表示方式，它可以作為對輸入或輸出運行該過濾。和其它產品結合，如rubydebug、graphite、fluent、nmap等等。

配置文件的含義

input
filebeat 傳入

filter
grok：數據結構化轉換工具
match：匹配條件格式
geoip：該過濾器從geoip中匹配ip字段，顯示該ip的地理位置
source：ip來源字段　
target：指定插入的logstash字段目標存儲為geoip　
add_field: 增加的字段，坐標經度　
add_field: 增加的字段，坐標緯度
mutate：數據的修改、刪除、類型轉換　
convert：將坐標轉為float類型　
replace：替換一個字段　
remove_field：移除message 的內容，因為數據已經過濾了一份，這里不必在用到該字段了，不然會相當於存兩份　
date: 時間處理，該插件很實用，主要是用你日志文件中事件的事件來對timestamp進行轉換　
match：匹配到timestamp字段后，修改格式為dd/MMM/yyyy:HH:mm:ss Z
mutate：數據修改　
remove_field：移除timestamp字段。

output
elasticsearch：輸出到es中
host：es的主機ip＋端口或者es 的FQDN＋端口
index：為日志創建索引logstash-nginx-access-*，這里也就是kibana那里添加索引時的名稱

Kibana 配置

注意：默認配置中Kibana的訪問日志會記錄在/var/log/message 中，使用logging.quiet參數關閉日志

[root@elk-node1 nginx]# egrep -v "*#|^$" /etc/kibana/kibana.yml 
server.port: 5601
server.host: "192.168.99.185"
elasticsearch.hosts: ["http://192.168.99.185:9200"]
kibana.index: ".kibana"
logging.quiet: true
i18n.locale: "zh-CN"
tilemap.url: 'http://webrd02.is.autonavi.com/appmaptile?lang=zh_cn&size=1&scale=1&style=7&x={x}&y={y}&z={z}'

配置“tilemap.url:”參數使Kibana使用高德地圖