From a2c60d319fd09edb646fca76d428b3dbd5979caf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Fri, 4 Sep 2026 23:46:49 +0800 Subject: [PATCH] =?UTF-8?q?task-181:=20task.created/running=20=E6=8C=87?= =?UTF-8?q?=E6=A0=87=E5=9F=8B=E7=82=B9=EF=BC=88TaskObservabilityMetrics=20?= =?UTF-8?q?recorder=EF=BC=9Acreated=20=E8=AE=A1=E6=95=B0=20+=20running=20g?= =?UTF-8?q?auge=20+=20moduleType/instanceId=20=E6=A0=87=E7=AD=BE=EF=BC=8C?= =?UTF-8?q?=E5=90=8D=E7=A7=B0=E5=86=BB=E7=BB=93=EF=BC=89+=208=20=E6=9D=A1?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 标准指标名 aiimage.task.created/running（spec11 §1）；moduleType/instanceId null 归 unknown；running 不为负 - Micrometer 可选：无注册表降级 debug 日志不抛错；recorder 为唯一接线面，各业务状态流转按 spec 待监控栈就绪后挂载 - 仅新增类与测试，不接调用点，零生产行为变化 --- .../metrics/TaskObservabilityMetrics.java | 171 ++++++++++++++++++ ...TaskObservabilityLifecycleMetricsTest.java | 92 ++++++++++ 2 files changed, 263 insertions(+) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/metrics/TaskObservabilityMetrics.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/metrics/TaskObservabilityLifecycleMetricsTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/metrics/TaskObservabilityMetrics.java b/backend-java/src/main/java/com/nanri/aiimage/metrics/TaskObservabilityMetrics.java new file mode 100644 index 00000000..72e98a88 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/metrics/TaskObservabilityMetrics.java @@ -0,0 +1,171 @@ +package com.nanri.aiimage.metrics; + +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Tag; +import io.micrometer.core.instrument.Timer; +import lombok.extern.slf4j.Slf4j; + +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 任务/结果文件/上传/临时磁盘统一指标 recorder（module 11，spec §1 标准名）。 + * + * 指标名与标签按 spec 11 §1 冻结：aiimage.task.* / aiimage.file-job.* / aiimage.upload.* / + * aiimage.tmpdisk.*，标签 moduleType/instanceId/stage；moduleType/instanceId null 一律归一 + * unknown；不记录用户级数据。Micrometer 注册表可选：未配置时事件仅降级 debug 日志（不引入 + * 新监控组件；生产接线按 spec 待监控基础设施就绪后挂载，本类为唯一接线面）。 + * + * 供各业务 Service 状态流转 / TaskResultFileJobWorker / TaskHeartbeatService / 上传 / 临时磁盘 + * 巡检调用；instanceId 由 InstanceMetadata 解析（双实例 server-110/server-121 区分）。 + */ +@Slf4j +public class TaskObservabilityMetrics { + + private static final String UNKNOWN = "unknown"; + + private final String instanceId; + private final MeterRegistry registry; + + private final ConcurrentMap runningValues = new ConcurrentHashMap<>(); + private final ConcurrentMap heartbeatAgeValues = new ConcurrentHashMap<>(); + private final ConcurrentMap capacityValues = new ConcurrentHashMap<>(); + private final ConcurrentMap usedValues = new ConcurrentHashMap<>(); + + public TaskObservabilityMetrics(String instanceId, MeterRegistry registry) { + this.instanceId = instanceId == null || instanceId.isBlank() ? UNKNOWN : instanceId.trim(); + this.registry = registry; + } + + // ---- 任务生命周期 task.* ---- + + /** 任务创建计数（moduleType/instanceId 标签）。 */ + public void taskCreated(String moduleType) { + bump("aiimage.task.created", moduleType); + } + + /** 运行中计数增量：开始 +1 / 结束 -1，底稿是 gauge（never 负）。 */ + public void taskRunningDelta(String moduleType, int delta) { + AtomicLong value = runningValues.computeIfAbsent(moduleType, k -> new AtomicLong()); + long next = Math.max(0, value.get() + delta); + value.set(next); + if (registry != null) { + Gauge.builder("aiimage.task.running", value, AtomicLong::get) + .tags(baseTags(moduleType)) + .register(registry); + } + } + + /** 终态计数：terminal ∈ success/failed/cancelled。 */ + public void taskTerminal(String moduleType, String terminal) { + bump("aiimage.task." + terminal, moduleType); + } + + /** 任务耗时（毫秒）：Timer 记录，无注册表时降级 debug 日志。 */ + public void taskDuration(String moduleType, long durationMs) { + recordTimer("aiimage.task.duration", moduleType, durationMs, "任务耗时"); + } + + /** 心跳年龄 gauge（毫秒）：由 TaskHeartbeatService/巡检把"最后心跳距今"写入。 */ + public void taskHeartbeatAge(String moduleType, long ageMillis) { + AtomicLong value = heartbeatAgeValues.computeIfAbsent(moduleType, k -> new AtomicLong()); + value.set(Math.max(0, ageMillis)); + if (registry != null) { + Gauge.builder("aiimage.task.heartbeat.age", value, AtomicLong::get) + .tags(baseTags(moduleType)) + .register(registry); + } + } + + // ---- 结果文件 file-job.* ---- + + /** file-job 阶段计数：state ∈ pending/running/success/failed/retry。 */ + public void fileJobState(String moduleType, String state) { + bump("aiimage.file-job." + state, moduleType); + } + + public void fileJobDuration(String moduleType, long durationMs) { + recordTimer("aiimage.file-job.duration", moduleType, durationMs, "文件任务耗时"); + } + + // ---- 上传 upload.* ---- + + /** 上传统计：耗时 Timer + 大小 summary + 结果计数。 */ + public void upload(String moduleType, long bytes, long durationMs, boolean success) { + if (registry != null) { + Timer timer = Timer.builder("aiimage.upload.duration") + .tags(baseTags(moduleType)).register(registry); + timer.record(durationMs, TimeUnit.MILLISECONDS); + + DistributionSummary size = DistributionSummary.builder("aiimage.upload.size") + .tags(baseTags(moduleType)).register(registry); + size.record(bytes); + } else { + log.debug("[metrics][upload] 无监控栈，降级记录 moduleType={} bytes={} durationMs={} success={}", + moduleType, bytes, durationMs, success); + } + bump("aiimage.upload.result", moduleType); + bump("aiimage.upload.result." + (success ? "success" : "failure"), moduleType); + } + + // ---- 临时磁盘 tmpdisk.* ---- + + /** 临时磁盘容量/占用 gauge（标签 path）。 */ + public void disk(String path, long capacityBytes, long usedBytes) { + AtomicLong cap = capacityValues.computeIfAbsent(path, k -> new AtomicLong()); + AtomicLong used = usedValues.computeIfAbsent(path, k -> new AtomicLong()); + cap.set(capacityBytes); + used.set(usedBytes); + if (registry != null) { + Gauge.builder("aiimage.tmpdisk.capacity", cap, AtomicLong::get) + .tags(pathTags(path)).register(registry); + Gauge.builder("aiimage.tmpdisk.used", used, AtomicLong::get) + .tags(pathTags(path)).register(registry); + } + } + + // ---- helpers ---- + + private void bump(String name, String moduleType) { + if (registry == null) { + log.debug("[metrics] 无监控栈，降级记录 name={} moduleType={} instanceId={}", + name, moduleType, instanceId); + return; + } + Counter.builder(name).tags(baseTags(moduleType)).register(registry).increment(); + } + + private void recordTimer(String name, String moduleType, long durationMs, String label) { + long safe = Math.max(0, durationMs); + if (registry != null) { + Timer.builder(name).tags(baseTags(moduleType)).register(registry) + .record(safe, TimeUnit.MILLISECONDS); + } else { + log.debug("[metrics] 无监控栈，{}降级 moduleType={} durationMs={} instanceId={}", + label, moduleType, safe, instanceId); + } + } + + private List baseTags(String moduleType) { + return List.of(Tag.of("moduleType", normalize(moduleType)), + Tag.of("instanceId", instanceId)); + } + + private List pathTags(String path) { + return List.of(Tag.of("path", normalize(path))); + } + + private static String normalize(String value) { + return value == null || value.isBlank() ? UNKNOWN : value.trim(); + } + + public String instanceId() { + return instanceId; + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/metrics/TaskObservabilityLifecycleMetricsTest.java b/backend-java/src/test/java/com/nanri/aiimage/metrics/TaskObservabilityLifecycleMetricsTest.java new file mode 100644 index 00000000..3b4183c4 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/metrics/TaskObservabilityLifecycleMetricsTest.java @@ -0,0 +1,92 @@ +package com.nanri.aiimage.metrics; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * task-181：task.created/running 指标埋点契约（spec 11 §1）。 + * + * aiimage.task.created 计数与 aiimage.task.running gauge，带 moduleType/instanceId 标签； + * 无监控栈(registry=null) 时降级 debug 日志不抛错。运行计数不为负；指标名冻结。 + */ +class TaskObservabilityLifecycleMetricsTest { + + private final SimpleMeterRegistry registry = new SimpleMeterRegistry(); + private final TaskObservabilityMetrics metrics = new TaskObservabilityMetrics("server-110", registry); + + @Test + void createdIncrementedPerModule() { + metrics.taskCreated("similarasin"); + metrics.taskCreated("similarasin"); + metrics.taskCreated("publish"); + assertEquals(2.0, counter("aiimage.task.created", "similarasin").count()); + assertEquals(1.0, counter("aiimage.task.created", "publish").count()); + } + + @Test + void runningGaugeReflectsDelta() { + metrics.taskRunningDelta("similarasin", 3); + metrics.taskRunningDelta("similarasin", 1); + assertEquals(4.0, gauge("aiimage.task.running", "similarasin").value()); + } + + @Test + void runningGaugeNeverNegative() { + metrics.taskRunningDelta("similarasin", -5); + assertEquals(0.0, gauge("aiimage.task.running", "similarasin").value(), "运行计数不应为负"); + } + + @Test + void moduleLabelAttached() { + metrics.taskCreated("shopdatacrawl"); + assertEquals("shopdatacrawl", + registry.get("aiimage.task.created").counter().getId().getTag("moduleType")); + } + + @Test + void instanceLabelAttached() { + metrics.taskCreated("brand"); + assertEquals("server-110", + registry.get("aiimage.task.created").counter().getId().getTag("instanceId")); + } + + @Test + void counterAccumulatesAcrossCalls() { + for (int i = 0; i < 5; i++) { + metrics.taskCreated("ziniao"); + } + assertEquals(5.0, counter("aiimage.task.created", "ziniao").count()); + } + + @Test + void metricNamesFrozen() { + metrics.taskCreated("withdraw"); + metrics.taskRunningDelta("withdraw", 1); + boolean hasCreated = registry.getMeters().stream() + .anyMatch(m -> m.getId().getName().equals("aiimage.task.created")); + boolean hasRunning = registry.getMeters().stream() + .anyMatch(m -> m.getId().getName().equals("aiimage.task.running")); + assertTrue(hasCreated, "应存在 aiimage.task.created"); + assertTrue(hasRunning, "应存在 aiimage.task.running"); + } + + @Test + void registryOptionalLogsFallbackWithoutError() { + // 无监控栈(registry=null)：调用不抛错，事件降级 debug 日志 + TaskObservabilityMetrics withoutRegistry = new TaskObservabilityMetrics("server-121", null); + withoutRegistry.taskCreated("similarasin"); + withoutRegistry.taskRunningDelta("similarasin", 2); + assertEquals("server-121", withoutRegistry.instanceId()); + } + + private io.micrometer.core.instrument.Counter counter(String name, String moduleType) { + return registry.get(name).tag("moduleType", moduleType).counter(); + } + + private io.micrometer.core.instrument.Gauge gauge(String name, String moduleType) { + return registry.get(name).tag("moduleType", moduleType).gauge(); + } +}