From 4b71b5bb09ebb4361116ae3e7db7433f7f624d2d Mon Sep 17 00:00:00 2001 From: windyboy Date: Sat, 12 Sep 2026 20:52:24 +0800 Subject: [PATCH] feat(jobs): expose job heartbeat and scan backlog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 JobActivity 持有作业心跳与扫描积压，JobRunner 每轮 tick 上报（失败只计数、不刷新心跳），PipelineMetrics 注册 msgx.pipeline.job.heartbeat_age_seconds / ticks.total / failures.total / last_sweep_selected，新增 JobRunnerHealthIndicator（未启动 UP 并标 started=false；已启动但心跳缺失或超过 3×扫描周期 DOWN）。对应 reference 作业健康段与 [G-JOB-HEARTBEAT]。 验证：./gradlew test 145 tests / 0 fail / 0 skipped（新增 JobRunnerTest 2、JobRunnerHealthIndicatorTest 4、PipelineMetrics 指标注册 1）。 --- .../infra/health/JobRunnerHealthIndicator.kt | 57 ++++++++++++++ .../msgexchange/infra/metrics/JobActivity.kt | 65 ++++++++++++++++ .../infra/metrics/PipelineMetrics.kt | 21 ++++++ .../gzzn/omms/msgexchange/jobs/JobRunner.kt | 37 +++++++-- .../health/JobRunnerHealthIndicatorTest.kt | 70 +++++++++++++++++ .../infra/metrics/PipelineMetricsTest.kt | 14 ++++ .../omms/msgexchange/jobs/JobRunnerTest.kt | 75 +++++++++++++++++++ 7 files changed, 333 insertions(+), 6 deletions(-) create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicator.kt create mode 100644 src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/JobActivity.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicatorTest.kt create mode 100644 src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicator.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicator.kt new file mode 100644 index 0000000..2715609 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicator.kt @@ -0,0 +1,57 @@ +package com.gzzn.omms.msgexchange.infra.health + +import com.gzzn.omms.msgexchange.infra.metrics.JobActivity +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 +import java.time.Clock +import java.time.Duration +import java.time.Instant + +/** + * 维护作业的存活判定：作业线程是回填的唯一驱动，停摆必须可见，而不是只留一条日志。 + * + * - 从未启动（例如 `msgx.pipeline.autostart=false`）→ `UP`，详情标 `started=false`，避免误报； + * - 已启动但心跳缺失或超过 `3 × 扫描周期`（周期是 `JobRunner` 的代码常量，见 reference）→ `DOWN`； + * - 其余 → `UP`，并给出心跳年龄、上一轮耗时、tick/失败数与扫描积压。 + */ +@Singleton +class JobRunnerHealthIndicator( + private val activity: JobActivity, + private val clock: Clock, +) : HealthIndicator { + + override fun getResult(): Publisher = + Publishers.just(jobRunnerHealth(activity.snapshot(), clock.instant())) +} + +/** 纯判定，便于单测；`now` 由调用方传入注入时钟的时刻。 */ +internal fun jobRunnerHealth(snapshot: JobActivity.Snapshot, now: Instant): HealthResult { + val ageSeconds = snapshot.lastTickAt?.let { Duration.between(it, now).seconds } + val stale = snapshot.started && (ageSeconds == null || ageSeconds > STALE_AFTER_SECONDS) + val message = when { + !snapshot.started -> "job runner not started (started=false)" + stale -> "job heartbeat stale: age=${ageSeconds ?: "never"}s > ${STALE_AFTER_SECONDS}s" + else -> "job heartbeat ok: age=${ageSeconds}s ticks=${snapshot.ticks}" + } + return HealthResult.builder("job-runner") + .status(if (stale) HealthStatus.DOWN else HealthStatus.UP) + .details( + mapOf( + "message" to message, + "started" to snapshot.started, + "lastTickAgeSeconds" to (ageSeconds ?: -1L), + "lastTickDurationMs" to snapshot.lastTickMillis, + "ticks" to snapshot.ticks, + "failures" to snapshot.failures, + "lastSweepSelected" to snapshot.lastSweepSelected, + ), + ) + .build() +} + +/** 3 × 30 秒扫描周期：连续两轮都没跑完才算停摆。 */ +internal const val STALE_AFTER_SECONDS = 90L diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/JobActivity.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/JobActivity.kt new file mode 100644 index 0000000..bfabbd6 --- /dev/null +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/JobActivity.kt @@ -0,0 +1,65 @@ +package com.gzzn.omms.msgexchange.infra.metrics + +import jakarta.inject.Singleton +import java.time.Instant +import java.util.concurrent.atomic.AtomicLong + +/** + * 维护作业的进程内活动事实（作业心跳、回填扫描积压），供 `/metrics` 与 `/health` 取用。 + * + * 放在 `infra` 是为了让 `jobs`（写）与指标/健康（读）都零依赖：作业线程只更新这个持有者， + * 指标名与暴露方式由 [PipelineMetrics] 决定，存活判定由 `JobRunnerHealthIndicator` 决定。 + * 计数在重启后归零；跨重启的累计值由指标后端聚合，这里不做持久化。 + */ +@Singleton +class JobActivity { + @Volatile + private var started = false + + @Volatile + private var lastTickAt: Instant? = null + + @Volatile + private var lastTickMillis = 0L + + /** 上一轮回填扫描选中的待办条数（扫描积压的即时值）。 */ + @Volatile + var lastSweepSelected = 0 + private set + + private val ticks = AtomicLong(0) + private val failures = AtomicLong(0) + + fun started() { + started = true + } + + fun stopped() { + started = false + } + + /** 一次作业 tick 正常跑完：记录心跳时刻、耗时与本次扫描积压。 */ + fun tickFinished(now: Instant, durationMillis: Long, sweepSelected: Int) { + lastTickAt = now + lastTickMillis = durationMillis + lastSweepSelected = sweepSelected + ticks.incrementAndGet() + } + + /** 一次作业 tick 抛错：只累计失败，**不更新心跳时刻**——停摆由心跳年龄暴露。 */ + fun tickFailed() { + failures.incrementAndGet() + } + + fun snapshot(): Snapshot = + Snapshot(started, lastTickAt, lastTickMillis, lastSweepSelected, ticks.get(), failures.get()) + + data class Snapshot( + val started: Boolean, + val lastTickAt: Instant?, + val lastTickMillis: Long, + val lastSweepSelected: Int, + val ticks: Long, + val failures: Long, + ) +} 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 9aac021..2259799 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 @@ -23,6 +23,9 @@ import java.time.Duration * - `msgx.pipeline.backfill.oldest_unmarked_seconds`：最老一条仍待自动回填的年龄 * - `msgx.pipeline.watermark.lag`：水位落后信箱最新 ID 的距离 * - `msgx.pipeline.hole.aged_out.total`：永久空洞放行次数 + * - `msgx.pipeline.job.heartbeat_age_seconds`：距上一次作业 tick 完成的秒数（未跑过为 -1） + * - `msgx.pipeline.job.ticks.total` / `msgx.pipeline.job.failures.total`：作业 tick 完成/抛错次数 + * - `msgx.pipeline.job.last_sweep_selected`：上一轮回填扫描选中的待办条数（扫描积压） * * 取数统一走 [BacklogSnapshotProvider]（30 秒 TTL），因此指标抓取不会打穿数据库。 * 无法取数时以 `NaN` 上报（Micrometer 的惯例表示"本次无值"），而不是伪造 0。 @@ -41,6 +44,7 @@ class PipelineMetrics( private val cursor: BeanProvider, private val mailbox: BeanProvider, private val counters: PipelineCounters, + private val activity: JobActivity, private val clock: Clock, ) { @@ -74,6 +78,23 @@ class PipelineMetrics( Gauge.builder("msgx.pipeline.late_arrival.detected.total", counters) { it.lateArrivalDetectedCount().toDouble() } .strongReference(true) .register(registry) + + // 作业心跳与扫描积压：作业线程是回填的唯一驱动，停摆必须能被 /metrics 与 /health 看见。 + Gauge.builder("msgx.pipeline.job.heartbeat_age_seconds", activity) { a -> + a.snapshot().lastTickAt?.let { Duration.between(it, clock.instant()).seconds.toDouble() } ?: -1.0 + }.strongReference(true).register(registry) + + Gauge.builder("msgx.pipeline.job.ticks.total", activity) { it.snapshot().ticks.toDouble() } + .strongReference(true) + .register(registry) + + Gauge.builder("msgx.pipeline.job.failures.total", activity) { it.snapshot().failures.toDouble() } + .strongReference(true) + .register(registry) + + Gauge.builder("msgx.pipeline.job.last_sweep_selected", activity) { it.snapshot().lastSweepSelected.toDouble() } + .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/jobs/JobRunner.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt index 9e9c038..5dd75e0 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt @@ -1,6 +1,7 @@ package com.gzzn.omms.msgexchange.jobs import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.infra.metrics.JobActivity import com.gzzn.omms.msgexchange.processing.BackfillService import jakarta.inject.Singleton import java.time.Clock @@ -23,6 +24,7 @@ class JobRunner( private val backfill: BackfillService, private val historySweep: HistorySweepJob, private val clock: Clock, + private val activity: JobActivity, operationDayProps: OperationDayProps, ) { private val log = org.slf4j.LoggerFactory.getLogger(JobRunner::class.java) @@ -39,28 +41,46 @@ class JobRunner( fun start() { if (thread != null) return running = true + activity.started() thread = Thread.ofPlatform().name("msgx-jobs").daemon(true).start { loop() } - log.info("job runner started (backfill 30s, history daily 03:30 {})", zone) + log.info("job runner started (backfill {}s, history daily 03:30 {})", TICK_PERIOD.seconds, zone) } fun stop() { running = false thread?.interrupt() thread = null + activity.stopped() } internal fun loop() { while (running) { try { - backfill.sweep() - maybeHistorySweep() + tickOnce() } catch (e: InterruptedException) { Thread.currentThread().interrupt() return - } catch (e: Exception) { - log.warn("job tick failed: {}", e.message ?: e.javaClass.simpleName) } - if (running) runCatching { Thread.sleep(Duration.ofSeconds(30).toMillis()) } + if (running) runCatching { Thread.sleep(TICK_PERIOD.toMillis()) } + } + } + + /** + * 跑一轮作业并上报心跳（拉出来单独可测，避免测试里 sleep）。 + * 回填扫描的选中条数即"扫描积压"的即时值；失败只累计计数，不刷新心跳时刻。 + */ + internal fun tickOnce() { + val startedAt = clock.instant() + val startedNanos = System.nanoTime() + try { + val selected = backfill.sweep(startedAt) + maybeHistorySweep() + activity.tickFinished(startedAt, (System.nanoTime() - startedNanos) / 1_000_000, selected) + } catch (e: InterruptedException) { + throw e + } catch (e: Exception) { + activity.tickFailed() + log.warn("job tick failed: {}", e.message ?: e.javaClass.simpleName) } } @@ -78,4 +98,9 @@ class JobRunner( } } } + + private companion object { + /** 作业循环周期：回填扫描与历史作业共用的代码常量（reference 记为 `backfill-scan-period`）。 */ + val TICK_PERIOD: Duration = Duration.ofSeconds(30) + } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicatorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicatorTest.kt new file mode 100644 index 0000000..9290b51 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicatorTest.kt @@ -0,0 +1,70 @@ +package com.gzzn.omms.msgexchange.infra.health + +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.infra.metrics.JobActivity +import io.micronaut.health.HealthStatus +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +/** + * 作业存活判定：没启动不误报；启动了但心跳缺失或超过 3 个扫描周期必须 DOWN； + * 正常心跳保持 UP 并把扫描积压等事实带进详情。 + */ +class JobRunnerHealthIndicatorTest { + + private val now = MutableClock.BASE + + private fun details(result: io.micronaut.management.health.indicator.HealthResult): Map { + @Suppress("UNCHECKED_CAST") + return result.details as Map + } + + @Test + fun `a job runner that never started is up but flagged`() { + val result = jobRunnerHealth(JobActivity().snapshot(), now) + + assertEquals(HealthStatus.UP, result.status) + assertEquals(false, details(result)["started"]) + assertEquals(-1L, details(result)["lastTickAgeSeconds"]) + } + + @Test + fun `a started job with a fresh heartbeat is up`() { + val activity = JobActivity().apply { + started() + tickFinished(now.minusSeconds(30), durationMillis = 12, sweepSelected = 3) + } + + val result = jobRunnerHealth(activity.snapshot(), now) + + assertEquals(HealthStatus.UP, result.status) + assertEquals(1L, details(result)["ticks"]) + assertEquals(3, details(result)["lastSweepSelected"]) + assertEquals(12L, details(result)["lastTickDurationMs"]) + } + + @Test + fun `a started job that never ticked or went stale is down`() { + val neverTicked = JobActivity().apply { started() } + assertEquals(HealthStatus.DOWN, jobRunnerHealth(neverTicked.snapshot(), now).status) + + val stale = JobActivity().apply { + started() + tickFinished(now.minusSeconds(STALE_AFTER_SECONDS + 1), durationMillis = 5, sweepSelected = 0) + } + assertEquals(HealthStatus.DOWN, jobRunnerHealth(stale.snapshot(), now).status) + } + + @Test + fun `a failed tick counts up without refreshing the heartbeat`() { + val activity = JobActivity().apply { started() } + + activity.tickFailed() + activity.tickFailed() + + val snapshot = activity.snapshot() + assertEquals(2L, snapshot.failures) + assertEquals(0L, snapshot.ticks) + assertEquals(null, snapshot.lastTickAt) + } +} 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 fc4583b..15645a6 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 @@ -58,6 +58,20 @@ class PipelineMetricsTest { assertEquals(before + 2.0, gauge("msgx.pipeline.hole.aged_out.total"), 0.001) } + @Test + fun `job heartbeat gauges are registered and reflect job activity`() { + val activity = ctx.getBean(JobActivity::class.java) + val ticksBefore = gauge("msgx.pipeline.job.ticks.total") + val failuresBefore = gauge("msgx.pipeline.job.failures.total") + + activity.tickFinished(Instant.now(), durationMillis = 12, sweepSelected = 7) + + assertEquals(ticksBefore + 1.0, gauge("msgx.pipeline.job.ticks.total"), 0.001) + assertEquals(failuresBefore, gauge("msgx.pipeline.job.failures.total"), 0.001) + assertEquals(7.0, gauge("msgx.pipeline.job.last_sweep_selected"), 0.001) + org.junit.jupiter.api.Assertions.assertTrue(gauge("msgx.pipeline.job.heartbeat_age_seconds") >= 0.0) + } + private fun gauge(name: String): Double = requireNotNull(registry.find(name).gauge()) { "gauge not registered: $name" }.value() } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt new file mode 100644 index 0000000..7c94785 --- /dev/null +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt @@ -0,0 +1,75 @@ +package com.gzzn.omms.msgexchange.jobs + +import com.gzzn.omms.msgexchange.MutableClock +import com.gzzn.omms.msgexchange.config.HistoryProps +import com.gzzn.omms.msgexchange.config.MailboxProps +import com.gzzn.omms.msgexchange.config.OperationDayProps +import com.gzzn.omms.msgexchange.config.PipelineProps +import com.gzzn.omms.msgexchange.domain.ProcStatus +import com.gzzn.omms.msgexchange.infra.metrics.JobActivity +import com.gzzn.omms.msgexchange.infra.persistence.BackfillDue +import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository +import com.gzzn.omms.msgexchange.infra.stub.StubFlightState +import com.gzzn.omms.msgexchange.infra.stub.StubInbox +import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineLock +import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTx +import com.gzzn.omms.msgexchange.infra.stub.StubProcState +import com.gzzn.omms.msgexchange.processing.BackfillService +import com.gzzn.omms.msgexchange.processing.MessageLifecycleGate +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test +import java.time.Instant + +/** + * 作业心跳的接线：一轮 tick 跑完必须上报心跳与扫描积压；tick 抛错只累计失败、 + * **不刷新心跳时刻**（停摆由心跳年龄暴露）。用 `TestClocks` 与内存适配器，不 sleep。 + */ +class JobRunnerTest { + + private val clock = MutableClock(MutableClock.BASE) + private val props = PipelineProps() + + private fun historySweep() = HistorySweepJob( + StubFlightState(), StubMsgEvents(), HistoryProps().apply { historyStoreEnabled = false }, + OperationDayProps(), StubPipelineTx(), StubPipelineLock(), + ) + + private fun runner(proc: ProcStateRepository, activity: JobActivity) = JobRunner( + BackfillService(proc, StubInbox(), MailboxProps(), props, clock, MessageLifecycleGate()), + historySweep(), clock, activity, OperationDayProps(), + ) + + @Test + fun `a completed tick reports the heartbeat and the backfill scan backlog`() { + val proc = StubProcState() + proc.insertIfAbsent(1L, MutableClock.BASE.minusSeconds(30L * 86400)) + proc.markTerminal(1L, ProcStatus.SUCCEEDED, now = MutableClock.BASE) // 终态未打标 → 本轮到期待办 + val activity = JobActivity() + + runner(proc, activity).tickOnce() + + val snapshot = activity.snapshot() + assertEquals(1, snapshot.lastSweepSelected) + assertEquals(1L, snapshot.ticks) + assertEquals(0L, snapshot.failures) + assertEquals(MutableClock.BASE, snapshot.lastTickAt) + } + + @Test + fun `a failing tick counts the failure and leaves the heartbeat untouched`() { + val broken = object : ProcStateRepository by StubProcState() { + override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List = + throw IllegalStateException("db-down") + } + val activity = JobActivity() + + runner(broken, activity).tickOnce() + + val snapshot = activity.snapshot() + assertNull(snapshot.lastTickAt) + assertEquals(0L, snapshot.ticks) + assertEquals(1L, snapshot.failures) + } +}