写在前面

在前面我们已经给trace-one和trace-two项目引入了Spring Cloud Sleuth的基础模块spring-cloud-starter-sleuth，实现了在各个微服务的日志信息中添加跟踪信息的功能。

但是问题来了，这些日志文件都是分散存储在各个服务实例的文件系统上，因此还需要借助于一些工具来帮助集中收集、存储和搜索这些跟踪信息。一般来说可以使用基于日志的分析系统，如ELK，它可以很轻松的帮助我们收集和存储这些跟踪日志，同时在需要的时候也可以根据TraceID来轻松地搜索出对应请求链路相关的明细日志。

ELK由ElasticSearch、Logstash和Kibana三个开源工具组成。其中ElasticSearch是一个开源的分布式搜索引擎，它可以支持分布式、零配置、自动发现、索引自动分片、索引副本机制、RESTful风格接口，多数据源、自动搜索负载等。

Logstash是一个完全开源的工具，它可以对日志进行收集、过滤，并将其进行存储以被后续使用。

Kibana也是一个开源工具，它可以为Logstash和ElasticSearch提供日志分析的Web界面，也可以帮助汇总和搜索重要数据日志。

当然一般来说，考虑到Logstash对资源的占用情况，我们都会使用Filebeat来采集数据，然后输出到Logstash进行数据的过滤或者存储等过程。

Windows安装ELKB

一般来说都是在Linux系统上安装ELKB，但是考虑到笔者日常的开发环境是Windows，因此本篇介绍如何在Windows系统上安装ELKB，注意ELKB这四者的版本号必须保持一致，否则会出现各种难以预料的问题，这里使用7.5.2的版本。

ElasticSearch安装

第一步，下载安装java8及以上版本，之后进行环境变量的设置：
（1）JAVA_HOME值为D:\Application\java\jdk1.8；（2）在系统变量的Path处设置两个值，分别是：%JAVA_HOME%\bin和%JAVA_HOME%\jre\bin。（3）使用java -version命令来检验是否安装成功，输出版本信息则表明Java安装成功：

第二步，新建对应的文件，建议将四者都放在同一个文件夹下面，如下所示：

第三步，下载ElasticSearch安装包，点击 ElasticSearch镜像地址，选择合适的64位版本后进行下载。

第四步，设置hostname。修改C:\Windows\System32\drivers\etc文件夹下面的hosts文件，在里面添加如下映射：

1
127.0.0.1   elk-1

第五步，创建数据及日志文件。开发者需要在ELK目录下创建数据和日志 文件，如下所示：

第六步，修改config/elasticsearch.yml配置文件，如下所示：

1
2
3
4
5
6
7
8
9
cluster.name: elk  #集群名
node.name: node-1  #节点名
path.data: D:/Application/ELK/ELK/data  # 数据路径
path.logs: D:/Application/ELK/ELK/logs  # 日志路径
bootstrap.memory_loacl: false # 锁内存，尽量不使用交换内存
network.host: 0.0.0.0 # 网络主机
http.port: 9200 # 端口号
discovery.seed_hosts: ["elk-1"]  # 发现集群hosts
cluster.inital_master_nodes: ["node-1"]  # 设置集群master节点

第七步，启动ElasticSearch。开发者可以进入到ElasticSearch的bin目录，然后直接执行如下命令即可在后台启动ElasticSearch：

1
D:\Application\ELK\ElasticSearch\elasticsearch-7.5.2\bin>start /min elasticsearch

之后直接打开浏览器，输入http://127.0.0.1:9200后显示下图所示信息，也表明elasticsearch已成功启动：

这样关于ElasticSearch的安装就完成了。

Kibana安装

第一步，下载Kibana安装包，点击 Kibana镜像地址，选择合适的64位版本后进行下载。

第二步，修改Kibana配置文件。修改config/kibana.yml配置文件，如下所示：

1
2
3
4
5
6
7
server.port: 5601   # 端口
server.host: "0.0.0.0"  # 访问限制
elasticsearch.hosts: ["http://127.0.0.1:9200"]  #修改为ES所在机器的ip
elasticsearch.ssl.verificationMode: none  #关闭ssl验证
kibana.index: ".kibana"  #去掉注释
logging.dest: D:/Application/ELK/ELK/log/kibana/kibana.log  #去掉注释，并新建相应文件目录
i18n.locale: "zh-CN"  # 去掉注释，并修改为"zh-CN"，支持中文显示

第三步，启动Kibana。开发者可以进入到Kibana的bin目录，然后直接执行如下命令即可在后台启动Kibana：

1
D:\Application\ELK\Kibana\kibana-7.5.2-windows-x86_64\bin>start /min kibana

之后直接打开浏览器，输入http://127.0.0.1:5601后显示下图所示信息，也表明Kibana已成功启动：

FileBeat安装

第一步，下载FileBeat安装包，点击 FileBeat镜像地址，选择合适的64位版本后进行下载。

第二步，修改FileBeat配置文件。修改filebeat.yml配置文件，这里我们需要修改两部分，第一部分是输入模块，里面涉及到两个参数，第一个是开启filebeat功能，第二个则是配置收集指定地址的日志文件：

第二部分是输出，这里我们需要将收集的日志文件输出到Logstash或者ElasticSearch中，因此这两个模块开启其中一个即可，由于是单机部署，因此IP地址都是本机IP，只是端口号不同而已（这里选择将其输出到Logstash）：

最后就是输出到Kibana的设置了，如下所示：

第三步，测试配置文件。接下来需要测试第二步中书写的配置文件是否正确，需要进入到filebeat.yml配置文件所在的目录下执行如下命令：

1
filebeat test config –e

第四步，启动FileBeat。开发者可以进入到FileBeat目录，然后直接执行如下命令即可在后台启动FileBeat：

1
D:\Application\ELK\FileBeat\filebeat-7.5.2-windows-x86_64>start /min filebeat -e -c filebeat.yml

此时虽然可以启动，但是由于Logstash还尚未安装，因此会出现无法连接到5044端口的错误，这是正常现象。

Logstash安装

第一步，下载Logstash安装包，点击 Logstash镜像地址，选择合适的64位版本后进行下载。

第二步，复制一份logstash-sample.conf文件并将其修改为logstash-hello.conf文件，修改的信息如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
input {
  beats {
    port => 5044
  }
}

output {
  elasticsearch {
    hosts => ["http://127.0.0.1:9200"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
}

注意上面其实只需要修改IP地址，当然如果是本机可以不做修改，因为就是localhost，我们前面搭建的ElasticSearch是没有设置用户名和密码的，因此这里就可以直接忽略。

第三步，修改Logstash配置文件。将Logstash的logstash.yml配置文件中的第190和195行，去掉两行前面的注释，并将其修改如下：

1
2
http.host: "127.0.0.1"
http.port: 9600-9700

第四步，启动Logstash。开发者可以使用如下命令来启动Logstash：

1
D:\Application\ELK\Logstash\logstash-7.5.2\bin>start /min logstash -f ../config/logstash-sample.conf

这样ELK日志分析平台就搭建成功了。

其实SpringCloud Sleuth在与ELK平台整合使用时，实际上只要实现与负责日志收集的Logstash完成数据对接即可，所以我们需要为Logstash准备JSON格式的日志输出即可。我们知道Spring Boot应用默认支持使用logback来记录日志，而Logstash自身也有对logback日志工具的支持，因此开发者可以直接在logback的配置中增加对Logstash的Appender，就可以非常方便的将日志转换成以JSON格式进行存储和输出了。

快速整合

Spring Cloud Sleuth与Logstash的直接集成有两种方式：（1）Logstash与微服务应用安装在一起，直接收集日志，如下所示：

（2）Logstash独立部署，微服务节点通过网络向Logstash发送日志信息，此时如下所示：

当然上面介绍的只是最常用的两种直接方式，还存在间接形式的变种，如微服务将日志发送给Redis或者MQ，再由它们去对接Logstash。

第一步，在之前创建的两个项目trace-one和trace-two的pom.xml配置文件中添加logstash-logback-encoder依赖，如下所示：

1
2
3
4
5
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>6.4</version>
</dependency>

请注意logstash-logback-encoder依赖的版本不能太低，否则项目无法启动。

第二步，在两个项目trace-one和trace-two的resource目录下分别新建bootstrap.properties启动文件，并将spring.application.name=trace-one和spring.application.name=trace-two配置移动到该文件中。之所以这么做，是因为logback-spring.xml的加载顺序是在application.properties配置文件之前，所以按照之前的配置，logback-spring.xml是无法获取到所需的spring.application.name属性，因此必须将spring.application.name属性放到最先加载的bootstrap.properties启动文件中。

第三步，在两个项目trace-one和trace-two的resource目录下分别新建logback-spring.xml日志配置文件，其中的代码如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
<?xml version="1.0" encoding="UTF-8"?>
<!--该配置将日志级别不同的log信息保存到不同的文件中 -->
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml" />

    <springProperty scope="context" name="springAppName"
                    source="spring.application.name" />

    <!-- 日志在工程中的输出位置 -->
    <property name="LOG_FILE" value="${BUILD_FOLDER:-build}/${springAppName}" />

    <!-- 控制台的日志输出样式 -->
    <property name="CONSOLE_LOG_PATTERN"
              value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}" />

    <!-- 控制台输出 -->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFO</level>
        </filter>
        <!-- 日志输出编码 -->
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
            <charset>utf8</charset>
        </encoder>
    </appender>

    <!--Logstash输出的JSON格式的Appender -->
    <appender name="logstash"
              class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_FILE}.json</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--日志文件输出的文件名 -->
            <fileNamePattern>${LOG_FILE}.json.%d{yyyy-MM-dd}.gz</fileNamePattern>
            <!--日志文件保留天数 -->
            <MaxHistory>3</MaxHistory>
        </rollingPolicy>
        <!-- 日志输出编码 -->
        <encoder
                class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <timestamp>
                    <timeZone>UTC</timeZone>
                </timestamp>
                <pattern>
                    <pattern>
                        {
                        "severity": "%level",
                        "service": "${springAppName:-}",
                        "trace": "%X{X-B3-TraceId:-}",
                        "span": "%X{X-B3-SpanId:-}",
                        "exportable": "%X{X-Span-Export:-}",
                        "pid": "${PID:-}",
                        "thread": "%thread",
                        "class": "%logger{40}",
                        "rest": "%message"
                        }
                    </pattern>
                </pattern>
            </providers>
        </encoder>
    </appender>

    <!-- 日志输出级别 -->
    <root level="INFO">
        <appender-ref ref="console" />
        <appender-ref ref="logstash" />
    </root>
</configuration>

可以发现上述对于Logstash的支持主要是通过名为logstash的Appender来实现的，其内容并不复杂，主要是对日志信息的格式化处理，为了后续测试和查看，笔者先将JSON格式的日志输出到文件中。

第四步，启动eureka-server服务注册中心和上述两个项目trace-one和trace-two，然后在浏览器中访问http://localhost:5001/trace-one链接得到返回值”This is the trace two!”;，如下所示：

同时在trace-one和trace-two这两个项目的目录下有一个build目录，里面分别创建了以各自应用名称命名的JSON文件，该文件就是在logback-spring.xml中配置的名为logstash的Appender输出的日志文件，其中记录了类似下面格式的日志：

1
2
3
4
5
6
7
8
9
10
11
12
{
	"@timestamp": "2020-05-11T02:50:47.706Z",
	"severity": "INFO",
	"service": "trace-one",
	"trace": "",
	"span": "",
	"exportable": "",
	"pid": "9876",
	"thread": "main",
	"class": "c.n.d.s.r.aws.ConfigClusterResolver",
	"rest": "Resolving eureka endpoints via configuration"
}

也就是说现在JSON格式的日志信息已经有了，接下来就是如何这些数据发送给Logstash？这里就以之前提到的两种最直接的方式来进行演示。

日志收集

（1）Logstash与微服务应用安装在一起，直接收集日志，如下所示：

此处需要在微服务各个节点安装Logstash，并在Logstash的config目录下新建logstash-trace-one.conf文件，此时的配置如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  file {
    type => "server"
	path => "D:/SpringCloud/cloudgo/trace-one/build/trace-one.json"
  }
}
filter {
  #Only matched data are send to output.
}

output {
  elasticsearch {
    hosts => ["http://127.0.0.1:9200"]
	index => "applog"
	action => "index"
    #index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
}

这种方式配置较为简单，因为Logstash是直接监控本服务的日志文件，并将其推送给ElasticSearch的接收地址。

（2）Logstash独立部署，微服务节点通过网络向Logstash发送日志信息，此时如下所示：

这种方式较为第一种有些麻烦，因为Logstash是独立部署的，因此必须要配置日志输入方式为TCP，此时需要在一个空闲的节点上安装Logstash，并在Logstash的config目录下新建logstash-trace-two.conf文件，此时的配置如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  tcp {
    mode => "server"
	host => "192.168.0.100"
	# 此处的host为需要发送日志的项目IP，也就是trace-two项目的IP，假设此时已经不在本地
	port => 9250
  }
}
filter {
  #Only matched data are send to output.
}

output {
  elasticsearch {
    hosts => ["http://127.0.0.1:9200"]  #ElasticSearch host, can be array.
	index => "applog"
	action => "index"
    #index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
}

也就是说此时Logstash需要监听192.168.0.100:9250地址，并将接收到的日志信息转发给ElasticSearch的接收地址。仅仅是这样还不行的，还需要改造之前的logback-spring.xml日志配置文件，目前的代码只是将log信息的记录方式修改为json格式，还不具备向Logstash发送日志的功能，此时logback-spring.xml日志配置文件中的代码为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
<?xml version="1.0" encoding="UTF-8"?>
<!--该配置将日志级别不同的log信息保存到不同的文件中 -->
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml" />

    <springProperty scope="context" name="springAppName"
                    source="spring.application.name" />

    <!-- 日志在工程中的输出位置 -->
    <property name="LOG_FILE" value="${BUILD_FOLDER:-build}/${springAppName}" />

    <!-- 控制台的日志输出样式 -->
    <property name="CONSOLE_LOG_PATTERN"
              value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}" />

    <!-- 控制台输出 -->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFO</level>
        </filter>
        <!-- 日志输出编码 -->
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
            <charset>utf8</charset>
        </encoder>
    </appender>

    <!--Logstash输出的JSON格式的Appender -->
    <appender name="logstash"
              class="ch.qos.logback.core.rolling.RollingFileAppender">    
        <!--配置日志发送地址，也就是之前Logstash的地址-->   
        <destination>192.168.0.100:9250</destination>
        <!-- 日志输出编码 -->
        <encoder
                class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <timestamp>
                    <timeZone>UTC</timeZone>
                </timestamp>
                <pattern>
                    <pattern>
                        {
                        "severity": "%level",
                        "service": "${springAppName:-}",
                        "trace": "%X{X-B3-TraceId:-}",
                        "span": "%X{X-B3-SpanId:-}",
                        "exportable": "%X{X-Span-Export:-}",
                        "pid": "${PID:-}",
                        "thread": "%thread",
                        "class": "%logger{40}",
                        "rest": "%message"
                        }
                    </pattern>
                </pattern>
            </providers>
        </encoder>
    </appender>

    <!-- 日志输出级别 -->
    <root level="INFO">
        <appender-ref ref="console" />
        <appender-ref ref="logstash" />
    </root>
</configuration>

无论上述哪种方式都是将采集的数据发送到ElasticSearch中，因此需要修改之前FileBeat的配置，注释掉之前输出到logstash的内容，改为输出到elasticsearch中，如下所示：

1
2
3
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["127.0.0.1:9200"]

之后在浏览器中访问http://localhost:5001/trace-one链接，可以看到任务链请求到trace-one，trace-one调用到trace-two的服务，整个任务链的日志信息都经过logstash收集发送给了elasticsearch，搜索elasticsearch可以获得本次任务链的所有节点的日志信息。

这样关于Spring Cloud Sleuth分布式服务跟踪整合ELK的学习就到此为止，后续开始学习如何整合Zipkin。