一、ELK介紹
官網(wǎng)簡(jiǎn)介
ELK是三個(gè)開(kāi)源軟件的縮寫(xiě),分別表示:Elasticsearch , Logstash, Kibana , 它們都是開(kāi)源軟件。新增了一個(gè)FileBeat,它是一個(gè)輕量級(jí)的日志收集處理工具(Agent),F(xiàn)ilebeat占用資源少,適合于在各個(gè)服務(wù)器上搜集日志后傳輸給Logstash,官方也推薦此工具。
Elasticsearch是個(gè)開(kāi)源分布式搜索引擎,提供搜集、分析、存儲(chǔ)數(shù)據(jù)三大功能。它的特點(diǎn)有:分布式,零配置,自動(dòng)發(fā)現(xiàn),索引自動(dòng)分片,索引副本機(jī)制,restful風(fēng)格接口,多數(shù)據(jù)源,自動(dòng)搜索負(fù)載等。
Logstash 主要是用來(lái)日志的搜集、分析、過(guò)濾日志的工具,支持大量的數(shù)據(jù)獲取方式。一般工作方式為c/s架構(gòu),client端安裝在需要收集日志的主機(jī)上,server端負(fù)責(zé)將收到的各節(jié)點(diǎn)日志進(jìn)行過(guò)濾、修改等操作在一并發(fā)往elasticsearch上去。
Kibana 也是一個(gè)開(kāi)源和免費(fèi)的工具,Kibana可以為 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面,可以幫助匯總、分析和搜索重要數(shù)據(jù)日志。
Filebeat隸屬于Beats。目前Beats包含四種工具:
- Packetbeat(搜集網(wǎng)絡(luò)流量數(shù)據(jù))
- Topbeat(搜集系統(tǒng)、進(jìn)程和文件系統(tǒng)級(jí)別的 CPU 和內(nèi)存使用情況等數(shù)據(jù))
- Filebeat(搜集文件數(shù)據(jù)),公司經(jīng)常使用這個(gè)工具來(lái)收集文件日志內(nèi)容
- Winlogbeat(搜集 Windows 事件日志數(shù)據(jù))
系統(tǒng)日志的搜集現(xiàn)在主流采用 FileBeat的方式進(jìn)行搜集。
官方文檔
Filebeat
https://www.elastic.co/cn/products/beats/filebeat
https://www.elastic.co/guide/en/beats/filebeat/5.6/index.html
Logstash
https://www.elastic.co/cn/products/logstash
https://www.elastic.co/guide/en/logstash/5.6/index.html
Kibana
https://www.elastic.co/cn/products/kibana
https://www.elastic.co/guide/en/kibana/5.5/index.html
Elasticsearch
https://www.elastic.co/cn/products/elasticsearch
https://www.elastic.co/guide/en/elasticsearch/reference/5.6/index.html
elasticsearch中文社區(qū)
https://elasticsearch.cn/

各服務(wù)器部署filebeat收集系統(tǒng)日志,將收集的日志放到redis里面緩存,然后logstash收集redis推送的日志內(nèi)容,過(guò)濾后存入es,然后在kibana web上面展示查看。logstash也可以將數(shù)據(jù)直接推送到數(shù)據(jù)庫(kù)中存儲(chǔ)。
二、各組件的工作原理
采摘自官網(wǎng)
Filebeat工作原理
Filebeat由兩個(gè)主要組件組成:prospectors 和 harvesters。這兩個(gè)組件協(xié)同工作將文件變動(dòng)發(fā)送到指定的輸出中。

1. Harvester(收割機(jī))
負(fù)責(zé)讀取單個(gè)文件內(nèi)容。每個(gè)文件會(huì)啟動(dòng)一個(gè)Harvester,每個(gè)Harvester會(huì)逐行讀取各個(gè)文件,并將文件內(nèi)容發(fā)送到制定輸出中。Harvester負(fù)責(zé)打開(kāi)和關(guān)閉文件,意味在Harvester運(yùn)行的時(shí)候,文件描述符處于打開(kāi)狀態(tài),如果文件在收集中被重命名或者被刪除,F(xiàn)ilebeat會(huì)繼續(xù)讀取此文件。所以在Harvester關(guān)閉之前,磁盤(pán)不會(huì)被釋放。默認(rèn)情況filebeat會(huì)保持文件打開(kāi)的狀態(tài),直到達(dá)到close_inactive(如果此選項(xiàng)開(kāi)啟,filebeat會(huì)在指定時(shí)間內(nèi)將不再更新的文件句柄關(guān)閉,時(shí)間從harvester讀取最后一行的時(shí)間開(kāi)始計(jì)時(shí)。若文件句柄被關(guān)閉后,文件發(fā)生變化,則會(huì)啟動(dòng)一個(gè)新的harvester。關(guān)閉文件句柄的時(shí)間不取決于文件的修改時(shí)間,若此參數(shù)配置不當(dāng),則可能發(fā)生日志不實(shí)時(shí)的情況,由scan_frequency參數(shù)決定,默認(rèn)10s。Harvester使用內(nèi)部時(shí)間戳來(lái)記錄文件最后被收集的時(shí)間。例如:設(shè)置5m,則在Harvester讀取文件的最后一行之后,開(kāi)始倒計(jì)時(shí)5分鐘,若5分鐘內(nèi)文件無(wú)變化,則關(guān)閉文件句柄。默認(rèn)5m)。
2. Prospector(勘測(cè)者)
負(fù)責(zé)管理Harvester并找到所有讀取源。
filebeat.prospectors:
- input_type: log
paths:
- /apps/logs/*/info.log
Prospector會(huì)找到/apps/logs/*目錄下的所有info.log文件,并為每個(gè)文件啟動(dòng)一個(gè)Harvester。Prospector會(huì)檢查每個(gè)文件,看Harvester是否已經(jīng)啟動(dòng),是否需要啟動(dòng),或者文件是否可以忽略。若Harvester關(guān)閉,只有在文件大小發(fā)生變化的時(shí)候Prospector才會(huì)執(zhí)行檢查。只能檢測(cè)本地的文件。
Filebeat如何記錄文件狀態(tài)
將文件狀態(tài)記錄在文件中(默認(rèn)在/var/lib/filebeat/registry)。此狀態(tài)可以記住Harvester收集文件的偏移量。若連接不上輸出設(shè)備,如ES等,filebeat會(huì)記錄發(fā)送前的最后一行,并再可以連接的時(shí)候繼續(xù)發(fā)送。Filebeat在運(yùn)行的時(shí)候,Prospector狀態(tài)會(huì)被記錄在內(nèi)存中。Filebeat重啟的時(shí)候,利用registry記錄的狀態(tài)來(lái)進(jìn)行重建,用來(lái)還原到重啟之前的狀態(tài)。每個(gè)Prospector會(huì)為每個(gè)找到的文件記錄一個(gè)狀態(tài),對(duì)于每個(gè)文件,F(xiàn)ilebeat存儲(chǔ)唯一標(biāo)識(shí)符以檢測(cè)文件是否先前被收集。
Filebeat如何保證事件至少被輸出一次
Filebeat之所以能保證事件至少被傳遞到配置的輸出一次,沒(méi)有數(shù)據(jù)丟失,是因?yàn)閒ilebeat將每個(gè)事件的傳遞狀態(tài)保存在文件中。在未得到輸出方確認(rèn)時(shí),filebeat會(huì)嘗試一直發(fā)送,直到得到回應(yīng)。若filebeat在傳輸過(guò)程中被關(guān)閉,則不會(huì)再關(guān)閉之前確認(rèn)所有時(shí)事件。任何在filebeat關(guān)閉之前為確認(rèn)的時(shí)間,都會(huì)在filebeat重啟之后重新發(fā)送。這可確保至少發(fā)送一次,但有可能會(huì)重復(fù)??赏ㄟ^(guò)設(shè)置shutdown_timeout 參數(shù)來(lái)設(shè)置關(guān)閉之前的等待事件回應(yīng)的時(shí)間(默認(rèn)禁用)。
Logstash工作原理
Logstash事件處理有三個(gè)階段:inputs → filters → outputs。是一個(gè)接收,處理,轉(zhuǎn)發(fā)日志的工具。支持系統(tǒng)日志,webserver日志,錯(cuò)誤日志,應(yīng)用日志,總之包括所有可以拋出來(lái)的日志類型。

1. Input:輸入數(shù)據(jù)到logstash
一些常用的輸入為:
- file:從文件系統(tǒng)的文件中讀取,類似于tial -f命令
- syslog:在514端口上監(jiān)聽(tīng)系統(tǒng)日志消息,并根據(jù)RFC3164標(biāo)準(zhǔn)進(jìn)行解析
- redis:從redis service中讀取
- beats:從filebeat中讀取
Filters:數(shù)據(jù)中間處理,對(duì)數(shù)據(jù)進(jìn)行操作
一些常用的過(guò)濾器為:
- grok:解析任意文本數(shù)據(jù),Grok 是 Logstash 最重要的插件。它的主要作用就是將文本格式的字符串,轉(zhuǎn)換成為具體的結(jié)構(gòu)化的數(shù)據(jù),配合正則表達(dá)式使用。內(nèi)置120多個(gè)解析語(yǔ)法。
- mutate:對(duì)字段進(jìn)行轉(zhuǎn)換。例如對(duì)字段進(jìn)行刪除、替換、修改、重命名等。
- drop:丟棄一部分events不進(jìn)行處理。
- clone:拷貝 event,這個(gè)過(guò)程中也可以添加或移除字段。
- geoip:添加地理信息(為前臺(tái)kibana圖形化展示使用)
2. Outputs:outputs是logstash處理管道的最末端組件。
一個(gè)event可以在處理過(guò)程中經(jīng)過(guò)多重輸出,但是一旦所有的outputs都執(zhí)行結(jié)束,這個(gè)event也就完成生命周期。
一些常見(jiàn)的outputs為:
- elasticsearch:可以高效的保存數(shù)據(jù),并且能夠方便和簡(jiǎn)單的進(jìn)行查詢。
- file:將event數(shù)據(jù)保存到文件中。
- graphite:將event數(shù)據(jù)發(fā)送到圖形化組件中,一個(gè)很流行的開(kāi)源存儲(chǔ)圖形化展示的組件。
- Codecs:codecs 是基于數(shù)據(jù)流的過(guò)濾器,它可以作為input,output的一部分配置。
Codecs可以幫助你輕松的分割發(fā)送過(guò)來(lái)已經(jīng)被序列化的數(shù)據(jù)。
一些常見(jiàn)的codecs:
- json:使用json格式對(duì)數(shù)據(jù)進(jìn)行編碼/解碼。
- multiline:將匯多個(gè)事件中數(shù)據(jù)匯總為一個(gè)單一的行。比如:java異常信息和堆棧信息。
Redis在架構(gòu)里的作用
- 存儲(chǔ)日志,可以是nginx,apache,tomcat等其他只要產(chǎn)生都可以存儲(chǔ),只要打上標(biāo)簽,logstash在input時(shí)就會(huì)分好類,就像京東快遞倉(cāng)庫(kù)一樣,把所有的快遞全部集中到一起,只要寫(xiě)好目標(biāo)地址,就會(huì)送到你手里
- 提高冗余性,若redis后面的全部宕機(jī)了,也不至于數(shù)據(jù)丟失
- 加快日志的讀取速度
- 全部日志集中一起,打好標(biāo)簽,便于操作管理
三、ELK安裝部署

1. Redis安裝部署
參考其他資料
2. Filebeat安裝配置
1、 下載并解壓縮
wget [https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.3.2-linux-x86_64.tar.gz](https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.3.2-linux-x86_64.tar.gz)
tar -zxvf filebeat-6.2.4-linux-x86_64.tar.gz
2、配置日志的輸入輸出
示例:
input
- type: log
enabled: true
paths:
- D:\idea\apache-tomcat-8.5.35\bin\logs\duty-web\infoFile.log
include_lines: ['userOperationLog']
tags: ["duty_audit"]
fields:
type: web_log
- type: log
enabled: true
paths:
- D:\idea\apache-tomcat-8.5.35\bin\logs\police-web\infoFile.log
include_lines: ['userOperationLog']
tags: ["police_audit"]
fields:
type: web_log
fields_under_root: true
output.redis
output.redis:
hosts: ["192.168.10.1"]
key: web_log
datatype: list
將多個(gè)系統(tǒng)日志文件 都推送到 redis中
3.Logstash安裝配置
解壓
tar -zxvf /home/wyk/logstash-7.7.0.tar.gz -C /opt/app/
解壓即是安裝結(jié)束,下面是個(gè)小例子:
新增一個(gè)配置文件: 讀取日志文件打印到窗口中
input {
redis {
host => "192.168.10.1"
port => 6379
db => "0"
data_type => "list"
key => web_log
type => web_log
}
}
filter {
mutate {
remove_field => ["@version","@timestamp","host","agent","offset","log","input","ecs","fields"]
gsub => [ "message"," "," "]
gsub => [ "message","="," "]
gsub => [ "message","null",'""']
}
grok {
match => {
"message" => "%{DATESTAMP:date} %{LOGLEVEL:level} %{NOTSPACE:class} %{NOTSPACE:logresult} - %{NOTSPACE:log_type} %{GREEDYDATA:json_log}"
}
}
json {
source => "json_log"
}
date {
match => [ "operationStarttime", "yyyy-MM-dd HH:mm:ss" ]
target => "operationStarttime_date"
}
date {
match => [ "operationEndtime", "yyyy-MM-dd HH:mm:ss"]
target => "operationEndtime_date"
}
mutate {
remove_field => ["date","level","class","logresult","log_type","message"]
}
}
output {
stdout { codec => rubydebug }
if [type] == "web_log" {
if [tags][0] == "duty_audit" {
elasticsearch {
hosts => ["192.168.10.1:9200"]
index => "duty_audit"
}
}
}
if [type] == "web_log" {
if [tags][0] == "police_audit" {
elasticsearch {
hosts => ["192.168.10.1:9200"]
index => "police_audit"
}
}
}
}
執(zhí)行文件啟動(dòng)
bin/logstash -f config/logstash-csdn.yml
也可以按日期配置es索引
elasticsearch {
hosts => ["192.168.10.1:9200"]
index => "police_audit-%{+YYYY.MM.dd}"
}
因?yàn)镋S保存日志是永久保存,所以需要定期刪除一下日志,下面命令為刪除指定時(shí)間前的日志
curl -X DELETE http://xx.xx.com:9200/logstash-*-`date +%Y-%m-%d -d "-$n days"`
4.ES安裝配置
解壓
tar zxvf elasticsearch-7.7.0-linux-x86_64.tar.gz -C /opt/app/
創(chuàng)建data和logs目錄
mkdir /opt/app/elasticsearch-7.7.0/data
mkdir /opt/app/elasticsearch-7.7.0/logs
配置文件
vim /opt/app/elasticsearch-7.7.0/config/elasticsearch.yml
修改es配置
cluster.name: MyES #集群名稱
node.name: node01 #本節(jié)點(diǎn)名稱
path.data: /opt/app/elasticsearch-7.7.0/data #數(shù)據(jù)存儲(chǔ)目錄
path.logs: /opt/app/elasticsearch-7.7.0/logs #日志存儲(chǔ)目錄
network.host: 0.0.0.0 #所有機(jī)器都可監(jiān)聽(tīng)
http.port: 9200 #默認(rèn)端口
cluster.initial_master_nodes: ["node01"] #主節(jié)點(diǎn)名稱,與上面配置的保持一致
JVM配置
因?yàn)镋S是基于Java開(kāi)發(fā)的,所以依賴于JVM,有需求可以修改JVM屬性:
vim /opt/app/elasticsearch-7.7.0/config/jvm.options
-Xms4g
-Xmx4g
創(chuàng)建elk用戶
# 創(chuàng)建用戶
useradd elk
# 為用戶設(shè)置密碼
passwd elk
# 刪除用戶
userdel -r elk
# 配置用戶的文件夾權(quán)限
chown -R elk:elk elasticsearch-7.7.0
啟動(dòng)
su - elk
$ES_HOME/bin/elasticsearch
報(bào)錯(cuò)
ERROR: [3] bootstrap checks failed
[1]: max file descriptors [4096] for elasticsearch process is too low, increase to at least [65535]
[2]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]
ERROR: Elasticsearch did not exit normally - check the logs at /opt/app/elasticsearch-7.7.0/logs/MyES.log
查看機(jī)器限制
ulimit -Sn
ulimit -Hn
修改限制:
vim /etc/security/limits.conf
* soft nofile 65535
* hard nofile 65535
elk soft nproc 4096
elk hard nproc 4096
執(zhí)行命令sysctl -p另其生效
sysctl -p
切換elk用戶再次啟動(dòng)ES,在瀏覽器中輸入node01:9200驗(yàn)證,出現(xiàn)下面的結(jié)果說(shuō)明已正常安裝并啟動(dòng):
啟停腳本
編寫(xiě)啟停腳本 start-es.sh以及stop-es.sh,編寫(xiě)后記得切換到elk用戶刷新一下環(huán)境變量,否則會(huì)找不到$ES_HOME:
start-es.sh
sudo -u elk nohup $ES_HOME/bin/elasticsearch >> $ES_HOME/output.log 2>&1 &
stop-es.sh
#!/bin/bash
PROCESS=`ps -ef | grep 'elasticsearch-7.7.0' |grep -v grep | grep -v PPID | awk '{ print $2}'`
echo $PROCESS
for i in $PROCESS
do
echo "Kill the elasticsearch process [ $i ]"
kill -9 $i
done
1.es啟動(dòng)不能使用root用戶
2.es-header優(yōu)先推薦使用google瀏覽器插件安裝
5. Kibana安裝使用
解壓
tar -zxvf /home/wyk/kibana-7.7.0-linux-x86_64.tar.gz -C /opt/app/
配置
vim config/kibana.yml
server.port: 5601
server.host: "0.0.0.0"
elasticsearch.hosts: ["http://localhost:9200"] #ES所在的ip
#elasticsearch.username: "kibana" #如果ES有配置用戶則這里需要配置用戶密碼
#elasticsearch.password: "123456"
給用戶分配文件夾操作權(quán)限
chown -R elk:elk /opt/app/kibana-7.7.0-linux-x86_64
啟動(dòng)
$KB_HOME/bin/kibana
后臺(tái)啟動(dòng),啟動(dòng)后會(huì)監(jiān)聽(tīng)5601端口
nohup ./kibana &
啟停腳本
start-kb.sh
#!/bin/bash
sudo -u elk nohup $KB_HOME/bin/kibana > $KB_HOME/output.log 2>&1 &
stop-kb.sh
#!/bin/bash
PROCESS=`ps -ef | grep 'kibana' |grep -v grep | grep -v PPID | awk '{ print $2}'`
echo $PROCESS
for i in $PROCESS
do
echo "Kill the kibana process [ $i ]"
kill -9 $i
done
四、日志格式
為實(shí)現(xiàn)對(duì)應(yīng)用系統(tǒng)用戶操作日志的收集、記錄、分析并最終實(shí)現(xiàn)審計(jì)功能,需要應(yīng)用系統(tǒng)按照約定的格式、內(nèi)容、和編碼方式將應(yīng)用系統(tǒng)用戶的操作日志輸出到指定目錄文件中。同時(shí),為了在應(yīng)用系統(tǒng)出現(xiàn)運(yùn)行故障,無(wú)法正常運(yùn)行時(shí),現(xiàn)場(chǎng)技術(shù)人員能夠借助ELK平臺(tái)快速檢索錯(cuò)誤日志,及時(shí)排查和解決系統(tǒng)問(wèn)題,需要應(yīng)用系統(tǒng)將錯(cuò)誤日志照約定的格式、內(nèi)容、和編碼方式輸出到指定目錄文件中。
1、 輸出路徑:
????Tomcat應(yīng)用服務(wù)器部署路徑/EHL_logs/應(yīng)用系統(tǒng)標(biāo)識(shí)/infoFile.log
????應(yīng)用系統(tǒng)標(biāo)識(shí)應(yīng)唯一標(biāo)識(shí)系統(tǒng)。
????示例:tomcat8.0/EHL_logs/EHL_EW/infoFile.log。
2、 日志文件編碼格式:UTF-8。
3、 輸出標(biāo)識(shí):
用戶操作日志添加前綴,與其他系統(tǒng)日志區(qū)別。
示例:userOperationLog={"requestIp":"10.2.113.44","userId":"001"……}。
4、 輸出格式:
????用戶操作日志內(nèi)容以json格式輸出。
????示例:2019/07/31 18:10:49 INFO LogAspect:77 userOperationLog={"logResult":"成功","logType":"QUERY",…}。

五、Kibana使用簡(jiǎn)要說(shuō)明
1. Stack Management
????Index management ????索引管理
????Index Patterns????索引檢索
2. Discover
????根據(jù)第一步增加的索引檢索,查詢索引日志
3. Dev tools
????Grok Debugger
????grok 語(yǔ)法測(cè)試,可以在這將輸出日志使用grok語(yǔ)法測(cè)試分割
六、Grok語(yǔ)法補(bǔ)充說(shuō)明
在線文檔
ELK Stack權(quán)威指南
https://www.bookstack.cn/read/logstash-best-practice-cn/README.md
logstash插件grok地址:
https://www.elastic.co/guide/en/logstash/7.6/plugins-filters-grok.html
正則在線調(diào)試:
http://grokdebug.herokuapp.com/
官方預(yù)定義的 grok 表達(dá)式:
https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns
grok·ELKstack中文指南:
https://elkguide.elasticsearch.cn/logstash/plugins/filter/grok.html
自定義grok表達(dá)式
%{HOSTNAME},匹配請(qǐng)求的主機(jī)名
%{TIMESTAMP_ISO8601:time},代表時(shí)間戳
%{LOGLEVEL},代表日志級(jí)別
%{URIPATHPARAM},代表請(qǐng)求路徑
%{INT},代表字符串整數(shù)數(shù)字大小
%{NUMBER}, 可以匹配整數(shù)或者小數(shù)
%{UUID},匹配類似091ece39-5444-44a1-9f1e-019a17286b48
%{IP}, 匹配ip
%{WORD}, 匹配請(qǐng)求的方式
%{GREEDYDATA},匹配所有剩余的數(shù)據(jù)
(?([\S+]*)),自定義正則
\s*或者\(yùn)s+,代表多個(gè)空格
\S+或者\(yùn)S*,代表多個(gè)字符
大括號(hào)里面:xxx,相當(dāng)于起別名
(?<class_info>([\S+]*)), 自定義正則匹配多個(gè)字符
logstash配置
filter {
grok {
match => { "message" =>"%{TIMESTAMP_ISO8601:timestamp}\s*\|\-%{LOGLEVEL:log_level}\s*\[%{DATA:thread}\]\s*(?m)(?<msg>.*|\s)" }
remove_field => ["message","_source"]
}
}
具體配置使用可在Kibana-->Grok Debugger中進(jìn)行語(yǔ)法測(cè)試,調(diào)試好后配置在logstash中