對於logstash和filebeat來說,均有采集log的功能,elk如果采集的日志無需處理,可以使用lfilebeat采集日志直接輸出至elasticsearch(非常不建議這么做,如此收集的日志輸出的名稱均為fieebeat+time,消息很多時候都放在了同一個標簽message下)
我們使用的日志采集為elk+filebeat,如果你項提升速度可以使用filebeat采集輸出至redis或者其他中間件來緩解logstash的壓力,logstash格式話日志大量的話非常消耗cpu資源,如果你可以和開發人員協商直接輸出json話的格式就可以舍棄掉logstash中 filter/grok 配置,對於nginx之類應用日志來說,
1,elasticesearch日志采集后均會輸出為json話的日志格式輸出至kibana,如果未經處理你會看見這樣的日志:
順便貼上日志樣式:
這里日志采集的filebeat放在了本機的windows下,采集到Logstash后未作任何處理便輸出到了ES,所有的日志消息全部生成到了message下,而且采集的時間和日志輸出的時間都是不一樣的,這顯然無法使用:
為此logstash提供了一系列的插件來處理這些日志格式:
http://doc.yonyoucloud.com/doc/logstash-best-practice-cn/input/file.html詳細可以參見這篇文章。
2,那么下面我們就一步步的來調試這部分log:
2.1,首先我們應該多行合並,logstash提供了一個叫multiline的插件,
查詢logstash安裝的插件:
/usr/share/logstash/bin/logstash-plugin list
如果沒有安裝則需要安裝:
/usr/share/logstash/bin/logstash-plugin install logstash-filter-multiline
2.2,合並日志:
input{ beats{ port => "5046" } } filter{ multiline { pattern => "^\====" negate => true what => "previous" } } output{ elasticsearch{ hosts => ["127.0.0.1:9200"] index => "windows-%{+YYYY.MM.dd}" } }
2.3,再看前端顯示的日志:
2.4,grok日志格式化:
ELK已經設置好了很多匹配格式,很多時候直接拿來用就行 https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns
調試可能需要調試很久,好在kibana提供了調式工具,
這里我調試好了日志格式
multiline固然合並了換行,但是在ES里輸出的時候仍舊為兩行的日志,所以這里我自定義了個TEST的匹配模式匹配了換行符,那么我們把它寫在配置文件中:
input{ beats{ port => "5046" } } filter{ multiline { pattern => "^\====" negate => true what => "previous" } grok { patterns_dir => "/etc/logstash/patterns/logstash.patterns" match => {"message" => "%{DATE_EU:date} %{TIME:time} %{USER:TakingID} %{USER:Loglevel}%{TEST:imformation}"} remove_field =>["message"] } } output{ elasticsearch{ hosts => ["127.0.0.1:9200"]
這里用remove_filed刪除了message字段;否則數據實質上就相當於是重復存儲了兩份
3,關於日志刪除
日志放在es里是要做定時刪除的,除了要清理客戶端的日志外,還要清理es存儲 的日志:
查詢 curl 'localhost:9200/_cat/indices?v'
刪除 curl -XDELETE 'localhost:9200/windows-2019.07.26'
刪除后在查詢:
4,到這里ELK基本能滿足日常使用了,后面在詳細說說logstash多入多出