基於elk 實現nginx日誌收集與數據分析。

時間 2019-12-09

標籤基於 elk 實現 nginx 日誌收集數據分析欄目 Nginx 简体版

原文原文鏈接

一。背景html

前端web服務器爲nginx，採用filebeat + logstash + elasticsearch + granfa 進行數據採集與展現，對客戶端ip進行地域統計，監控服務器響應時間等。前端

二。業務總體架構：linux

nginx日誌落地——》filebear——》logstash——》elasticsearch——》grafna（展現）nginx

三。先上個效果圖，慢慢去一步步實現git

如上只是簡單的幾個實用的例子，實際上有多維度的數據以後還能夠定製不少須要的內容，如終端ip訪問數，國家、地區佔比，訪問前十的省份，請求方法佔比，referer統計，user_agent統計，慢響應時間統計，更有世界地圖座標展現等等，只要有數據就能多維度進行展現。這裏提供模板搜索位置你們能夠查找參考：https://grafana.com/dashboardsgithub

四，準備條件web

須要具有以下條件：正則表達式

1.nginx日誌落地，須要主要落地格式，以及各個字段對應的含義。json

2.安裝filebeat。 filebeat輕量，功能相比較logstash而言比較單一。ruby

3.安裝logstash 做爲中繼服務器。這裏須要說明一下的是，起初設計階段並無計劃使用filebeat，而是直接使用logstash發往elasticsearch，可是當前端機數量增長以後logstash數量也隨之增長，同時發往elasticsearch的數量增大，logstash則會拋出因爲elasticsearch 限制致使的錯誤，你們遇到後搜索相應錯誤的代碼便可。爲此只採用logstash做爲中繼。

4.elasticsearch 集羣。坑點是index templates的建立會影響新手操做 geoip模塊。後文會有。

5.grafana安裝，取代傳統的kibana，grafana有更友好、美觀的展現界面。

五。實現過程

1.nginx日誌落地配置

nginx日誌格式、字段的內容和順序都是高度可定製化的，將須要收集的字段內容排列好。定義一個log_format

定義的形勢實際上直接決定了logstash配置中對於字段抽取的模式，這裏有兩種可用，一種是直接在nginx日誌中拼接成json的格式，在logstash中用codec => "json"來轉換，

一種是常規的甚至是默認的分隔符的格式，在logstash中須要用到grok來進行匹配，這個會是相對麻煩些。兩種方法各有優勢。直接拼接成json的操做比較簡單，可是在轉碼過程當中

會遇到諸如 \x 沒法解析的狀況。這裏我也遇到過，若有必要後續會詳談。採用grok來匹配的方法相對難操做，可是準確性有提高。咱們這裏採用的是第一種方法，下面logstash部分

也會給出採用grok的例子。 nginx日誌中日誌格式定義以下：

log_format access_json   '{"timestamp":"$time_iso8601",'
                        '"hostname":"$hostname",'
                        '"ip":"$remote_addrx",'
                        '"request_method":"$request_method",'
                        '"domain":"XXXX",'
                        '"size":$body_bytes_sent,'
                        '"status": $status,'
                        '"responsetime":$request_time,'
                        '"sum":"1"'
                        '}';

2.filebeat配置文件

關於filebeat更多內容請參考https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-overview.html

配置文件內容：filebeat.yml 這裏應該不會遇到坑。

filebeat.prospectors:
- input_type: log
  paths:
    - /data0/logs/log_json/*.log      #nginx日誌路徑

output.logstash:
    hosts: ["xxxx.xxxx.xxxx.xxx:12121"]   #logstash 服務器地址

3.logstahs配置文件內容：

這裏是針對json已經拼接號，直接進行json轉碼的狀況：

須要注意以下：

1）date模塊必須有，不然會形成數據沒法回填致使最終的圖像出現鋸齒狀影響穩定性（緣由是排列時間並非日誌產生的時間，而是進入logstash的時間）。這裏後面的 yyyy-MM-dd'T'HH:mm:ssZZ 須要根據你日誌中的日期格式進行匹配匹配規則見：https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html

　2）須要說明的是我下面去掉了好多的字段（remove_field），緣由是咱們數據量大，es服務器有限。能夠根據須要隨時調整收集的字段。

　3) geoip 僅須要指定源ip字段名稱便可，fields並非必須的，我加入的緣由仍是因爲資源有限致使的。

input {
    beats {
    port => 12121
    host => "10.13.0.80"
    codec => "json"
    }
}

filter {
    date {
      match => [ "timestamp", "yyyy-MM-dd'T'HH:mm:ssZZ" ]
      #timezone => "Asia/Shanghai"
      #timezone => "+00:00"
    }
    mutate {
      convert => [ "status","integer" ]
      convert => [ "sum","integer" ]
      convert => [ "size","integer" ]
      remove_field => "message"
      remove_field => "source"
      remove_field => "tags"
      remove_field => "beat"
      remove_field => "offset"
      remove_field => "type"
      remove_field => "@source"
      remove_field => "input_type"
      remove_field => "@version"
      remove_field => "host"
      remove_field => "client"
      #remove_field => "request_method"
      remove_field => "size"
      remove_field => "timestamp"
      #remove_field => "domain"
      #remove_field => "ip"
    }
    geoip {
      source => "ip"
      fields => ["country_name", "city_name", "timezone","region_name","location"]
    }
}

output {
    elasticsearch {
       hosts => ["xxx:19200","xxx:19200","xxx:19200"]
       user => "xxx"
       password => "xxx"
       index => "logstash-suda-alllog-%{+YYYY.MM.dd}"
       flush_size => 10000
       idle_flush_time => 35
    }
}

　　下面給出一個採用grok的例子：

其中match內的內容不用參考個人，須要根據你的字段個數，以及格式來定製。這裏實際上是正則表達式。自帶了一部分幾乎就能夠知足全部的需求了無需本身寫了，能夠參考：https://github.com/elastic/logstash/blob/v1.4.0/patterns/grok-patterns 直接用便可。

input {
  file {
  type => "access"
  path => ["/usr/local/nginx/logs/main/*.log"]
  }
}
 
filter {
  if [type] == "access" {
  if [message] =~ "^#" {
    drop {}
  }
 
 
  grok {
    match => ["message", "\[%{HTTPDATE:log_timestamp}\] %{HOSTNAME:server_name} \"%{WORD:request_method} %{NOTSPACE:query_string} HTTP/%{NUMBER:httpversion}\" \"%{GREEDYDATA:http_user_agent}\" %{NUMBER:status} %{IPORHOST:server_addr} \"%{IPORHOST:remote_addr}\" \"%{NOTSPACE:http_referer}\" %{NUMBER:body_bytes_sent} %{NUMBER:time_taken} %{GREEDYDATA:clf_body_bytes_sent} %{NOTSPACE:uri} %{NUMBER:m_request_time}"]
  }
 
 
  date {
    match => [ "log_timestamp", "dd/MMM/yyyy:mm:ss:SS Z" ]
    timezone => "Etc/UTC"
  }
     
  mutate {
  convert => [ "status","integer" ]
  convert => [ "body_bytes_sent","integer" ]
  convert => [ "m_request_time","float" ]
  }

在提供一個高級的用法：

ruby：強大的模塊，能夠進行諸如時間轉換，單位計算等，多個能夠用分號隔開。

  ruby {
  code => "event.set('logdateunix',event.get('@timestamp').to_i);event.set('request_time', event.get('m_request_time') / 1000000 )"  
  }
 
  mutate {
   add_field => {
        "http_host" => "%{server_name}"
        "request_uri" => "%{uri}"
        }

在windowns上使用lgostsh須要注意的是：（win上收集iis的日誌，我想正常環境是不會用到的，可是我確實用到了。。。。。。）

path路徑必定要採用 linux中的分割符來拼接路徑，用win的格式則正則不能實現，你們能夠測試下。其餘配置則無區別。

input {
  file {
    type => "access"
    path => ["C:/WINDOWS/system32/LogFiles/W3SVC614874788/*.log"]
  }
}

4.elasticsearch配置，集羣的安裝以及啓動，調優這裏很少說（說不完），須要注意的一個是，geoip location的格式，我這裏採用的是index templates來實現的以下：

最重要的是 "location" 定義 （不然geoip_location字段格式有問題，沒法拼接成座標），其餘能夠根據狀況自定：

{
  "order": 0,
  "version": 50001,
  "index_patterns": [
    "suda-*"
  ],
  "settings": {
    "index": {
      "number_of_shards": "5",
      "number_of_replicas": "1",
      "refresh_interval": "200s"
    }
  },
  "mappings": {
    "_default_": {
      "dynamic_templates": [
        {
          "message_field": {
            "path_match": "message",
            "mapping": {
              "norms": false,
              "type": "text"
            },
            "match_mapping_type": "string"
          }
        },
        {
          "string_fields": {
            "mapping": {
              "norms": false,
              "type": "text",
              "fields": {
                "keyword": {
                  "ignore_above": 256,
                  "type": "keyword"
                }
              }
            },
            "match_mapping_type": "string",
            "match": "*"
          }
        }
      ],
      "properties": {
        "@timestamp": {
          "type": "date"
        },
        "geoip": {
          "dynamic": true,
          "properties": {
            "ip": {
              "type": "ip"
            },
            "latitude": {
              "type": "half_float"
            },
 "location": { "type": "geo_point" },
            "longitude": {
              "type": "half_float"
            }
          }
        },
        "@version": {
          "type": "keyword"
        }
      }
    }
  },
  "aliases": {}
}

5 grafana 安裝以及模板建立，這個比較簡單，安裝完直接寫語句便可附一個例子以下：

這裏的變量須要自定義：

經過上面應該可以完成一個完整的收集和展現狀況，這裏實際上提供了一種可行的方法，並麼有太大的具體操做。

但願多多交流

推薦內容：kibana中文指南（有ibook版本，看着挺方便）三斗室著

　　　　　https://www.elastic.co/cn/

　　　　 https://grafana.com/dashboards

　　　　 https://grafana.com/