<!-- SkyWalking 集成 Logback 日志，自动注入追踪ID -->
<dependency>
    <groupId>org.apache.skywalking</groupId>
    <artifactId>apm-toolkit-logback-1.x</artifactId>
    <version>8.9.0</version>
</dependency>

在 logback-spring.xml 中配置 TraceId 输出格式

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
    <!-- 引入 SkyWalking 日志转换器 -->
    <conversionRule conversionWord="tid" converterClass="org.apache.skywalking.apm.toolkit.log.logback.v1.x.TraceIdPatternConverter"/>

    <!-- 控制台输出 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <!-- %tid 就是 SkyWalking 的 TraceId，核心字段 -->
            <pattern>%d{yyyy-MM-dd HH:mm:ss} [%tid] [%thread] %-5level %logger{50} - %msg%n</pattern>
            <charset>UTF-8</charset>
        </encoder>
    </appender>

    <!-- 根日志配置 -->
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>
</configuration>

2 启动 FileBeat 收集本地日志

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - f:\logs\cmmall\*.log
    # 多行日志合并（Java 异常栈专用）
    multiline.type: pattern
    multiline.pattern: '^\d{4}\-\d{2}\-\d{2}\s\d{2}\:\d{2}\:\d{2}\.\d{3}'
    multiline.negate: true
    multiline.match: after

# 输出到 Logstash
output.logstash:
  hosts: ["192.168.xx.xxx:xxxx"]

#启动 FileBeats
filebeat.exe -e -c filebeat.yml

input {
  beats {
    port => XXXX
    codec => "json"
  }
}

filter {
  # 解析带 SkyWalking TraceId 的 Java 日志
  grok {
    match => { "message" => '%{TIMESTAMP_ISO8601:log_time}\s+%{LOGLEVEL:level}\s+\[TID:%{DATA:trace_id}\]\s+\[%{DATA:thread}\]\s+%{DATA:class}\s+-\s+%{GREEDYDATA:content}' }
  }

  # 把 log_time 转成 ES 可识别的时间格式
  date {
    match => [ "log_time", "yyyy-MM-dd HH:mm:ss.SSS" ]
    target => "@timestamp"
  }

  # 删除无用字段，节省存储空间
  mutate {
    remove_field => [ "message", "tags", "host", "@version" ]
  }
}

output {
  elasticsearch {
    hosts => ["192.168.XX.XXX:XXXX"]
    index => "cmmall-log-%{+YYYY.MM.dd}"  # 按天分索引，更规范
  }
}