在實際的項目中需要對線上日志做實時分析跟統計，這一套方案可以用現有的ELK(ElasticSearch, Logstash, Kibana)方案既可以滿足，關於這個方案的具體的步驟可以參考網上的解決方案。但如果只想統計某個錯誤碼(http狀態碼，業務錯誤碼)在指定時間內出現多少次然后就觸發一個告警或者某個指令動作(郵件或者是調用已經寫好的http接口，例如例如微信平台來通知告警信息等)，這種需求可以用logstash進行實現，並且這種方案比較輕量級，很容易實現。這里以在linux平台為主。
下載
首先下載 最新的logstash 壓縮包，https://download.elastic.co/logstash/logstash/logstash-2.3.2.zip ，下載后用unzip命令進行解壓只指定的目錄，解壓后會生成一個目錄logstash-2.3.2 這個目錄，里面包含下面這些文件：
基本配置
開始可以通過一個小的介紹了解logstash是怎么回事，鏈接里面有一些基本的demo，有興趣可以參考里面的寫一下。鏈接如下：
http://udn.yyuap.com/doc/logstash-best-practice-cn/get_start/hello_world.html
運行命令如下：
假設在與logstash 平級目錄新建一個 配置文件，sample.conf ,這個配置文件就是定義 input filter output 這3塊，其主要表示 input的來源是什么(文件或命令之類的),filter 代表對於輸入的內容按照規則怎么來過濾，output 就是按照過濾的規則進行輸出 。
最簡單的寫法如下：
# bin/logstash -e 'input{stdin{}}output{stdout{codec=>rubydebug}}'
如果是在sample.conf 中寫入則如下：
input {
stdin{}
}
filter {
}
output {
stdout {}
}
# bin/logstash -f sample.conf
特殊配置
在實際的業務中可能往往要對日志做過濾，然后根據某些關鍵字段進行統計，通過統計結果來決定需要做什么。logstash 本身對於syslog,redis等日志在過濾的時候一般不需要自己寫正則表達式去過濾，這些都是通用的，在filter中可以直接通過%{定義的常量}來獲取日志內容，已經自動支持的過濾的日志包含下面這些：
這個列表不全，對於已經支持的這些可以直接到github 上進行搜索(https://github.com/logstash/logstash/tree/v1.4.2/patterns)，上面都有這些日志過濾的正則表達式,這些是屬於通用配置。
往往在業務日志中，所打印的業務關鍵錯誤碼以及格式並非與通用格式相匹配，那么此時就需要自己針對自己的業務日志格式，寫出正則表達式來過濾自己想要的信息。
這里就以日志中的errno 來為例，所實現的目的就是 當這個錯誤碼在30s內出現3次就把這個錯誤輸出出來或告警。
在logstash中正則表達式叫做grok表達式，其規則跟perl與ruby的正則表達式一樣。 也有在線的grok正則表達式校驗工具：http://grokdebug.herokuapp.com/
這里就以下面的日志格式為例：log.txt
begin 10001 end begin 10001 end begin 10001 end
在sample.conf 中，grok 是寫在filter 中的，這里既可以直接寫正則表達式，也可以引用外部文件的正則表達。
方式1：
grok { match => { "message" => "\s+(?<code>\d+?)\s+" } }
方式2就是預先在 patterns/ 目錄下創建一個文件，這個文件里面含有正則表達式，例如創建文件的名稱為 test ，內容如下：
ERRORCODE \s+(?<code>\d+?)\s+
grok { patterns_dir => "../../patterns/test" match => ["message", "%{ERRORCODE:code:int}"] }：
很顯然這里我們習慣采用方式2，因為正則匹配相關的可以單獨在一個文件，在sample.conf 就直接引用正則匹配中字段值，例如上面就直接用"ERRORCODE"， message 中%后面花括號中的寫法規則如下：
%{PATTERN_NAME:capture_name:data_type}
以上面完子為例，完整的 sample.conf 內容如下：
input { # 以日志文件作為來源 file { # 文件路徑 path => "../../log.txt" } } filter { # 正則解析 grok { # 增加自定義的正則，這里我把自定義的正則文件，加在了這個目錄下 patterns_dir => "../../patterns/test" # 從message字段中解析出 code match => ["message", "%{ERRORCODE:code:int}"] # 刪除原來的message字段 remove_field => [ "message" ] # 增加WARN，作為標記 add_tag => "warn" } # 計數 metrics { # 計數器數據保存的字段名 code的值就是上面解析出來的 meter => "events_%{code}" # 增加"metric"，作為標記 add_tag => "metric" # 3秒內的message數據才統計，避免延遲 # ignore_older_than => 3 # 每隔300秒統計一次 flush_interval => "30" # 每301秒清空計數器 clear_interval => "31" } # 如果event中標記為“metric”的 if "metric" in [tags] { # 執行ruby代碼 ruby { # 如果count為10001的數量小於3條，就忽略此事件(即不發送任何消息)。 code => "event.cancel if event['events_10001']['count'] < 3" } } } output { # 輸出到console stdout { codec => rubydebug } }
上面配置結束后，由於開啟 debug 模式，所以當每次往 log.txt 寫入 "begin 10001 end" 在控制台下都會有輸出，輸入內容下：
{ "@version" => "1", "@timestamp" => "2016-11-14T13:14:14.374Z", "path" => "../../log.txt", "host" => "xxxx", "code" => "10001", "tags" => [ [0] "warn" ] }
當在30s內輸入的次數大於3次的時候，輸出的內容如下：
{ "@version" => "1", "@timestamp" => "2016-11-14T13:14:20.148Z", "message" => "xxx", "events_" => { "WARN" => { "count" => 3, "rate_1m" => 2.1864058457089883, "rate_5m" => 3.814999397602868, "rate_15m" => 4.1951645878441255 } }, "tags" => [ [0] "metric" ] }
通過上面的結構可以看出我們在輸出的時候還可以寫表達式，例如只有當30s內大於3次才輸出信息的話可以在 output 中加一個判斷，邏輯如下：
output { # 如果event中標記為“metric”，表示只發送計數的消息。 if "metric" in [tags] { # 執行一個簡單的命令 exec { command => "echo \"hello: %{events_10001.count}\"" } } }
到這里已經結束，關於logstash中的 統計相關的規則可以參考：http://udn.yyuap.com/doc/logstash-best-practice-cn/filter/metrics.html
輸出的形式也是多種的 可以對接ES,觸發http接口，這里就以輸出為例。