使用Filebeat + ES + Kibina的組合進(jìn)行日志收集的一個(gè)優(yōu)點(diǎn)就是輕量級悠汽，因?yàn)槿サ袅吮恐氐膌ogstash, 占用資源更少。但這也引入了一個(gè)問題芥驳，即filebeat并沒有l(wèi)ogstash那樣強(qiáng)大的日志解析能力柿冲，往往只能把整條日志當(dāng)成一個(gè)整體扔到ES中。好消息是兆旬，ES從5.x版本開始引入了Ingest Node假抄，即允許你在對文檔進(jìn)行索引之前進(jìn)行預(yù)處理，且支持logstash的Grok語法。因此我們可以在ES中針對我們的日志格式創(chuàng)建一個(gè)預(yù)處理pipeline, 通過配置pipeline中的processor完成日志解析宿饱。
以下面這條日志為基礎(chǔ)舉例:
[2019-02-19 17:04:28:017] http-nio-8050-exec-2 INFO c.b.o.xxx.ms.api.TaskController - response = {"jobId":"123","ms":10}
我們期望能夠?qū)⑦@條日志中的時(shí)間2019-02-19 17:04:28:017熏瞄、線程http-nio-8050-exec-2、日志級別INFO谬以、Java類名c.b.o.xxx.ms.api.TaskController和日志正文response = {"jobId":"123","ms":10}分別提取出來方便我們?nèi)蘸笤趉ibana中做篩選統(tǒng)計(jì), 同時(shí)時(shí)間要以日志中打印的時(shí)間為基準(zhǔn)而不是filebeat發(fā)送消息時(shí)的時(shí)間强饮。為了實(shí)現(xiàn)這一目標(biāo)，我們可以向ES發(fā)一個(gè)HTTP請求創(chuàng)建一個(gè)名為xxx-log的pipeline:
PUT /_ingest/pipeline/xxx-log HTTP/1.1
Host: localhost:8200
Content-Type: application/json
{
"description" : "xxx-log",
"processors": [
{
"grok": {
"field": "message",
"patterns": ["\\[%{TIMESTAMP_ISO8601:log_time}\\] %{NOTSPACE:thread} %{NOTSPACE:log_level} %{NOTSPACE:java_class} - %{GREEDYDATA:content}"]
}
},
{
"date": {
"field": "log_time",
"formats": ["yyyy-MM-dd HH:mm:ss:SSS"],
"timezone": "Asia/Shanghai",
"target_field": "@timestamp"
}
}
]
}
在這里我們定義了兩個(gè)processor为黎，第一個(gè)為grok處理器邮丰，用于解析日志字符串提取其中關(guān)鍵字段; 第二個(gè)是日期處理器，功能為把log_time字段以yyyy-MM-dd HH:mm:ss:SSS格式解析成日期碍舍，然后將結(jié)果保存到@timestamp字段中柠座。
創(chuàng)建完processor以后，我們只需要配置filebeat在輸出日志到ES時(shí)使用這個(gè)名為xxx-log的預(yù)處理器即可:
這樣就完成了所有的工作片橡。這時(shí)啟動filebeat, 如果如出以下錯(cuò)誤信息
ERROR pipeline/output.go:92 Failed to publish events: temporary bulk send failure
大概率是因?yàn)槟惆l(fā)送的日志格式無法與grok表達(dá)式匹配妈经，修改processor定義json即可。也可以在啟動filebeat時(shí)添加-d "*"參數(shù)來查看具體的錯(cuò)誤原因捧书。
下圖是日志在kibana中的展示效果:
可以看到主要的字段都已經(jīng)被正確解析吹泡。
