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完成日志的可視化查詢展示

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"}

分析詳解：

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配置】

【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成功存儲索引名稱

http://192.168.37.136:5601   接下來在kibana創建索引，如下圖所示

最后我們可以看到。從最開始的自定義日志格式，到索引存儲，最終到Kibana展示，過程一點毛病都沒有，展示的日志都 是我們自定義的樣子，依舊是13個字段，很有條理性，我們還可以在左側進行刪選操作，這樣更加的高效~ok，打完收工~