基于LOGBACK实现的分布式日志跟踪

站长

2024年03月11日 18:40 · 阅读数 15

背景是随着微服务架构的兴起，应用程序被拆分为多个独立的服务，每个服务负责特定的功能。在这种架构下，一个请求可能会经过多个不同的服务，跨越多个服务器和实例。为了追踪和关联请求在分布式系统中的流转路径，分布式日志跟踪变得必要。分布式日志跟踪是在分布式系统中追踪和关联多个服务之间的请求的过程。它允许你在整个分布式系统中跟踪请求的流转路径，以便进行故障排查、性能分析和日志分析。

常见的分布式日志跟踪方案

生成全局唯一的 Trace ID：在每个请求的入口处生成一个全局唯一的 Trace ID。可以使用 UUID 或其他唯一标识符生成算法来生成 Trace ID。
传递 Trace ID：在分布式系统的服务之间传递 Trace ID。可以通过请求头、消息队列消息的属性或其他适当的方式在服务之间传递 Trace ID。确保在请求的转发过程中，始终将 Trace ID 传递给下游服务。
记录 Trace ID：在每个服务的日志中记录 Trace ID。在日志中包含 Trace ID 可以帮助你跟踪和关联请求的流转路径。可以使用日志框架（如 Logback）的上下文信息（如 MDC）或自定义的方式将 Trace ID 添加到日志中。
日志存储与查询：将日志集中存储到适当的位置，如日志服务器、分布式日志存储系统或日志分析平台。确保你的日志存储方案能够支持根据 Trace ID 进行查询和分析。
日志关联和查询：使用 Trace ID 进行日志关联和查询。通过在日志存储中搜索特定的 Trace ID，你可以获取与该 Trace ID 相关的所有日志记录，从而跟踪请求的流转路径，分析请求的处理过程，以及排查潜在的问题。

基于LOGBACK的分布式日志跟踪方案

LOGBACK介绍

Logback 是一个功能强大、灵活且高效的日志框架，它是由 Ceki Gülcü 创建的，也是 Log4j 框架的继任者。Logback 的设计目标是成为 Java 应用程序的首选日志框架，它具有以下主要特点：

高性能：Logback 在日志记录和处理方面具有优异的性能。它通过利用异步日志记录、可配置的缓冲区大小和高效的算法来实现高吞吐量和低开销的日志操作。
灵活配置：Logback 提供了丰富的配置选项，允许开发人员根据不同的需求进行灵活的日志配置。它支持通过 XML、Groovy 或编程方式进行配置，使用户能够根据具体的应用环境和需求进行定制。
多种输出目标：Logback 可以将日志输出到多种目标，包括控制台、文件、远程服务器、各种数据库和消息队列等。它支持多种输出格式，如文本、JSON、XML 等。
级别和过滤器：Logback 提供了丰富的日志级别（如 DEBUG、INFO、WARN、ERROR 等）和过滤器选项，使开发人员能够根据需要对日志进行过滤和筛选。
上下文支持：Logback 支持多线程和多上下文的环境，并提供了线程安全的日志记录器和上下文对象，以确保并发环境下的正确日志输出。

引入LOGBACK

基于spring boot 首先引入maven相关配置

<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
    <version>1.2.10</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-ext</artifactId>
    <version>1.7.30</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>log4j-over-slf4j</artifactId>
    <version>1.7.30</version>
</dependency>

接着配置logback.xml文件，配置文件如下

<?xml version="1.0" encoding="UTF-8"?><!DOCTYPE configuration>
<configuration debug="false" scan="true" scanPeriod="30 seconds">
    <contextName>hello.world</contextName>
    <property name="log.base" value="${work.home}/logs"/>
    <property name="module" value="test"/>
    <shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
    <conversionRule conversionWord="msg" converterClass="com.toby.dynamic.data.source.logback.MessageFilterConverter"/>
    <conversionRule conversionWord="localIP" converterClass="com.toby.dynamic.data.source.logback.LocalIPConverter"/>
    <conversionRule conversionWord="thid" converterClass="com.toby.dynamic.data.source.logback.ThreadIdConverter"/>

    <turboFilter class="com.toby.dynamic.data.source.logback.TraceIdTurboFilter"/>


    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度,${module}:模块,%thid:线程ID,
            %localIP:本地IP,%X{traceId}:分布式跟踪TRACEID,$%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS}|%-5level|${module}|%thid|%localIP|%X{traceId}|%.-51200msg|%logger%n
            </pattern>
        </encoder>
    </appender>

    <!--文件日志， 按照每天生成日志文件 -->
    <appender name="RUNNING_LOG" class="com.toby.dynamic.data.source.logback.SpRollingFileAppender">
        <file>${log.base}/${module}_running.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <!--日志文件输出的文件名-->
            <FileNamePattern>${log.base}/archive/${module}_running.%d{yyyy-MM-dd}.%i.log</FileNamePattern>
            <!--日志文件保留天数-->
            <MaxHistory>30</MaxHistory>
            <MaxFileSize>100MB</MaxFileSize>
        </rollingPolicy>
        <immediateFlush>false</immediateFlush>
        <idleTimeForceFlush>300</idleTimeForceFlush>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度,${module}:模块,%thid:线程ID,
            %localIP:本地IP,%X{traceId}:分布式跟踪TRACEID,$%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS}|%-5level|${module}|%thid|%localIP|%X{traceId}|%.-51200msg|%logger%n
            </pattern>
            <charset>UTF-8</charset>
        </encoder>
        <!--日志文件最大的大小-->
        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
            <MaxFileSize>100MB</MaxFileSize>
        </triggeringPolicy>
    </appender>
    <!--文件日志， 按照每天生成日志文件 -->
    <appender name="RUNNING_LOG_ASYNC" class="ch.qos.logback.classic.AsyncAppender">
        <discardingThreshold>0</discardingThreshold>
        <queueSize>1024</queueSize>
        <appender-ref ref="RUNNING_LOG"/>
    </appender>

    <appender name="INTERFACE_LOG" class="com.toby.dynamic.data.source.logback.SpRollingFileAppender">
        <file>${log.base}/${module}_interface.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
            <!--日志文件输出的文件名-->
            <FileNamePattern>${log.base}/archive/${module}_interface.%i.log</FileNamePattern>
            <!--日志文件保留天数-->
            <minIndex>1</minIndex>
            <maxIndex>10</maxIndex>
        </rollingPolicy>
        <immediateFlush>false</immediateFlush>
        <idleTimeForceFlush>3000</idleTimeForceFlush>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度,${module}:模块,%thid:线程ID,
            %localIP:本地IP,%X{traceId}:分布式跟踪TRACEID,$%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS}|%-5level|${module}|%thid|%localIP|%X{traceId}|%.-51200msg|%logger%n
            </pattern>
            <charset>UTF-8</charset>
        </encoder>
        <!--日志文件最大的大小-->
        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
            <MaxFileSize>100MB</MaxFileSize>
        </triggeringPolicy>
    </appender>
    <!--文件日志， 按照每天生成日志文件 -->
    <appender name="INTERFACE_LOG_ASYNC" class="ch.qos.logback.classic.AsyncAppender">
        <discardingThreshold>0</discardingThreshold>
        <queueSize>1024</queueSize>
        <appender-ref ref="INTERFACE_LOG"/>
    </appender>

    <logger name ="com.toby.dynamic" additivity="false" level="INFO">
        <appender-ref ref="INTERFACE_LOG_ASYNC"/>
        <appender-ref ref="RUNNING_LOG_ASYNC"/>
    </logger>

    <root level="DEBUG">
        <appender-ref ref="STDOUT"/>
        <appender-ref ref="RUNNING_LOG_ASYNC"/>
    </root>
</configuration>

相关定制改造

定制基于`TurboFilter`的`TraceIdTurboFilter`

核心是将上下文中的traceID放到LOGBACK的MDC上下文中核心代码如下：

public class TraceIdTurboFilter extends TurboFilter {
    private static final String TRACE_ID = "traceId";
    @Override
    public FilterReply decide(Marker marker, Logger logger, Level level, String s, Object[] objects, Throwable throwable) {
        String traceId = Context.getParameter(TRACE_ID,null);
        if (StringUtil.isEmpty(traceId)) {
            traceId = MDC.get(TRACE_ID);
            if (StringUtil.isEmpty(traceId)) {
                MDC.put(TRACE_ID, IdGenerator.createID());
            }
        } else {
            MDC.put(TRACE_ID, traceId);
        }
        return FilterReply.NEUTRAL;
    }
}

上下文实现如下，主要逻辑是将rpc调用上下文或者http调用header中的traceID塞入到线程本地变量中

public class Context {
    private static final ThreadLocal<Context> CONTEXT = new ThreadLocal<Context>() {
        @Override
        protected Context initialValue() {
            return new Context();
        }
    };

    private final Map<String, String> parameters = new HashMap<>();
    private final Map<String, String> localValues = new HashMap<>();

    public Context() {

    }

    public static void setContext(Context context) {
        CONTEXT.set(context);
    }

    public static Context getContext() {
        return CONTEXT.get();
    }

    public static void setParameter(String key, String value) {
        CONTEXT.get().parameters.put(key, value);
    }

    public static String getParameter(String key, String defaultValue) {
        return CONTEXT.get().parameters.getOrDefault(key, defaultValue);
    }

    public static void set(String key, String value) {
        CONTEXT.get().localValues.put(key, value);
    }

    public static String get(String key, String defaultValue) {
        return CONTEXT.get().localValues.getOrDefault(key, defaultValue);
    }

    public static String remove(String key) {
        return CONTEXT.get().localValues.remove(key);
    }

    public static void clear() {
        CONTEXT.remove();
    }

}

这一块是分布式日志跟踪的核心时间，一个请求从前端进入，生成唯一的traceId，后端各个微服务的请求日志中的该笔请求的日志都是基于该traceID的后续可以在ELK等平台很容易通过该traceId来找到对应的所有日志

定制日志刷盘处理

基于LOGBACK的RollingFileAppender，如果设置immediateFlush为false，不会立刻刷盘，需要等日志缓冲区满了才会刷盘。这要会导致有些场景，日志不会立刻刷入磁盘，这里需要进行相关改造，增加定制刷盘逻辑

public class SpRollingFileAppender extends RollingFileAppender<ILoggingEvent> {
    private long idleTimeForceFlush = 0;

    private volatile long lastFlushTime = System.currentTimeMillis();

    private Timer timer = null;

    public long getIdleTimeForceFlush() {
        return idleTimeForceFlush;
    }

    public void setIdleTimeForceFlush(long idleTimeForceFlush) {
        this.idleTimeForceFlush = idleTimeForceFlush;
    }

    @Override
    public void start() {
        super.start();
        if (needForceFlush()) {
            timer = new Timer(true);
            timer.schedule(new FlushTask(), 1000L, idleTimeForceFlush);
        }
    }

    @Override
    protected void subAppend(ILoggingEvent event) {
        super.subAppend(event);
        if (needForceFlush()) {
            lastFlushTime = System.currentTimeMillis();
        }
    }

    private boolean needForceFlush() {
        return !isImmediateFlush() && idleTimeForceFlush > 0;
    }

    private void flush() {
        lock.lock();
        try {
            OutputStream outputStream = getOutputStream();
            if (!Objects.isNull(outputStream)) {
                outputStream.flush();
            }
        } catch (Throwable e) {
            addStatus(new ErrorStatus("flush IO failure in timer task.", this, e));
        } finally {
            lock.unlock();
        }
    }

    private class FlushTask extends TimerTask {

        @Override
        public void run() {
            if (System.currentTimeMillis() - lastFlushTime > idleTimeForceFlush) {
                flush();
            }
        }
    }
}

定制日志打印相关信息

打印本地IPLocalIPConverter 对文本进行二次处理MessageFilterConverter 获取线程IDThreadIdConverter

因上述基于实现比较简单，不再贴相关代码实现

日志打印最终效果

test_running.log打印效果如下

2023-09-20 17:15:04.374|WARN |test|22|127.0.0.1|205c279beb804a14894d8d33294034c7|removeAbandoned is true, not use in production.|com.alibaba.druid.pool.DruidDataSource
2023-09-20 17:15:04.458|INFO |test|22|127.0.0.1|205c279beb804a14894d8d33294034c7|{dataSource-4} inited|com.alibaba.druid.pool.DruidDataSource
2023-09-20 17:15:04.820|INFO |test|22|127.0.0.1|205c279beb804a14894d8d33294034c7|Started UserPointBalanceMapperTest in 3.69 seconds (JVM running for 5.555)|com.toby.dynamic.data.source.db.dao.UserPointBalanceMapperTest
2023-09-20 17:15:05.050|INFO |test|22|127.0.0.1|205c279beb804a14894d8d33294034c7|changeDataSource to data source=db_w_2|com.toby.dynamic.data.source.db.ShardingDataSourceAspect
2023-09-20 17:15:05.071|INFO |test|22|127.0.0.1|205c279beb804a14894d8d33294034c7|dataSource=db_w_2|com.toby.dynamic.data.source.db.ShardingDataSource
2023-09-20 17:15:05.138|INFO |test|22|127.0.0.1|205c279beb804a14894d8d33294034c7|changeDataSource end.|com.toby.dynamic.data.source.db.ShardingDataSourceAspect
2023-09-20 17:15:05.138|INFO |test|22|127.0.0.1|205c279beb804a14894d8d33294034c7|UserPointBalance{id=null, uid=3, type='2', value=100, expireTime=Sun Dec 31 00:00:00 CST 2023, country='CN'}|com.toby.dynamic.data.source.db.dao.UserPointBalanceMapperTest