docker实战之ELK

docker搭建ELK

首先我们搭建ELK环境, 然后再引入缓存Kafka, 数据流为:

log-->logstash-agent-->kafka-->logstash-->es-->kibana

首先全局了解下文件结构

.
├── docker-compose.yml
├── kafka
│   ├── data
│   │   └── kafka-logs-e07760a64ec8
│   │       ├── cleaner-offset-checkpoint
│   │       ├── log-start-offset-checkpoint
│   │       ├── meta.properties
│   │       ├── recovery-point-offset-checkpoint
│   │       ├── replication-offset-checkpoint
│   │       └── test-0
│   │           ├── 00000000000000000000.index
│   │           ├── 00000000000000000000.log
│   │           ├── 00000000000000000000.timeindex
│   │           └── leader-epoch-checkpoint
│   └── docker.sock
├── kibana
│   └── kibana.yml
├── logstash
│   └── logstash.conf
└── zookeeper
    ├── data
    └── datalog

docker-compose-elk.yml

version: "3"
services:
    elasticsearch:
        image: elasticsearch:7.7.0
        container_name: elasticsearch
        hostname: elasticsearch
        environment:
            - "discovery.type=single-node"
        ports:
            - 9200:9200
            - 9300:9300
        networks:
            - elknetwork
    kibana:
        image: kibana:7.7.0
        container_name: kibana
        hostname: kibana
        ports:
            - 5601:5601
        volumes:
            - ./kibana:/usr/share/kibana/config/
        links:
            - elasticsearch:elasticsearch
        depends_on:
          - elasticsearch
        networks:
          - elknetwork
    logstash:
        image: logstash:7.7.0
        container_name: logstash
        hostname: logstash
        ports:
            - 9600:9600
            - 8089:8089
        volumes:
          - ./logstash:/usr/share/logstash/pipeline/
        links:
            - elasticsearch:elasticsearch
        depends_on:
          - elasticsearch
        networks:
          - elknetwork
    zookeeper:
        image: wurstmeister/zookeeper
        container_name: zookeeper
        restart: unless-stopped
        ports:
          - "2181:2181"
        volumes:
          - ./zookeeper/data:/data
          - ./zookeeper/datalog:/datalog
        networks:
          - elknetwork
    kafka:
        image: wurstmeister/kafka
        container_name: kafka
        restart: unless-stopped
        ports:
          - "9092:9092"
        environment:
          KAFKA_ADVERTISED_HOST_NAME: 192.168.124.5   # 用ifconfig查询，或直接填写kafka
          KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
          KAFKA_CREATE_TOPICS: "test:1:1"
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.124.5:9092   ## 修改:宿主机IP
        volumes:
          - "./kafka/docker.sock:/var/run/docker.sock"
          - "./kafka/data/:/kafka"
        depends_on:
          - zookeeper
        networks:
          - elknetwork
networks:
    elknetwork:
        driver: bridge

注意: logstash没数据源时会主动停止进程, 勿慌

kibana.yml

server.name: kibana
server.host: "0"
elasticsearch.hosts: [ "http://elasticsearch:9200" ]
monitoring.ui.container.elasticsearch.enabled: true
i18n.locale: "zh-CN"

logstash.conf

input {
  tcp {
    port => 8089
  }
}

output {
  elasticsearch { hosts => ["elasticsearch:9200"] }
}

docker中logstash默认配置(/usr/share/logstash/config/pipelines.yml)

- pipeline.id: main
  path.config: "/usr/share/logstash/pipeline"

docker中logstash默认配置(/usr/share/logstash/config/logstash.yml)

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]

熟悉docker中默认配置

# 进入docker容器
docker exec -it logstash /bin/bash
# 查看文件目录
ls
# 查看docker中配置文件
more config/logstash.yaml

• logstash中 /usr/share/logstash/config/pipelines.yml 将配置文件指向 /usr/share/logstash/pipeline 目录下, 该目录下是我们真实的配置文件目录
• kibana中 /usr/share/kibana/config/kibana.yml 可以修改为中文

log4j2通过tcp将日志输出到logstash

pom.xml

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-logging</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-log4j2</artifactId>
        </dependency>
<!--         <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.6.2</version>
        </dependency> -->

log4j2-spring.xml

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="OFF" monitorInterval="60">
    <Appenders>
        <!-- Console 日志，只输出 level 及以上级别的信息，并配置各级别日志输出颜色 -->
        <Console name="Console" target="SYSTEM_OUT">
            <!--控制台只输出level及以上级别的信息（onMatch），其他的直接拒绝（onMismatch）-->
            <ThresholdFilter level="info" onMatch="ACCEPT" onMismatch="DENY"/>
            <PatternLayout pattern="%highlight{%d{yyyy.MM.dd 'at' HH:mm:ss z} %-5level %class{36} %M() @%L - %msg%n}{FATAL=Bright Red, ERROR=Bright Magenta, WARN=Bright Yellow, INFO=Bright Green, DEBUG=Bright Cyan, TRACE=Bright White}"/>
        </Console>
        <!-- socket 日志，输出日志到 Logstash 中做日志收集 -->
        <Socket name="Socket" host="127.0.0.1" port="8089" protocol="TCP">
            <JsonLayout properties="true" compact="true" eventEol="true" />
            <PatternLayout pattern="%d{yyyy.MM.dd 'at' HH:mm:ss z} %-5level %class{36} %M() @%L - %msg%n"/>
        </Socket>
    </Appenders>
    <Loggers>
        <Root level="INFO">
            <property name="hostName">cuishiying</property>
            <property name="applicationName">elk-demo</property>
            <appender-ref ref="Console"/>
            <appender-ref ref="Socket"/>
        </Root>
    </Loggers>
</Configuration>

测试

@RestController
@RequestMapping("/test")
public class LogController {

    private Logger logger = LogManager.getLogger(LogController.class);

    // http://127.0.0.1:8080/test/log4j2
    @RequestMapping(value = "/log4j2", method = RequestMethod.GET)
    public String testLog(){
        try {
            MDC.put("traceid", String.valueOf(System.currentTimeMillis()));
            logger.info("Hello 这是 info message. 信息");
            logger.error("Hello 这是 error message. 报警");
            logger.warn("Hello 这是 warn message. 警告");
            logger.debug("Hello 这是 debug message. 调试");
            List<String> list = new ArrayList<>();
            System.out.println(list.get(2));
        } catch (Exception e) {
            logger.error("testLog", e);
        } finally {
            MDC.clear();
        }
        return "";
    }
}

项目启动后，点击management可以看到:
logstash会按默认规则自动在ES中创建索引。我们需要手动创建kibana索引，
然后在discover中即可看到日志

用socket方式将日志传输给logstash，如果把logstash停掉然后再启动，日志就无法继续传输了，也就是说socket无法自动重连，这在生产环境中，当然是个隐患。所以生产环境一般会用Logstash-gelf

<?xml version="1.0" encoding="UTF-8"?>
<Configuration>
    <Properties>
        <Property name="LOG_PATTERN">{"logger": "%logger", "level": "%level", "msg": "%message"}%n</Property>
    </Properties>
    <Appenders>
        <Console name="Console" target="SYSTEM_OUT" follow="true">
            <PatternLayout pattern="${LOG_PATTERN}"/>
        </Console>
        <Gelf name="logstash-gelf" host="udp:127.0.0.1" port="4567" version="1.1" ignoreExceptions="true">
            <Field name="timestamp" pattern="%d{yyyy-MM-dd HH:mm:ss.SSS}" />
            <Field name="logger" pattern="%logger" />
            <Field name="level" pattern="%level" />
            <Field name="simpleClassName" pattern="%C{1}" />
            <Field name="className" pattern="%C" />
            <Field name="server" pattern="%host" />
        </Gelf>
    </Appenders>
    <Loggers>
        <Root level="INFO">
            <AppenderRef ref="Console"/>
            <AppenderRef ref="logstash-gelf" />
        </Root>
    </Loggers>
</Configuration>

log4j2将日志输出到kafka

log4j2-spring.xml

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="OFF" monitorInterval="60">
    <Properties>
        <property name="log_pattern_console">%highlight{%d{yyyy.MM.dd 'at' HH:mm:ss z} %-5level %class{36} %M() @%L - %msg%n}{FATAL=Bright Red, ERROR=Bright Magenta, WARN=Bright Yellow, INFO=Bright Green, DEBUG=Bright Cyan, TRACE=Bright White}</property>
        <!-- 日志文件默认输出格式;%X{traceid}:链路id;%C:大写,类名;%M:方法名;%l:行号,影响性能;%m:信息;%n:换行 -->
        <property name="log_pattern">%d{yyyy-MM-dd HH:mm:ss.SSS} [%X{traceid}] [%-5level] %l - %m%n</property>
    </Properties>
    <Appenders>
        <!-- Console 日志，只输出 level 及以上级别的信息，并配置各级别日志输出颜色 -->
        <Console name="Console" target="SYSTEM_OUT">
            <!--控制台只输出level及以上级别的信息（onMatch），其他的直接拒绝（onMismatch）-->
            <ThresholdFilter level="info" onMatch="ACCEPT" onMismatch="DENY"/>
            <PatternLayout pattern="${log_pattern_console}"/>
        </Console>
        <!-- kafka 日志，输出日志到 Logstash 中做日志收集 -->
        <Kafka name="Kafka" topic="test">
            <PatternLayout pattern="${log_pattern}"/>
            <Property name="bootstrap.servers">localhost:9092</Property>
        </Kafka>
    </Appenders>
    <Loggers>
        <Root level="INFO">
            <property name="hostName">cuishiying</property>
            <property name="app_name">elk-demo</property>
            <appender-ref ref="Console"/>
            <!--<appender-ref ref="Socket"/>-->
            <AppenderRef ref="Kafka"/>
        </Root>
    </Loggers>
</Configuration>

此时，日志可以直接输出到kafka中

log4j2将日志通过kafka输出到elk中

修改logstash.conf, 将输入源设置为kafka, 输出设置为es， 以kafka的topic和日期创建es索引

logstash.conf

input {
    kafka{
        bootstrap_servers => ["kafka:9092"]
        topics_pattern  => "test*"
        group_id => "logstash-test-group"
        consumer_threads => 5
        codec => "json"   # logstash-kafka 插件输入和输出默认codec为json格式
        decorate_events => true  #此属性会将当前topic、offset、group、partition等信息也带到message中
        auto_offset_reset => "latest"
        type => "test"  #所有插件通用属性,尤其在input里面配置多个数据源时很有用
    }
    kafka{
        bootstrap_servers => ["kafka:9092"]
        topics_pattern  => "ocpc*"
        group_id => "logstash-ocpc-group"
        consumer_threads => 5
        codec => "json"
        decorate_events => true  #此属性会将当前topic、offset、group、partition等信息也带到message中
        auto_offset_reset => "latest"
        type => "ocpc"  #所有插件通用属性,尤其在input里面配置多个数据源时很有用
    }
}

filter {
   json {
       source => "message"
   }
   mutate{
       remove_field => "@version"
   }
}

output {
    if[type] == "test" {
        elasticsearch {
            hosts => ["elasticsearch:9200"]
            index => "%{[@metadata][kafka][topic]}-%{app_name}-%{+YYYYMMdd}"

            manage_template => false  # 取消logstash自定义模板功能，进而强制使用es的内置模板
        }
        stdout {
            codec => rubydebug {metadata => true}  #logstash控制台输出日志和@metadata信息
      }
    }  
    if[type] == "ocpc" {
        elasticsearch {
            hosts => ["elasticsearch:9200"]
            index => "%{[@metadata][kafka][topic]}-%{+YYYYMMdd}"

            manage_template => false  # 取消logstash自定义模板功能，进而强制使用es的内置模板
        }
        stdout {
            codec => rubydebug {metadata => true}  #logstash控制台输出日志和@metadata信息
      }
    } 
}

重启docker

docker-compose restart logstash

访问 http://localhost:5601, 可以看到日志已正常输出到ELK中, 包括异常栈。

日志链路追踪

上边的 MDC 放入traceid的操作可以通过过滤器统一放入。

package com.easyliao.auth.common.filter;
​
​
import com.easyliao.auth.common.utils.IdUtils;
import com.easyliao.auth.common.utils.IpUtil;
import com.easyliao.auth.common.utils.RequestUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
​
import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebFilter;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
​
@Slf4j
@WebFilter(urlPatterns = "/*", filterName = "logFilter")
@Order(value = Ordered.HIGHEST_PRECEDENCE)
@Component
public class LogFilter extends OncePerRequestFilter {
​
    private final String TRACE_ID = "traceid";
    private final String IP = "ip";
    private final String DEFAULT_TRACE_ID = "0";
​
    @Override
    protected void doFilterInternal(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, FilterChain filterChain) throws ServletException, IOException {
        try {
            // traceId初始化
            initTraceId(httpServletRequest);
            // 初始化ip
            initIp(httpServletRequest);
            // 入口信息
            log(httpServletRequest, httpServletResponse);
            // 执行后续过滤器
            filterChain.doFilter(httpServletRequest,httpServletResponse);
        } finally {
            afterLog(httpServletRequest, httpServletResponse);
        }
    }
​
    private void log(HttpServletRequest request, HttpServletResponse response){
        if (!log.isInfoEnabled()) {
            return;
        }
        log.info("\n请求地址: [{}] \n请求参数: [{}]",
                request.getRequestURL().toString(),
                RequestUtils.getFormParams(request)
        );
    }
​
    private void afterLog(HttpServletRequest req, HttpServletResponse response) {
        MDC.remove(TRACE_ID);
        MDC.remove(IP);
    }
​
    private void initIp(HttpServletRequest servletRequest) {
        MDC.put(IP, IpUtil.getIpAddr(servletRequest));
    }
​
    private void initTraceId(HttpServletRequest request) {
        //尝试获取http请求中的traceId
        String traceId = request.getParameter(TRACE_ID);
​
        //如果当前traceId为空或者为默认traceId，则生成新的traceId
        if (StringUtils.isBlank(traceId) || this.defaultTraceId(traceId)){
            traceId = this.genTraceId();
        }
​
        //设置traceId
        this.setTraceId(traceId);
    }
​
    public Boolean defaultTraceId(String traceId) {
        return DEFAULT_TRACE_ID.equals(traceId);
    }
​
    public String genTraceId() {
        return IdUtils.uuid();
    }
​
    public void setTraceId(String traceId) {
        //如果参数为空，则设置默认traceId
        traceId = StringUtils.isBlank(traceId) ? DEFAULT_TRACE_ID : traceId;
        //将traceId放到MDC中
        MDC.put(TRACE_ID, traceId);
    }
    public String getTraceId() {
        //获取
        String traceId = MDC.get(TRACE_ID);
        //如果traceId为空，则返回默认值
        return StringUtils.isBlank(traceId) ? DEFAULT_TRACE_ID : traceId;
    }
}

logback+kafka+elk

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.4.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>elk-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>elk-demo</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>


    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.github.danielwegener</groupId>
            <artifactId>logback-kafka-appender</artifactId>
            <version>0.2.0-RC2</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

logback-spring.xml

<?xml version="1.0" encoding="UTF-8"?>
<!-- scan：当此属性设置为true时，配置文件如果发生改变，将会被重新加载，默认值为true。 scanPeriod：设置监测配置文件是否有修改的时间间隔，如果没有给出时间单位，
    默认单位是毫秒当scan为true时，此属性生效。默认的时间间隔为1分钟。 debug：当此属性设置为true时，将打印出logback内部日志信息，实时查看logback运行状态。
    默认值为false。 -->
<!-- <configuration scan="false" scanPeriod="60 seconds" debug="false"> -->
<configuration>

    <!-- 日志文件默认输出格式;%X{traceid}:链路id;%C:大写,类名;%M:方法名;%l:行号,影响性能;%m:信息;%n:换行 -->
    <property name="log_pattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%X{traceid}] [%-5level] %L - %m%n"/>

    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${log_pattern}</pattern>
        </encoder>
    </appender>

    <!-- This is the kafkaAppender -->
    <appender name="kafkaAppender" class="com.github.danielwegener.logback.kafka.KafkaAppender">
        <encoder>
            <pattern>${log_pattern}</pattern>
        </encoder>
        <topic>test</topic>
        <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy" />
        <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />

        <!-- Optional parameter to use a fixed partition -->
        <!-- <partition>0</partition> -->

        <!-- Optional parameter to include log timestamps into the kafka message -->
        <!-- <appendTimestamp>true</appendTimestamp> -->

        <!-- each <producerConfig> translates to regular kafka-client config (format: key=value) -->
        <!-- producer configs are documented here: https://kafka.apache.org/documentation.html#newproducerconfigs -->
        <!-- bootstrap.servers is the only mandatory producerConfig -->
        <producerConfig>bootstrap.servers=localhost:9092</producerConfig>

        <!-- this is the fallback appender if kafka is not available. -->
        <appender-ref ref="STDOUT" />
    </appender>

    <root level="info">
        <appender-ref ref="STDOUT" />
        <appender-ref ref="kafkaAppender" />
    </root>
</configuration>

application.properties

logging.config=classpath:logback-spring.xml

最后

本文到此结束，感谢阅读。如果您觉得不错，请关注公众号【当我遇上你】，您的支持是我写作的最大动力。

参考

• https://medium.com/@harisshafiq08/elk-stack-deployment-through-docker-compose-98ce40ff2fb6
• https://2much2learn.com/centralized-logging-with-kafka-and-elk-stack/
• https://yangbingdong.com/2018/spring-boot-docker-elk/
• https://github.com/Haris3243/docker-elkstack
• http://liupzmin.com/2018/03/20/bigdata/logstash-consume-from-kafka-to-elasticsearch/