前言
最近公司分了個ELK相關的任務給我，在一邊學習一邊工作之余，總結下這些天來的學習歷程和踩坑記錄。
首先介紹下使用ELK的項目背景：在項目的數據庫里有個表用來存儲消息隊列的消費日志，這些日志用於開發者日后的維護。每當客戶端生產一條消息並發送到消息隊列后，就會插入一條對應的記錄到數據庫里。當這條消息被消費之后，又會更新數據庫里對應的記錄的幾個column的值，比如status、updated_on這些常用的column。
由於客戶每天生產消費的消息很多，導致數據庫里的這個表里的數據很多，長年累月下來，會達到數以億計。領導決定不再把這些消費日志保存到數據庫，而是改為通過Log4j2 + ELK架構把這些日志保存到Elasticsearch里。
ELK簡介
ELk是Elasticsearch + Logstash + Kibana的縮寫，ELK一般用來收集分布式架構下各個節點的日志，並進行統一地管理。
Elasticsearch是個開源分布式搜索引擎，提供搜集、分析、存儲數據三大功能。它的特點有：分布式，零配置，自動發現，索引自動分片，索引副本機制，restful風格接口，多數據源，自動搜索負載等。
Logstash主要是用來日志的搜集、分析、過濾日志的工具，支持大量的數據獲取方式。一般工作方式為c/s架構，client端安裝在需要收集日志的主機上，server端負責將收到的各節點日志進行過濾、修改等操作在一並發往elasticsearch上去。
Kibana也是一個開源和免費的工具，Kibana可以為Logstash和ElasticSearch提供的日志分析友好的Web界面，可以幫助匯總、分析和搜索重要數據日志。
上面的官方介紹可能會比較抽象，按我個人的理解，可以簡單將ELK理解為一個MVC架構的Java web應用：Elasticsearch對應M，Logstash對應C，Kibana對應V。
由於項目使用的是6.4.2版本的Elasticsearch，所以整個ELK都采用了同樣的版本6.4.2。這三個軟件都可以直接從官網下載到，下面是官網地址。
→ 官方下載地址
ELK的下載安裝與快速入門
本文只是基於Windows平台下，進行簡單的快速入門，先搭建好ELK框架並測試通過，后續文章再記錄更多的細節。
Elasticsearch 6.4.2
從官網下載了6.4.2版本的Elasticsearch的壓縮版后，解壓即可使用，使用默認的配置即可。
在Elasticsearch的安裝目錄下，進入/bin目錄，可以看到有兩個文件：
- elaticsearch
- elaticsearch.bat
這兩個文件都可以啟動Elasticsearch，暫時沒發現在Windows平台下通過這兩個文件啟動Elasticsearch有什么不同。我一般使用沒有后綴名的那個文件來啟動Elasticsearch。
啟動成功后，在瀏覽器輸入127.0.0.1:9200，如果訪問成功會反饋信息：
{
"name" : "erwbgE5",
"cluster_name" : "elasticsearch",
"cluster_uuid" : "QQvV3hBnSCSGsf-ycD3fng",
"version" : {
"number" : "6.4.2",
"build_flavor" : "default",
"build_type" : "zip",
"build_hash" : "04711c2",
"build_date" : "2018-09-26T13:34:09.098244Z",
"build_snapshot" : false,
"lucene_version" : "7.4.0",
"minimum_wire_compatibility_version" : "5.6.0",
"minimum_index_compatibility_version" : "5.0.0"
},
"tagline" : "You Know, for Search"
}
注意，如果使用Elasticsearch5.X及以上的版本，需要使用jdk 1.8；5.X以下版本使用jdk 1.6或1.7。
Logstash 6.4.2
同樣從官網下載6.4.2版本的Logstash安裝包，解壓之后進入/config目錄，創建一個配置文件tcp.conf，內容如下：
input {
stdin {
}
}
filter {
}
output {
stdout {
codec => rubydebug
}
}
接着進入/bin目錄，運行命令如下：
logstash -f ../config/test.conf
當看到Successfully started Logstash API endpoint的字眼時表示啟動成功，此時輸入任意字符，比如輸入hello，可以得到相應的反饋，如下：
{
"message" => "hello\r",
"@timestamp" => 2019-05-09T14:48:04.033Z,
"host" => "DESKTOP-S7HJJKD",
"@version" => "1"
}
這里解釋下，Logstash的配置非常簡單，就是一套流程：input -> filter -> output。
input用來收集信息，這里配置的是stdin插件，即標准輸入，也就是剛剛在控制台里輸入的字符串。
filter表示過濾信息，這里沒有進行任何過濾。
output表示輸出信息，這里配置的是stdout插件，即標准輸出，也就是將信息輸出到控制台上。這里的codec指明使用rubydebug作為編解碼器。
接着是運行的命令，使用了-f參數來指定使用某個配置文件。如果想要熱加載的效果，可以加上-r參數，這樣就可以在運行Logstash的時候去修改配置文件並自動重加載生效。這個-r參數等同於--config.reload.automatic。如下：
logstash -f ../config/test.conf -r
logstash -f ../config/test.conf --config.reload.automatic
注意，如果在輸入源里使用了stdin或者syslog等輸入插件，是不支持熱加載的，會一直報錯。
Kibana 6.4.2
從官網上下載Kibana6.4.2的壓縮包，解壓后即可使用。接着進入/bin目錄，運行kibana.bat。運行成功后，在瀏覽器輸入localhost:5601，即可訪問Kibana的頁面，之后就可以通過這個Kibana提供的web界面來對Elasticsearch里的文檔進行各種操作。
Logstash + Log4j2的快速搭建用例其一
配置tcp插件並啟動Logstash
修改之前創建的Logstash的配置文件test.config，內容如下：
input {
tcp {
mode => "server"
host => "127.0.0.1"
port => 4567
}
}
filter {
}
output {
stdout {
codec => rubydebug
}
}
然后運行命令logstash -f ../config/test.conf -r來啟動Logstash。由於我們這里通過-r來啟用了熱加載功能，所以可以在運行中直接修改配置並生效，比如修改input里的port。熱加載成功后會看到如下字眼：
Reloading pipeline {"pipeline.id"=>:main}
使用了Socket Appender的Log4j2項目demo
接着准備一個使用了Log4j2的項目demo，如下是一個測試類Test.java：
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
public class Test {
public static final Logger LOGGER = LogManager.getLogger("elk.test");
public static void main(final String[] args) {
LOGGER.info("Hello world!");
}
}
這里使用的是2.11.1版本的Log4j2，Maven依賴如下：
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.11.1</version>
</dependency>
接着是配置Log4j2的配置文件log4j2.xml，如下：
<?xml version="1.0" encoding="UTF-8"?>
<configuration xmlns:xi="http://www.w3.org/2001/XInclude" monitorInterval="30">
<Properties>
<Property name="LOG_PATTERN">{"logger": "%logger", "level": "%level", "msg": "%message"}%n</Property>
</Properties>
<Appenders>
<Console name="stdout" target="SYSTEM_OUT">
<PatternLayout pattern="${LOG_PATTERN}" />
</Console>
<Socket name="logstash-tcp" host="127.0.0.1" port="4567" protocol="TCP">
<PatternLayout pattern="${LOG_PATTERN}"/>
</Socket>
</Appenders>
<Loggers>
<Logger name="elk.test" level="info" additivity="false">
<AppenderRef ref="stdout" />
<AppenderRef ref="logstash-tcp" />
</Logger>
<Root level="error">
<AppenderRef ref="stdout" />
</Root>
</Loggers>
</configuration>
從配置文件中可以看到，這里使用的是Socket Appender來將日志打印的信息發送到Logstash。
注意了，Socket的Appender必須要配置到下面的Logger才能將日志輸出到Logstash里！
另外這里的host是部署了Logstash服務端的地址，並且端口號要和你在Logstash里配置的一致才行。
運行該項目demo，可以看到Logstash的控制台收集到了數據，如下：
{
"host" => "127.0.0.1",
"message" => "{\"logger\": \"elk.test\", \"level\": \"INFO\", \"msg\": \"Hello world!\"}\r",
"@timestamp" => 2019-05-09T16:20:35.940Z,
"@version" => "1",
"port" => 49781
}
注意
這里由於使用的是Socket方式來連接Logstash的服務端，如果在連接期間，Logstash的服務停止了或者斷掉了，就算接下來重啟了Logstash，項目工程也無法自動重新連接上Logstash，除非重啟項目工程。
在生產環境中，Logstash自然是有可能半路出問題重啟的，所以不能使用這種Socket方式來傳輸日志。
可以使用gelf的方式來傳輸日志到Logstash，用例如下所示。
Logstash + Log4j2的快速搭建用例其二
配置gelf插件並啟動Logstash
修改之前創建的Logstash的配置文件test.config，內容如下：
input {
gelf {
host => "127.0.0.1"
port => 4567
use_tcp => true
}
}
filter {
}
output {
stdout {
codec => rubydebug
}
}
運行命令logstash -f ../config/test.conf -r啟動Logstash。
在Log4j2項目中使用Gelf Appender
將之前的項目工程里的log4j2.xml的Socket Appender改為使用Gelf Appender，如下：
<?xml version="1.0" encoding="UTF-8"?>
<configuration xmlns:xi="http://www.w3.org/2001/XInclude" monitorInterval="30">
<Properties>
<Property name="LOG_PATTERN">{"logger": "%logger", "level": "%level", "msg": "%message"}%n</Property>
</Properties>
<Appenders>
<Console name="stdout" target="SYSTEM_OUT">
<PatternLayout pattern="${LOG_PATTERN}" />
</Console>
<Gelf name="logstash-gelf" host="tcp:localhost" port="4567" version="1.1" ignoreExceptions="true"
extractStackTrace="true" filterStackTrace="false">
<Field name="timestamp" pattern="%d{yyyy-MM-dd'T'HH:mm:ss.SSSZ}" />
<Field name="level" pattern="%level" />
<Field name="simpleClassName" pattern="%C{1}" />
<Field name="className" pattern="%C" />
<Field name="server" pattern="%host" />
</Gelf>
</Appenders>
<Loggers>
<Logger name="elk.test" level="info" additivity="false">
<AppenderRef ref="stdout" />
<AppenderRef ref="logstash-gelf" />
</Logger>
<Root level="error">
<AppenderRef ref="stdout" />
</Root>
</Loggers>
</configuration>
另外，這個Gelf Appender需要導入另一個依賴，如下：
<dependency>
<groupId>biz.paluch.logging</groupId>
<artifactId>logstash-gelf</artifactId>
<version>1.11.1</version>
</dependency>
接着運行項目工程，可以看到Logstash的控制台已經把收集到的日志打印出來了：
{
"message" => "Hello world!",
"@timestamp" => 2019-05-10T14:09:43.267Z,
"className" => "lewky.cn.Test",
"source_host" => "127.0.0.1",
"timestamp" => "2019-05-10T22:09:43.211+0800",
"simpleClassName" => "Test",
"facility" => "logstash-gelf",
"level" => "INFO",
"host" => "DESKTOP-S7HJJKD",
"@version" => "1",
"server" => "DESKTOP-S7HJJKD"
}
ELK + Log4j2快速搭建用例
接下來就可以把Logstash收集到的日志輸出到Elasticsearch，並通過Kibana顯示到界面上。
Logstash配置Elasticsearch插件
修改配置文件如下：
input {
gelf {
host => "127.0.0.1"
port => 4567
use_tcp => true
}
}
filter {
}
output {
stdout {
codec => rubydebug
}
elasticsearch {
hosts => ["127.0.0.1:9200"]
document_id => "%{docId}"
index => "%{indexName}"
}
}
output里添加了elasticsearch插件：
- hosts里配置Elasticsearch server的地址
- document_id是index到ES時使用的索引id
- index是index到ES是使用的索引名字
修改log4j2.xml和項目代碼
在項目的log4j2.xml里的Gelf Appender加上兩個個新的Field：indexName和docId，如下：
<Gelf name="logstash-gelf" host="tcp:localhost" port="4567" version="1.1" ignoreExceptions="true"
extractStackTrace="true" filterStackTrace="false">
<Field name="timestamp" pattern="%d{yyyy-MM-dd'T'HH:mm:ss.SSSZ}" />
<Field name="level" pattern="%level" />
<Field name="simpleClassName" pattern="%C{1}" />
<Field name="className" pattern="%C" />
<Field name="server" pattern="%host" />
<Field name="indexName" mdc="indexName" />
<Field name="docId" mdc="docId" />
</Gelf>
這里添加的兩個新的Field對應於上邊Logstash配置文件里的兩個變量，然后這里用到了mdc，這個是Log4j2里的ThreadContext的東西，有興趣可以去了解下Log4j2里的MDC和NDC。
接着修改測試類的代碼，如下：
public class Test {
public static final Logger LOGGER = LogManager.getLogger("elk.test");
public static void main(final String[] args) {
ThreadContext.put("docId", "1");
ThreadContext.put("indexName", "test");
LOGGER.info("Hello world!");
}
}
接着依次啟動ELK三個軟件，然后運行項目，可以發現Logstash控制台里收集到了日志信息：
{
"simpleClassName" => "Test",
"server" => "DESKTOP-S7HJJKD",
"message" => "Hello world!",
"@version" => "1",
"source_host" => "127.0.0.1",
"indexName" => "test",
"level" => "INFO",
"@timestamp" => 2019-05-11T16:40:14.996Z,
"facility" => "logstash-gelf",
"className" => "lewky.cn.Test",
"timestamp" => "2019-05-12T00:40:14.877+0800",
"docId" => 1,
"host" => "DESKTOP-S7HJJKD"
}
而在我們的IDE控制台（我用的是Eclipse）里也可以看到輸出了信息：
{"logger": "elk.test", "level": "INFO", "msg": "Hello world!"}
配置Kibana查看Elasticsearch的index數據
接下來就是最后一步了，通過Kibana來查看我們剛剛index到Elasticsearch里的數據。
啟動了Kibana后，在瀏覽器訪問localhost:5601，進入界面后，操作如下：
- Management -> Index Patterns
- 輸入index的名字，我們這里填的是
test；然后點擊Next step - 在
Time Filter field name下方的下拉框里選擇timestamp作為我們的一個排序字段，默認是desc，即遞減排序 - 最后點擊
Create index pattern
現在已經配置好了Index pattern，我們就可以直接在左側菜單欄里的Discover去查看對應的index里的數據了。如果不出意外，現在在Discover里已經看到剛剛被我們index進去的日志信息了。
默認只會顯示Time和_source兩個字段的數據，Time就是排序字段，它的值和之前我們選擇的那個timestamp一樣。_source里則是所有字段的數據總和。
可以根據需要，在顯示字段的左側把任意的字段add到右側以顯示出來。當你添加了新的字段之后，_source字段會自動消失。
這就是最簡單的一個ELK快速搭建例子，有興趣的可以接着看后續的文章以了解更多和ELK相關的問題或知識。