共计 10716 个字符，预计需要花费 27 分钟才能阅读完成。
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 来进行日志分析了。
