情境描述

微服务架构中，除了Spring Cloud所需的组件，如网关、Eureka注册中心、配置中心等，还有大量经过业务拆分生成的微服务节点。如何有效地收集汇总各个微服务节点的日志，对于应对微服务架构的复杂性有很大的帮助。
一个良好的微服务日志中心需具备方便查询、可视化展示等特点。

技术选型

针对上述情景，最终选定使用ELK构建日志中心

ElasticSearch

ElasticSearch是一个开源的分布式的搜索引擎，它负责存储数据，并且提供了许多灵活而实用的Rest API，所以，上层应用可以根据需要去查询数据，使用数据，分析数据。在日志中心，所有的日志数据都存储到ElasticSearch中，借助其强大的搜索能力，可以很灵活地查询日志。

Logstash

Logstash主要用于收集数据，并将数据保存到ElasticSearch中。
Logstash提供许多插件，易于扩展。Logstash收集到数据后，可以做很多处理，最终再将数据输出到ElasticSearch中。在日志中心，它主要负责采集应用的日志。
2.3 Kibana
Kibana主要负责读取ElasticSearch中的数据，并进行可视化展示。它还自带Tool，可以方便调用ElasticSearch的Rest API。在日志中心，我们通过Kibana查看日志。

搭建日志中心

为快速搭建日志中心，我们选择采用Docker部署ELK。

1. 首先需要修改系统的vm.max_map_count参数，否则ELK容器启动后会由于vm.max_map_count数值偏小，而自动停止运行

$ vim /etc/sysctl.conf
#末尾添加一行
vm.max_map_count=65536
$ sysctl -p
vm.max_map_count=65536

2. 启动ELK容器

$ docker run -p 5601:5601 -p 9200:9200 -p 5044:5044 --name log-center -d sebp/elk

3. 修改logstash配置
   由于elk中logstash的默认input方式是filebeat，此处我们将其input方式改为input方式。

# 进入容器
$ docker exec -it log-center /bin/bash
$ vim /etc/logstash/conf.d/02-beats-input.conf
input {        
   tcp {                 
      port => 5044                 
      codec => json_lines                 
   }     
} 
output{      
   elasticsearch {
      hosts => ["localhost:9200"]
      # 为传入的日志索引命名
      index => "app-log-%{+YYYY.MM.dd}"
   }     
}
$ docker restart log-center

微服务应用改造

在pom.xml文件中添加依赖

<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>5.3</version>
</dependency>

在项目的resources文件夹下添加logback.xml文件

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml" />
    <conversionRule conversionWord="ip" converterClass="com.example.elk.LocalIPConfig" />
    <springProperty scope="context" name="springAppName" source="spring.application.name" />
    <!-- 为logstash输出的JSON格式的Appender -->
    <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>192.168.153.138:5044</destination>
        <!-- 日志输出编码 -->
        <encoder
                class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <timestamp>
                    <timeZone>UTC</timeZone>
                </timestamp>
                <pattern>
                    <pattern>
                        {
                        "host": "%ip",
                        "level": "%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}",
                        "log": "%message"
                        }
                    </pattern>
                </pattern>
            </providers>
        </encoder>
    </appender>

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

附录：

1. 如何让ELK展示的日志显示服务名
A: 通过在logback.xml中配置 :
    <springProperty scope="context" name="springAppName" source="spring.application.name" />
    该配置会读取配置文件application.yml的spring.application.name的值，获取到服务名
    <encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <pattern>
                    <pattern>
                        {
                        "service": "${springAppName:-}"
                        }
                    </pattern>
                </pattern>
            </providers>
    </encoder>
2. 如何让ELK展示的日志显示服务所在主机IP
A: 编写一个类继承`ClassicConverter`
package com.example.elk;

import ch.qos.logback.classic.pattern.ClassicConverter;
import ch.qos.logback.classic.spi.ILoggingEvent;

import java.net.InetAddress;
import java.net.UnknownHostException;

public class LocalIPConfig extends ClassicConverter {
    @Override
    public String convert(ILoggingEvent event) {
        try {
            // 获取主机IP
            return InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
        }
        return null;
    }
}
然后在logback.xml中配置：
<conversionRule conversionWord="ip" converterClass="com.example.elk.LocalIPConfig" />
<encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
        <providers>
            <pattern>
                <pattern>
                    {
                    "host": "%ip"
                    }
                </pattern>
            </pattern>
        </providers>
</encoder>

编写一个接口

package com.example.elk;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TestController {
    private Logger logger = LoggerFactory.getLogger(TestController.class);

    @GetMapping("/test")
    public String testlogstash() {
        logger.info("ELK测试");
        return "hello word";
    }
}

此时启动应用。通过postman向/test接口发送请求
登陆kibana，即可看到此时日志可以只展示host、service、log三部分内容，有效地减轻了我们使用微服务时日志查看时的工作量