ELK收集Nginx自定義日誌格式輸出

時間 2019-11-10

原文原文鏈接

1.ELK收集日誌的有兩種經常使用的方式：html

1.1:不修改源日誌格式，簡單的說就是在logstash中轉經過 grok方式進行過濾處理，將原始無規則的日誌轉換爲規則日誌（Logstash自定義日誌格式）node

1.2:修改源日誌格式，將須要的日誌格式進行規則輸出，logstash只負責日誌的收集和傳輸，不對日誌作任何過濾處理（filebeat生產者自定義日誌格式）nginx

優缺點：web

首先咱們來看下不修改源日誌格式，這樣Logstash會經過grok來處理分析，對線上業務無任何影響；可是在高壓環境下，Logstash中的grok會成爲性能瓶頸，最終會阻塞正常的日誌輸出，因此，在Logsatsh中，儘可能不要使用grok過濾功能正則表達式

第二種是修改源日誌格式，也就是在收集生產日誌的過程當中，自定義日誌格式，雖然有必定的工做量，可是優點很明顯，由於是實現定義好了日誌輸出格式，logstash那就只負責收集和傳輸了，這樣大大減輕了logstash負擔，能夠更高效的收集和傳輸日誌；是企業首選方案json

下圖是引用了網上與之本次實驗效果圖bootstrap

最前方是一臺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}"    #自定義指定索引名稱
   } 
 }
}