所有ELK的安裝包都可以去官網下載，雖然速度稍慢，但還可以接受，官網地址：https://www.elastic.co/

logstash

在Logstash1.5.1版本，pattern的目錄已經發生改變，存儲在/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-0.1.10/目錄下，但是好在配置引用的時候是可以對patterns的目錄進行配置的，所以本人在Logstash的根目錄下新建了一個patterns目錄。而配置目錄在1.5.1版本中也不存在了，如果是rpm包安裝的，可以在/etc/logstash/conf.d/下面進行配置，但個人測試多次，這樣啟動經常性的失敗，目前還沒有去分析原因(個人不推薦使用RPM包安裝)。所以大家可以采用Nohup或者screen的方式進行啟動
專屬nginx的pattern配置：

由于是測試環境，我這里使用logstash讀取nginx日志文件的方式來獲取nginx的日志，并且僅讀取了nginx的access log，對于error log沒有關心。

使用的logstash版本為2.2.0，在log stash程序目錄下創建conf文件夾，用于存放解析日志的配置文件，并在其中創建文件test.conf，文件內容如下：

input {  file {    path => ["/var/log/nginx/access.log"]  }}filter {  grok {    match => {      "message" => "%{IPORHOST:clientip} /[%{HTTPDATE:time}/] /"%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}/" %{NUMBER:http_status_code} %{NUMBER:bytes} /"(?<http_referer>/S+)/" /"(?<http_user_agent>/S+)/" /"(?<http_x_forwarded_for>/S+)/""    }  }  }output {  elasticsearch {    hosts => ["10.103.17.4:9200"]    index => "logstash-nginx-test-%{+YYYY.MM.dd}"    workers => 1    flush_size => 1    idle_flush_time => 1    template_overwrite => true  }  stdout{codec => rubydebug}} 

需要說明的是，filter字段中的grok部分，由于nginx的日志是格式化的，logstash解析日志的思路為通過正則表達式來匹配日志，并將字段保存到相應的變量中。logstash中使用grok插件來解析日志，grok中message部分為對應的grok語法，并不完全等價于正則表達式的語法，在其中增加了變量信息。

具體grok語法不作過多介紹，可以通過logstash的官方文檔中來了解。但grok語法中的變量類型如IPORHOST并未找到具體的文檔，只能通過在logstash的安裝目錄下通過grep -nr "IPORHOST" .來搜索具體的含義。

配置文件中的stdout部分用于打印grok解析結果的信息，在調試階段一定要打開。

可以通過這里來驗證grok表達式的語法是否正確，編寫grok表達式的時候可以在這里編寫和測試。

對于elasticsearch部分不做過多介紹，網上容易找到資料。

elk收集分析nginx access日志

使用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格式了。

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

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

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語句如下：

好了，現在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}';