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 index 2715609..6498d70 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicator.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicator.kt @@ -45,6 +45,7 @@ internal fun jobRunnerHealth(snapshot: JobActivity.Snapshot, now: Instant): Heal "started" to snapshot.started, "lastTickAgeSeconds" to (ageSeconds ?: -1L), "lastTickDurationMs" to snapshot.lastTickMillis, + "lastFailureAgeSeconds" to (snapshot.lastFailureAt?.let { Duration.between(it, now).seconds } ?: -1L), "ticks" to snapshot.ticks, "failures" to snapshot.failures, "lastSweepSelected" to snapshot.lastSweepSelected, 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 index bfabbd6..77eff44 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/JobActivity.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/metrics/JobActivity.kt @@ -22,6 +22,9 @@ class JobActivity { @Volatile private var lastTickMillis = 0L + @Volatile + private var lastFailureAt: Instant? = null + /** 上一轮回填扫描选中的待办条数(扫描积压的即时值)。 */ @Volatile var lastSweepSelected = 0 @@ -46,18 +49,20 @@ class JobActivity { ticks.incrementAndGet() } - /** 一次作业 tick 抛错:只累计失败,**不更新心跳时刻**——停摆由心跳年龄暴露。 */ - fun tickFailed() { + /** 一次作业 tick 抛错:记下最近失败时刻,**不更新心跳时刻**——停摆由心跳年龄暴露。 */ + fun tickFailed(now: Instant) { + lastFailureAt = now failures.incrementAndGet() } fun snapshot(): Snapshot = - Snapshot(started, lastTickAt, lastTickMillis, lastSweepSelected, ticks.get(), failures.get()) + Snapshot(started, lastTickAt, lastTickMillis, lastFailureAt, lastSweepSelected, ticks.get(), failures.get()) data class Snapshot( val started: Boolean, val lastTickAt: Instant?, val lastTickMillis: Long, + val lastFailureAt: Instant?, 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 2259799..ff1a85d 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 @@ -24,6 +24,7 @@ import java.time.Duration * - `msgx.pipeline.watermark.lag`:水位落后信箱最新 ID 的距离 * - `msgx.pipeline.hole.aged_out.total`:永久空洞放行次数 * - `msgx.pipeline.job.heartbeat_age_seconds`:距上一次作业 tick 完成的秒数(未跑过为 -1) + * - `msgx.pipeline.job.last_failure_age_seconds`:距最近一次作业 tick 失败的秒数(从未失败为 -1) * - `msgx.pipeline.job.ticks.total` / `msgx.pipeline.job.failures.total`:作业 tick 完成/抛错次数 * - `msgx.pipeline.job.last_sweep_selected`:上一轮回填扫描选中的待办条数(扫描积压) * @@ -84,6 +85,10 @@ class PipelineMetrics( a.snapshot().lastTickAt?.let { Duration.between(it, clock.instant()).seconds.toDouble() } ?: -1.0 }.strongReference(true).register(registry) + Gauge.builder("msgx.pipeline.job.last_failure_age_seconds", activity) { a -> + a.snapshot().lastFailureAt?.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) 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 5dd75e0..762cd62 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunner.kt @@ -79,7 +79,7 @@ class JobRunner( } catch (e: InterruptedException) { throw e } catch (e: Exception) { - activity.tickFailed() + activity.tickFailed(startedAt) log.warn("job tick failed: {}", e.message ?: e.javaClass.simpleName) } } 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 index 9290b51..bca15f6 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicatorTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/health/JobRunnerHealthIndicatorTest.kt @@ -59,12 +59,16 @@ class JobRunnerHealthIndicatorTest { fun `a failed tick counts up without refreshing the heartbeat`() { val activity = JobActivity().apply { started() } - activity.tickFailed() - activity.tickFailed() + activity.tickFailed(now.minusSeconds(5)) + activity.tickFailed(now) val snapshot = activity.snapshot() assertEquals(2L, snapshot.failures) assertEquals(0L, snapshot.ticks) assertEquals(null, snapshot.lastTickAt) + + val details = details(jobRunnerHealth(snapshot, now)) + assertEquals(0L, details["lastFailureAgeSeconds"]) + assertEquals(2L, details["failures"]) } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt index 7c94785..dda6a71 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/jobs/JobRunnerTest.kt @@ -69,6 +69,7 @@ class JobRunnerTest { val snapshot = activity.snapshot() assertNull(snapshot.lastTickAt) + assertEquals(MutableClock.BASE, snapshot.lastFailureAt) assertEquals(0L, snapshot.ticks) assertEquals(1L, snapshot.failures) }