shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: support tdengine gb32960 raw queries
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Details

Browse Source
This commit is contained in:
lingniu
2026-06-29 20:19:44 +08:00
parent 12de83e37f
commit 80e96d8075
13 changed files with 398 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
modules/apps/gb32960-ingest-app/src/main/resources/application.yml
View File

@@ -44,7 +44,7 @@ lingniu:
- username: ${GB32960_PLATFORM_USER_HYUNDAI:Hyundai}
password: ${GB32960_PLATFORM_PWD_HYUNDAI:f2e3445d7cda409fb4f278f6fb890734}
allowed-ips:
- ${GB32960_PLATFORM_IP_HYUNDAI:8.134.95.166}
- ${GB32960_PLATFORM_IP_HYUNDAI:}
description: 外部下级平台 - Hyundai
tls:
enabled: ${GB32960_TLS_ENABLED:false}

6
modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/auth/Gb32960PlatformAuthorizer.java
View File

@@ -62,7 +62,11 @@ public final class Gb32960PlatformAuthorizer {
if (!Objects.equals(entry.getPassword(), password)) return Result.DENY_BAD_PASSWORD;
Set<String> allowed = entry.getAllowedIps() == null
? Set.of()
: Set.copyOf(entry.getAllowedIps());
: entry.getAllowedIps().stream()
.filter(Objects::nonNull)
.map(String::trim)
.filter(ip -> !ip.isBlank())
.collect(java.util.stream.Collectors.toSet());
if (!allowed.isEmpty() && (peerIp == null || !allowed.contains(peerIp))) {
return Result.DENY_IP_NOT_ALLOWED;
}

19
modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960AccessServiceTest.java
View File

@@ -42,6 +42,25 @@ class Gb32960AccessServiceTest {
.accepted()).isFalse();
}
@Test
void authenticatePlatformLogin_ignoresBlankAllowedIps() {
Gb32960Properties.Auth auth = new Gb32960Properties.Auth();
Gb32960Properties.Auth.Platform platform = new Gb32960Properties.Auth.Platform();
platform.setUsername("lingniu");
platform.setPassword("secret");
platform.setAllowedIps(List.of("", " ", "\t"));
auth.setPlatforms(List.of(platform));
Gb32960AccessService service = new Gb32960AccessService(
new Gb32960VinAuthorizer(auth),
new Gb32960PlatformAuthorizer(auth.getPlatforms()));
EmbeddedChannel channel = new EmbeddedChannel();
channel.connect(new InetSocketAddress("115.29.187.205", 9001));
assertThat(service.authenticatePlatformLogin("lingniu", "secret", channel)
.accepted()).isTrue();
}
private static Gb32960AccessService newService(boolean vinAuthEnabled) {
Gb32960Properties.Auth auth = new Gb32960Properties.Auth();
auth.setEnabled(vinAuthEnabled);

152
modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java
View File

@@ -9,6 +9,11 @@ import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.model.Gb32960Message;
import com.lingniu.ingest.protocol.gb32960.model.InfoBlock;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdenginePage;
import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -19,7 +24,9 @@ import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
@@ -43,8 +50,10 @@ public final class Gb32960DecodedFrameService {
private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
private static final int OUTPUT_DOUBLE_SCALE = 6;
private static final String RAW_ARCHIVE_EVENT_TYPE = "RAW_ARCHIVE";
private static final ZoneId DEFAULT_ZONE = ZoneId.of("Asia/Shanghai");
private final EventFileStore store;
private final TdengineHistoryReader tdengineReader;
private final Gb32960MessageDecoder decoder;
private final Path archiveRoot;
private final ObjectMapper objectMapper;
@@ -53,8 +62,25 @@ public final class Gb32960DecodedFrameService {
Gb32960MessageDecoder decoder,
Path archiveRoot,
ObjectMapper objectMapper) {
if (store == null) {
throw new IllegalArgumentException("store must not be null");
this(store, null, decoder, archiveRoot, objectMapper);
}
public Gb32960DecodedFrameService(TdengineHistoryReader tdengineReader,
Gb32960MessageDecoder decoder,
Path archiveRoot,
ObjectMapper objectMapper) {
this(null, tdengineReader, decoder, archiveRoot, objectMapper);
}
public Gb32960DecodedFrameService(EventFileStore store,
TdengineHistoryReader tdengineReader,
Gb32960MessageDecoder decoder,
Path archiveRoot,
ObjectMapper objectMapper) {
if (store == null && tdengineReader == null) {
log.info("gb32960 decoded frame service has no history index backend; rawUri direct lookup remains available");
} else if (store == null) {
log.info("gb32960 decoded frame service using TDengine raw_frames index");
}
if (decoder == null) {
throw new IllegalArgumentException("decoder must not be null");
@@ -63,6 +89,7 @@ public final class Gb32960DecodedFrameService {
throw new IllegalArgumentException("archiveRoot must not be null");
}
this.store = store;
this.tdengineReader = tdengineReader;
this.decoder = decoder;
this.archiveRoot = archiveRoot.toAbsolutePath().normalize();
this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
@@ -86,6 +113,22 @@ public final class Gb32960DecodedFrameService {
String vin,
String platformAccount) throws IOException {
int frameLimit = Math.max(1, Math.min(limit, 1000));
List<DecodedFrame> fromEventStore = store == null ? List.of() : queryFromEventStore(
dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, frameLimit, vin, platformAccount);
if (!fromEventStore.isEmpty() || tdengineReader == null) {
return fromEventStore;
}
return queryFromTdengine(dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, frameLimit, vin, platformAccount);
}
private List<DecodedFrame> queryFromEventStore(LocalDate dateFrom,
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
EventFileQuery.Order order,
int frameLimit,
String vin,
String platformAccount) throws IOException {
// 只查 RAW_ARCHIVE：REALTIME/LOCATION 等派生事件不再作为 32960 历史查询的数据源。
List<EventFileRecord> records = store.query(new EventFileQuery(
ProtocolId.GB32960,
@@ -118,6 +161,40 @@ public final class Gb32960DecodedFrameService {
return out;
}
private List<DecodedFrame> queryFromTdengine(LocalDate dateFrom,
LocalDate dateTo,
java.time.Instant eventTimeFrom,
java.time.Instant eventTimeTo,
EventFileQuery.Order order,
int frameLimit,
String vin,
String platformAccount) throws IOException {
if (vin == null || vin.isBlank()) {
return List.of();
}
TdenginePage<TdengineRawFrameRow> page = tdengineReader.queryRawFrames(new TdengineRawFrameQuery(
"GB32960",
vin.trim(),
fromInstant(dateFrom, eventTimeFrom),
toExclusiveInstant(dateTo, eventTimeTo),
tdengineOrder(order),
frameLimit,
null));
List<DecodedFrame> out = new ArrayList<>(page.items().size());
LinkedHashSet<String> seenUris = new LinkedHashSet<>();
for (TdengineRawFrameRow row : page.items()) {
String rawArchiveUri = row.rawUri();
if (rawArchiveUri == null || rawArchiveUri.isBlank() || !seenUris.add(rawArchiveUri)) {
continue;
}
DecodedFrame frame = decodeIfAvailable(row, platformAccount);
if (frame != null) {
out.add(frame);
}
}
return out;
}
public List<LogicalSnapshot> snapshots(QueryTimeRange range,
EventFileQuery.Order order,
int limit,
@@ -182,9 +259,11 @@ public final class Gb32960DecodedFrameService {
if (rawArchiveUri == null || rawArchiveUri.isBlank()) {
return null;
}
EventFileRecord record = store.findByRawArchiveUri(rawArchiveUri);
if (record != null) {
return decode(record, rawArchiveUri(record), platformAccount);
if (store != null) {
EventFileRecord record = store.findByRawArchiveUri(rawArchiveUri);
if (record != null) {
return decode(record, rawArchiveUri(record), platformAccount);
}
}
return decodeWithoutIndex(rawArchiveUri, platformAccount);
}
@@ -247,6 +326,16 @@ public final class Gb32960DecodedFrameService {
}
}
private DecodedFrame decodeIfAvailable(TdengineRawFrameRow row, String platformAccount) throws IOException {
try {
return decode(row, platformAccount);
} catch (NoSuchFileException e) {
log.warn("skip gb32960 frame because raw archive is missing frameId={} rawArchiveUri={}",
row.frameId(), row.rawUri());
return null;
}
}
private DecodedFrame decode(EventFileRecord record, String rawArchiveUri, String platformAccount) throws IOException {
Path rawPath = archivePath(rawArchiveUri);
byte[] rawBytes = Files.readAllBytes(rawPath);
@@ -270,6 +359,30 @@ public final class Gb32960DecodedFrameService {
blocks);
}
private DecodedFrame decode(TdengineRawFrameRow row, String platformAccount) throws IOException {
Path rawPath = archivePath(row.rawUri());
byte[] rawBytes = Files.readAllBytes(rawPath);
Gb32960Message message = decoder.decode(ByteBuffer.wrap(rawBytes), platformAccount(row, platformAccount));
List<Map<String, Object>> blocks = new ArrayList<>();
for (InfoBlock block : message.infoBlocks()) {
blocks.add(blockJson(block));
}
Instant ingestTime = row.receivedAt() == null ? Files.getLastModifiedTime(rawPath).toInstant() : row.receivedAt();
return new DecodedFrame(
row.frameId(),
message.header().vin(),
message.header().command().name(),
message.header().responseFlag().name(),
message.header().protocolVersion().name(),
message.header().encryptType().name(),
message.header().dataLength(),
message.header().eventTime() == null ? null : message.header().eventTime().toString(),
ingestTime.toString(),
row.rawUri(),
rawBytes.length,
blocks);
}
private DecodedFrame decodeWithoutIndex(String rawArchiveUri, String platformAccount) throws IOException {
Path rawPath = archivePath(rawArchiveUri);
byte[] rawBytes = Files.readAllBytes(rawPath);
@@ -460,6 +573,35 @@ public final class Gb32960DecodedFrameService {
return record.metadata().get("platformAccount");
}
private String platformAccount(TdengineRawFrameRow row, String override) {
if (override != null && !override.isBlank()) {
return override;
}
if (row.metadataJson() == null || row.metadataJson().isBlank()) {
return null;
}
try {
return objectMapper.readValue(row.metadataJson(), MAP_TYPE).getOrDefault("platformAccount", "").toString();
} catch (IOException ignored) {
return null;
}
}
private static Instant fromInstant(LocalDate dateFrom, Instant eventTimeFrom) {
return eventTimeFrom != null ? eventTimeFrom : dateFrom.atStartOfDay(DEFAULT_ZONE).toInstant();
}
private static Instant toExclusiveInstant(LocalDate dateTo, Instant eventTimeTo) {
Instant inclusive = eventTimeTo != null
? eventTimeTo
: dateTo.plusDays(1).atStartOfDay(DEFAULT_ZONE).toInstant().minusMillis(1);
return inclusive.plusMillis(1);
}
private static TdengineQueryOrder tdengineOrder(EventFileQuery.Order order) {
return order == EventFileQuery.Order.ASC ? TdengineQueryOrder.ASC : TdengineQueryOrder.DESC;
}
private static String rawArchiveEventId(String rawArchiveUri) {
if (rawArchiveUri == null || rawArchiveUri.isBlank()) {
return "";

10
modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
View File

@@ -81,13 +81,15 @@ public class EventHistoryAutoConfiguration {
}
@Bean
@ConditionalOnBean({EventFileStore.class, Gb32960MessageDecoder.class})
@ConditionalOnBean(Gb32960MessageDecoder.class)
@ConditionalOnMissingBean
public Gb32960DecodedFrameService gb32960DecodedFrameService(EventFileStore store,
public Gb32960DecodedFrameService gb32960DecodedFrameService(ObjectProvider<EventFileStore> store,
ObjectProvider<TdengineHistoryReader> reader,
Gb32960MessageDecoder decoder,
SinkArchiveProperties archiveProperties) {
// snapshot/frame 查询通过 EventFileStore 找到 rawArchiveUri，再从 archiveRoot 读取 .bin 即时解码。
return new Gb32960DecodedFrameService(store, decoder, archiveRoot(archiveProperties.getPath()), null);
// 优先使用 EventFileStore 索引；TDengine-only 高吞吐运行时可直接从 raw_frames 找 rawUri。
return new Gb32960DecodedFrameService(store.getIfAvailable(), reader.getIfAvailable(),
decoder, archiveRoot(archiveProperties.getPath()), null);
}
@Bean

98
modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java
View File

@@ -14,6 +14,14 @@ import com.lingniu.ingest.protocol.gb32960.codec.parser.vendor.guangdong.GdFcSta
import com.lingniu.ingest.protocol.gb32960.codec.profile.Gb32960ProfileRegistry;
import com.lingniu.ingest.protocol.gb32960.codec.profile.VendorExtensionCatalog;
import com.lingniu.ingest.protocol.gb32960.codec.profile.VendorExtensionSelector;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery;
import com.lingniu.ingest.tdenginehistory.TdengineLocationRow;
import com.lingniu.ingest.tdenginehistory.TdenginePage;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -113,6 +121,24 @@ class Gb32960DecodedFrameServiceTest {
.containsExactly("VEHICLE", "GD_FC_DCDC");
}
@Test
void findsSingleFrameDirectlyByRawArchiveUriWhenEventFileStoreIsDisabled() throws Exception {
String key = "2026/06/22/GB32960/LNVFC000000000001/1792527837000000.bin";
writeArchive(key, realtimeFrame());
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
(EventFileStore) null,
decoder(),
archiveRoot,
null);
Gb32960DecodedFrameService.DecodedFrame frame =
service.findByRawArchiveUri("archive://" + key, "Hyundai");
assertThat(frame.eventId()).isEqualTo("1792527837000000");
assertThat(frame.vin()).isEqualTo("LNVFC000000000001");
}
@Test
void findsSingleFrameByRawArchiveUriIndexWhenAvailable() throws Exception {
String key = "2026/06/22/GB32960/LNVFC000000000001/1792527837000001.bin";
@@ -266,6 +292,53 @@ class Gb32960DecodedFrameServiceTest {
}
}
@Test
void snapshotsCanUseTdengineRawFramesWhenEventFileStoreIsDisabled() throws Exception {
String vin = "LNVFC000000000001";
String key = "2026/06/22/GB32960/" + vin + "/tdengine-raw.bin";
String uri = "archive://" + key;
writeArchive(key, realtimeFrame());
CapturingTdengineReader reader = new CapturingTdengineReader(List.of(new TdengineRawFrameRow(
Instant.parse("2026-06-22T10:00:00Z"),
"tdengine-raw",
Instant.parse("2026-06-22T10:00:01Z"),
0x02,
0,
Instant.parse("2026-06-22T10:00:00Z"),
uri,
"sha256:test",
realtimeFrame().length,
"SUCCEEDED",
"",
"127.0.0.1:32960",
"{\"platformAccount\":\"Hyundai\"}",
"GB32960",
vin,
vin,
"")));
Gb32960DecodedFrameService service = new Gb32960DecodedFrameService(
reader,
decoder(),
archiveRoot,
null);
List<Gb32960DecodedFrameService.LogicalSnapshot> snapshots = service.snapshots(
QueryTimeRange.parse("2026-06-22T18:00:00", "2026-06-22T18:00:00"),
EventFileQuery.Order.DESC,
10,
vin,
null);
assertThat(reader.lastQuery.protocol()).isEqualTo("GB32960");
assertThat(reader.lastQuery.vehicleKey()).isEqualTo(vin);
assertThat(snapshots).hasSize(1);
assertThat(snapshots.getFirst().sourceFrames()).extracting(Gb32960DecodedFrameService.SourceFrame::rawArchiveUri)
.containsExactly(uri);
assertThat(snapshots.getFirst().blocks()).extracting(block -> block.get("type"))
.containsExactly("VEHICLE", "GD_FC_DCDC");
}
@Test
void snapshotsKeepMissingStackCellsAsNullsUpToCellCount() throws Exception {
String vin = "LNVFC000000000001";
@@ -561,4 +634,29 @@ class Gb32960DecodedFrameServiceTest {
.orElse(null);
}
}
private static final class CapturingTdengineReader implements TdengineHistoryReader {
private final List<TdengineRawFrameRow> rows;
private TdengineRawFrameQuery lastQuery;
private CapturingTdengineReader(List<TdengineRawFrameRow> rows) {
this.rows = rows;
}
@Override
public TdenginePage<TdengineRawFrameRow> queryRawFrames(TdengineRawFrameQuery query) {
lastQuery = query;
return new TdenginePage<>(rows, java.util.Optional.empty());
}
@Override
public TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) {
throw new UnsupportedOperationException();
}
@Override
public TdenginePage<TdengineTelemetryFieldRow> queryTelemetryFields(TdengineTelemetryFieldQuery query) {
throw new UnsupportedOperationException();
}
}
}

17
modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
View File

@@ -66,6 +66,23 @@ class EventHistoryAutoConfigurationTest {
});
}
@Test
void createsGb32960FrameBeansWithTdengineReaderWhenEventFileStoreIsDisabled() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withPropertyValues(
"lingniu.ingest.event-history.enabled=true",
"lingniu.ingest.sink.archive.path=/tmp/lingniu-test-archive")
.withBean(Gb32960MessageDecoder.class, () -> mock(Gb32960MessageDecoder.class))
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
.withBean(SinkArchiveProperties.class, SinkArchiveProperties::new)
.run(context -> {
assertThat(context).doesNotHaveBean(EventFileStore.class);
assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class);
assertThat(context).hasSingleBean(Gb32960FrameController.class);
});
}
@Test
void createsJt808LocationHistoryControllerWhenTdengineReaderExists() {
contextRunner