Docker安装ELK并实现JSON格式日志分析的方法-巨人网络通讯

Docker安装ELK并实现JSON格式日志分析的方法

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/qq13112...

拉取镜像

有了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来进行日志分析了。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

上一篇：Linux sftp命令用法总结
下一篇：linux touch命令用法实例

Docker安装ELK并实现JSON格式日志分析的方法

ELK是什么 ELK是elastic公司提供的一套完整的日志收集以及前端展示的解决方案，是三个产品的首字母缩写，分别是ElasticSearch、Logstash和Kibana。其中Logstash负责对日志进行处理，如日志的Docker,安装,ELK,并,实现,JSON,...

电话机器人怎么辨别真假（电话机器人试用）

本文目录一览：1、在安徽,各类型的电话机器人五花八门,贵的一万多一年,便宜的一千多...

电销专用卡购买-电销卡卡商

电销专用卡购买-电销卡卡商巨人网络通讯主营企业通信业务，专售电销卡、电销专用卡...

外呼系统怎么合作（外呼系统好做吗）

本文目次一览： 1、新型外呼零碎，怎样操纵？ 2、怎样抉择外呼零碎？ 3、外呼零碎是...

江西电销语音机器人公司（江西电销语音机器人公司排名

本文目录一览： 1、电销行业想使用语音机器人，有什么好的推荐吗2、电销机器人功能...

扬州云外呼系统（网络外呼系统）

本篇文章给我们谈谈扬州云外呼体系，以及网络外呼体系对应的知识点，期望对各位有...

Win8.1怎么看Wifi密码？Win8.1查看已连接成功的wifi密码方法

最近一亲朋好友来小编家玩，亲友想通过智能手机连接家里的无线网络，于是询问笔者...

疾控中心打电话机器人接听（疾控中心主动打电话）

本文目录一览：1、打电话机器人通话是对方设置了吗2、“疾控中心”来电说你是密接...

联通物联网卡怎么回事

什么是联通物联网卡联网物联网卡业务是中国联通为企业物联网用户提供的，采用物联...

电话机器人央视（呼叫电话机器人）

本文目录一览： 1、我给央视旧事公共号回响反映,为什么公共号认为是板滞人？ 2、我...

廊坊电话外呼系统供应商（外呼电话线路）

今天给各位分享廊坊电话外呼系统供应商的知识，其中也会对外呼电话线路进行解释，...

接听400电话收费吗400电话收费标准是什么

(接听400电话收费吗)(400电话收费标准是什么)以下内容由巨人小编整理发布。接听400电...

加盟400电话业务所带来的收益

随着现在通讯业的发达，电话成为了一个公司必不可少的一部分，但是现有的固定电话...

5g联通物联网卡_套餐优势_采购须知_平台管理

现如今，我们正处于网络高速发展的时代，从原先的2G到现如今的5G网络时代，除了手机...

详解用Python把PDF转为Word方法总结

先讲一下为啥要写这个文章，网上其实很多这种PDF转化的代码和软件。我一直想用Pyt...

福州不封号电销防封软件怎么办理-请看！

福州不封号电销防封软件怎么办理然后把有意义的电话自动转接给客服，提高了呼叫效...

Docker安装ELK并实现JSON格式日志分析的方法

全 部 栏 目

全部栏目