Spring Cloud ELK+kafka日誌分析平臺（二）優化
一、概述
在筆者的上一篇部落格介紹了Spring Cloud ELK+kafka日誌分析平臺的搭建，http://xuyangyang.club/articles/2018/05/24/1527176074152.html，但是筆者在測試環境中發現，在logstash採用了grok外掛去處理日誌埋點和解析的時候發現了高資源佔用，在阿里雲8核16G的伺服器部署後，測試環境大概每秒不超過幾百條的日誌的解析下竟然CPU佔用高達95%左右，筆者分析了其中的原因，首先由於幾個服務的日誌格式相關配置還沒有落地導致了日誌格式無法解析和解析慢，其次grok天生解析正則表示式非常耗CPU。在筆者分析原因後立馬進行了改造，將無法解析的格式不做任何處理直接輸出到elasticsearch,另外多起了同一個group的基於kafka的logstash去消費日誌，發現CPU佔用雖然降低了，但任然在高達70%的CPU佔用，這使得筆者不得不放棄原先的logstash外掛grok，採用了ruby來解析日誌和處理日誌埋點，並且這次同時對docker啟動shell指令碼做相應修改，原先由於書寫filter導致指令碼雜亂無法檢視改為啟動傳入配置檔案的方式。
二、改造logstash
本次主要的問題主要發生在logstash的grok外掛，在筆者查閱相關資料中發現，logstash提供了grok、mutate、ruby、json等等多種filter外掛，由於我們的日誌非表中的json格式，所以放棄了測試json外掛，而筆者在ruby外掛的介紹中發現官方對其頗為稱讚，它支援ruby語法，對一些字串、迴圈、判斷能更方便，並且可以操作event物件（其實就是logstash物件），所以筆者採用了ruby外掛。以下為筆者所寫的logstash配置檔案示例：

filter {
    ruby{
        code => "
            array=event.get('message').split('|')
            if array.length==9 
                event.set('localDate',array.at(0))
                event.set('level',array.at(1))
                event.set('thread',array.at(2))
                event.set('class'
 
,array.at(3))
                event.set('method',array.at(4))
                event.set('line',array.at(5))
                event.set('service',array.at(6))
                event.set('ip',array.at(7))
                event.set('content',array.at(8))
            elsif array.length==10
                event.
 
set('localDate',array.at(0))
                event.set('level',array.at(1))
                event.set('thread',array.at(2))
                event.set('class',array.at(3))
                event.set('method',array.at(4))
                event.set('line',array.at(5))
                event.set('service',array.at(6))
                event.set('ip',array.at(7))
                event.set('uid',array.at(8))
                event.set('content',array.at(9))
            else
                content= event.get('message')
                event.set('content',content)
            end
            "
        remove_field=>"message"
    }
}

筆者對日誌輸出做了修改，將日誌用“|”分割了一下，並日志埋點了uid,在ruby中解析到字串陣列後，通過判斷它的length來判斷它是哪種型別的資料，從而進行解析處理，最後將冗餘欄位message刪除，需要注意的時在解析日誌之前最好手動去建立elasticsearch的index,如果不建立,logstash在解析到日誌會自動建立，在採用如上配置的解析後，最後用kibana分析的時候發現欄位確實解析和處理了，但是型別都是String，例如日誌埋點的uid或者其他屬性需要做一些過濾就不是很好用了，因此需要提前手動建立並設定型別。最後再說一下kibana，在第一次配置完索引後欄位不會馬上加載出來，也可能點開具體檢視的時候我們解析的fileld報感嘆號，這是因為kibana還沒載入到索引的全部欄位，可以去kibana的index管理中去refresh下就好了。以下附上建立elasticsearch索引的相關程式碼示例（在kibana中可以直接除錯使用）：

PUT /logstash-index
{  
 "mappings" : {  
  "_default_" : {  
   "properties" : {  
    "uid" : {"type": "long"},
    "localDate" : { "type" : "text" },  
    "level" : { "type" : "text" }  ,
    "thread" : { "type" : "text" } ,
    "class" : { "type" : "text" } ,
    "method" : { "type" : "text" } ,
    "service" : { "type" : "text" } ,
    "ip" : { "type" : "text" } ,
    "line" : { "type" : "text" } ,
    "content" : { "type" : "text" }
   }  
  }  
 }  
}

三、改造啟動指令碼
原先因為筆者想盡快測試使用所以採用了docker -e的方式，結果導致指令碼看起來雜亂不堪，現在附上新的docker啟動方式，通過傳配置檔案路徑進行docker部署啟動

docker run -idt -v /usr/local/elk/logstash.conf:/usr/local/logstash.conf --name logstash logstash -f /usr/local/logstash.conf

其中通過-v的方式對映配置檔案路徑,方式為 -v 宿主機路徑:容器內路徑，-f為logstash的啟動命令，可以傳入配置檔案路徑。