[大数据]ELK分布式日志收集-企业级日志中心

传统项目中，如果需要在生产环境定位异常的话，我们常常需要在服务器上使用命令的方式查询。而很多情况我们需要用到微服务架构或集群架构，日志被分散在不同的机器上，使得日志的查询变得异常困难。工欲善其事，必先利其器。如果此时有一个统一的实时日志分析平台，那可谓是绝渡逢舟，必定能够提高我们排查线上问题的效率。本文我们就来一起学习下开源的实时日志分析平台 ELK 的搭建及使用。

〓?ELK 简介

ELK 是一个开源的实时日志分析平台，它主要由 Elasticsearch、Logstash 和 Kibana 三部分组成。

常见模式如下：

1）datasource->logstash->elasticsearch->kibana

2）datasource->logstash->redis/kafka->logstash->elasticsearch->kibana

3）datasource->filebeat->logstash-> elasticsearch->kibana

4）datasource->filebeat->redis/kafka->logstash-> elasticsearch->kibana

5）datasource->filebeat->logstash->redis/kafka->logstash->elasticsearch->kibana

Ps:datasource可以是log4j产生的file文件，也可以是mysql，kafka等

这里我们把elk传输处理场景进行一个，从数据源开始，如何采集，用什么工具采集，采集到哪里，经过怎样的处理过滤，传输到哪里，怎样进行展示。本文介绍第2和第4种方案，推荐第4种方案。

▍Logstash

Logstash 主要用于收集服务器日志，它是一个开源数据收集引擎，具有实时管道功能。Logstash 可以动态地将来自不同数据源的数据统一起来，并将数据标准化到您所选择的目的地。

必须说的是Logstash很占用资源，比较重。它有两个角色shipper[日志收集]和indexer[日志存储]，而一台服务器只能存在一个Logstash实例。基于此，如果你一台服务器已经用了Logstash的indexer角色存储日志到ES，再用Filebeat代替Logstash的shipper角色来收集日志，就显得非常有价值，而且它非常轻量，不会加重服务器负担，且Filebeat还非常稳定，基本不会有宕机问题。

Logstash 收集数据的过程主要分为以下三个部分：

??输入：数据（包含但不限于日志）往往都是以不同的形式、格式存储在不同的系统中，而 Logstash 支持从多种数据源中收集数据（File、Syslog、MySQL、消息中间件等等）。

??过滤器：实时解析和转换数据，识别已命名的字段以构建结构，并将它们转换成通用格式。

??输出：Elasticsearch 并非存储的唯一选择，Logstash 提供很多输出选择。

▍Elasticsearch

Elasticsearch （ES）是一个分布式的 Restful 风格的搜索和数据分析引擎，它具有以下特点：

??查询：允许执行和合并多种类型的搜索 — 结构化、非结构化、地理位置、度量指标 — 搜索方式随心而变。

??分析：Elasticsearch 聚合让您能够从大处着眼，探索数据的趋势和模式。

??速度：很快，可以做到亿万级的数据，毫秒级返回。

??可扩展性：可以在笔记本电脑上运行，也可以在承载了 PB 级数据的成百上千台服务器上运行。

??弹性：运行在一个分布式的环境中，从设计之初就考虑到了这一点。

灵活性：具备多个案例场景。支持数字、文本、地理位置、结构化、非结构化，所有的数据类型都欢迎。

▍Kibana

Kibana 可以使海量数据通俗易懂。它很简单，基于浏览器的界面便于您快速创建和分享动态数据仪表板来追踪 Elasticsearch 的实时数据变化。其搭建过程也十分简单，您可以分分钟完成 Kibana 的安装并开始探索 Elasticsearch 的索引数据?—?没有代码、不需要额外的基础设施。

在 ELK 中，三大组件的大概工作流程如下图所示，由 Logstash 从各个服务中采集日志并存放至 Elasticsearch 中，然后再由 Kiabana 从 Elasticsearch 中查询日志并展示给终端用户。

▍插入介绍 Filebeat

Filebeat是一个日志文件托运工具，在服务器上安装客户端后，Filebeat会监控日志目录或者指定的日志文件，追踪读取这些文件（追踪文件的变化，不停的读），并且转发这些信息到Logstarsh中存放。上图的收集日志的过程，实际中我们往往会使用Filebeat来完成。

Filebeat 由两个主要组件组成：harvester 和 prospector ：

??采集器 harvester 的主要职责是读取单个文件的内容。读取每个文件，并将内容发送到 the output。每个文件启动一个 harvester，harvester 负责打开和关闭文件，这意味着在运行时文件描述符保持打开状态。如果文件在读取时被删除或重命名，Filebeat 将继续读取文件。

??查找器 prospector 的主要职责是管理 harvester 并找到所有要读取的文件来源。如果输入类型为日志，则查找器将查找路径匹配的所有文件，并为每个文件启动一个 harvester。每个 prospector 都在自己的 Go 协程中运行。

注：Filebeat prospector只能读取本地文件，没有功能可以连接到远程主机来读取存储的文件或日志。

Filebeat由以上两个组件一起工作来读取文件（就如tail -f）并将事件数据发送到您指定的输出。

〓?ELK 实现方案

通常情况下我们的服务都部署在不同的服务器上，那么如何从多台服务器上收集日志信息就是一个关键点了。本篇文章中提供的解决方案如下图所示：

如上图所示，整个 ELK 的运行流程如下：

1. 在微服务（产生日志的服务）上部署一个??Shipper 角色的?Logstash，或者部署一个Filebeat，主要负责对所在机器上的服务产生的日志文件进行数据采集，并将消息推送到 Redis 消息队列。

2. 另用一台服务器部署一个 Indexer 角色的 Logstash，主要负责从 Redis 消息队列中读取数据，并在 Logstash 管道中经过 Filter 的解析和处理后输出到 Elasticsearch 集群中存储。

3. Elasticsearch 主副节点之间数据同步。

4. 单独一台服务器部署 Kibana 读取 Elasticsearch 中的日志数据并展示在 Web 页面。

通过这张图，相信您已经大致清楚了我们将要搭建的 ELK 平台的工作流程，以及所需组件。下面就让我们一起开始搭建起来吧。

〓?ELK 平台搭建

本节主要介绍搭建 ELK 日志平台，包括安装 Indexer 角色的 Logstash，Elasticsearch 以及 Kibana 三个组件。完成本小节，您需要做如下准备：

1. 一台 Centos 机器或虚拟机，作为入门教程，此处省略了 Elasticsearch 集群的搭建，且将 Logstash(Indexer)、Elasticsearch 以及 Kibana 安装在同一机器上。

2. 在?Centos?上安装 JDK，注意 Logstash 要求 JDK 在 1.7 版本以上。

3. Logstash、Elasticsearch、Kibana 、Filebeat安装包下载。

▍安装 Logstash

Logstash下载：https://www.elastic.co/cn/downloads/logstash

解压压缩包：

tar?-xzvf?logstash-7.13.4.tar.gz

Logstash 的启动命令咱们先忽略，后面的案例我再讲。

▍安装 Elasticsearch

Elasticsearch下载：https://www.elastic.co/cn/downloads/elasticsearch

解压安装包：

tar?-xzvf?elasticsearch-7.13.4-linux-x86_64.tar.gz

启动 Elasticsearch：

cd elasticsearch-7.13.4/./bin/elasticsearch

在启动 Elasticsearch 的过程中，有一个问题你应该会碰到：

如果您是以 root 用户启动的话，就会报下图所示的错误。解决方案自然就是添加一个新用户启动 Elasticsearch。

Root 用户启动 Elasticsearch 报错

如何添加新用户并且授权es？

1. 创建elk 用户组

groupadd gelk

2. 创建用户elk（账号密码自己修改）

useradd?elk?passwd?elk1234

3.将elk用户添加到elk组

usermod?-G?gelk?elk

4.设置sudo权限

visudo

找到root ALL=(ALL) ALL一行，添加elk用户，如下，然后【esc】【:wq!】保存退出

##?Allow?root?to?run?any?commands?anywhere
root?ALL=(ALL)?ALL
elk??ALL=(ALL)?ALL

5.为用户分配权限

chown?-R?elk:gelk?/opt/elk/elasticsearch-7.13.4

6.登录

su -l ues
#输入密码：
?
#然后可以启动es
./bin/elasticsearch

Ps:后面我们Logstash(Indexer)、Kibana也一样用这个账号，

#?后面再执行
chown -R elk:gelk /opt/elk/logstash-7.13.4
chown -R elk:gelk /opt/elk/kibana-7.13.4-linux-x86_64

启动成功后，另起一个会话窗口执行 curl http://localhost:9200 命令，如果出现如下结果，则代表 Elasticsearch 安装成功。

• 检查 Elasticsearch 是否启动成功

{
  "name" : "iZwz933r7t5dod0qy5l2e6Z",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "FM5fLGPbR3q4ak0MtcmkVQ",
  "version" : {
    "number" : "7.13.4",
    "build_flavor" : "default",
    "build_type" : "tar",
    "build_hash" : "c5f60e894ca0c61cdbae4f5a686d9f08bcefc942",
    "build_date" : "2021-07-14T18:33:36.673943207Z",
    "build_snapshot" : false,
    "lucene_version" : "8.8.2",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

▍安装 Kibana

Kibana 下载页：https://www.elastic.co/cn/downloads/kibana

解压安装包：

tar?-xzvf?kibana-7.13.4-linux-x86_64.tar.gz

修改配置文件 config/kibana.yml ，主要指定 Elasticsearch 的信息。

Kibana 配置信息#Elasticsearch主机地址

elasticsearch.hosts: "http://127.0.0.1:9200"
# 允许远程访问
server.host: "0.0.0.0"
# Elasticsearch用户名 这里其实就是我在服务器启动Elasticsearch的用户名
elasticsearch.username:?"elk"
# Elasticsearch鉴权密码 这里其实就是我在服务器启动Elasticsearch的密码
elasticsearch.password:?"elk1234"

启动 Kibana：

cd?kibana-7.13.4-linux-x86_64/bin
./kibana

在浏览器中访问 http://ip:5601 ，若出现以下界面，则表示 Kibana 安装成功。

Kibana 启动成功界面

▍安装 Filebeat

Filebeat下载：https://www.elastic.co/cn/downloads/beats/filebeat

解压安装包：

tar?-xzvf?filebeat-7.13.4-linux-x86_64.tar.gz

检查配置文件：

./filebeat test config -c filebeat.yml
Config OK

Filebeat调试启动，退出终端或 ctrl+c 会退出运行：

./filebeat -c filebeat.yml -e -d "*"

Ps:碰到问题一定记得用这个命令看看.

线上环境建议用如下命令：

nohup ./filebeat -e -c filebeat.yml >filebeat.log 2>&1 &

ELK 日志平台安装完成后，下面我们就将通过具体的例子来看下如何使用 ELK，下文将分别介绍如何将 Spring Boot 日志（其实就是file日志）和 Nginx 日志交由 ELK 分析。

〓?在 SpringBoot?或 Netty?等项目中使用 ELK

首先什么项目其实无所谓，只要log4j输出的文件日志，例如app/logs/*.log（或者*.txt格式不重要，只要是普通文本就行）这种。我们这里用一个基于Netty的Smqtt集群项目作为例子来讲。

▍修改并部署 Smqtt 项目

logback.xml 配置文件如下

<?xml version="1.0" encoding="UTF-8"?>
<configuration?scan="true" debug="true">
    <!-- 日志格式 -->
    <property name="LOG_PATTERN" value="[%d{yyyy-MM-dd HH:mm:ss.SSS}][%-5level][%thread][%45logger] %msg%n"/>
    <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>
    </appender>
    <appender name="file" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
????????????<fileNamePattern>logs/smqtt.%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>30</maxHistory>
        </rollingPolicy>
        <encoder>
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>
    </appender>
    <root level="DEBUG">
        <appender-ref ref="stdout"/>
        <appender-ref ref="file"/>
????</root>
</configuration>

在上面的配置中我们定义了一个名为 file 的 Appender 往日志文件中输出指定格式的日志。而上面的 pattern 标签正是具体日志格式的配置，通过上面的配置，我们指定输出了时间、日志级别、线程、logger（通常为日志打印所在类的全路径）以及具体日志内容等信息。

将项目打包部署到2台 Centos 服务器上。

查看日志文件， logback 配置文件中我将日志存放在?logs/smqtt.xxxx-xx-xx.log?文件中，执行?tail -f?logs/smqtt.xxxx-xx-xx.log 命令。

▍ 配置 Shipper 角色 Logstash

smqtt项目部署成功之后，我们还需要在分别在2台服务部署的机器上安装并配置 Shipper 角色的 Logstash。Logstash 的安装过程在 ELK 平台搭建小节中已有提到，这里不再赘述。安装完成后，我们需要编写 Logstash 的配置文件，以支持从日志文件中收集日志并输出到 Redis 消息管道中，Shipper 的配置如下所示。

Shipper 角色的 Logstash 的配置（看看就好，先不动手）

input {
    file {
        path => [
            # 这里填写需要监控的文件
            "/root/smqtt/logs/*.log"
        ]
    }
}
?
output {
    # 输出到redis
    redis {
????????host?=>?"x.x.x.x"???#?redis主机地址
        port => 6379              # redis端口号
????????password?=>?"xxx"
        db => 4                   # redis数据库编号
        data_type => "channel"    # 使用发布/订阅模式
        key => "logstash_list_smqtt_0"  # 发布通道名称
    }
}

其实 Logstash 的配置是与前面提到的 Logstash 管道中的三个部分（输入、过滤器、输出）一一对应的，只不过这里我们不需要过滤器所以就没有写出来。上面配置中 Input 使用的数据源是文件类型的，只需要配置上需要收集的本机日志文件路径即可。Output 描述数据如何输出，这里配置的是输出到 Redis。

Redis 的配置 data_type 可选值有 channel 和 list 两个。channel 是 Redis 的发布/订阅通信模式，而 list 是 Redis 的队列数据结构，两者都可以用来实现系统间有序的消息异步通信。channel 相比 list 的好处是，解除了发布者和订阅者之间的耦合。举个例子，一个 Indexer 在持续读取 Redis 中的记录，现在想加入第二个 Indexer，如果使用 list ，就会出现上一条记录被第一个 Indexer 取走，而下一条记录被第二个 Indexer 取走的情况，两个 Indexer 之间产生了竞争，导致任何一方都没有读到完整的日志。channel 就可以避免这种情况。这里 Shipper 角色的配置文件和下面将要提到的 Indexer 角色的配置文件中都使用了 channel 。

但是这个小节的内容，我要废弃，原因是logstash基于java的jvm，很重，非常消耗资源，只用于转运数据，应该用更轻量的工具，我们改用我们前面安装的Filebeat。

▍配置?Filebeat 替代 Shipper 角色的 Logstash

安装过程前面已经有介绍，我们分别在2台smqtt服务的机子上安装Filebeat

第1台的filebeat.yml配置文件，主要修改以下内容，省略部分保存默认不变。

filebeat.inputs:
-?type:?log
??enabled:?true
  paths:
    - /root/smqtt/logs/*.log
?
...
?
  fields:
    log_src: smqtt-test-1 # logstash判断日志来源    
?
#?输出到redis，默认是list形式，我们改成channel
output.redis:
  hosts: ["x.x.x.x:6379"]
??key:?logstash_list_smqtt_0??#?redis中的键，值默认是一个list，俩表中存储的一行一行的上面两个应用的日志，logstash中会用到这个参数
  password: xxx # redis访问密码
  db: 0
  datatype: channel
?
?...
?
?processors:
  # 将自带的一些字段可以去除
  # - add_host_metadata:
  #     when.not.contains.tags: forwarded
  # - add_cloud_metadata: ~
  # - add_docker_metadata: ~
  # - add_kubernetes_metadata: ~
  # 在这里可以设置要去除的字段
  - drop_fields:
      # when: 可以设置去除的条件
      #   condition
      fields: ["log","host","input","agent","ecs"]
      ignore_missing: false

第2台服务filebeat.yml，同第一台，仅仅修改一处

...
?
  fields:
????log_src:?smqtt-test-2?#?logstash判断日志来源??
??
...

▍?配置 Indexer 角色 Logstash

上面我们已经配置好转运数据部分，我们还需要配置 Indexer 角色 Logstash 以支持从 Redis 接收日志数据，并通过过滤器解析后存储到 Elasticsearch 中，其配置内容如下所示。

Indexer 角色的 Logstash 的配置

input {
    redis {
        host => "x.x.x.x"   # redis主机地址
        port => 6379              # redis端口号
        password => "xxx"
        db => 0                   # redis数据库编号
        data_type => "channel"    # 使用发布/订阅模式
        key => "logstash_list_smqtt_0"  # 发布通道名称
    }
}
?
filter {
?????#【建议】不要拆分message字段数据
     #定义数据的格式
     #grok {
       #正则拆分message字段数据，赋值到自定义字段
       #[%d{yyyy-MM-dd HH:mm:ss.SSS}][%-5level][%thread][%45logger] %msg%n
       #match => { "message" => "\[%{TIMESTAMP_ISO8601:log_time}\]\[%{NOTSPACE:log_app}\]\[%{DATA:log_level}\]\[%{NOTSPACE:log_threadName}\]\[%{DATA:log_logger}\] %{GREEDYDATA:log_msg}"}
       #删除message字段
       #remove_field => [ "version","host","path" ]
     #}
?
    #Rename some of the beats fields due to 6.3.0 beats changes
????#以下规则，处理这个问题：Could not index event to Elasticsearch,如果Filebeat删除host可能就不需要了
    mutate {
        rename => { "[host][name]" => "host" }
    }
}
?
output {
    stdout {}
    elasticsearch {
        hosts => "localhost:9200"
        index => "smqtt"
    }
?
}

与 Shipper 不同的是，Indexer 的管道中我们定义了过滤器，也正是在这里将日志解析成结构化的数据。下面是我截取的一条 logback 的日志内容：

smqtt 项目输出的一条日志

[2021-08-05 14:49:20.862][INFO ][sc-cluster-io-epoll-1][io.github.quickmsg.biz.bridge.CounterInterceptor] 终端数量=0

如果在 Filter 中我们使用 Grok 插件从上面这条日志中解析出了时间、日志级别、线程、logger（通常为日志打印所在类的全路径）以及具体日志内容几个字段。Grok 又是如何工作的呢？[Ps：暂时不推荐用这个，费性能，且目前用处也不大，不过上面注释掉的代码都是调试过的，也方便有需要的朋友少走弯路]

1. message 字段是 Logstash 存放收集到的数据的字段， match = {"message" => ...} 代表是对日志内容做处理。

2. Grok 实际上也是通过正则表达式来解析数据的，上面出现的 TIMESTAMP_ISO8601 、NOTSPACE 等都是 Grok 内置的 patterns。

3. 我们编写的解析字符串可以使用 Grok Debugger 来测试是否正确，这样避免了重复在真实环境中校验解析规则的正确性。

▍?查看效果

经过上面的步骤，我们已经完成了整个 ELK 平台的搭建以及 smqtt 项目的接入。下面我们按照以下步骤执行一些操作来看下效果。

启动 Elasticsearch，启动命令在 ELK 平台搭建 小节中有提到，这里不赘述（Kibana 启动同）。启动 Indexer 角色的 Logstash。

# 进入到 Logstash 的解压目录，然后执行下面的命令
bin/logstash -f indexer-smqtt.conf

启动 Kibana。

[忽略]启动 Shipper 角色的 Logstash。

# 进入到 Logstash 的解压目录，然后执行下面的命令
./bin/logstash -f shipper-logstash.conf

调用 smqtt 业务输出日志，此时应该已经有数据写入到 ES 中了。

在浏览器中访问 http://ip:5601 ，打开 Kibana 的 Web 界面，并且如下图所示添加 logback 索引。

在 Kibana 中添加 Elasticsearch 索引

进入 Discover 界面，选择 logback 索引，就可以看到日志数据了，这里我们看到2台服务的日志都进来了。然后为了看起来比较清爽，我勾选了Available Fields中的log_src字段和message做显示。如下图所示

〓?在其它项目中使用 ELK

通过上面的步骤我们已经成功的搭建起了自己的 ELK 实时日志平台，并且接入了 smqtt项目的日志。但是实际场景下，几乎不可能只有一种类型的日志，例如我们要看Nginx日志，下面我们简单讲讲多输入源的情况。

Nginx 的访问日志默认在 /var/log/nginx/access.log 文件中，支持两种日志输入的 Indexer 角色的 Logstash 配置

input {
    redis {
        type      => "logback"
        ...
    }
    redis {
       type       => "nginx"
       ...
    }
}
?
filter {
     if [type] == "logback" {
         ...
     }
     if [type] == "nginx" {
         ...
     }
}
?
output {
    if [type] == "logback" {
        ...
    }
    if [type] == "nginx" {
       ...
    }
}

〓?ELK 启动

在上面的步骤中，ELK 的启动过程是我们一个一个的去执行三大组件的启动命令的。而且还是在前台启动的，意味着如果我们关闭会话窗口，该组件就会停止导致整个 ELK 平台无法使用，这在实际工作过程中是不现实的，我们剩下的问题就在于如何使 ELK 在后台运行。根据 《Logstash 最佳实践》 一书的推荐，我们将使用 Supervisor 来管理 ELK 的启停。首先我们需要安装 Supervisor，在 Centos 上执行 yum install supervisor 即可。安装成功后，我们还需要在 Supervisor 的配置文件中配置 ELK 三大组件（其配置文件默认为 /etc/supervisor/supervisord.conf 文件），但是配置文件本身已经很多内容，我们看到里面有一行代码

[include]
files = supervisord.d/*.ini

所以，为了比较清晰，我们可以去/etc/supervisord.d/目录下创建*.ini文件即可。我们创建一个elk.ini文件

elk.ini文件内容

[program:elasticsearch]
environment=JAVA_HOME="/opt/java/jdk1.8.0_202/"
directory=/opt/elk/elasticsearch-7.13.4
user=elk
command=/opt/elk/elasticsearch-7.13.4/bin/elasticsearch
?
[program:logstash-indexer]
environment=JAVA_HOME="/opt/java/jdk1.8.0_202/"
directory=/opt/elk/logstash-7.13.4
user=elk
command=/opt/elk/logstash-7.13.4/bin/logstash -f /opt/elk/logstash-7.13.4/config/indexer-smqtt.conf
?
[program:kibana]
environment=LS_HEAP_SIZE=5000m
directory=/opt/elk/kibana-7.13.4-linux-x86_64
user=elk
command=/opt/elk/kibana-7.13.4-linux-x86_64/bin/kibana

按照以上内容配置完成后，执行 sudo supervisorctl reload 即可完成整个 ELK 的启动，而且其默认是开机自启。当然，我们也可以使用 sudo supervisorctl start/stop/restart [program_name] 来管理单独的应用。

〓 写在最后，一个FAQ

elk分布式日志架构下为啥需要用redis这样的消息队列，而不是直接Logstash 存入ES里呢？为什么部署不是各个应用服务器都去安装logstash或者filebeat。logstash采集本地log4j产生的日志，然后输入进es里，而中间加一层redis的目的是什么呢？还有一个问题是，log4j有些插件可以直接把产生的日志写进es里，直接用这样的不是更方便么？连logstash都省去了。请问这样做有什么不好的么？

回答：

logstash、redis（一般是用list）都是队列，起到缓冲作用，也就是削峰填谷。如果直接写到ES的话，由于ES的HTTP API处理能力有限，在日志写入频繁的情况下可能会超时、丢失，所以用队列来做缓冲。还有些是用RabbitMQ、Kafka，以及一些更先进的消息队列来缓冲日志的。

都看到这里了，老铁可不可以给个关注呀！