一、說明
ELK是當下流行的日志監控系統。ELK是Elasticsearch、Logstash、Kibana三個軟件的統稱。
在ELK日志監控系統中，Logstash負責讀取和結構化各類日志+發送給Elasticsearch，Elasticsearch負責存儲Logstash發送過來的日志+響應Kibana的查詢，Kibana負責從Elasticsearch查詢內容+在web界面中向用戶展示。
二、ELK安裝與啟動
ELK看是三個工具，但是都相當於”綠色軟件“，基本都是下載解壓就能運行了。
1.1 安裝jdk
下載地址：https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
1.2 安裝elasticsearch
下載地址：https://www.elastic.co/downloads/elasticsearch
tar -zxf elasticsearch-6.4.2.tar.gz cd elasticsearch-6.4.2/ # 以后台方式啟動 bin/elasticsearch -d -p pid.log # 啟動需要點時間，稍等一下再執行下一句 curl http://localhost:9200/
如果有如圖json結果返回，那就說明安裝成功。
可以弄個啟停腳本，省得每次啟停都得想一個需不需要帶參數帶什么參數：
# 啟動腳本 echo './elasticsearch -d -p pid.log' > start_elasticsearch chmod u+x start_elasticsearch # 停止腳本，直接用kill -9殺除進程 echo 'kill -9 `cat ../pid.log`' > stop_elasticsearch chmod u+x stop_elasticsearch
1.3 安裝logstash
下載地址：https://www.elastic.co/downloads/logstash
cd logstash-6.4.2/ cat > config/logstash.conf << EOF input { file { path => ["/var/dpkg.log"] start_position => "beginning" } } filter { if [path] =~ "access" { mutate { replace => { "type" => "apache_access" } } grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } } date { match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ] } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } } EOF nohup bin/logstash -f config/logstash.conf &
可以弄個啟停腳本，省得每次啟停都得想一個需不需要帶參數帶什么參數：
# 啟動腳本，配置文件名改成自己的 echo 'nohup ./logstash -f ../config/logstash.conf &' > start_logstash chmod u+x start_logstash # 殺除進程腳本 echo 'kill -9 `ps -ef |grep logstash |grep -v grep| awk '{print $2}'|head -n 1`' > stop_logstash chmod u+x stop_logstash
1.4 安裝kibana
下載地址：https://www.elastic.co/downloads/kibana
解壓后編緝config/kibana.yml，將server.host項修改為本機ip，將elasticsearch.url項賦值為elasticsearch地址。如我這里是：
server.host: "10.10.6.92" elasticsearch.url: "http://localhost:9200"
保存后啟動：
nohup bin/kibana &
啟動后默認監聽5601端口，使用http://{ip}:5601訪問即可，界面如下圖所示：
可以直接弄個啟停腳本，省得每次啟停都得想一個需不需要帶參數帶什么參數：
# 啟動腳本 echo 'nohup ./kibana &' > start_kibana chmod u+x start_kibana # 停止腳本，直接用kill -9殺除進程 echo 'kill -9 `ps -ef |grep ./../node/bin|grep -v grep|awk '{print $2}'|head -n 1`' > stop_kibana chmod u+x stop_kibana
三、ELK具體使用
在上一節中我們對ELK安裝配置並運行了起來，這節我們進一步探究ELK更高級的配置和用法。
3.1 logstash配置說明
本節主要參考頁面：https://www.elastic.co/guide/en/logstash/current/configuration-file-structure.html
logstash配置主要就是config/logstash.conf文件中的input/filter/output三大節區，其中input用於指示logstash從何處讀取，filter節區指示logstash如何結構化（格式化）輸入，output節區指示logstash輸出到哪里。這三大節區的成員都是插件，下邊分別講述各節區支持的插件及插件配置。
3.1.1 input節區插件及配置
input節區所有插件參見：https://www.elastic.co/guide/en/logstash/current/input-plugins.html
在1.3節中我們使用的就是file插件，這里就以file插件為例講解其為什么可以這么配置，還可以怎么配置。
在上邊插件頁面中點擊file插件鏈接即可進入file插件說明頁面，如下所示
進入后直接下接到"File Input Configuration Options"節（xxx插件就下拉找到xxx Input Configuration Options節）
Setting列----插件支持的配置項的名稱
Input type列----該配置項的值的格式。包插number（數字）、string（字符串）、array（數組）等幾種類型
Required列----該配置項是否必須配置。No表示非必須，Yes表示必須
往下拉可以看到我們前面file插件配置path和start_position，可以看到path是數組類型且是必須的，而start_position是字符串類型（只能是”beginning“或”end“）不是必須的。
再點擊配置項鏈接能跳轉到配置項的詳細說明。各插件都類似以上步驟查看即可。
input最常用的插件感覺是file和stdin，格式如下：
input { file { path => ["/home/ls/test.txt"] start_position => "beginning" } stdin { } }
3.1.2 filter節區插件及配置
filter節區所有插件參見：https://www.elastic.co/guide/en/logstash/current/filter-plugins.html
在1.3中我們配置了mutate、grok和date三個插件，多個插件組成一個插件鏈；比如這里輸入先被mutate處理，再被傳到grok處理，再被傳到date處理。
所謂的”處理“是一個關鍵，我們將config/logstash.conf修改為如下內容（表示從標控制台獲取輸入、不使用任何插件、輸出到控制台）並重新啟動logstash
input { stdin { } } filter { } output { stdout { codec => rubydebug } }
我們輸入如下一條http訪問日志：
127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] "GET /xampp/status.php HTTP/1.1" 200 3891 "http://cadenza/xampp/navi.php" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0"
如下{}中的json就是logstash的輸出，共message（信息原文）、@version（信息版本）、@timestamp（信息接收時間）、host（信息來源主機）四項。
首先，我們在filter{}中加入以下一個規則（指示以apache日志格式進行解析message項），並重啟logstash並再次輸入前邊的http日志
grok { match => { "message" => "%{COMBINEDAPACHELOG}" } }
輸出結果如下，可以看到在原來那四項的基礎上，增加了很多項
然后，我們在filter{}中再加入以下一個規則（指示以"dd/MMM/yyyy:HH:mm:ss Z"解析timestamp項，並重寫為"yyyy-MMM-dd HH:mm:ss Z"格式），並重啟logstash並再次輸入前邊的http日志
date { match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ,"yyyy-MMM-dd HH:mm:ss Z"] }
可以看到@timestamp項被成功按指定格式重寫
最后，在filter{}中再加入以下一個規則（指示新增一個叫new_filed的項，其值為固定值”this is a new field“），此時完整配置文件如下：
input { stdin { } } filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } date { match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ] } environment { add_field => { "new_field" => "this is a new field" } } } output { stdout { } }
由於environment插件默認沒有安裝我們先安裝一下（所有插件都可以使用logstash-plugin命令進行管理，類似yum或apt）
bin/logstash-plugin install logstash-filter-environment
此時重啟logstash並再次輸入前邊的http日志再次在控制台輸入上邊的http訪問日志，輸出結果如下。可以看到按預期新增了一個叫”new_field“的項
從如上分析可以得出以下結論：
logstash默認對傳來的信息進行初步處理，生成message（信息原文）、@version（信息版本）、@timestamp（信息接收時間）、host（信息來源主機）四個鍵
后面的插件前面的插件生成的結果，以鍵值對為操作單無，或是修改已有鍵值對或是新增鍵值對。
最后，所有filter插件都支持以下選項
filter最常用的是將非結構化數據進行結構化的grok，由於篇幅較長分離了出去，見后邊第四大點“從自定義文件解析IP並可視化”。
3.1.3 ouput節區插件及配置
output節區所有插件參見：https://www.elastic.co/guide/en/logstash/current/output-plugins.html
在ELK中最常用的就是我們1.3中設置的，輸出到elasticsearch
elasticsearch { hosts => ["localhost:9200"] }
3.2 elasticsearch配置說明
elasticsearch會自動完成數據接收，然后kibana來查詢時會自動響應，就單個來說沒有其他太多的配置。主要的配置是在集群方面（cluster）但集群沒研究所以這里就不介紹了。
另外一個是配聽公網地址，elasticsearch默認只監聽127.0.0.1，如果要監聽其他地址需要打開其config/ elasticsearch.yml，修改network.host的值。比直接監聽所有地址：
network.host: 0.0.0.0
但是不是隨便改完重啟就完了，elasticsearch默認認為當前是開發環境，如果修改network.host監聽127.0.0.1以外的地址，那么elasticsearch就認為是遷移到了生產環境。
在開發環境中如果這些項沒有按要求配置那么只視為異常，但到生產環境這些項如果沒按要求配置那將會被視為錯誤elasticsearch將拒絕啟動。也就是說要修改監聽地址成功，除了要修改network.host項還要保證這些項已按要求配置。
3.3 kibana使用操作
在2.4中我們只是把kibana運行起來，雖說kibana接收了elasticsearch的數據，但並沒有展示這些數據，下面我們就來介紹如何展示數據。
3.3.1 總體菜單
3.3.2 設置索引
3.3.3 Discover菜單
切換到Discover菜單，默認會顯示kinaba接收到的所有數據，”Available fields“那里是當前所有數據出現過的所有項，點擊它們可通過他們進行數據篩選。
3.3.4 Visualize菜單
即可看到所有kibana支持的圖表，我們這里以pie為例
Filter表示輸入過濾表達式，篩選出要參與圖表繪制的數據集，我們這里直接點下方建議的“logstash-*”(這是我們前面3.2建立索引的所有數據)
進入后正中央一個大環表示所有數據、左下方的“Split Slices”和“Split Chart”用於添加分割指標、左上方播放按鈕用於分割指標效果預覽，右上方“Save”菜單用於保存圖表。
所謂的分割指標其實主要就是我們logstash生成的好些項，把他們以大於、小於、等於某個值進行分類。
以下是一個建立完的效果圖
在可視化圖表中，最主要的事情就是兩件：一是指定如何分類（Buckets），二是指定以什么作為度量（Metrics）。
比如在上圖中我們以timestamp項以年（Yearly）為基准進行分類，然后以數量（Metrics中設置為Count）作為度量（具體到這里就是timestamp項各年出現的次數），形成了上邊的餅狀圖。
分類（Buckets）和度量（Metrics）一般都是先選Aggregation再選Field，再選Field的一些細化選項。
Aggregation----類型的意思比如是日期類型（date histogram）還是范圍類型（range），這只是一個大體的分類，隨便選選就行了。
Filed----就是前面說的logstash的那些項（當然logstash的項和kibana中的項只有在建索引后才能算是一回事，見下文4.3 kibana IP地理可視化），比如timestamp啦host啦，按這些項分類或排序或統計可視化圖表怎么建立和配置就很明了了。
Field的一些細化選項----就是一些細化限制，自己隨便多點點就知道了。
3.3.5 Dashboard
所謂儀表盤就是，就是可視化圖表的集合。用3.3.4的方法創建一個pie（命名為test pie）再創建一個bar（命名為test bar）我們這里以把它們加載到新建的儀表盤為例
全點擊加載上來
我們可以隨意改變加載進來的圖表的位置和大小，最后點擊“Save”保存即可。這就是所謂的儀表盤，感覺和zabbix是一樣的玩法。
四、從自定義文件解析IP並可視化
數據庫下載地址：https://dev.maxmind.com/geoip/geoip2/geolite2/
在3.1.2中我們說明了三個關鍵點：多個插件是以插件鏈形式運行的，后續插件以其前插件生成的項為操作單元，及如何查看插件文檔。但自己想了一下通過看文檔知道插件選項怎么寫這個答案是不能令人滿意的，同時grok解析自定義文本文檔是常見的需要我們就以此為例進行演示。
4.1 使用grok從自定義文本中提取ip、端口和協議
生成的內容形如下，我們使用grok插件編寫規則，解析出ip、端口、協議三項：
75.207.190.84:1530 udp
在config目錄下創建patterns文件夾，再在patterns下創建patterns.conf文件，在其下寫入以下內容（其實就相當於正則宏定義）：
正則表達式在線測試網址：http://tool.chinaz.com/regex/
grok解析表達式在線測試網址（推薦在這里測過了再寫）：http://grokdebug.herokuapp.com
MYIP (\d+\.+){3}\d+
MYPORT \d{1,5}
MYPROTO (t|u){1}(c|d){1}p
在logstash.conf中寫入以下內容（格式要點說明都在注釋中，另外注意按自己的修改文件路徑）：
input { stdin { } } filter { # 在真正使用時通常會接收來自不同地方不同格式的文件，此時就要限定插件對哪些來源啟用而對哪些來源不啟用，因為使用格式去解析不匹配的文件那是解析不了的就會報錯
# 后邊啟用些限制，只有path項中為test.txt文件才使用此規則。這里是在控制台調試所以先注釋掉 # if [path] =~ "test.txt" { grok { # 這里是自定義正則的文件夾 patterns_dir => ["/usr/myapp/elk/logstash-6.4.2/config/patterns"] # 這條規則的意思是，針對前面的“message”項的內容，使用"%{MYIP:myip}:%{MYPORT:myport} %{MYPROTO:myproto}"格式進行解析 # MYIP/MYPORT/MYPROTO我們自定義的正則規則，myip/myport/myproto三項表示其前正則解析出的內容賦給這些項 # =>后的規則不要求能解析完整個項，但一定要是從頭連續解析；比如可以只要match => { "message" => "%{MYIP:myip}} # 但不能是跳過端口：match => { "message" => "%{MYIP:myip} %{MYPROTO:myproto}"} # 其實的：和空格不是隨意的是因為不用正則的地方要原樣保留75.207.190.84:1530 udp中我們的正則沒處理其中的：和空格要原樣保留 match => { "message" => "%{MYIP:myip}:%{MYPORT:myport} %{MYPROTO:myproto}"} # 如果不想用文件夾也可以直接把規則寫在這里，比如上邊這條規則直接寫在這里的等價形式為 # match => { "message" => "(?<myip>(\d+\.+){3}\d+):(?<myport>(\d{1,5}) %(?<myproto>((t|u){1}(c|d){1}p)"} } # } } output { stdout { } }
在下圖中可以看到對於輸入“75.207.190.84:1530 udp”，ip、端口和協議都已按預期成功提取
另外我們還輸入了按%{MYIP:myip}:%{MYPORT:myport} %{MYPROTO:myproto}格式不能解析的“dd”，此時報錯[0] "_grokparsefailure"。這就是上方配置注釋中強調針對來源啟用插件的原因。
4.2 使用geoip將ip轉換為地理位置
首先要下載GeoLite2數據庫，下載地址：https://dev.maxmind.com/geoip/geoip2/geolite2/
這幾個的區別是精度的區別ASN精確到自治域，Country精確到國家，City精確到城市。我們直接下載城市版，然后解壓
按流程來講還是得要在stdin{}和stdout{}調試一翻，完后再寫真正配置，但這里為了節省篇幅混合在一起進行。
修改logstash.conf為以下內容，然后啟動logstash：
input { file { # 存放python生成的那些東西的文件的位置，根據自己的修改 path => ["/home/ls/test.txt"] start_position => "beginning" } } filter { # 只有path項中為test.txt文件才使用此規則 if [path] =~ "test.txt" { grok { # 這里是自定義正則的文件夾，根據自己的修改 patterns_dir => ["/usr/myapp/elk/logstash-6.4.2/config/patterns"] # 這條規則的意思是，針對前面的“message”項的內容，使用"%{MYIP:myip}:%{MYPORT:myport} %{MYPROTO:myproto}"格式進行解析 # MYIP/MYPORT/MYPROTO我們自定義的正則規則，myip/myport/myproto三項表示其前正則解析出的內容賦給這些項 # =>后的規則不要求能解析完整個項，但一定要是從頭連續解析；比如可以只要match => { "message" => "%{MYIP:myip}} # 但不能是跳過端口：match => { "message" => "%{MYIP:myip} %{MYPROTO:myproto}"} # 其實的：和空格不是隨意的是因為不用正則的地方要原樣保留75.207.190.84:1530 udp中我們的正則沒處理其中的：和空格要原樣保留 match => { "message" => "%{MYIP:myip}:%{MYPORT:myport} %{MYPROTO:myproto}"} # 如果不想用文件夾也可以直接把規則寫在這里，比如上邊這條規則直接寫在這里的等價形式為 # match => { "message" => "(?<myip>(\d+\.+){3}\d+):(?<myport>(\d{1,5}) %(?<myproto>((t|u){1}(c|d){1}p)"} } # 這里可以還是需要放if內進行限制，我們要對“myip”鍵進行處理，只有經過上邊grok處理的才會有myip鍵 # 其他消息沒有myip，處理時不會自動跳運而是一樣報[0] "_geoip_lookup_failure" geoip{ # 指定要進行地址轉換的字段----當然只能是ip字段，我們這里是myip source => "myip" # 下載的geoip數據庫文件的位置，根據自己的修改 database => "/usr/myapp/elk/logstash-6.4.2/config/GeoLite2-City_20181016/GeoLite2-City.mmdb" } } } output { elasticsearch { hosts => ["localhost:9200"] } # 輸出到控制台是為了調試用的，到真正配時注釋掉 stdout{} }
我們使用以下python代碼生成多一些和”75.207.190.84:1530 udp“格式一樣的數據，生成后保存到/home/ls/test.txt（保存到哪都可以只要和input中file插件設置的位置一致）
for i in range(100): print(f"{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}:{random.randint(1000,2000)} tcp") print(f"{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}:{random.randint(1000,2000)} udp")
由於我們輸出到stdout{}(用nohub啟動的可以查看nohub.out)，此時可以在控制台看到類似以下輸出。“110.40.27.195”這個IP被成功解析為中國北京。
4.3 kibana IP地理可視化
在第三大節中我們創建了index，所謂index就是給logstash生成的那些鍵在kibana中設定為索引。現在myip等項和geoip的各項都是新加進來的，在kibana中就還沒有對應索引所以我們需要重新建一個索引。
對於任何新增的項，都需要新建index才能將將其作為可視化圖表中的“維”來使用。
取定好數據集，我這里取定為logstash-2018.10.*，除了昨天和今天的數據集logstash-2018.10.18和logstash-2018.10.19，明天傳來的數據會保存為logstash-2018.10.20，由於能匹配上所以也會自動包含進去。11月的數據為logstash-2018.11.x由於匹配不上所以就不會被包含了，這就是所謂“index pattern”的意思。
此時我們再到Visualize創建一個Region map
選用新建的索引
如圖所示，我們以國家代碼（geoip.country_code2）進行分類，然后以數量為其度量，就生成了以下圖表
參考：
https://blog.csdn.net/KingBoyWorld/article/details/78555120
https://blog.csdn.net/kingboyworld/article/details/77814528
https://www.elastic.co/downloads/logstash