From 9bf879ff7676d8d1684c8de21d5907c361ca14b9 Mon Sep 17 00:00:00 2001 From: windyboy Date: Mon, 21 Sep 2026 12:40:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(metrics):=20=E6=8A=95=E9=80=92=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E4=B8=8E=E6=AD=BB=E4=BF=A1=E5=8F=AF=E8=A7=82=E6=B5=8B?= =?UTF-8?q?=E6=8C=87=E6=A0=87=EF=BC=8C=E4=BF=AE=20flushSchd=20=E9=A6=96?= =?UTF-8?q?=E6=AC=A1=E5=A4=B1=E8=B4=A5=E5=8E=9F=E5=9B=A0=E4=B8=A2=E5=A4=B1?= =?UTF-8?q?=EF=BC=88ACM2-98=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PipelineCounters 按 target 累计发送失败；Dispatcher.retryOrDead 记账（OPS-2） - 新增 deadByTarget 查询与 DeliveryDeadSnapshotProvider（TTL 缓存），死信存量按 target 暴露 - /metrics 注册 send_failures.total{target} 与 dead{target}，登记进 reference.md「指标与健康」 - flushSchd 失败改记真实异常原因，不再退回固定 send-failed --- docs/reference.md | 2 + .../omms/msgexchange/delivery/Dispatcher.kt | 9 ++-- .../health/DeliveryDeadSnapshotProvider.kt | 44 +++++++++++++++++++ .../infra/metrics/PipelineCounters.kt | 11 +++++ .../infra/metrics/PipelineMetrics.kt | 22 ++++++++++ .../infra/persistence/Repositories.kt | 3 ++ .../persistence/jdbc/JdbcPgRepositories.kt | 7 +++ .../infra/stub/StubRepositories.kt | 3 ++ .../delivery/DispatcherTickTest.kt | 15 +++++-- .../infra/metrics/PipelineMetricsTest.kt | 32 ++++++++++++++ 10 files changed, 142 insertions(+), 6 deletions(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/DeliveryDeadSnapshotProvider.kt diff --git a/docs/reference.md b/docs/reference.md index 453f800..bcb9978 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -95,6 +95,8 @@ | `msgx.pipeline.codec.srvt_seen.total` | 收到含 `SRVT` 段的消息数 | 非零：核对 `G-SRVT-VIPF` | | `msgx.pipeline.codec.vipf_seen.total` | 收到含 `VIPF` 段的消息数 | 非零：核对 `G-SRVT-VIPF` | | `msgx.pipeline.processing.ignored.total` | 被 `IgnoreRules` 跳过的消息数 | 清单含 `REGN`、`RSTA`、`EROR`，与 `US-13`、`US-09` 冲突 | +| `msgx.pipeline.delivery.send_failures.total{target}` | 该投递目标累计发送失败次数（进程内，重启归零） | 持续增长且 `dead` 非零：投递链路故障（`OPS-2`） | +| `msgx.pipeline.delivery.dead{target}` | 该投递目标当前死信（`DEAD`）行数 | 非零即告警：死信需人工处置（`OPS-2`；状态语义见 [implementation.md](implementation.md)「状态与错误分类」） | | `msgx.pipeline.watermark.lag` | 信箱最新编号比已扫描编号大多少 | 改用处理时间筛选后删除（`G-SCAN-PREDICATE`） | 未处理消息的统计使用同一份缓存；统计不可用显示 `NaN`，没有记录可比时年龄与编号差显示 `-1`。作业计数在进程重启后归零；作业已启动却连续三个检查周期没有成功运行时，`/health` 报 `DOWN`。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt index 93ae612..712724a 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/delivery/Dispatcher.kt @@ -45,6 +45,7 @@ class Dispatcher( private val port: DeliveryPort, private val props: PipelineProps, private val scheduler: FailureScheduler, + private val counters: com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters, ) { private val log = org.slf4j.LoggerFactory.getLogger(Dispatcher::class.java) @@ -162,7 +163,7 @@ class Dispatcher( lastFlush = scheduler.now() return } - val failures = mutableListOf() + val failures = mutableListOf>() for (e in batch) { try { when (e.eventType) { @@ -172,15 +173,17 @@ class Dispatcher( // 条件确认：读取时刻的代次（EVENT_ID + STATE_VERSION）被新写入覆盖时不标记，留待下一轮重发。 e.eventId?.let { msgEvents.markSentIfVersion(it, e.stateVersion, scheduler.now()) } } catch (ex: Exception) { - failures.add(e) + // 首次失败就带上真实原因；不能退回字面量把根因抹掉 + failures += e to (ex.message ?: ex.javaClass.simpleName) } } - failures.forEach { retryOrDead(it, it.lastError ?: "send-failed") } + failures.forEach { (e, error) -> retryOrDead(e, error) } lastFlush = scheduler.now() } /** 一条事件发失败之后怎么走：重试次数加一，到上限就标成 DEAD(EXHAUSTED) 留作死信（次数落库便于追查），否则按退避推到下次再发。 */ private fun retryOrDead(e: MsgEvent, lastError: String) { + counters.sendFailedAdd(e.target) // OPS-2：发送失败按 target 计数，供 /metrics 告警 val eventId = e.eventId ?: return val attempts = e.attempts + 1 if (scheduler.exhausted(attempts)) { diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/DeliveryDeadSnapshotProvider.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/DeliveryDeadSnapshotProvider.kt new file mode 100644 index 0000000..f9f7ee4 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/DeliveryDeadSnapshotProvider.kt @@ -0,0 +1,44 @@ +package com.gzzn.omms.msgexchange.infra.health + +import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository +import io.micronaut.context.BeanProvider +import io.micronaut.context.annotation.Value +import jakarta.inject.Singleton +import java.time.Clock +import java.time.Duration +import java.time.Instant + +/** + * 投递死信统计（每个 target 的 `DEAD` 行数）的唯一取数点，带 TTL 缓存。 + * + * 与 [BacklogSnapshotProvider] 同一设计：`/metrics` 抓取可能高频发生， + * `GROUP BY target` 聚合不该每次直打数据库；数字最多滞后一个缓存窗口，对告警信号足够。 + * 缓存窗口沿用 `msgx.health.backlog-cache-ttl-ms`。 + */ +@Singleton +class DeliveryDeadSnapshotProvider( + private val msgEvents: BeanProvider, + private val clock: Clock, + @Value("\${msgx.health.backlog-cache-ttl-ms:30000}") private val ttlMs: Long = 30_000, +) { + @Volatile + private var cached: Map? = null + + @Volatile + private var cachedAt: Instant? = null + + /** @return target → 死信行数；仓储未绑定时返回 null，由调用方按"未绑定"处理。 */ + fun snapshot(): Map? { + if (!msgEvents.isPresent) return null + val now = clock.instant() + val at = cachedAt + val value = cached + if (ttlMs > 0 && at != null && value != null && Duration.between(at, now).toMillis() < ttlMs) { + return value + } + val fresh = msgEvents.get().deadByTarget() + cached = fresh + cachedAt = now + return fresh + } +} diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt index d3235f7..45abf98 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineCounters.kt @@ -16,6 +16,7 @@ class PipelineCounters { private val srvtSeen = AtomicLong(0) private val vipfSeen = AtomicLong(0) private val ignored = AtomicLong(0) + private val sendFailures = java.util.concurrent.ConcurrentHashMap() /** * 入站记录里出现 `SRVT`/`VIPF` 段的条数（`[G-SRVT-VIPF]`）。 @@ -36,4 +37,14 @@ class PipelineCounters { fun ignoredAdd(count: Int = 1) { ignored.addAndGet(count.toLong()) } fun ignoredCount(): Long = ignored.get() + + /** + * 按投递目标（`KAFKA:msg`/`KAFKA:schd`）累计的发送失败次数（`OPS-2`）。 + * 每次记退避或死信时累加；重启归零，跨重启的累计由指标后端聚合。 + */ + fun sendFailedAdd(target: String) { + sendFailures.computeIfAbsent(target) { AtomicLong(0) }.incrementAndGet() + } + + fun sendFailedCount(target: String): Long = sendFailures[target]?.get() ?: 0L } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt index be4ded1..5659841 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetrics.kt @@ -1,6 +1,7 @@ package com.gzzn.omms.msgexchange.infra.metrics import com.gzzn.omms.msgexchange.infra.health.BacklogSnapshotProvider +import com.gzzn.omms.msgexchange.infra.health.DeliveryDeadSnapshotProvider import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository import com.gzzn.omms.msgexchange.infra.persistence.InboxCursorRepository import io.micronaut.context.BeanProvider @@ -29,6 +30,8 @@ import java.time.Duration * - `msgx.pipeline.codec.srvt_seen.total` / `msgx.pipeline.codec.vipf_seen.total`：入站记录里出现 * `SRVT`/`VIPF` 段的条数（尚未落明细表，`[G-SRVT-VIPF]`；> 0 表示真实流量确有该段） * - `msgx.pipeline.processing.ignored.total`：命中 US-04 忽略清单的报文条数 + * - `msgx.pipeline.delivery.send_failures.total{target}`：按 target 的投递发送失败次数（`OPS-2`） + * - `msgx.pipeline.delivery.dead{target}`：按 target 的死信（`DEAD`）存量行数（`OPS-2`，非零需人工处置） * * 取数统一走 [BacklogSnapshotProvider]（30 秒 TTL），因此指标抓取不会打穿数据库。 * 无法取数时以 `NaN` 上报（Micrometer 的惯例表示"本次无值"），而不是伪造 0。 @@ -44,6 +47,7 @@ import java.time.Duration class PipelineMetrics( private val registry: MeterRegistry, private val backlogs: BacklogSnapshotProvider, + private val deadEvents: DeliveryDeadSnapshotProvider, private val cursor: BeanProvider, private val mailbox: BeanProvider, private val activity: JobActivity, @@ -106,6 +110,24 @@ class PipelineMetrics( Gauge.builder("msgx.pipeline.processing.ignored.total", counters) { it.ignoredCount().toDouble() } .strongReference(true) .register(registry) + + // OPS-2：投递失败可观测。发送失败按 target 计数（进程内，重启归零）； + // 死信存量按 target 从库端聚合取（带 TTL 缓存）。非零死信是人工处置信号。 + for (target in listOf(com.gzzn.omms.msgexchange.domain.Targets.KAFKA_MSG, com.gzzn.omms.msgexchange.domain.Targets.KAFKA_SCHD)) { + Gauge.builder("msgx.pipeline.delivery.send_failures.total", counters) { it.sendFailedCount(target).toDouble() } + .tag("target", target) + .strongReference(true) + .register(registry) + + Gauge.builder("msgx.pipeline.delivery.dead", deadEvents) { provider -> + val dead = provider.snapshot() + // 仓储未绑定按无值（NaN）上报；已绑定但该 target 无死信行是确定的 0 + if (dead == null) Double.NaN else (dead[target] ?: 0).toDouble() + } + .tag("target", target) + .strongReference(true) + .register(registry) + } } private fun backlogGauge(name: String, value: (com.gzzn.omms.msgexchange.infra.persistence.Backlog) -> Double) { diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt index 1a3c9b8..d504fd3 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt @@ -186,6 +186,9 @@ interface MsgEventRepository { fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int? = null) + /** 死信统计：每个 target 的 `STATE='DEAD'` 行数（投递告警信号，`OPS-2`）。 */ + fun deadByTarget(): Map + /** * 删除已确认投递且超过保留期的事件行：`STATE='SENT' AND SENT_AT < cutoff`。 * 带 `STATE='SENT'` 条件，避免与 upsertSchd 的并发写入冲突。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 869e511..8359b61 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -562,6 +562,13 @@ class JdbcMsgEventRepository( ) } + override fun deadByTarget(): Map = + ds.query( + "SELECT target, count(*) AS total FROM msg_event WHERE state = 'DEAD' GROUP BY target", + { /* 无绑定参数 */ }, + { rs -> rs.getString("target") to rs.getInt("total") }, + ).toMap() + override fun deleteExpiredSent(cutoff: Instant, limit: Int): Int = ds.update( """ diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index 8f2b56c..8c52265 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -304,6 +304,9 @@ class StubMsgEvents : MsgEventRepository { ) } + override fun deadByTarget(): Map = + rows.values.filter { it.state == EventStatus.DEAD }.groupingBy { it.target }.eachCount() + override fun deleteExpiredSent(cutoff: Instant, limit: Int): Int { val ids = rows.values .filter { it.state == EventStatus.SENT && it.sentAt != null && it.sentAt < cutoff } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt index 0d30b59..e086c00 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/delivery/DispatcherTickTest.kt @@ -8,6 +8,7 @@ import com.gzzn.omms.msgexchange.domain.EventType import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.Targets import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository +import com.gzzn.omms.msgexchange.infra.metrics.PipelineCounters import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler import com.gzzn.omms.msgexchange.infra.stub.StubDeliveryPort import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents @@ -28,8 +29,12 @@ class DispatcherTickTest { private val clock = MutableClock(MutableClock.BASE) - private fun dispatcher(repo: MsgEventRepository, port: DeliveryPort, p: PipelineProps = PipelineProps()): Dispatcher = - Dispatcher(repo, port, p, FailureScheduler(p, clock)) + private fun dispatcher( + repo: MsgEventRepository, + port: DeliveryPort, + p: PipelineProps = PipelineProps(), + counters: PipelineCounters = PipelineCounters(), + ): Dispatcher = Dispatcher(repo, port, p, FailureScheduler(p, clock), counters) private fun ev(id: Long, target: String, key: String, payload: String, version: Long = 0) = MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload, createdAt = MutableClock.BASE) @@ -123,11 +128,14 @@ class DispatcherTickTest { fun `schd send failure schedules backoff and goes DEAD at limit`() { val repo = StubMsgEvents() val p = PipelineProps() - val d = dispatcher(repo, FailingSchdPort(), p) + val counters = PipelineCounters() + val d = dispatcher(repo, FailingSchdPort(), p, counters) repo.insertAll(listOf(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}""", 1))) d.flushSchd() assertEquals(1, repo.rows[1]!!.attempts) + // OPS-2：发送失败按 target 计数，且失败原因不是固定字面量 + assertEquals(1L, counters.sendFailedCount(Targets.KAFKA_SCHD)) repeat(p.pipeline.maxAttempts - 1) { clock.advance(p.pipeline.backoffFor(repo.rows[1]!!.attempts) + 1) @@ -135,6 +143,7 @@ class DispatcherTickTest { } val final = repo.rows[1]!! assertTrue(final.attempts >= p.pipeline.maxAttempts || final.state.name == "DEAD") + assertEquals(0L, counters.sendFailedCount(Targets.KAFKA_MSG)) // msg 目标不受 schd 失败影响 } @Test diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt index a0b23a1..e16b2a2 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/metrics/PipelineMetricsTest.kt @@ -1,6 +1,10 @@ package com.gzzn.omms.msgexchange.infra.metrics +import com.gzzn.omms.msgexchange.domain.ErrorClass +import com.gzzn.omms.msgexchange.domain.MsgEvent import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.domain.Targets +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents import com.gzzn.omms.msgexchange.infra.stub.StubProcState import io.micrometer.core.instrument.MeterRegistry import io.micronaut.context.ApplicationContext @@ -74,6 +78,34 @@ class PipelineMetricsTest { assertEquals(vipfBefore + 1.0, gauge("msgx.pipeline.codec.vipf_seen.total"), 0.001) } + @Test + fun `delivery send-failure and dead-letter gauges are registered per target`() { + val counters = ctx.getBean(PipelineCounters::class.java) + val events = ctx.getBean(StubMsgEvents::class.java) + val failuresBefore = taggedGauge("msgx.pipeline.delivery.send_failures.total", Targets.KAFKA_SCHD) + + counters.sendFailedAdd(Targets.KAFKA_SCHD) + val ids = events.insertAll( + listOf( + MsgEvent(target = Targets.KAFKA_MSG, partitionKey = "F1", stateVersion = 1, payloadJson = "{}", createdAt = Instant.now()), + ), + ) + events.markDead(ids.single(), ErrorClass.EXHAUSTED, "boom", attempts = 5) + + assertEquals( + failuresBefore + 1.0, + taggedGauge("msgx.pipeline.delivery.send_failures.total", Targets.KAFKA_SCHD), + 0.001, + ) + assertEquals(1.0, taggedGauge("msgx.pipeline.delivery.dead", Targets.KAFKA_MSG), 0.001) + assertEquals(0.0, taggedGauge("msgx.pipeline.delivery.dead", Targets.KAFKA_SCHD), 0.001) + } + private fun gauge(name: String): Double = requireNotNull(registry.find(name).gauge()) { "gauge not registered: $name" }.value() + + private fun taggedGauge(name: String, target: String): Double = + requireNotNull(registry.find(name).tag("target", target).gauge()) { + "gauge not registered: $name(target=$target)" + }.value() }