fix: tighten ingest metrics protocol labels

modules/core/observability/pom.xml

@@ -16,10 +16,6 @@
             <groupId>com.lingniu.ingest</groupId>
             <artifactId>ingest-api</artifactId>
         </dependency>
-        <dependency>
-            <groupId>com.lingniu.ingest</groupId>
-            <artifactId>ingest-core</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>
@@ -28,5 +24,15 @@
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-registry-prometheus</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

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<ProtocolId, Counter> frameCounters = new ConcurrentHashMap<>();
-    private final ConcurrentMap<ProtocolId, Counter> errorCounters = new ConcurrentHashMap<>();
-    private final ConcurrentMap<ProtocolId, Timer> frameTimers = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, Counter> errorCounters = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, Timer> frameTimers = new ConcurrentHashMap<>();
     private final ConcurrentMap<String, Counter> 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();
+    }
 }

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"));
+    }
+}