當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

转: 基于elk 实现nginx日志收集与数据分析

發布時間：2025/3/15 编程问答 18 豆豆

生活随笔收集整理的這篇文章主要介紹了转: 基于elk 实现nginx日志收集与数据分析小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

原文鏈接:https://www.cnblogs.com/wenchengxiaopenyou/p/9034213.html

一。背景

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

二。業務整體架構：

? ? ? nginx日志落地——》filebear——》logstash——》elasticsearch——》grafna（展示）

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

如上只是簡單的幾個實用的例子，實際上有多維度的數據之后還可以定制很多需要的內容，如終端ip訪問數，國家、地區占比，訪問前十的省份，請求方法占比，referer統計，user_agent統計，慢響應時間統計，更有世界地圖坐標展示等等，只要有數據就能多維度進行展示。這里提供模板搜索位置大家可以查找參考：https://grafana.com/dashboards

四，準備條件

需要具備如下條件：

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

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

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日志中日志格式定義如下：

12345678910

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? 這里應該不會遇到坑。

1234567

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并不是必須的，我加入的原因還是由于資源有限導致的。

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051

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字段格式有問題，無法拼接成坐標），其他可以根據情況自定：

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172

{??"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/

分類: 運維知識積累標簽: ELK, nginx日志收集, grafana ? 上一篇：python-num18 － django進階一 posted @ 2018-05-14 01:13 文成小盆友閱讀(305) 評論(0) 編輯收藏

轉載于:https://www.cnblogs.com/wangdaijun/p/9110556.html

總結

以上是生活随笔為你收集整理的转: 基于elk 实现nginx日志收集与数据分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Qt的index 用方法static_c
下一篇：解决centos下缺少sasl.h的问题