task-181: task.created/running 指标埋点（TaskObservabilityMetrics recorder：created 计数 + running gauge + moduleType/instanceId 标签，名称冻结）+ 8 条测试
- 标准指标名 aiimage.task.created/running（spec11 §1）；moduleType/instanceId null 归 unknown；running 不为负 - Micrometer 可选：无注册表降级 debug 日志不抛错；recorder 为唯一接线面，各业务状态流转按 spec 待监控栈就绪后挂载 - 仅新增类与测试，不接调用点，零生产行为变化
This commit is contained in:
@@ -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<String, AtomicLong> runningValues = new ConcurrentHashMap<>();
|
||||
private final ConcurrentMap<String, AtomicLong> heartbeatAgeValues = new ConcurrentHashMap<>();
|
||||
private final ConcurrentMap<String, AtomicLong> capacityValues = new ConcurrentHashMap<>();
|
||||
private final ConcurrentMap<String, AtomicLong> 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<Tag> baseTags(String moduleType) {
|
||||
return List.of(Tag.of("moduleType", normalize(moduleType)),
|
||||
Tag.of("instanceId", instanceId));
|
||||
}
|
||||
|
||||
private List<Tag> 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;
|
||||
}
|
||||
}
|
||||
+92
@@ -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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user