diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java
index 47aada88..3fce1d5b 100644
--- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java
+++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java
@@ -231,6 +231,7 @@ class MavenModuleProfileTest {
@Test
void observabilityBuildSurfaceUsesMicrometerWithoutUnusedOpenTelemetryBom() throws Exception {
Document pom = rootPom();
+ Document observabilityPom = modulePom("modules/core/observability/pom.xml");
assertThat(hasProjectProperty(pom, "micrometer.version"))
.isTrue();
@@ -240,6 +241,9 @@ class MavenModuleProfileTest {
.isFalse();
assertThat(hasDependency(pom, "io.opentelemetry", "opentelemetry-bom"))
.isFalse();
+ assertThat(hasDependency(observabilityPom, "com.lingniu.ingest", "ingest-core"))
+ .as("observability is an interceptor API module and should not depend on core implementation")
+ .isFalse();
assertThat(productionJavaContains("io.opentelemetry"))
.isFalse();
}
diff --git a/modules/core/observability/pom.xml b/modules/core/observability/pom.xml
index 88cbaa32..e5e14c6e 100644
--- a/modules/core/observability/pom.xml
+++ b/modules/core/observability/pom.xml
@@ -16,10 +16,6 @@
com.lingniu.ingest
ingest-api
-
- com.lingniu.ingest
- ingest-core
-
org.springframework.boot
spring-boot-starter-actuator
@@ -28,5 +24,15 @@
io.micrometer
micrometer-registry-prometheus
+
+ org.junit.jupiter
+ junit-jupiter
+ test
+
+
+ org.assertj
+ assertj-core
+ test
+
diff --git a/modules/core/observability/src/main/java/com/lingniu/ingest/observability/IngestMetrics.java b/modules/core/observability/src/main/java/com/lingniu/ingest/observability/IngestMetrics.java
index 57125bdf..8769c1c0 100644
--- a/modules/core/observability/src/main/java/com/lingniu/ingest/observability/IngestMetrics.java
+++ b/modules/core/observability/src/main/java/com/lingniu/ingest/observability/IngestMetrics.java
@@ -28,11 +28,12 @@ import java.util.concurrent.ConcurrentMap;
public class IngestMetrics implements IngestInterceptor, Ordered {
private static final String ATTR_TIMER_SAMPLE = "metrics.timer.sample";
+ private static final String ATTR_PROTOCOL = "metrics.protocol";
private final MeterRegistry registry;
private final ConcurrentMap frameCounters = new ConcurrentHashMap<>();
- private final ConcurrentMap errorCounters = new ConcurrentHashMap<>();
- private final ConcurrentMap frameTimers = new ConcurrentHashMap<>();
+ private final ConcurrentMap errorCounters = new ConcurrentHashMap<>();
+ private final ConcurrentMap frameTimers = new ConcurrentHashMap<>();
private final ConcurrentMap eventCounters = new ConcurrentHashMap<>();
public IngestMetrics(MeterRegistry registry) {
@@ -41,7 +42,8 @@ public class IngestMetrics implements IngestInterceptor, Ordered {
@Override
public boolean before(RawFrame frame, IngestContext ctx) {
- ProtocolId p = frame.protocolId();
+ ProtocolId p = frame.protocolId() == null ? ProtocolId.UNKNOWN : frame.protocolId();
+ ctx.attr(ATTR_PROTOCOL, p);
frameCounters.computeIfAbsent(p, k ->
Counter.builder("ingest_frames_total").tag("protocol", k.name()).register(registry)).increment();
Timer.Sample sample = Timer.start(registry);
@@ -61,24 +63,42 @@ public class IngestMetrics implements IngestInterceptor, Ordered {
Timer.Sample s = ctx.attr(ATTR_TIMER_SAMPLE);
if (s != null) {
- Timer t = frameTimers.computeIfAbsent(event.source(), p ->
- Timer.builder("ingest_frame_duration_seconds").tag("protocol", p.name()).register(registry));
- s.stop(t);
- ctx.attr(ATTR_TIMER_SAMPLE, null);
+ stopTimer(ctx, event.source() == null ? "unknown" : event.source().name());
}
}
@Override
public void onError(Throwable error, IngestContext ctx) {
- // 无协议上下文时归到 UNKNOWN
- Counter c = errorCounters.computeIfAbsent(
- ProtocolId.GB32960, // 默认桶:由上层拦截器设置具体协议更精确
- p -> Counter.builder("ingest_errors_total").tag("protocol", "unknown").register(registry));
- c.increment();
+ String protocol = protocolTag(ctx);
+ errorCounters.computeIfAbsent(protocol, p ->
+ Counter.builder("ingest_errors_total").tag("protocol", p).register(registry)).increment();
+ stopTimer(ctx, protocol);
}
@Override
public int getOrder() {
return 10; // 尽早介入,晚于 Tracing(order=0)
}
+
+ private void stopTimer(IngestContext ctx, String protocol) {
+ if (ctx == null) {
+ return;
+ }
+ Timer.Sample sample = ctx.attr(ATTR_TIMER_SAMPLE);
+ if (sample == null) {
+ return;
+ }
+ Timer timer = frameTimers.computeIfAbsent(protocol, p ->
+ Timer.builder("ingest_frame_duration_seconds").tag("protocol", p).register(registry));
+ sample.stop(timer);
+ ctx.attr(ATTR_TIMER_SAMPLE, null);
+ }
+
+ private static String protocolTag(IngestContext ctx) {
+ if (ctx == null) {
+ return "unknown";
+ }
+ ProtocolId protocol = ctx.attr(ATTR_PROTOCOL);
+ return protocol == null || protocol == ProtocolId.UNKNOWN ? "unknown" : protocol.name();
+ }
}
diff --git a/modules/core/observability/src/test/java/com/lingniu/ingest/observability/IngestMetricsTest.java b/modules/core/observability/src/test/java/com/lingniu/ingest/observability/IngestMetricsTest.java
new file mode 100644
index 00000000..85003585
--- /dev/null
+++ b/modules/core/observability/src/test/java/com/lingniu/ingest/observability/IngestMetricsTest.java
@@ -0,0 +1,65 @@
+package com.lingniu.ingest.observability;
+
+import com.lingniu.ingest.api.ProtocolId;
+import com.lingniu.ingest.api.pipeline.IngestContext;
+import com.lingniu.ingest.api.pipeline.RawFrame;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class IngestMetricsTest {
+
+ @Test
+ void countsErrorsAgainstTheFrameProtocolCapturedInBefore() {
+ SimpleMeterRegistry registry = new SimpleMeterRegistry();
+ IngestMetrics metrics = new IngestMetrics(registry);
+ IngestContext context = new IngestContext("trace-1");
+
+ metrics.before(frame(ProtocolId.JT808), context);
+ metrics.onError(new IllegalStateException("boom"), context);
+
+ assertThat(registry.counter("ingest_errors_total", "protocol", "JT808").count())
+ .isEqualTo(1.0);
+ assertThat(registry.find("ingest_errors_total").tag("protocol", "unknown").counter())
+ .isNull();
+ }
+
+ @Test
+ void recordsFrameDurationWhenProcessingFailsAfterBefore() {
+ SimpleMeterRegistry registry = new SimpleMeterRegistry();
+ IngestMetrics metrics = new IngestMetrics(registry);
+ IngestContext context = new IngestContext("trace-2");
+
+ metrics.before(frame(ProtocolId.MQTT_YUTONG), context);
+ metrics.onError(new RuntimeException("boom"), context);
+
+ assertThat(registry.timer("ingest_frame_duration_seconds", "protocol", "MQTT_YUTONG").count())
+ .isEqualTo(1);
+ }
+
+ @Test
+ void countsErrorsAsUnknownWhenThereIsNoFrameProtocolContext() {
+ SimpleMeterRegistry registry = new SimpleMeterRegistry();
+ IngestMetrics metrics = new IngestMetrics(registry);
+
+ metrics.onError(new IllegalStateException("boom"), new IngestContext("trace-3"));
+
+ assertThat(registry.counter("ingest_errors_total", "protocol", "unknown").count())
+ .isEqualTo(1.0);
+ }
+
+ private static RawFrame frame(ProtocolId protocolId) {
+ return new RawFrame(
+ protocolId,
+ 0x0200,
+ 0,
+ null,
+ new byte[]{0x01},
+ Map.of(),
+ Instant.parse("2026-07-01T00:00:00Z"));
+ }
+}