From 8b7a1ae76beb12343bc1f6f06ed7d661d97812cc Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 21:17:34 +0800 Subject: [PATCH] feat: add gb32960 session diagnostics --- .../actuator/Gb32960SessionsEndpoint.java | 27 +++++ .../src/main/resources/application.yml | 2 +- .../Gb32960IngestAppDefaultsTest.java | 4 +- .../actuator/Gb32960SessionsEndpointTest.java | 33 +++++ .../config/Gb32960AutoConfiguration.java | 11 +- .../inbound/Gb32960ChannelHandler.java | 14 ++- .../inbound/Gb32960ConnectionDiagnostics.java | 113 ++++++++++++++++++ .../gb32960/inbound/Gb32960NettyServer.java | 6 +- .../Gb32960ChannelHandlerAckBoundaryTest.java | 3 +- .../Gb32960ConnectionDiagnosticsTest.java | 78 ++++++++++++ 10 files changed, 285 insertions(+), 6 deletions(-) create mode 100644 modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpoint.java create mode 100644 modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpointTest.java create mode 100644 modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnostics.java create mode 100644 modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnosticsTest.java diff --git a/modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpoint.java b/modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpoint.java new file mode 100644 index 00000000..d44ba2dd --- /dev/null +++ b/modules/apps/gb32960-ingest-app/src/main/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpoint.java @@ -0,0 +1,27 @@ +package com.lingniu.ingest.gb32960app.actuator; + +import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960ConnectionDiagnostics; +import org.springframework.boot.actuate.endpoint.annotation.Endpoint; +import org.springframework.boot.actuate.endpoint.annotation.ReadOperation; +import org.springframework.stereotype.Component; + +import java.util.Map; + +@Component +@Endpoint(id = "gb32960-sessions") +public class Gb32960SessionsEndpoint { + + private final Gb32960ConnectionDiagnostics diagnostics; + + public Gb32960SessionsEndpoint(Gb32960ConnectionDiagnostics diagnostics) { + this.diagnostics = diagnostics; + } + + @ReadOperation + public Map sessions() { + var sessions = diagnostics.activeSessions(); + return Map.of( + "activeCount", sessions.size(), + "sessions", sessions); + } +} diff --git a/modules/apps/gb32960-ingest-app/src/main/resources/application.yml b/modules/apps/gb32960-ingest-app/src/main/resources/application.yml index 25836691..ce7666ba 100644 --- a/modules/apps/gb32960-ingest-app/src/main/resources/application.yml +++ b/modules/apps/gb32960-ingest-app/src/main/resources/application.yml @@ -119,7 +119,7 @@ management: endpoints: web: exposure: - include: health,info,metrics,prometheus,env + include: health,info,metrics,prometheus,env,gb32960-sessions health: redis: enabled: ${MANAGEMENT_HEALTH_REDIS_ENABLED:false} diff --git a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java index 2796b0f1..69382d0b 100644 --- a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java +++ b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java @@ -30,7 +30,9 @@ class Gb32960IngestAppDefaultsTest { .containsEntry("lingniu.ingest.event-file-store.enabled", false) .containsEntry("lingniu.ingest.event-history.enabled", false) .containsEntry("lingniu.ingest.vehicle-state.enabled", false) - .containsEntry("lingniu.ingest.vehicle-stat.enabled", false); + .containsEntry("lingniu.ingest.vehicle-stat.enabled", false) + .containsEntry("management.endpoints.web.exposure.include", + "health,info,metrics,prometheus,env,gb32960-sessions"); } private static Properties applicationProperties() { diff --git a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpointTest.java b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpointTest.java new file mode 100644 index 00000000..72d475b8 --- /dev/null +++ b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/actuator/Gb32960SessionsEndpointTest.java @@ -0,0 +1,33 @@ +package com.lingniu.ingest.gb32960app.actuator; + +import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960ConnectionDiagnostics; +import com.lingniu.ingest.protocol.gb32960.model.CommandType; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class Gb32960SessionsEndpointTest { + + @Test + void returnsActiveGb32960Sessions() { + Gb32960ConnectionDiagnostics diagnostics = new Gb32960ConnectionDiagnostics(); + diagnostics.opened("session-1", "127.0.0.1:65469"); + diagnostics.received("session-1", CommandType.PLATFORM_LOGIN, ""); + diagnostics.platformLoginAccepted("session-1", "Hyundai"); + + Gb32960SessionsEndpoint endpoint = new Gb32960SessionsEndpoint(diagnostics); + + Map result = endpoint.sessions(); + + assertThat(result).containsEntry("activeCount", 1); + assertThat(result.get("sessions")).isInstanceOf(List.class); + assertThat((List) result.get("sessions")).singleElement() + .hasFieldOrPropertyWithValue("sessionId", "session-1") + .hasFieldOrPropertyWithValue("peer", "127.0.0.1:65469") + .hasFieldOrPropertyWithValue("lastCommand", "PLATFORM_LOGIN") + .hasFieldOrPropertyWithValue("platformAccount", "Hyundai"); + } +} diff --git a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java index 2641cc54..513b3f51 100644 --- a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java +++ b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/config/Gb32960AutoConfiguration.java @@ -42,6 +42,7 @@ import com.lingniu.ingest.protocol.gb32960.codec.profile.VendorExtensionSelector import com.lingniu.ingest.protocol.gb32960.handler.Gb32960RealtimeHandler; import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960AccessService; import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960AckService; +import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960ConnectionDiagnostics; import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960FrameDiagnostics; import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer; import com.lingniu.ingest.protocol.gb32960.mapper.Gb32960EventMapper; @@ -220,6 +221,12 @@ public class Gb32960AutoConfiguration { props.getDiagnostics().getMaxLoggedFrameKeysPerChannel()); } + @Bean + @ConditionalOnMissingBean + public Gb32960ConnectionDiagnostics gb32960ConnectionDiagnostics() { + return new Gb32960ConnectionDiagnostics(); + } + @Bean @ConditionalOnMissingBean @ConditionalOnProperty(prefix = "lingniu.ingest.gb32960.server", name = "enabled", @@ -229,7 +236,8 @@ public class Gb32960AutoConfiguration { Dispatcher dispatcher, Gb32960AccessService accessService, Gb32960AckService ackService, - Gb32960FrameDiagnostics diagnostics) { + Gb32960FrameDiagnostics diagnostics, + Gb32960ConnectionDiagnostics connectionDiagnostics) { return new Gb32960NettyServer( props.getPort(), props.getWorkerThreads(), @@ -239,6 +247,7 @@ public class Gb32960AutoConfiguration { accessService, ackService, diagnostics, + connectionDiagnostics, props.getTls()); } } diff --git a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java index 8285f6aa..437a5539 100644 --- a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java +++ b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java @@ -64,26 +64,31 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { private final Gb32960AccessService accessService; private final Gb32960AckService ackService; private final Gb32960FrameDiagnostics diagnostics; + private final Gb32960ConnectionDiagnostics connectionDiagnostics; public Gb32960ChannelHandler(Gb32960MessageDecoder decoder, Dispatcher dispatcher, Gb32960AccessService accessService, Gb32960AckService ackService, - Gb32960FrameDiagnostics diagnostics) { + Gb32960FrameDiagnostics diagnostics, + Gb32960ConnectionDiagnostics connectionDiagnostics) { this.decoder = decoder; this.dispatcher = dispatcher; this.accessService = accessService; this.ackService = ackService; this.diagnostics = diagnostics; + this.connectionDiagnostics = connectionDiagnostics; } @Override public void channelActive(ChannelHandlerContext ctx) { + connectionDiagnostics.opened(sessionId(ctx), addr(ctx)); log.info("[gb32960] connection opened peer={}", addr(ctx)); } @Override public void channelInactive(ChannelHandlerContext ctx) { + connectionDiagnostics.closed(sessionId(ctx)); log.info("[gb32960] connection closed peer={}", addr(ctx)); } @@ -94,6 +99,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent idle && idle.state() == IdleState.READER_IDLE) { + connectionDiagnostics.readIdle(sessionId(ctx)); log.warn("[gb32960] read idle peer={} noInboundFrameFor>=thresholdSeconds (see idleReadSeconds)", addr(ctx)); } @@ -115,6 +121,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { String vin = msg.header().vin(); CommandType cmd = msg.header().command(); + connectionDiagnostics.received(sessionId(ctx), cmd, vin); // 原样回显请求帧的 17 字节 VIN（frame[4..20]），避免平台登入 VIN 全 0 // 被 vinPadded 重写为空格导致对端判定 VIN 不匹配。 byte[] rawVin = new byte[17]; @@ -339,6 +346,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { addr(ctx), pl.username(), pl.encryptRule(), pl.serialNo(), pl.loginTime(), result); // 把 username 钉到 channel attribute，供后续 0x02/0x03 帧的 vendor profile 路由 accessService.bindPlatformAccount(ctx.channel(), pl.username()); + connectionDiagnostics.platformLoginAccepted(sessionId(ctx), pl.username()); return true; } @@ -494,4 +502,8 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler { return "unknown"; } + private static String sessionId(ChannelHandlerContext ctx) { + return ctx.channel().id().asLongText(); + } + } diff --git a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnostics.java b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnostics.java new file mode 100644 index 00000000..a7be8bd5 --- /dev/null +++ b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnostics.java @@ -0,0 +1,113 @@ +package com.lingniu.ingest.protocol.gb32960.inbound; + +import com.lingniu.ingest.protocol.gb32960.model.CommandType; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +public class Gb32960ConnectionDiagnostics { + + private final Clock clock; + private final ConcurrentMap activeSessions = new ConcurrentHashMap<>(); + + public Gb32960ConnectionDiagnostics() { + this(Clock.systemUTC()); + } + + Gb32960ConnectionDiagnostics(Clock clock) { + this.clock = clock; + } + + public void opened(String sessionId, String peer) { + Instant now = clock.instant(); + activeSessions.put(sessionId, new MutableSession(sessionId, peer, now)); + } + + public void received(String sessionId, CommandType command, String vin) { + activeSessions.computeIfPresent(sessionId, (ignored, session) -> { + session.lastInboundAt = clock.instant(); + session.lastCommandCode = command.code(); + session.lastCommand = command.name(); + session.vin = vin == null ? "" : vin; + return session; + }); + } + + public void platformLoginAccepted(String sessionId, String platformAccount) { + activeSessions.computeIfPresent(sessionId, (ignored, session) -> { + session.platformAccount = platformAccount == null ? "" : platformAccount; + return session; + }); + } + + public void readIdle(String sessionId) { + activeSessions.computeIfPresent(sessionId, (ignored, session) -> { + session.readIdleCount++; + return session; + }); + } + + public void closed(String sessionId) { + activeSessions.remove(sessionId); + } + + public List activeSessions() { + Instant now = clock.instant(); + return activeSessions.values().stream() + .map(session -> session.snapshot(now)) + .sorted(Comparator.comparing(SessionSnapshot::connectedAt)) + .toList(); + } + + public record SessionSnapshot( + String sessionId, + String peer, + Instant connectedAt, + Instant lastInboundAt, + long idleSeconds, + int lastCommandCode, + String lastCommand, + String vin, + String platformAccount, + long readIdleCount) { + } + + private static final class MutableSession { + private final String sessionId; + private final String peer; + private final Instant connectedAt; + private Instant lastInboundAt; + private int lastCommandCode; + private String lastCommand = ""; + private String vin = ""; + private String platformAccount = ""; + private long readIdleCount; + + private MutableSession(String sessionId, String peer, Instant connectedAt) { + this.sessionId = sessionId; + this.peer = peer; + this.connectedAt = connectedAt; + this.lastInboundAt = connectedAt; + } + + private SessionSnapshot snapshot(Instant now) { + long idleSeconds = Math.max(0, Duration.between(lastInboundAt, now).toSeconds()); + return new SessionSnapshot( + sessionId, + peer, + connectedAt, + lastInboundAt, + idleSeconds, + lastCommandCode, + lastCommand, + vin, + platformAccount, + readIdleCount); + } + } +} diff --git a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960NettyServer.java b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960NettyServer.java index 50df20b5..49e67501 100644 --- a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960NettyServer.java +++ b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960NettyServer.java @@ -50,6 +50,7 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean { private final Gb32960AccessService accessService; private final Gb32960AckService ackService; private final Gb32960FrameDiagnostics diagnostics; + private final Gb32960ConnectionDiagnostics connectionDiagnostics; private final Gb32960Properties.Tls tlsConfig; private EventLoopGroup boss; @@ -63,6 +64,7 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean { Gb32960AccessService accessService, Gb32960AckService ackService, Gb32960FrameDiagnostics diagnostics, + Gb32960ConnectionDiagnostics connectionDiagnostics, Gb32960Properties.Tls tlsConfig) { this.port = port; this.workerThreads = workerThreads; @@ -72,6 +74,7 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean { this.accessService = accessService; this.ackService = ackService; this.diagnostics = diagnostics; + this.connectionDiagnostics = connectionDiagnostics; this.tlsConfig = tlsConfig; } @@ -113,7 +116,8 @@ public class Gb32960NettyServer implements InitializingBean, DisposableBean { .addLast("frame", new Gb32960FrameDecoder()) .addLast("dispatch", new Gb32960ChannelHandler(messageDecoder, dispatcher, - accessService, ackService, diagnostics)); + accessService, ackService, diagnostics, + connectionDiagnostics)); } }); this.serverChannel = b.bind(port).sync().channel(); diff --git a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java index d2ce859d..4522bb79 100644 --- a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java +++ b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java @@ -139,7 +139,8 @@ class Gb32960ChannelHandlerAckBoundaryTest { dispatcher, accessService(), new Gb32960AckService(), - new Gb32960FrameDiagnostics(16)); + new Gb32960FrameDiagnostics(16), + new Gb32960ConnectionDiagnostics()); } private static Gb32960MessageDecoder decoder() { diff --git a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnosticsTest.java b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnosticsTest.java new file mode 100644 index 00000000..cc68c65c --- /dev/null +++ b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ConnectionDiagnosticsTest.java @@ -0,0 +1,78 @@ +package com.lingniu.ingest.protocol.gb32960.inbound; + +import com.lingniu.ingest.protocol.gb32960.model.CommandType; +import org.junit.jupiter.api.Test; + +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; + +import static org.assertj.core.api.Assertions.assertThat; + +class Gb32960ConnectionDiagnosticsTest { + + private static final Instant NOW = Instant.parse("2026-06-29T10:00:00Z"); + + @Test + void tracksActiveSessionCommandPlatformAccountAndReadIdleCount() { + MutableClock clock = new MutableClock(NOW); + Gb32960ConnectionDiagnostics diagnostics = new Gb32960ConnectionDiagnostics(clock); + + diagnostics.opened("session-1", "127.0.0.1:65469"); + clock.advanceSeconds(5); + diagnostics.received("session-1", CommandType.PLATFORM_LOGIN, ""); + diagnostics.platformLoginAccepted("session-1", "Hyundai"); + clock.advanceSeconds(30); + diagnostics.readIdle("session-1"); + + assertThat(diagnostics.activeSessions()).singleElement().satisfies(session -> { + assertThat(session.sessionId()).isEqualTo("session-1"); + assertThat(session.peer()).isEqualTo("127.0.0.1:65469"); + assertThat(session.connectedAt()).isEqualTo(NOW); + assertThat(session.lastInboundAt()).isEqualTo(NOW.plusSeconds(5)); + assertThat(session.idleSeconds()).isEqualTo(30); + assertThat(session.lastCommandCode()).isEqualTo(0x05); + assertThat(session.lastCommand()).isEqualTo("PLATFORM_LOGIN"); + assertThat(session.vin()).isEmpty(); + assertThat(session.platformAccount()).isEqualTo("Hyundai"); + assertThat(session.readIdleCount()).isEqualTo(1); + }); + } + + @Test + void removesSessionWhenChannelCloses() { + Gb32960ConnectionDiagnostics diagnostics = new Gb32960ConnectionDiagnostics(Clock.fixed(NOW, ZoneOffset.UTC)); + + diagnostics.opened("session-1", "127.0.0.1:65469"); + diagnostics.closed("session-1"); + + assertThat(diagnostics.activeSessions()).isEmpty(); + } + + private static final class MutableClock extends Clock { + private Instant instant; + + private MutableClock(Instant instant) { + this.instant = instant; + } + + void advanceSeconds(long seconds) { + instant = instant.plusSeconds(seconds); + } + + @Override + public ZoneOffset getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(java.time.ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return instant; + } + } +}