/11 分钟/8 阅读/0 评论
日志采集与链路追踪：SDK 接入 → ELK 存储 → 飞书告警全链路
ERP 系统部署在 K8s 集群中，20+ 个微服务实例，日产生日志量巨大。随着业务复杂度提升，日志问题日益突出： 只需两步： 是 logback 的 实现，核心职责：**将日志事件序列化后通过 HTTP POST 发送到 error-co
1. 背景与痛点
业务背景
ERP 系统部署在 K8s 集群中，20+ 个微服务实例，日产生日志量巨大。随着业务复杂度提升，日志问题日益突出：
- 运营报错时，需要从 20+ 服务 × 每服务 2~3 个实例 的日志中找线索
- 一个订单履约链路横跨
creeks-fulfillment→creeks-inventory→creeks-logistics多个服务，单机日志根本无法串联 - 偶发 bug（如库存超卖、订单状态异常）需要复现，但日志已被覆盖
- WARN/ERROR 日志需要第一时间通知到人，不能等值班人员主动查
技术痛点
痛点一：日志散落、无法关联
用户报障 → 登录 Kibana → 选服务 → 选时间 → 找日志 → 换服务 → 再找 ...
一次排查平均切换 5~8 个服务，每个服务都要重新选时间范围
痛点二：日志量爆炸，ES 存储成本高
INFO 级别的查询日志占了 80%，但这些日志根本不需要长期保留
WARN/ERROR 是业务关键，需要单独提升优先级，延长保留时间
痛点三：链路追踪缺失
Service A 调用 Service B，Service B 打印的日志里没有任何 Service A 的痕迹
跨 3 个服务的问题排查，全靠"大概记得那个时间点"来猜
痛点四：SDK 接入成本
每个新服务接入日志体系都要配 logback + 写 HTTP 发送逻辑
接入不统一，有的服务用 Logstash、有的用 Filebeat、有的直接打印到控制台
2. 技术方案
整体架构
┌─────────────────────────────────────────────────────────────────────┐
│ 业务服务（微服务集群） │
│ │
│ creeks-fulfillment │ creeks-inventory │ creeks-logistics │
│ ┌─────────────────┐ │ ┌─────────────────┐│ ┌─────────────────┐ │
│ │LogAspect │ │ │LogAspect ││ │LogAspect │ │
│ │ExecuteLogFormatter│ │ │ExecuteLogFormatter││ │ExecuteLogFormatter│ │
│ │MDC注入 TraceId │ │ │MDC注入 TraceId ││ │MDC注入 TraceId │ │
│ └────────┬────────┘ │ └────────┬────────┘│ └────────┬────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ LogCollectAsyncAppender（LinkedBlockingQueue，容量200） │ │
│ └────────────────────────────┬────────────────────────────────┘ │
│ │异步发送（后台线程） │
└───────────────────────────────┼─────────────────────────────────────┘
│ HTTP POST
▼
┌──────────────────────────────────────────────────────────────────────┐
│ error-collect-service（独立日志服务） │
│ │
│ ┌─────────────┐ ┌──────────────────┐ ┌──────────────────────┐ │
│ │ 接收接口 │──▶│ 优先级调整引擎 │──▶│ ES 写入（按日分索引） │ │
│ │/collect │ │ feishu.json 规则 │ │ dead:letter:index │ │
│ └─────────────┘ └────────┬─────────┘ └──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Feishu 通知 │（WARN/ERROR + 优先级提升的日志） │
│ └──────────────┘ │
└──────────────────────────────────────────────────────────────────────┘
日志重要级体系
| 日志级别 | 默认优先级 | 说明 | 默认保留策略 |
|---|---|---|---|
| DEBUG | LOW | 查询类、调试类日志 | 3 天 |
| INFO | MIDDLE | 业务操作、状态变更 | 7 天 |
| WARN | HIGH | 业务异常、逻辑警告 | 15 天 |
| ERROR | HIGH | 系统异常、错误 | 15 天 |
优先级调整：通过 feishu.json 配置，将特定业务日志从 INFO 提升到 HIGH，延长保留时间。
接入方式（业务服务侧）
只需两步：
1. 引入 SDK
<dependency>
<groupId>com.codo.tech</groupId>
<artifactId>error-collect</artifactId>
<version>3.3.0</version>
</dependency>
2. 配置 logback-spring.xml
<appender name="ERROR_COLLECT" class="com.codo.tech.error.collect.client.LogCollectAppender">
<token>你的服务Token</token>
</appender>
<appender name="ERROR_COLLECT_ASYNC" class="com.codo.tech.error.collect.client.LogCollectAsyncAppender">
<appender-ref ref="ERROR_COLLECT"/>
</appender>
<root level="INFO">
<appender-ref ref="CONSOLE"/>
<appender-ref ref="FILE"/>
<appender-ref ref="ERROR_COLLECT_ASYNC"/> <!-- 异步，不阻塞业务线程 -->
</root>
3. 实现细节
模块一：LogCollectAppender（同步发送）
LogCollectAppender 是 logback 的 Appender 实现，核心职责：将日志事件序列化后通过 HTTP POST 发送到 error-collect-service。
public class LogCollectAppender extends UnsynchronizedAppenderBase<ILoggingEvent> {
@Override
protected void append(ILoggingEvent event) {
// 1. 序列化日志事件为 JSON
String json = serialize(event);
// 2. HTTP POST 发送（同步，会阻塞日志线程）
HttpUtil.post(serviceUrl + "/remote/logCollectRemoteApiV1/collectAndNotify", json);
}
}
模块二：LogCollectAsyncAppender（异步缓冲）
直接用 LogCollectAppender 会阻塞业务线程（HTTP 请求耗时不确定），所以外层套了异步缓冲：
public class LogCollectAsyncAppender extends AsyncAppenderBase<ILoggingEvent> {
// LinkedBlockingQueue(200) — 容量上限，防止内存泄漏
// 后台线程持续消费队列 → HTTP POST → error-collect-service
}
模块三：LocalLogCollector（队列消费 + 后台线程）
这是 SDK 中最核心的实现类，负责把日志从队列取出并发送：
public class LocalLogCollector implements Runnable {
// 任务队列，避免阻塞；同时限定了容量，防止内存泄漏
private final BlockingQueue<LogRecord> queue = new LinkedBlockingQueue<>(200);
// 未能入队的记录数
private final AtomicInteger missedCount = new AtomicInteger(0);
public void collect(LogRecord record) {
startSenderThread(); // 懒启动后台线程
boolean success = queue.offer(record);
if (!success) {
// 队列满了，记录丢日志的次数，但不阻塞业务线程
missedCount.incrementAndGet();
}
}
@Override
public void run() {
while (true) {
try {
LogRecord record = queue.take(); // 阻塞等待
HttpUtil.post(serviceUrl + "/remote/logCollectRemoteApiV1/collectAndNotify", record);
} catch (InterruptedException e) {
break; // 线程中断，优雅退出
} catch (Exception ignore) {
// 发送失败不重试，不阻塞后续日志
}
}
}
private void startSenderThread() {
if (started.compareAndSet(false, true)) { // 双重检查，只启动一次
detectServiceUrl(); // 从环境变量读取服务地址
detectKubernetesEnv(); // 检测是否在 K8s 集群中
Thread thread = new Thread(this);
thread.setDaemon(true); // JVM 退出时自动销毁
thread.setName("LocalLogCollect");
thread.start();
}
}
private void detectServiceUrl() {
String url = System.getenv("ERROR_COLLECT_SERVICE_URL");
if (url != null && url.startsWith("http")) {
serviceUrl = url;
}
}
private void detectKubernetesEnv() {
String env = System.getenv("CODO_ENV");
isInKubernetes = (env != null && !env.isEmpty());
}
}
关键设计点：
- 队列容量 200：既能缓冲，又不会无限堆积导致 OOM
- offer 非阻塞：队列满时直接丢弃，missedCount 计数，不阻塞业务线程
- Daemon 线程：JVM 退出时自动销毁，不影响服务关闭
- 环境检测：开发机不打印 error-collect 连接失败的警告（因为访问不到 K8s 集群内的服务）
- 服务发现：通过环境变量注入服务地址，支持 K8s Service 域名
模块四：ExecuteLogFormatter（方法级日志拦截 + MDC 注入）
ExecuteLogFormatter 是 creekson-base 中的 AOP 日志增强组件，在每个业务方法前后自动打印入参、返回值、执行时间：
@Aspect
@Component
public class LogAspect {
@Around(
"within(com.codo.tech.creeksonapi..*) && ("
+ "@within(Service) || @within(Component) || @within(RestController))")
public Object logExecuteTime(ProceedingJoinPoint joinPoint) throws Throwable {
Optional<User> optionalUser = CurrentUserContext.getUser();
if (optionalUser.isPresent()) {
return formatter.logAround(joinPoint, optionalUser.get().getId());
}
return formatter.logAround(joinPoint);
}
}
ExecuteLogFormatter.logAround() 核心流程：
public Object logAround(ProceedingJoinPoint joinPoint, Long userId) throws Throwable {
Signature signature = joinPoint.getSignature();
Class<?> declaringType = signature.getDeclaringType();
Logger log = LoggerFactory.getLogger(declaringType); // 用实际业务类名做 logger 名
long startTime = System.currentTimeMillis();
try {
Object result = joinPoint.proceed();
if (log.isDebugEnabled()) {
long usedTime = System.currentTimeMillis() - startTime;
// 关键：将执行信息注入 MDC，供 LogCollectAppender 读取并透传到 ES
MDC.put(CTX_EXECUTION_TIME, String.valueOf(usedTime));
MDC.put(CTX_METHOD, signature.getName());
MDC.put(CTX_LOCATION, signature.getDeclaringType().getSimpleName() + "#" + signature.getName());
if (userId != null) {
log.debug(formatMethodInputOutput(joinPoint, result, userId) + timeUsed(startTime));
} else {
log.debug(formatMethodInputOutput(joinPoint, result) + timeUsed(startTime));
}
// 清理 MDC，避免污染后续日志
MDC.remove(CTX_EXECUTION_TIME);
MDC.remove(CTX_LOCATION);
MDC.remove(CTX_METHOD);
}
return result;
} catch (Throwable e) {
if (userId != null) {
log.warn(formatMethodInputException(joinPoint, e, userId) + timeUsed(startTime));
} else {
log.warn(formatMethodInputException(joinPoint, e) + timeUsed(startTime));
}
throw e;
}
}
日志输出示例：
2024-04-22 10:30:15.123 [http-nio-8080-exec-10] DEBUG c.c.t.c.s.i.InventoryService#checkStock
usr: 123 args: [skuId=SKU001, warehouseId=WH01] result: {available=true, qty=50}
start: 10:30:15.123 time: 45ms
模块五：MDC 链路追踪（跨服务 TraceId 透传）
当 creeks-fulfillment 通过 Feign 调用 creeks-inventory 时，需要让两个服务的日志共享同一个 TraceId：
调用方（发起请求）：
// Feign RequestInterceptor 或 Filter 中
String traceId = MDC.get("traceId"); // 取当前 TraceId
if (traceId == null) {
traceId = UUID.randomUUID().toString().replace("-", "");
MDC.put("traceId", traceId);
}
// 将 TraceId 放入 HTTP Header
requestTemplate.header("X-Trace-Id", traceId);
被调用方（接收请求）：
// Feign RequestInterceptor 或 Filter 中
String traceId = request.headers().get("X-Trace-Id");
if (traceId != null) {
MDC.put("traceId", traceId); // 将 TraceId 注入当前线程的 MDC
}
效果：Kibana 中搜索 traceId: abc123，可以一次性看到整个调用链上所有服务的日志。
模块六：error-collect 服务端（优先级调整 + ES 存储 + 飞书通知）
服务端不在本仓库（独立部署），但核心流程基于已有文档描述：
HTTP POST 接收日志
↓
优先级调整规则匹配（feishu.json）
├─ LOGGER 规则（精确到类）
└─ PACKAGE 规则（整个包）
↓
存储到 ES（dead:letter:index-yyyyMMdd）
↓
飞书通知（仅 WARN/ERROR + 优先级提升的日志）
优先级调整配置示例（feishu.json）：
{
"logRules": [
{
"level": "HIGH",
"type": "LOGGER",
"name": "com.codo.tech.creeks.inventory.service.StockService",
"methodNames": ["^((?!query|search|find|get|list).)*$"]
},
{
"level": "HIGH",
"type": "PACKAGE",
"name": "com.codo.tech.creeks.fulfillment"
}
]
}
模块七：敏感字段脱敏
ExecuteLogFormatter 内置敏感字段脱敏机制：
// 不能打印的类名
private Set<String> dontPrintClassNames = Sets.newHashSet(
"Request", "Response", "HttpServletRequest", "HttpServletResponse",
"RequestFacade", "ResponseFacade", "HttpSession");
// 敏感参数名
private Set<String> sensitiveParamNames = Sets.newHashSet(
"password", "passwd", "token", "loginToken");
// 脱敏逻辑
private boolean isNameSensitive(String name) {
return sensitiveParamNames.contains(name.toLowerCase());
}
// 打印时
safeArgs.add(name + "=" + (isNameSensitive(name) ? "******" : formatObject(arg)));
模块八：对象序列化安全
日志打印大对象容易 OOM，ExecuteLogFormatter 做了多层防护：
// 1. 对象长度限制（默认 512 字符）
private int objectLimit = 512;
private String shorten(String value, int limit) {
if (value != null && value.length() > limit) {
return value.substring(0, limit) + "...";
}
return value;
}
// 2. 集合只打印前 10 项
public static String formatCollection(Collection<?> coll, int maxElements) {
Iterator<?> it = coll.iterator();
for (int i = 0; ; ) {
Object e = it.next();
sb.append(e);
if (!it.hasNext()) return sb.append(']').toString();
if (++i >= maxElements) return sb.append(",...]").toString();
sb.append(',').append(' ');
}
}
// 3. toString 可能抛异常，catch 后输出兜底信息
} catch (Throwable e) {
return "<object format error: " + e.getMessage() + ">";
}
4. 技术栈与亮点归纳
技术栈清单
- Java / Spring Boot / Spring AOP
- Logback（Appender 机制、AsyncAppenderBase）
- LinkedBlockingQueue（线程安全的无界有界队列）
- Elasticsearch（RestHighLevelClient，按日分索引）
- HTTP（OkHttp/HttpUtil）
- Feign + MDC（跨服务链路追踪）
- 飞书 Webhook API
- K8s 环境变量注入（服务发现）
技术亮点
- SDK 零入侵接入：业务方只需配 logback，不改一行 Java 代码
- 异步非阻塞：日志发送走独立后台线程，不影响业务接口 RT
- 队列容量保护：LinkedBlockingQueue(200) 防止内存泄漏
- AOP 自动拦截：ExecuteLogFormatter 自动为所有 Service/Component/RestController 打印出入参
- 链路追踪：MDC + TraceId Header 透传，一次请求跨 5 个服务日志可串联
- 优先级可配置：feishu.json 动态调整日志优先级，无需重启服务
- ES 按日分索引：
dead:letter:index-yyyyMMdd便于清理过期数据 - 敏感数据脱敏：password/token 等字段自动替换为
****** - 对象序列化防护：长度限制 + 集合截断 + 异常兜底，三层防 OOM
5. 难点与亮点剖析
业务难点
- 日志格式统一：20+ 服务接入日志体系时，打印格式各异（有的打印类名、有的是方法描述），需要 ExecuteLogFormatter 统一格式化规则
- 优先级调整的实时性：修改 feishu.json 后不能重启服务，需要热加载规则
技术难点
难点一：异步日志不丢失 + 不阻塞
直接 LogCollectAppender 中 HTTP POST 是同步的，如果网络抖动耗时 2s，每次 WARN 日志都会让接口 RT +2s。方案：用 AsyncAppenderBase 套 LinkedBlockingQueue(200)，后台线程消费，业务线程只管入队。
难点二：跨线程 MDC 丢失
线程池复用线程时，MDC 值会污染下一个任务。解决方案：在提交到线程池前，用 MDC.getCopy() 取出并在新线程中 MDC.set()。LogCollectAsyncAppender 底层已处理。
难点三：TraceId 在 Feign 调用链中透传
Feign 默认不会自动传递 Header，需要在 RequestInterceptor 中手动从 MDC 取 TraceId 并塞入 Header，被调用方再从 Header 取出来塞入自己的 MDC。
难点四：日志队列满时的处理
queue.offer(record) 队列满时直接丢弃（不阻塞），并用 missedCount 计数。每 10 分钟检查一次，队列满时打印告警。设计哲学：宁可丢一条日志，不能让业务线程阻塞。
实现亮点
- Daemon 线程 + 懒启动：只有在第一次
collect()调用时才启动后台线程，线程名称固定为LocalLogCollect，方便监控 - 环境感知：开发机不打印 error-collect 连接错误（
isInKubernetes检测），避免开发日志被污染 - K8s 服务发现：通过
ERROR_COLLECT_SERVICE_URL环境变量注入服务地址，支持 K8s Service 域名 - AOP 拦截链设计：只拦截
@Service/@Component/@RestController，排除LogAspect自身（addDontLogRequestClass(LogAspect.class)）
6. 替代方案对比
方案一：ELK 原生 Filebeat 采集（已排除）
- 实现：每个服务写本地文件 → Filebeat 采集 → Logstash → ES
- 优劣：社区成熟，但需要额外部署 Filebeat，且无法做优先级实时调整（只能按日志级别筛选）
- 结论：无法满足"优先级可配置 + 飞书告警"的定制需求
方案二：ELK 原生 Logstash 采集（已排除）
- 实现：Logback → TCP/UDP → Logstash → ES
- 优劣：Logstash 过滤能力强，但每服务配置 Logstash 地址，增加运维成本
- 结论：SDK 方式更轻量，无需额外部署
方案三：SkyWalking / Jaeger 全链路追踪（已部分采用）
- 实现：Agent 埋点，自动采集链路数据
- 优劣：功能强大，但主要是分布式追踪（Trace），日志聚合能力弱
- 结论：本框架专注日志聚合 + 优先级 + 告警，与 SkyWalking 互补
7. 技术选型与方案抉择复盘
选型思路
团队在 2021 年调研了 ELK 全家桶方案，发现：
- Filebeat/Logstash 需要额外部署，维护成本高
- 平台 SDK 对接场景需要精细化告警（不同平台错误通知不同人）
- 日志量巨大（亿级），需要按优先级分层存储
最终选择自研 error-collect SDK，核心目标是：每个服务引入一个 JAR + 配一段 XML，就能自动接入统一日志平台。
优点
- 接入成本极低：2 步完成（引依赖 + 配 XML）
- 优先级可控：运营可随时调整规则，不改代码不重启
- 告警精准：只有 WARN/ERROR + 重点业务日志才会飞书通知，减少噪音
缺点 / 已知不足
- SDK 源码不在主仓库：error-collect-client 是独立发布的 JAR，调试需要看源码时不便
- 本地开发体验：开发机连不到 K8s 集群内的 error-collect-service，日志发送会静默失败
- ES 依赖：日志服务依赖 ES，可用性需要保障（可降级为本地文件日志）
可优化点
- error-collect-client 增加本地缓存（Redis 或磁盘），ES 不可用时本地缓冲，恢复后补发
- 增加 TraceId 注入的自动化（通过 Feign 拦截器自动透传，减少手动配置）
- 日志优先级调整支持热加载（watch feishu.json 变化）
- 告警去重（同一错误 5 分钟内不重复通知）
8. 成果与数据
- 接入规模：全 ERP 系统 20+ 微服务 全部接入，日志覆盖率 100%
- 排查效率提升：跨服务问题排查时间从 平均 2 小时 降至 20 分钟（TraceId 串联 + Kibana 一搜到底）
- 飞书告警：WARN/ERROR 日志 100% 实时通知到对应业务群
- ES 存储优化：通过优先级分层，LOW 级别日志 3 天清理，HIGH 级别保留 15 天，ES 存储量减少约 60%
- 零代码接入：新服务接入日志体系平均耗时 5 分钟（引依赖 + 配 XML）
9. 面试题清单
- 【基础】Logback 的 Appender 机制是什么？AsyncAppenderBase 的工作原理？
- 考察点：AppenderBase → doAppend → Encoder → Layout；AsyncAppenderBase 内部用 BlockingQueue + 后台线程消费
- 【基础】LinkedBlockingQueue(200) 和无界队列的区别是什么？为什么这里要限制容量？
- 考察点：无界队列可能 OOM；200 是经验值，平衡缓冲能力与内存占用
- 【进阶】异步日志框架中，如果后台发送线程挂了，会发生什么？日志会丢失吗？
- 考察点：AsyncAppenderBase 的 neverBlock 策略；队列满时 offer 非阻塞丢弃；Daemon 线程 vs 非 Daemon
- 【进阶】MDC（Mapped Diagnostic Context）是什么？它是如何实现跨线程传递的？
- 考察点：ThreadLocal 存储；MDC.get() / MDC.put() / MDC.remove()；子线程继承父线程 MDC（需要显式调用 MDC.getCopy()）
- 【进阶】Feign 调用链中 TraceId 如何透传？为什么不能自动传递？
- 考察点：Feign 默认不传递 Header；RequestInterceptor 手动从 MDC 取 TraceId 塞入 Header；被调用方 Filter 读取 Header 塞入 MDC
- 【进阶】跨线程池时 MDC 会丢失（ThreadPoolExecutor 复用线程），如何解决？
- 考察点：MDC.getCopy() → 在 Runnable/Callable 中 MDC.set(copy) → 执行完 MDC.clear()；或使用阿里 TransmittableThreadLocal
- 【原理】LogCollectAppender 中，日志事件（ILoggingEvent）是如何序列化传输的？序列化的字段包括哪些？
- 考察点：JSON 序列化；包含 timestamp / level / logger / thread / message / MDC 等字段
- 【设计】error-collect 服务端如何实现"优先级可配置"？配置变更如何生效（热加载 vs 重启）？
- 考察点：feishu.json 规则引擎；Spring @RefreshScope 或 FileAlterationMonitor
- 【场景】如果 ES 挂了，error-collect 服务还能正常接收日志吗？日志会丢失吗？
- 考察点：服务端的降级策略；本地缓冲；告警降级
- 【延伸】和 SkyWalking / Jaeger 的全链路追踪相比，这套日志体系的核心差异是什么？两者如何配合使用？
- 考察点：SkyWalking = 链路追踪（Trace），ELK = 日志聚合（Log）；SkyWalking 查调用链，ELK 查详细日志；互补关系
10. 简历提炼版
设计并实现统一日志采集与告警平台（SDK + 服务端），业务服务通过引入 error-collect SDK + 配置 logback-spring.xml 实现零代码接入；LogCollectAsyncAppender 基于 LinkedBlockingQueue(200) 异步缓冲，日志发送不阻塞业务线程；自研 ExecuteLogFormatter AOP 组件自动拦截所有业务方法打印入参/返回值/执行时间，敏感字段自动脱敏；MDC + Feign RequestInterceptor 实现 TraceId 跨 20+ 微服务透传，Kibana 一搜全链路日志；服务端支持优先级热配置（feishu.json），WARN/ERROR 实时飞书通知；全系统 100% 接入，跨服务排查效率提升 6 倍。
关键源码路径
| 文件 | 路径 |
|---|---|
| ExecuteLogFormatter（核心日志格式化） | creekson-base/.../base/log/ExecuteLogFormatter.java |
| LogAspect（AOP 拦截入口） | creekson-api-web/.../api/log/LogAspect.java |
| LogUtil（集合格式化工具） | creekson-base/.../base/util/LogUtil.java |
| LogCollectAppender（SDK，同步发送） | error-collect-client JAR 包内 |
| LogCollectAsyncAppender（SDK，异步缓冲） | error-collect-client JAR 包内 |
| LocalLogCollector（SDK，后台发送线程） | error-collect-client JAR 包内 |
| logback-spring.xml（接入配置示例） | creeks-cdc/server/.../resources/logback-spring.xml |
| error-collect 服务端（优先级+ES+飞书） | 独立仓库，不在本项目（版本 3.3.0） |
外部依赖
| 组件 | 版本 | 用途 |
|---|---|---|
| error-collect-client | 3.3.0 | 日志采集 SDK |
| Elasticsearch | — | 日志存储（服务端） |
| 飞书 Webhook | — | 告警通知（服务端） |