elk收集分析nginx access日志

首先elk的搭建按照這篇文章使用elk+redis搭建nginx日志分析平台說的，使用redis的push和pop做隊列，然后有個logstash_indexer來從隊列中pop數據分析插入elasticsearch。這樣做的好處是可擴展，logstash_agent只需要收集log進入隊列即可，比較可能會有瓶頸的log分析使用logstash_indexer來做，而這個logstash_indexer又是可以水平擴展的，我可以在單獨的機器上跑多個indexer來進行日志分析存儲。

好了，現在進一步配置了。

nginx中的日志存儲格式

nginx由於有get請求，也有post請求，get請求的參數是會直接顯示在日志的url中的，但是post請求的參數呢，卻不會在access日志中體現出來。那么我想要post的參數也進行存儲紀錄下來。就需要自己定義一個log格式了。

log_format logstash '$http_host $server_addr $remote_addr [$time_local] "$request" $request_body $status $body_bytes_sent "$http_referer" "$http_user_agent" $request_time $upstream_response_time';

這里的\(request_body里面存放的就是POST請求的body了，然后GET請求的參數在\)request里面。具體怎么分析，我們在indexer中再想。

這里的server_addr存放的是當前web機器的IP，存這個IP是為了分析日志的時候可以分析日志的原始來源。

下面是一個GET請求的例子：

api.yejianfeng.com 10.171.xx.xx 100.97.xx.xx [10/Jun/2015:10:53:24 +0800] "GET /api1.2/qa/getquestionlist/?limit=10&source=ios&token=12343425324&type=1&uid=304116&ver=1.2.379 HTTP/1.0" - 200 2950 "-" "TheMaster/1.2.379 (iPhone; iOS 8.3; Scale/2.00)" 0.656 0.654

下面是一個POST請求的例子：

api.yejianfeng.com 10.171.xx.xx 100.97.xx.xx [10/Jun/2015:10:53:24 +0800] "POST /api1.2/user/mechanicupdate/ HTTP/1.0" start_time=1276099200&lng=110.985723&source=android&uid=328910&lat=35.039471&city=140800 200 754 "-" "-" 0.161 0.159

順便說下，這里知識在nginx.conf中定義了一個日志格式，還要記得在具體的服務中加入日志存儲。比如

listen       80;
server_name api.yejianfeng.com;
access_log /mnt/logs/api.yejianfeng.com.logstash.log logstash;

log_agent的配置

這個配置就是往redis隊列中塞入日志就行。output的位置設置為redis就行。

input {
        file {
                type => "nginx_access"
                path => ["/mnt/logs/api.yejianfeng.com.logstash.log"]
        }
}
output {
        redis {
                host => "10.173.xx.xx"
                port => 8001
                password => pass
                data_type => "list"
                key => "logstash:redis"
        }
}

log_indexer的配置

log_indexer的配置就比較麻煩了，需要配置的有三個部分

• input: 負責從redis中獲取日志數據
• filter: 負責對日志數據進行分析和結構化
• output: 負責將結構化的數據存儲進入elasticsearch

input部分

input {
        redis {
                host => "10.173.xx.xx"
                port => 8001
                password => pass
                data_type => "list"
                key => "logstash:redis"
        }
}

其中的redis配置當然要和agent的一致了。

filter部分

解析文本可以使用grokgrok debug進行分析，參照着之前的log格式，需要一個個進行日志分析比對。這個grok語法寫的還是比較復雜的，還好有在線grok比對工具可以使用。比對前面的GET和POST的日志格式，修改出來的grok語句如下：

%{IPORHOST:http_host} %{IPORHOST:server_ip} %{IPORHOST:client_ip} \[%{HTTPDATE:timestamp}\] \"%{WORD:http_verb} (?:%{PATH:baseurl}\?%{NOTSPACE:params}(?: HTTP/%{NUMBER:http_version})?|%{DATA:raw_http_request})\" (%{NOTSPACE:params})?|- %{NUMBER:http_status_code} (?:%{NUMBER:bytes_read}|-) %{QS:referrer} %{QS:agent} %{NUMBER:time_duration:float} %{NUMBER:time_backend_response:float}

這里使用了一點小技巧，params的使用，為了讓GET和POST的參數都反映在一個參數上，在對應的GET和POST的參數的地方，都設計使用params這個參數進行對應。

好了，現在params中是請求的參數。比如source=ios&uid=123。但是呢，最后做統計的時候，我希望得出的是“所有source值為ios的調用”，那么就需要對參數進行結構化了。而且我們還希望如果接口中新加入了一個參數，不用修改logstash_indexer就可以直接使用，方法就是使用kv，kv能實現對一個字符串的結構進行k=v格式的拆分。其中的參數prefix可以為這個key在統計的時候增加一個前綴，include_keys可以設置有哪些key包含在其中，exclude_keys可以設置要排除哪些key。

kv {
    prefix => "params."
    field_split => "&"
    source => "params"
}

好了，現在還有一個問題，如果請求中有中文，那么日志中的中文是被urlencode之后存儲的。我們具體分析的時候，比如有個接口是/api/search?keyword=我們，需要統計的是keyword被查詢的熱門順序，那么就需要解碼了。logstash牛逼的也有urldecode命令，urldecode可以設置對某個字段，也可以設置對所有字段進行解碼。

urldecode {
    all_fields => true
}

看起來沒事了，但是實際上在運行的時候，你會發現一個問題，就是存儲到elasticsearch中的timestamp和請求日志中的請求時間不一樣。原因是es中的請求日志使用的是日志結構存放進入es的時間，而不是timestamp的時間，這里想要吧es中的時間和請求日志中的時間統一怎么辦呢？使用date命令。具體設置如下：

date {
        locale => "en"
        match => ["timestamp" , "dd/MMM/YYYY:HH:mm:ss Z"]
}

具體的logstash_indexer中的全部配置如下：

filter {
        grok {
            match => [
                        "message", "%{IPORHOST:http_host} %{IPORHOST:server_ip} %{IPORHOST:client_ip} \[%{HTTPDATE:timestamp}\] \"%{WORD:http_verb} (?:%{PATH:baseurl}\?%{NOTSPACE:params}(?: HTTP/%{NUMBER:http_version})?|%{DATA:raw_http_request})\" (%{NOTSPACE:params})?|- %{NUMBER:http_status_code} (?:%{NUMBER:bytes_read}|-) %{QS:referrer} %{QS:agent} %{NUMBER:time_duration:float} %{NUMBER:time_backend_response:float}"
            ]
        }
        kv {
            prefix => "params."
            field_split => "&"
            source => "params"
        }
        urldecode {
                  all_fields => true
        }
        date {
                locale => "en"
                match => ["timestamp" , "dd/MMM/YYYY:HH:mm:ss Z"]
        }

}

output部分

這里就是很簡單往es中發送數據

output {
        elasticsearch {
                embedded => false
                protocol => "http"
                host => "localhost"
                port => "9200"
                user => "yejianfeng"
                password => "yejianfeng"
        }
}

這里有個user和password，其實elasticsearch加上shield就可以強制使用用戶名密碼登錄了。這里的output就是配置這個使用的。

查詢elasticsearch

比如上面的例子，我要查詢某段時間的params.source(其實是source參數，但是前面的params是前綴)調用情況

$url = 'http://xx.xx.xx.xx:9200/logstash-*/_search';
$filter = '
{
  "query": {
        "range" : {
            "@timestamp" : {
                "gt" : 123213213213,
                "lt" : 123213213213
            }
        }
   },
   "aggs" : {
        "group_by_source" : {"terms" : {"field" : "params.source"}}
    },
    "size": 0
}';

具體使用參考elasticsearch的文檔