From 6de3fa6c9c9b1829a755e66aa25d2daef80eac69 Mon Sep 17 00:00:00 2001 From: windyboy Date: Sun, 6 Sep 2026 21:02:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(observability):=20U12=20=E5=8F=AF=E8=A7=82?= =?UTF-8?q?=E6=B5=8B=E5=BA=95=E5=BA=A7=EF=BC=88ACM2-10=20U12/R05/N31b?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - logback：logstash 通道先加 AsyncAppender（queueSize 4096、neverBlock、discardingThreshold 0） ——logstash 不可达不阻塞业务线程（N31b 顺序约束：先降级再补日志）；MDC 增加 traceId/target - 结构化日志 + MDC traceId（TraceLog.withTrace=cminmsgsId/eventId）： InboxService 收报 INFO；MessageProcessor SUCCEEDED/SKIPPED INFO、FAILED/解码失败 WARN、 DEAD(MALFORMED/毒丸/EXHAUSTED) ERROR；SnapshotFlow staging/CAS WARN、快照成功 INFO； Dispatcher 发送 DEBUG、逐条/整批重试 WARN、DEAD(DLQ) ERROR、flushSchd 批次 INFO； ReplayService 重放 INFO/非法类 WARN；PipelineLifecycle 起停 INFO - 自定义健康指示器（infra/health）：redis-flight-store / kafka-delivery——BeanProvider 可选解析， 缺 bean（未 stub 也未实装）报 DOWN 而非启动失败；/health 聚合 - 保留：micrometer 队列深度/投递延迟 gauge 依赖版本选型（micronaut-micrometer 与平台 BOM 5.1.3 对齐待锁），随数据层/真实 client 批次补；U13 毒丸超时告警与 DLQ 巡检入口仍属 WP2 --- .../msgexchange/nextgen/PipelineLifecycle.kt | 3 ++ .../nextgen/delivery/Dispatcher.kt | 8 ++++ .../nextgen/infra/health/HealthIndicators.kt | 48 +++++++++++++++++++ .../msgexchange/nextgen/infra/log/TraceLog.kt | 18 +++++++ .../nextgen/infra/retry/ReplayService.kt | 10 +++- .../nextgen/ingress/InboxService.kt | 3 ++ .../msgexchange/nextgen/processing/Pump.kt | 31 ++++++++---- .../nextgen/processing/SnapshotFlow.kt | 5 ++ src/main/resources/logback.xml | 17 +++++-- 9 files changed, 130 insertions(+), 13 deletions(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/health/HealthIndicators.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/log/TraceLog.kt diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt index 6cc2330..1cea278 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/PipelineLifecycle.kt @@ -20,6 +20,7 @@ class PipelineLifecycle( private val pump: Pump, private val dispatcher: Dispatcher, ) { + private val log = org.slf4j.LoggerFactory.getLogger(PipelineLifecycle::class.java) private val threads = mutableListOf() @Volatile @@ -35,6 +36,7 @@ class PipelineLifecycle( started = true threads += spawn("msgx-pump", pump::loop) threads += spawn("msgx-dispatcher", dispatcher::loop) + log.info("pipeline loops started (pump, dispatcher)") } private fun spawn(name: String, body: () -> Unit): Thread = @@ -46,5 +48,6 @@ class PipelineLifecycle( dispatcher.stop() threads.forEach { it.interrupt() } // 解除 Thread.sleep 阻塞，加速退出 threads.forEach { runCatching { it.join(3000) } } + log.info("pipeline loops stopped") } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt index a30b550..f5e89fb 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt @@ -38,6 +38,8 @@ class Dispatcher( private val props: PipelineProps, private val scheduler: FailureScheduler, ) { + private val log = org.slf4j.LoggerFactory.getLogger(Dispatcher::class.java) + @Volatile private var running = true @@ -76,6 +78,7 @@ class Dispatcher( try { deliver(t, head) msgEvents.markSent(head.eventId!!) + log.debug("sent target={} eventId={}", t, head.eventId) if (t == Targets.ES_FLIGHT_HTS && props.phase == PipelineProps.Phase.B) { // 定案 2：ES 投递成功 → 同线程同步 enqueue 删除事件（不轮询 ack） msgEvents.insertSync(listOf(MsgEvent(target = Targets.REDIS_FLIGHT_INFO, payloadJson = deleteOf(head)))) @@ -96,8 +99,10 @@ class Dispatcher( private fun retryOrDead(e: MsgEvent, lastError: String) { val attempts = e.attempts + 1 if (scheduler.exhausted(attempts)) { + log.error("event DEAD(DLQ) eventId={} attempts={} lastError={}", e.eventId, attempts, lastError) msgEvents.markDead(e.eventId!!, ErrorClass.EXHAUSTED, lastError, attempts) } else { + log.warn("event retry scheduled eventId={} attempts={} nextAttemptAt={}", e.eventId, attempts, scheduler.nextAttemptAt(attempts)) msgEvents.scheduleRetry(e.eventId!!, scheduler.nextAttemptAt(attempts), attempts) } } @@ -128,13 +133,16 @@ class Dispatcher( return // 队首仍在退避：整批等待，不推进 lastFlush（到期再试） } val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]") + log.debug("flushSchd batch size={} payloadLen={}", pending.size, payload.length) try { port.sendKafka("schd", payload) } catch (e: Exception) { + log.warn("flushSchd send failed batch={} -> batch retryOrDead: {}", pending.size, e.message) pending.forEach { retryOrDead(it, "schd-send: ${e.message ?: "unknown"}") } return // 不推进 lastFlush：整批退避（含 DEAD 出队）后到期重试 } msgEvents.markAllSent(pending.mapNotNull { it.eventId }) + log.info("flushSchd sent batch={}", pending.size) lastFlush = scheduler.now() } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/health/HealthIndicators.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/health/HealthIndicators.kt new file mode 100644 index 0000000..387999e --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/health/HealthIndicators.kt @@ -0,0 +1,48 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.health + +import com.gzzn.omms.msgexchange.nextgen.delivery.DeliveryPort +import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient +import io.micronaut.context.BeanProvider +import io.micronaut.core.async.publisher.Publishers +import io.micronaut.health.HealthStatus +import io.micronaut.management.health.indicator.HealthIndicator +import io.micronaut.management.health.indicator.HealthResult +import jakarta.inject.Singleton +import org.reactivestreams.Publisher + +/** + * U12（R05）：阶段 A 关键依赖的自定义健康指示器—— + * Redis（flightInfo 权威存储）与 Kafka（投递端口）。依赖经 BeanProvider 可选解析： + * 缺 bean（如未用 stub 也未实装）时指示 DOWN 而非启动失败。 + */ +@Singleton +class FlightRedisHealthIndicator( + private val redis: BeanProvider, +) : HealthIndicator { + + override fun getResult(): Publisher = + Publishers.just(resultOf("redis-flight-store", "flight store", redis.isPresent, + redis.isPresent.takeIf { it }?.let { redis.get()::class.java.simpleName })) +} + +@Singleton +class KafkaDeliveryHealthIndicator( + private val port: BeanProvider, +) : HealthIndicator { + + override fun getResult(): Publisher = + Publishers.just(resultOf("kafka-delivery", "delivery port", port.isPresent, + port.isPresent.takeIf { it }?.let { port.get()::class.java.simpleName })) +} + +private fun resultOf(name: String, what: String, up: Boolean, impl: String?): HealthResult { + val builder = HealthResult.builder(name) + .status(if (up) HealthStatus.UP else HealthStatus.DOWN) + .details( + mapOf( + "message" to if (up) "$what present" else "$what bean missing (stub off, impl pending)", + "implementation" to (impl ?: "none"), + ), + ) + return builder.build() +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/log/TraceLog.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/log/TraceLog.kt new file mode 100644 index 0000000..6d99e99 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/log/TraceLog.kt @@ -0,0 +1,18 @@ +package com.gzzn.omms.msgexchange.nextgen.infra.log + +import org.slf4j.MDC + +/** + * U12（R05）：处理路径入口写入 MDC traceId（=cminmsgsId/eventId）， + * 使一条消息全链路日志可串（logback %X{traceId} + logstash includeMdcKeyName）。 + */ +object TraceLog { + fun withTrace(id: Any, body: () -> T): T { + MDC.put("traceId", id.toString()) + return try { + body() + } finally { + MDC.remove("traceId") + } + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt index aec8e9d..6cca1a4 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt @@ -12,6 +12,7 @@ import jakarta.inject.Singleton class ReplayService( private val procState: ProcStateRepository, ) { + private val log = org.slf4j.LoggerFactory.getLogger(ReplayService::class.java) /** 可恢复错误类：codec 修复可重放 / 未实装补齐可重放 / 基础设施抖动可重放 / 重试耗尽后人工复核可重放。 */ val replayableErrorClasses: Set = setOf(ErrorClass.CODEC_ERROR, ErrorClass.UNSUPPORTED, ErrorClass.INFRA, ErrorClass.EXHAUSTED) @@ -19,8 +20,13 @@ class ReplayService( /** 只重放白名单内的类；请求含 MALFORMED 等非法类时静默忽略该类。 */ fun replay(requested: Collection): Int { val allowed = requested.filter { it in replayableErrorClasses } - if (allowed.isEmpty()) return 0 - return procState.requeueByErrorClasses(allowed) + if (allowed.isEmpty()) { + log.warn("replay requested only non-replayable classes: {}", requested) + return 0 + } + val n = procState.requeueByErrorClasses(allowed) + log.info("replayed rows={} classes={}", n, allowed) + return n } /** 默认入口：重放全部可恢复类。 */ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt index 7504fb6..a641bce 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/ingress/InboxService.kt @@ -13,12 +13,15 @@ class InboxService( // TODO(阶段1后续): 事务边界（@Transactional）随 Micronaut Data 实装补齐； // pump 唤醒仅加速，崩溃后主泵 1s 轮询兜底。 ) { + private val log = org.slf4j.LoggerFactory.getLogger(InboxService::class.java) + data class Receipt(val cminmsgsId: Long, val receivedAt: Instant) fun accept(rawXml: String): Receipt { val id = inbox.insertRaw(rawXml) procState.insert(id) // 同事务（实装后）；接收层无唯一约束（I3） wakePump() + log.info("accepted cminmsgsId={} (tx1)", id) return Receipt(id, Instant.now()) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt index ad22324..9a492ee 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt @@ -31,6 +31,8 @@ class Pump( private val jobExecutor: JobExecutor, private val props: PipelineProps, ) { + private val log = org.slf4j.LoggerFactory.getLogger(Pump::class.java) + @Volatile private var running = true @@ -65,6 +67,7 @@ class Pump( head == null -> sleepQuietly(props.pipeline.pollInterval) head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() -> if (poisoned(head)) { + log.error("poison -> DEAD id={} attempts={} lastError={}", head.cminmsgsId, head.attempts, head.lastError) procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "head-deadline-exceeded") } else { @@ -116,27 +119,34 @@ class MessageProcessor( private val procFailure: ProcFailure, private val props: PipelineProps, ) { + private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java) + fun processOne(head: ProcState) { - try { - processInternal(head) - } catch (e: InterruptedException) { - Thread.currentThread().interrupt() - throw e - } catch (e: Exception) { - // U08：边界化——异常归于本条 head，写 FAILED/DEAD，而不是穿出杀 pump - procFailure.fail(head, ErrorClass.INFRA, e.message ?: e.javaClass.simpleName) + com.gzzn.omms.msgexchange.nextgen.infra.log.TraceLog.withTrace(head.cminmsgsId) { + try { + processInternal(head) + } catch (e: InterruptedException) { + Thread.currentThread().interrupt() + throw e + } catch (e: Exception) { + // U08：边界化——异常归于本条 head，写 FAILED/DEAD，而不是穿出杀 pump + log.warn("processOne unexpected failure id={} ec=INFRA msg={}", head.cminmsgsId, e.message ?: e.javaClass.simpleName) + procFailure.fail(head, ErrorClass.INFRA, e.message ?: e.javaClass.simpleName) + } } } private fun processInternal(head: ProcState) { // 守卫：手工/遗留 FAILED 行若 attempts 已达上限，直接终态（防止退避到期后无限重试） if (head.state == ProcStatus.FAILED && procFailure.scheduler.exhausted(head.attempts)) { + log.error("head exhausted at entry -> DEAD id={} attempts={}", head.cminmsgsId, head.attempts) procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "max-attempts") return } val raw = inbox.rawOf(head.cminmsgsId) ?: run { + log.error("raw missing -> DEAD(MALFORMED) id={}", head.cminmsgsId) procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing") return } @@ -145,9 +155,11 @@ class MessageProcessor( is com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult.Err -> { // T06（U11）：MALFORMED（报文非法）→ DEAD 不重试；CODEC_ERROR（可随 codec 修复重放）→ FAILED 退避 if (r.failure.errorClass == ErrorClass.MALFORMED) { + log.error("decode MALFORMED -> DEAD id={} detail={}", head.cminmsgsId, r.failure.detail) procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = r.failure.detail) } else { + log.warn("decode {} -> FAILED id={} detail={}", r.failure.errorClass, head.cminmsgsId, r.failure.detail) procFailure.fail(head, r.failure.errorClass, r.failure.detail) } return @@ -159,6 +171,7 @@ class MessageProcessor( val identity = Identity.of(decoded, props.identity) if (!procState.tryBindIdentity(head.cminmsgsId, identity)) { val owner = procState.ownerOfIdentity(identity) ?: -1L + log.info("duplicate-of:{} -> SKIPPED id={}", owner, head.cminmsgsId) procState.update(head.cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner") return } @@ -168,6 +181,7 @@ class MessageProcessor( val handler = handlers.registry.dispatcherFor(decoded) if (handler == null) { // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态 + log.warn("no-handler:{} -> FAILED(UNSUPPORTED) id={}", decoded.typeTag, head.cminmsgsId) procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}") return } @@ -192,6 +206,7 @@ class MessageProcessor( msgEvents.insertAll(events) inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn) procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) + log.info("SUCCEEDED id={} events={}", head.cminmsgsId, events.size) // 阶段 B：flightState.apply(decision.flightChanges) 进入同一事务；投影事件（ES/REDIS）追加。 } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt index fa58d4b..993951a 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt @@ -25,10 +25,13 @@ class SnapshotFlow( private val redis: FlightRedisClient, private val procFailure: ProcFailure, ) { + private val log = org.slf4j.LoggerFactory.getLogger(SnapshotFlow::class.java) + fun publishSnapshot(head: ProcState, msg: DecodedMessage) { // 1) staging：流式解析 + 整包校验（TODO(阶段2): 流式 codec；千级 FLTR 为 MB 级，内存瞬态） val staged = StageResult.stagingOf(msg) // 骨架：TODO 解析 FLTR 集与重组（KEEP 现役 MAFL/登机桥规则） if (staged is StageResult.Invalid) { + log.warn("staging not implemented -> FAILED(UNSUPPORTED) id={} reason={}", head.cminmsgsId, staged.reason) procFailure.fail(head, ErrorClass.UNSUPPORTED, staged.reason) // U10：未实装 → 可重放，非终态 return } @@ -47,11 +50,13 @@ class SnapshotFlow( // 重放路径：version 已是目标值 → no-op 视为成功 val again = refData.getGen(day) if (again == null || again.version != expected + 1) { + log.warn("gen CAS conflict -> FAILED(INFRA) id={}", head.cminmsgsId) procFailure.fail(head, ErrorClass.INFRA, "gen-cas-conflict") // N06/N28：带退避，禁止紧循环 return } } procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) + log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size) } /** staging 结果（骨架）。 */ diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml index 6eb9b7b..1ba57c3 100644 --- a/src/main/resources/logback.xml +++ b/src/main/resources/logback.xml @@ -1,6 +1,7 @@ - + @@ -8,17 +9,27 @@ - + ${MSGX_LOGSTASH_HOST:-127.0.0.1}:${MSGX_LOGSTASH_PORT:-5044} traceId + target 10 seconds + + + 4096 + 0 + true + false + + + - +