S小魚仔S ELK 收集 Fortigate 60D Syslog 使用 Grok 產生 字段

關於「ELK」簡單來說就是一個「LOG」收集分析軟件，屬於「開源」基於「Linux」環境 架設與佈署，Windows Log、Linux Log、Syslog 照單全收，需要建置角色「Elasticsearch」、「Logstash」、「Kibana」三個套件，個人還是喜歡「Splunk」因為簡單好上手。

由於「Fortigate 60D - Syslog」 送往「ELK」並不會自動做「字段分析」，需要使用「logstash」=>「filter」插件功能 =>啟用「grok」，同時感謝「料理佳餚」作者 分享「ELK」設定「Grok」方法。

先讓各位看一下「Grok」編寫方法

「ELK」收集「字段」顯示

使用「Grok」功能，我們需要先獲取「Fortigate 60D」Message

<189>date=2017-06-03 time=05:02:07 devname=FGT60D1112219335 devid=FGT60D1112219335 logid=0000000013 type=traffic subtype=forward level=notice vd=root srcip=192.168.2.100 srcport=50086 srcintf="internal" dstip=117.185.116.152 dstport=80 dstintf="wan1" poluuid=08a76f2c-1110-51e7-d44e-8c9dc0f52380 sessionid=300676 proto=6 action=close policyid=1 policytype=policy dstcountry="China" srccountry="Reserved" trandisp=snat transip=192.168.1.5 transport=50086 service="HTTP" appcat="unknown" applist="default" duration=1 sentbyte=703 rcvdbyte=519 sentpkt=6 rcvdpkt=5 wanin=307 wanout=383 lanin=383 lanout=307

天啊..這麼多訊息，我們要怎麼進行「字段」分析，別擔心前往「Grok Debug」將「內文」貼上，使用「Discover」智能分析「關鍵」字段，由於「畢竟」是「工具」許多「字段」都是寫死，必須改為「變數」。

將「Dicover」智能分析「結果」Copy 過來「Debugger」頁面，依序修改「變數」並且 所見及所得 ( What You See It Waht You Get ) ，一步一步 邁向成功之路.....

得到最後結果如下

「

%{SYSLOG5424PRI}date=20%{DATE} time=%{TIME:time} devname=%{HOSTNAME:devName} devid=%{HOSTNAME:devId} logid=%{NUMBER:logId} type=%{WORD:logType} subtype=%{WORD:logSubType} level=%{WORD:level} vd=%{DATA:virtualDomain} srcip=%{IP:srcIP} srcport=%{NUMBER:srcPort} srcintf=%{DATA:srcInterface} dstip=%{IP:dstIP} dstport=%{NUMBER:dstPort} dstintf=%{DATA:dstInterface} poluuid=%{DATA:poluuid} sessionid=%{NUMBER:sessionid} proto=%{INT:protocol} action=%{WORD:action} policyid=%{INT:policyId} policytype=%{WORD:policytype} dstcountry=%{QS:dstcountry} srccountry=%{QS:dstcountry} trandisp=%{WORD:trandisp} transip=%{IP:transip} transport=%{NUMBER:transport} service=%{QS:service} appcat=%{QS:appcat} applist=%{QS:applist} duration=%{NUMBER:duration} sentbyte=%{NUMBER:entbyte} rcvdbyte=%{NUMBER:rcvdbyte} sentpkt=%{NUMBER:sentpkt} rcvdpkt=%{NUMBER:rcvdpkt} wanin=%{NUMBER:wanin} wanout=%{NUMBER:wanout} lanin=%{NUMBER:lanin} lanout=%{NUMBER:lanout}

」

將「結果」貼到「Filter」內「grok」

filter {           

   grok { 

                  match => [

                  "message", "%{SYSLOG5424PRI}date=20%{DATE} time=%{TIME:time} devname=%{HOSTNAME:devName} devid=%{HOSTNAME:devId} logid=%{NUMBER:logId} type=%{WORD:logType} subtype=%{WORD:logSubType} level=%{WORD:level} vd=%{DATA:virtualDomain} srcip=%{IP:srcIP} srcport=%{NUMBER:srcPort} srcintf=%{DATA:srcInterface} dstip=%{IP:dstIP} dstport=%{NUMBER:dstPort} dstintf=%{DATA:dstInterface} poluuid=%{DATA:poluuid} sessionid=%{NUMBER:sessionid} proto=%{INT:protocol}

action=%{WORD:action} policyid=%{INT:policyId} policytype=%{WORD:policytype} dstcountry=%{QS:dstcountry} srccountry=%{QS:dstcountry} trandisp=%{WORD:trandisp} transip=%{IP:transip} transport=%{NUMBER:transport} service=%{QS:service} appcat=%{QS:appcat} applist=%{QS:applist} duration=%{NUMBER:duration} sentbyte=%{NUMBER:entbyte} rcvdbyte=%{NUMBER:rcvdbyte} sentpkt=%{NUMBER:sentpkt} rcvdpkt=%{NUMBER:rcvdpkt} wanin=%{NUMBER:wanin} wanout=%{NUMBER:wanout} lanin=%{NUMBER:lanin} lanout=%{NUMBER:lanout}"

                           ]

                }

        }

最終結果如下

重新啟動「ELK」(logstash)

找到 (logstash)「5043」端口「服務」

netstat -ltunp | grep "5043"

強制結束「PID」程序

kill -9 16741

重新啟動「logstash」

nohup /opt/logstash-5.4.0/bin/logstash -f /opt/logstash-5.4.0/config/nginx.yml &

登入「ELK」Web UI (Kibana) 更新「字段」

終於完成..這麼複雜的方法..我開始懷念「Splunk」好用的工具。

參考資料

「料理佳餚」-「ELK 用 syslog 插件收集網路設備的 Log」
「Logstash」-「How to use multiple geoip filter in one message」