ELK是什么
ELK是elastic公司提供的一套完整的日志收集以及前端展示的解決方案，是三個產品的首字母縮寫，分別是ElasticSearch、Logstash和Kibana。
其中Logstash負責對日志進行處理，如日志的過濾、日志的格式化等；ElasticSearch具有強大的文本搜索能力，因此作為日志的存儲容器；而Kibana負責前端的展示。
ELK搭建架構如下圖：
加入了filebeat用於從不同的客戶端收集日志，然后傳遞到Logstash統一處理。
ELK的搭建
因為ELK是三個產品，可以選擇依次安裝這三個產品。
這里選擇使用Docker安裝ELk。
Docker安裝ELk也可以選擇分別下載這三個產品的鏡像並運行，但是本次使用直接下載elk的三合一鏡像來安裝。
因此首先要保證已經有了Docker的運行環境，Docker運行環境的搭建請查看：https://blog.csdn.net/qq1311256696/article/details/85277220
拉取鏡像
有了Docker環境之后，在服務器運行命令：
docker pull sebp/elk
這個命令是在從Docker倉庫下載elk三合一的鏡像，總大小為2個多G，如果發現下載速度過慢，可以將Docker倉庫源地址替換為國內源地址。
下載完成之后，查看鏡像：
docker images
Logstash配置
在/usr/config/logstash目錄下新建beats-input.conf，用於日志的輸入：
input {
beats {
port => 5044
}
}
新建output.conf，用於日志由Logstash到ElasticSearch的輸出：
output {
elasticsearch {
hosts => ["localhost"]
manage_template => false
index => "%{[@metadata][beat]}"
}
}
其中的index為輸出到ElasticSearch后的index。
運行容器
有了鏡像之后直接啟動即可：
docker run -d -p 5044:5044 -p 5601:5601 -p 9203:9200 -p 9303:9300 -v /var/data/elk:/var/lib/elasticsearch -v /usr/config/logstash:/etc/logstash/conf.d --name=elk sebp/elk
-d的意思是后台運行容器；
-p的意思是宿主機端口:容器端口，即將容器中使用的端口映射到宿主機上的某個端口，ElasticSearch的默認端口是9200和9300，由於我的機器上已經運行了3台ElasticSearch實例，因此此處將映射端口進行了修改；
-v的意思是宿主機的文件|文件夾:容器的文件|文件夾，此處將容器中elasticsearch 的數據掛載到宿主機的/var/data/elk上，以防容器重啟后數據的丟失；並且將logstash的配置文件掛載到宿主機的/usr/config/logstash目錄。
--name的意思是給容器命名，命名是為了之后操作容器更加方便。
如果你之前搭建過ElasticSearch的話，會發現搭建的過程中有各種錯誤，但是使用docker搭建elk的過程中並沒有出現那些錯誤。
運行后查看容器：
docker ps
查看容器日志：
docker logs -f elk
進入容器：
docker exec -it elk /bin/bash
修改配置后重啟容器：
docker restart elk
查看kinaba
瀏覽器輸入http://my_host:5601/
即可看到kinaba界面。此時ElasticSearch中還沒有數據，需要安裝Filebeat采集數據到elk中。
Filebeat搭建
Filebeat用於采集數據並上報到Logstash或者ElasticSearch，在需要采集日志的服務器上下載Filebeat並解壓即可使用
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.2.1-linux-x86_64.tar.gz
tar -zxvf filebeat-6.2.1-linux-x86_64.tar.gz
修改配置文件
進入filebeat，修改filebeat.yml。
filebeat.prospectors:
- type: log
#需要設置為true配置才能生效
enabled: true
path:
#配置需要采集的日志路徑
- /var/log/*.log
#可以打一個tag以后分類使用
tag: ["my_tag"]
#對應ElasticSearch的type
document_type: my_type
setup.kibana:
#此處為kibana的ip及端口，即kibana:5601
host: ""
output.logstash:
#此處為logstash的ip及端口，即logstash:5044
host: [""]
#需要設置為true，否則不生效
enabled: true
#如果想直接從Filebeat采集數據到ElasticSearch，則可以配置output.elasticsearch的相關配置
運行Filebeat
運行：
./filebeat -e -c filebeat.yml -d "publish"
此時可以看到Filebeat會將配置的path下的log發送到Logstash；然后在elk中，Logstash處理完數據之后就會發送到ElasticSearch。但我們想做的是通過elk進行數據分析，因此導入到ElasticSearch的數據必須是JSON格式的。
這是之前我的單條日志的格式：
2019-10-22 10:44:03.441 INFO rmjk.interceptors.IPInterceptor Line:248 - {"clientType":"1","deCode":"0fbd93a286533d071","eaType":2,"eaid":191970823383420928,"ip":"xx.xx.xx.xx","model":"HONOR STF-AL10","osType":"9","path":"/applicationEnter","result":5,"session":"ef0a5c4bca424194b29e2ff31632ee5c","timestamp":1571712242326,"uid":"130605789659402240","v":"2.2.4"}
導入之后不好分析，之后又想到使用Logstash的filter中的grok來處理日志使之變成JSON格式之后再導入到ElasticSearch中，但是由於我的日志中的參數是不固定的，發現難度太大了，於是轉而使用Logback，將日志直接格式化成JSON之后，再由Filebeat發送。
Logback配置
我的項目是Spring Boot，在項目中加入依賴：
<dependency>
<groupId>net.logstash.logback</groupId>
<artifactId>logstash-logback-encoder</artifactId>
<version>5.2</version>
</dependency>
然后在項目中的resource目錄下加入logback.xml：
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<!--
說明：
1、日志級別及文件
日志記錄采用分級記錄，級別與日志文件名相對應，不同級別的日志信息記錄到不同的日志文件中
例如：error級別記錄到log_error_xxx.log或log_error.log（該文件為當前記錄的日志文件），而log_error_xxx.log為歸檔日志，
日志文件按日期記錄，同一天內，若日志文件大小等於或大於2M，則按0、1、2...順序分別命名
例如log-level-2013-12-21.0.log
其它級別的日志也是如此。
2、文件路徑
若開發、測試用，在Eclipse中運行項目，則到Eclipse的安裝路徑查找logs文件夾，以相對路徑../logs。
若部署到Tomcat下，則在Tomcat下的logs文件中
3、Appender
FILEERROR對應error級別，文件名以log-error-xxx.log形式命名
FILEWARN對應warn級別，文件名以log-warn-xxx.log形式命名
FILEINFO對應info級別，文件名以log-info-xxx.log形式命名
FILEDEBUG對應debug級別，文件名以log-debug-xxx.log形式命名
stdout將日志信息輸出到控制上，為方便開發測試使用
-->
<contextName>service</contextName>
<property name="LOG_PATH" value="logs"/>
<!--設置系統日志目錄-->
<property name="APPDIR" value="doctor"/>
<!-- 日志記錄器，日期滾動記錄 -->
<appender name="FILEERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 正在記錄的日志文件的路徑及文件名 -->
<file>${LOG_PATH}/${APPDIR}/log_error.log</file>
<!-- 日志記錄器的滾動策略，按日期，按大小記錄 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 歸檔的日志文件的路徑，例如今天是2013-12-21日志，當前寫的日志文件路徑為file節點指定，可以將此文件與file指定文件路徑設置為不同路徑，從而將當前日志文件或歸檔日志文件置不同的目錄。
而2013-12-21的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
<fileNamePattern>${LOG_PATH}/${APPDIR}/error/log-error-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<!-- 除按日志記錄之外，還配置了日志文件不能超過2M，若超過2M，日志文件會以索引0開始，
命名日志文件，例如log-error-2013-12-21.0.log -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<maxFileSize>2MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 追加方式記錄日志 -->
<append>true</append>
<!-- 日志文件的格式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger Line:%-3L - %msg%n</pattern>
<charset>utf-8</charset>
</encoder>
<!-- 此日志文件只記錄info級別的 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>error</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!-- 日志記錄器，日期滾動記錄 -->
<appender name="FILEWARN" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 正在記錄的日志文件的路徑及文件名 -->
<file>${LOG_PATH}/${APPDIR}/log_warn.log</file>
<!-- 日志記錄器的滾動策略，按日期，按大小記錄 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 歸檔的日志文件的路徑，例如今天是2013-12-21日志，當前寫的日志文件路徑為file節點指定，可以將此文件與file指定文件路徑設置為不同路徑，從而將當前日志文件或歸檔日志文件置不同的目錄。
而2013-12-21的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
<fileNamePattern>${LOG_PATH}/${APPDIR}/warn/log-warn-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<!-- 除按日志記錄之外，還配置了日志文件不能超過2M，若超過2M，日志文件會以索引0開始，
命名日志文件，例如log-error-2013-12-21.0.log -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<maxFileSize>2MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 追加方式記錄日志 -->
<append>true</append>
<!-- 日志文件的格式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger Line:%-3L - %msg%n</pattern>
<charset>utf-8</charset>
</encoder>
<!-- 此日志文件只記錄info級別的 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>warn</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!-- 日志記錄器，日期滾動記錄 -->
<appender name="FILEINFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 正在記錄的日志文件的路徑及文件名 -->
<file>${LOG_PATH}/${APPDIR}/log_info.log</file>
<!-- 日志記錄器的滾動策略，按日期，按大小記錄 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 歸檔的日志文件的路徑，例如今天是2013-12-21日志，當前寫的日志文件路徑為file節點指定，可以將此文件與file指定文件路徑設置為不同路徑，從而將當前日志文件或歸檔日志文件置不同的目錄。
而2013-12-21的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
<fileNamePattern>${LOG_PATH}/${APPDIR}/info/log-info-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<!-- 除按日志記錄之外，還配置了日志文件不能超過2M，若超過2M，日志文件會以索引0開始，
命名日志文件，例如log-error-2013-12-21.0.log -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<maxFileSize>2MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 追加方式記錄日志 -->
<append>true</append>
<!-- 日志文件的格式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level %logger Line:%-3L - %msg%n</pattern>
<charset>utf-8</charset>
</encoder>
<!-- 此日志文件只記錄info級別的 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>info</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<appender name="jsonLog" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 正在記錄的日志文件的路徑及文件名 -->
<file>${LOG_PATH}/${APPDIR}/log_IPInterceptor.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<fileNamePattern>${LOG_PATH}/${APPDIR}/log_IPInterceptor.%d{yyyy-MM-dd}.log</fileNamePattern>
</rollingPolicy>
<encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
<jsonFactoryDecorator class="net.logstash.logback.decorate.CharacterEscapesJsonFactoryDecorator">
<escape>
<targetCharacterCode>10</targetCharacterCode>
<escapeSequence>\u2028</escapeSequence>
</escape>
</jsonFactoryDecorator>
<providers>
<pattern>
<pattern>
{
"timestamp":"%date{ISO8601}",
"uid":"%mdc{uid}",
"requestIp":"%mdc{ip}",
"id":"%mdc{id}",
"clientType":"%mdc{clientType}",
"v":"%mdc{v}",
"deCode":"%mdc{deCode}",
"dataId":"%mdc{dataId}",
"dataType":"%mdc{dataType}",
"vid":"%mdc{vid}",
"did":"%mdc{did}",
"cid":"%mdc{cid}",
"tagId":"%mdc{tagId}"
}
</pattern>
</pattern>
</providers>
</encoder>
</appender>
<!-- 彩色日志 -->
<!-- 彩色日志依賴的渲染類 -->
<conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter"/>
<conversionRule conversionWord="wex"
converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter"/>
<conversionRule conversionWord="wEx"
converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter"/>
<!-- 彩色日志格式 -->
<property name="CONSOLE_LOG_PATTERN"
value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}"/>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!--encoder 默認配置為PatternLayoutEncoder-->
<encoder>
<pattern>${CONSOLE_LOG_PATTERN}</pattern>
<charset>utf-8</charset>
</encoder>
<!--此日志appender是為開發使用，只配置最底級別，控制台輸出的日志級別是大於或等於此級別的日志信息-->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>debug</level>
</filter>
</appender>
<!-- 指定項目中某個包，當有日志操作行為時的日志記錄級別 -->
<!-- rmjk.dao.mappe為根包，也就是只要是發生在這個根包下面的所有日志操作行為的權限都是DEBUG -->
<!-- 級別依次為【從高到低】：FATAL > ERROR > WARN > INFO > DEBUG > TRACE -->
<logger name="rmjk.dao.mapper" level="DEBUG"/>
<logger name="rmjk.service" level="DEBUG"/>
<!--顯示日志-->
<logger name="org.springframework.jdbc.core" additivity="false" level="DEBUG">
<appender-ref ref="STDOUT"/>
<appender-ref ref="FILEINFO"/>
</logger>
<!-- 打印json日志 -->
<logger name="IPInterceptor" level="info" additivity="false">
<appender-ref ref="jsonLog"/>
</logger>
<!-- 生產環境下，將此級別配置為適合的級別，以免日志文件太多或影響程序性能 -->
<root level="INFO">
<appender-ref ref="FILEERROR"/>
<appender-ref ref="FILEWARN"/>
<appender-ref ref="FILEINFO"/>
<!-- 生產環境將請stdout,testfile去掉 -->
<appender-ref ref="STDOUT"/>
</root>
</configuration>
其中的關鍵為：
<logger name="IPInterceptor" level="info" additivity="false">
<appender-ref ref="jsonLog"/>
</logger>
在需要打印的文件中引入slf4j：
private static final Logger LOG = LoggerFactory.getLogger("IPInterceptor");
MDC中放入需要打印的信息：
MDC.put("ip", ipAddress);
MDC.put("path", servletPath);
MDC.put("uid", paramMap.get("uid") == null ? "" : paramMap.get("uid").toString());
此時如果使用了LOG.info("msg")的話，打印的內容會輸入到日志的message中，日志格式如下：
修改Logstash配置
修改/usr/config/logstash目錄下的beats-input.conf：
input {
beats {
port => 5044
codec => "json"
}
}
只加了一句codec => "json"，但是Logstash會按照JSON格式來解析輸入的內容。
因為修改了配置，重啟elk：
docker restart elk
這樣，當我們的日志生成完畢之后，使用Filebeat導入到elk中，就可以通過Kibana來進行日志分析了。
轉評贊就是最大的鼓勵
