diff --git a/modules/apps/vehicle-history-app/pom.xml b/modules/apps/vehicle-history-app/pom.xml index 2316864c..67598da8 100644 --- a/modules/apps/vehicle-history-app/pom.xml +++ b/modules/apps/vehicle-history-app/pom.xml @@ -33,6 +33,10 @@ com.lingniu.ingest event-file-store + + com.lingniu.ingest + tdengine-history-store + com.lingniu.ingest event-history-service diff --git a/modules/apps/vehicle-history-app/src/main/resources/application.yml b/modules/apps/vehicle-history-app/src/main/resources/application.yml index aef384ca..7f38d578 100644 --- a/modules/apps/vehicle-history-app/src/main/resources/application.yml +++ b/modules/apps/vehicle-history-app/src/main/resources/application.yml @@ -66,6 +66,9 @@ lingniu: zone-id: ${EVENT_FILE_STORE_ZONE_ID:Asia/Shanghai} batch-size: ${EVENT_FILE_STORE_BATCH_SIZE:500} flush-interval-millis: ${EVENT_FILE_STORE_FLUSH_INTERVAL_MILLIS:1000} + tdengine-history: + enabled: ${TDENGINE_HISTORY_ENABLED:false} + database: ${TDENGINE_HISTORY_DATABASE:vehicle_history} event-history: enabled: true vehicle-state: diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java index e05ab364..0c54f2df 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java @@ -17,6 +17,8 @@ import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration; import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink; import com.lingniu.ingest.sink.mq.KafkaEventSink; import com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration; +import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema; +import com.lingniu.ingest.tdenginehistory.config.TdengineHistoryAutoConfiguration; import org.apache.kafka.clients.producer.KafkaProducer; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -39,6 +41,7 @@ class VehicleHistoryAppCompositionTest { .withConfiguration(AutoConfigurations.of( SinkArchiveAutoConfiguration.class, EventFileStoreAutoConfiguration.class, + TdengineHistoryAutoConfiguration.class, SinkMqAutoConfiguration.class, Gb32960AutoConfiguration.class, EventHistoryAutoConfiguration.class)) @@ -50,6 +53,8 @@ class VehicleHistoryAppCompositionTest { "lingniu.ingest.sink.archive.path=" + tempDir.resolve("archive"), "lingniu.ingest.event-file-store.enabled=true", "lingniu.ingest.event-file-store.path=" + tempDir.resolve("event-store"), + "lingniu.ingest.tdengine-history.enabled=true", + "lingniu.ingest.tdengine-history.database=vehicle_history_test", "lingniu.ingest.event-history.enabled=true", "lingniu.ingest.gb32960.enabled=true", "lingniu.ingest.gb32960.server.enabled=false", @@ -64,6 +69,7 @@ class VehicleHistoryAppCompositionTest { assertThat(context).hasSingleBean(RawArchiveEventSink.class); assertThat(context).hasSingleBean(EventFileStore.class); assertThat(context).hasSingleBean(EventFileStoreSink.class); + assertThat(context).hasSingleBean(TdengineHistorySchema.class); assertThat(context).hasSingleBean(KafkaEventSink.class); assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class); diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java index 9a7cafe3..38c47d24 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java @@ -24,6 +24,8 @@ class VehicleHistoryAppDefaultsTest { .containsEntry("lingniu.ingest.gb32960.server.enabled", false) .containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}") .containsEntry("lingniu.ingest.event-file-store.enabled", "${EVENT_FILE_STORE_ENABLED:true}") + .containsEntry("lingniu.ingest.tdengine-history.enabled", "${TDENGINE_HISTORY_ENABLED:false}") + .containsEntry("lingniu.ingest.tdengine-history.database", "${TDENGINE_HISTORY_DATABASE:vehicle_history}") .containsEntry("lingniu.ingest.event-history.enabled", true) .containsEntry("lingniu.ingest.vehicle-state.enabled", false) .containsEntry("lingniu.ingest.vehicle-stat.enabled", false) diff --git a/modules/sinks/tdengine-history-store/pom.xml b/modules/sinks/tdengine-history-store/pom.xml new file mode 100644 index 00000000..1612dcb0 --- /dev/null +++ b/modules/sinks/tdengine-history-store/pom.xml @@ -0,0 +1,52 @@ + + + 4.0.0 + + com.lingniu.ingest + lingniu-vehicle-ingest + 0.1.0-SNAPSHOT + ../../../pom.xml + + tdengine-history-store + tdengine-history-store + TDengine-backed hot history store schema and Kafka envelope mapping. + + + + com.lingniu.ingest + sink-mq + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-autoconfigure + + + org.springframework.boot + spring-boot-configuration-processor + true + + + com.fasterxml.jackson.core + jackson-databind + + + org.junit.jupiter + junit-jupiter + test + + + org.assertj + assertj-core + test + + + org.springframework.boot + spring-boot-starter-test + test + + + diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineBatchStatement.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineBatchStatement.java new file mode 100644 index 00000000..d634ae4d --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineBatchStatement.java @@ -0,0 +1,15 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.util.List; +import java.util.ArrayList; +import java.util.Collections; + +public record TdengineBatchStatement( + String createChildTableSql, + String insertSql, + List values +) { + public TdengineBatchStatement { + values = values == null ? List.of() : Collections.unmodifiableList(new ArrayList<>(values)); + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java new file mode 100644 index 00000000..8d38b9d9 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java @@ -0,0 +1,111 @@ +package com.lingniu.ingest.tdenginehistory; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.lingniu.ingest.sink.mq.proto.LocationPayload; +import com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload; +import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; + +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; + +public final class TdengineEnvelopeRows { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private TdengineEnvelopeRows() { + } + + public static Optional rawFrame(VehicleEnvelope envelope) { + if (envelope == null || !envelope.hasRawFrameFact()) { + return Optional.empty(); + } + RawFrameFactPayload raw = envelope.getRawFrameFact(); + Map metadata = new LinkedHashMap<>(); + metadata.putAll(envelope.getMetadataMap()); + metadata.putAll(raw.getMetadataMap()); + return Optional.of(new TdengineRawFrameRow( + instant(envelope.getEventTimeMs()), + raw.getFrameId(), + instant(envelope.getIngestTimeMs()), + raw.getMessageId(), + raw.getSubType(), + instant(envelope.getEventTimeMs()), + raw.getRawUri(), + raw.getChecksum(), + raw.getRawSizeBytes(), + raw.getParseStatus().name().replace("PARSE_STATUS_", ""), + raw.getParseError(), + raw.getPeer(), + json(metadata), + protocol(envelope), + firstNonBlank(raw.getVehicleKey(), envelope.getMetadataOrDefault("vehicle_key", ""), envelope.getVin()), + firstNonBlank(raw.getVin(), envelope.getVin()), + raw.getPhone() + )); + } + + public static Optional location(VehicleEnvelope envelope) { + if (envelope == null || !envelope.hasLocation()) { + return Optional.empty(); + } + LocationPayload location = envelope.getLocation(); + String rawUri = envelope.hasRawArchive() ? envelope.getRawArchive().getUri() : ""; + String frameId = envelope.getMetadataOrDefault("frame_id", envelope.getEventId()); + return Optional.of(new TdengineLocationRow( + instant(envelope.getEventTimeMs()), + envelope.getEventId(), + frameId, + instant(envelope.getIngestTimeMs()), + location.getLongitude(), + location.getLatitude(), + location.getAltitudeM(), + location.getSpeedKmh(), + location.getDirectionDeg(), + location.getAlarmFlag(), + location.getStatusFlag(), + totalMileage(envelope), + rawUri, + json(envelope.getMetadataMap()), + protocol(envelope), + firstNonBlank(envelope.getMetadataOrDefault("vehicle_key", ""), envelope.getVin()), + envelope.getVin(), + envelope.getMetadataOrDefault("phone", "") + )); + } + + private static Instant instant(long epochMillis) { + return Instant.ofEpochMilli(epochMillis); + } + + private static String protocol(VehicleEnvelope envelope) { + return firstNonBlank(envelope.getSource(), "UNKNOWN"); + } + + private static Double totalMileage(VehicleEnvelope envelope) { + String value = envelope.getMetadataOrDefault("total_mileage_km", ""); + if (value.isBlank()) { + return null; + } + return Double.parseDouble(value); + } + + private static String json(Map metadata) { + try { + return OBJECT_MAPPER.writeValueAsString(metadata == null ? Map.of() : metadata); + } catch (JsonProcessingException e) { + throw new IllegalArgumentException("metadata cannot be serialized as json", e); + } + } + + private static String firstNonBlank(String... values) { + for (String value : values) { + if (value != null && !value.isBlank()) { + return value; + } + } + return ""; + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java new file mode 100644 index 00000000..9887b0fe --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java @@ -0,0 +1,81 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.util.List; + +/** + * SQL model for the TDengine hot history store. + */ +public final class TdengineHistorySchema { + + private final String database; + + public TdengineHistorySchema(String database) { + this.database = TdengineIdentifier.database(database); + } + + public List bootstrapSql() { + return List.of( + "CREATE DATABASE IF NOT EXISTS " + database + " PRECISION 'ms'", + "USE " + database, + rawFramesStableSql(), + vehicleLocationsStableSql() + ); + } + + public String rawFrameTable(String protocol, String vehicleKey) { + return "raw_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey); + } + + public String locationTable(String protocol, String vehicleKey) { + return "loc_" + TdengineIdentifier.fragment(protocol) + "_" + TdengineIdentifier.hash16(vehicleKey); + } + + private static String rawFramesStableSql() { + return """ + CREATE STABLE IF NOT EXISTS raw_frames ( + ts TIMESTAMP, + frame_id NCHAR(64), + received_at TIMESTAMP, + message_id INT, + sub_type INT, + event_time TIMESTAMP, + raw_uri NCHAR(512), + checksum NCHAR(128), + raw_size_bytes BIGINT, + parse_status NCHAR(16), + parse_error NCHAR(512), + peer NCHAR(128), + metadata_json NCHAR(4096) + ) TAGS ( + protocol NCHAR(16), + vehicle_key NCHAR(128), + vin NCHAR(64), + phone NCHAR(32) + )"""; + } + + private static String vehicleLocationsStableSql() { + return """ + CREATE STABLE IF NOT EXISTS vehicle_locations ( + ts TIMESTAMP, + fact_id NCHAR(64), + frame_id NCHAR(64), + received_at TIMESTAMP, + longitude DOUBLE, + latitude DOUBLE, + altitude_m DOUBLE, + speed_kmh DOUBLE, + direction_deg DOUBLE, + alarm_flag BIGINT, + status_flag BIGINT, + total_mileage_km DOUBLE, + raw_uri NCHAR(512), + metadata_json NCHAR(4096) + ) TAGS ( + protocol NCHAR(16), + vehicle_key NCHAR(128), + vin NCHAR(64), + phone NCHAR(32) + )"""; + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java new file mode 100644 index 00000000..e0220666 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java @@ -0,0 +1,64 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.util.List; +import java.util.Arrays; + +public final class TdengineHistoryStatements { + + private static final String RAW_FRAME_COLUMNS = "ts, frame_id, received_at, message_id, sub_type, event_time, " + + "raw_uri, checksum, raw_size_bytes, parse_status, parse_error, peer, metadata_json"; + private static final String RAW_FRAME_PLACEHOLDERS = "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?"; + + private static final String LOCATION_COLUMNS = "ts, fact_id, frame_id, received_at, longitude, latitude, " + + "altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, raw_uri, metadata_json"; + private static final String LOCATION_PLACEHOLDERS = "?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?"; + + private final TdengineHistorySchema schema; + + public TdengineHistoryStatements(TdengineHistorySchema schema) { + if (schema == null) { + throw new IllegalArgumentException("schema must not be null"); + } + this.schema = schema; + } + + public TdengineBatchStatement rawFrame(TdengineRawFrameRow row) { + String table = schema.rawFrameTable(row.protocol(), row.vehicleKey()); + return new TdengineBatchStatement( + "CREATE TABLE IF NOT EXISTS " + table + " USING raw_frames TAGS (" + + tags(row.protocol(), row.vehicleKey(), row.vin(), row.phone()) + ")", + "INSERT INTO " + table + " (" + RAW_FRAME_COLUMNS + ") VALUES (" + RAW_FRAME_PLACEHOLDERS + ")", + values( + row.ts(), row.frameId(), row.receivedAt(), row.messageId(), row.subType(), + row.eventTime(), row.rawUri(), row.checksum(), row.rawSizeBytes(), row.parseStatus(), + row.parseError(), row.peer(), row.metadataJson() + ) + ); + } + + public TdengineBatchStatement location(TdengineLocationRow row) { + String table = schema.locationTable(row.protocol(), row.vehicleKey()); + return new TdengineBatchStatement( + "CREATE TABLE IF NOT EXISTS " + table + " USING vehicle_locations TAGS (" + + tags(row.protocol(), row.vehicleKey(), row.vin(), row.phone()) + ")", + "INSERT INTO " + table + " (" + LOCATION_COLUMNS + ") VALUES (" + LOCATION_PLACEHOLDERS + ")", + values( + row.ts(), row.factId(), row.frameId(), row.receivedAt(), row.longitude(), row.latitude(), + row.altitudeM(), row.speedKmh(), row.directionDeg(), row.alarmFlag(), row.statusFlag(), + row.totalMileageKm(), row.rawUri(), row.metadataJson() + ) + ); + } + + private static String tags(String protocol, String vehicleKey, String vin, String phone) { + return quote(protocol) + ", " + quote(vehicleKey) + ", " + quote(vin) + ", " + quote(phone); + } + + private static String quote(String value) { + return "'" + (value == null ? "" : value.replace("'", "''")) + "'"; + } + + private static List values(Object... values) { + return Arrays.asList(values); + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java new file mode 100644 index 00000000..e8d51a7c --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java @@ -0,0 +1,19 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.io.IOException; +import java.util.List; + +public interface TdengineHistoryWriter { + + default void appendRawFrame(TdengineRawFrameRow row) throws IOException { + appendRawFrames(List.of(row)); + } + + void appendRawFrames(List rows) throws IOException; + + default void appendLocation(TdengineLocationRow row) throws IOException { + appendLocations(List.of(row)); + } + + void appendLocations(List rows) throws IOException; +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineIdentifier.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineIdentifier.java new file mode 100644 index 00000000..5f1d730a --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineIdentifier.java @@ -0,0 +1,42 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; +import java.util.Locale; + +final class TdengineIdentifier { + + private TdengineIdentifier() { + } + + static String database(String value) { + String fragment = fragment(value); + if (fragment.isBlank()) { + throw new IllegalArgumentException("database must contain identifier characters"); + } + return fragment; + } + + static String fragment(String value) { + if (value == null) { + return "unknown"; + } + String normalized = value.trim().toLowerCase(Locale.ROOT) + .replaceAll("[^a-z0-9_]+", "_") + .replaceAll("_+", "_") + .replaceAll("^_|_$", ""); + return normalized.isBlank() ? "unknown" : normalized; + } + + static String hash16(String value) { + try { + byte[] digest = MessageDigest.getInstance("SHA-256") + .digest(String.valueOf(value).getBytes(StandardCharsets.UTF_8)); + return HexFormat.of().formatHex(digest, 0, 8); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 is not available", e); + } + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineLocationRow.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineLocationRow.java new file mode 100644 index 00000000..8a3d80cb --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineLocationRow.java @@ -0,0 +1,25 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.time.Instant; + +public record TdengineLocationRow( + Instant ts, + String factId, + String frameId, + Instant receivedAt, + double longitude, + double latitude, + double altitudeM, + double speedKmh, + double directionDeg, + long alarmFlag, + long statusFlag, + Double totalMileageKm, + String rawUri, + String metadataJson, + String protocol, + String vehicleKey, + String vin, + String phone +) { +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameRow.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameRow.java new file mode 100644 index 00000000..a9a473bd --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameRow.java @@ -0,0 +1,24 @@ +package com.lingniu.ingest.tdenginehistory; + +import java.time.Instant; + +public record TdengineRawFrameRow( + Instant ts, + String frameId, + Instant receivedAt, + int messageId, + int subType, + Instant eventTime, + String rawUri, + String checksum, + long rawSizeBytes, + String parseStatus, + String parseError, + String peer, + String metadataJson, + String protocol, + String vehicleKey, + String vin, + String phone +) { +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java new file mode 100644 index 00000000..4a7fc489 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java @@ -0,0 +1,23 @@ +package com.lingniu.ingest.tdenginehistory.config; + +import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema; +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; + +@AutoConfiguration +@EnableConfigurationProperties(TdengineHistoryProperties.class) +public class TdengineHistoryAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty( + prefix = "lingniu.ingest.tdengine-history", + name = "enabled", + havingValue = "true") + public TdengineHistorySchema tdengineHistorySchema(TdengineHistoryProperties properties) { + return new TdengineHistorySchema(properties.getDatabase()); + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryProperties.java b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryProperties.java new file mode 100644 index 00000000..03009ee2 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryProperties.java @@ -0,0 +1,33 @@ +package com.lingniu.ingest.tdenginehistory.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "lingniu.ingest.tdengine-history") +public class TdengineHistoryProperties { + + /** + * 默认关闭，避免未配置 TDengine 连接时影响现有历史查询服务启动。 + */ + private boolean enabled = false; + + /** + * TDengine 历史库名。 + */ + private String database = "vehicle_history"; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } +} diff --git a/modules/sinks/tdengine-history-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/modules/sinks/tdengine-history-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 00000000..2bd97e31 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +com.lingniu.ingest.tdenginehistory.config.TdengineHistoryAutoConfiguration diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java new file mode 100644 index 00000000..84a39e0f --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRowsTest.java @@ -0,0 +1,89 @@ +package com.lingniu.ingest.tdenginehistory; + +import com.lingniu.ingest.sink.mq.proto.LocationPayload; +import com.lingniu.ingest.sink.mq.proto.ParseStatusProto; +import com.lingniu.ingest.sink.mq.proto.RawArchiveRef; +import com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload; +import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; +import org.junit.jupiter.api.Test; + +import java.time.Instant; + +import static org.assertj.core.api.Assertions.assertThat; + +class TdengineEnvelopeRowsTest { + + @Test + void mapsRawFrameFactEnvelopeToRawFrameRow() { + VehicleEnvelope envelope = VehicleEnvelope.newBuilder() + .setEventId("evt-raw-1") + .setVin("VIN123") + .setSource("JT808") + .setEventTimeMs(1_772_000_001_234L) + .setIngestTimeMs(1_772_000_001_999L) + .putMetadata("channel", "tcp-808") + .setRawFrameFact(RawFrameFactPayload.newBuilder() + .setFrameId("frame-1") + .setVehicleKey("jt808:g7gps") + .setPhone("013800000000") + .setMessageId(0x0200) + .setSubType(0) + .setRawUri("archive://jt808/2026/06/29/frame-1.bin") + .setChecksum("sha256:abc") + .setRawSizeBytes(67) + .setParseStatus(ParseStatusProto.PARSE_STATUS_SUCCEEDED) + .setPeer("10.0.0.8:30001") + .putMetadata("auth", "passed")) + .build(); + + TdengineRawFrameRow row = TdengineEnvelopeRows.rawFrame(envelope).orElseThrow(); + + assertThat(row.ts()).isEqualTo(Instant.ofEpochMilli(1_772_000_001_234L)); + assertThat(row.receivedAt()).isEqualTo(Instant.ofEpochMilli(1_772_000_001_999L)); + assertThat(row.protocol()).isEqualTo("JT808"); + assertThat(row.vehicleKey()).isEqualTo("jt808:g7gps"); + assertThat(row.vin()).isEqualTo("VIN123"); + assertThat(row.phone()).isEqualTo("013800000000"); + assertThat(row.messageId()).isEqualTo(0x0200); + assertThat(row.rawUri()).isEqualTo("archive://jt808/2026/06/29/frame-1.bin"); + assertThat(row.metadataJson()).contains("\"channel\":\"tcp-808\"", "\"auth\":\"passed\""); + } + + @Test + void mapsLocationEnvelopeToVehicleLocationRow() { + VehicleEnvelope envelope = VehicleEnvelope.newBuilder() + .setEventId("evt-location-1") + .setVin("VIN32960") + .setSource("GB32960") + .setEventTimeMs(1_772_000_002_000L) + .setIngestTimeMs(1_772_000_002_321L) + .putMetadata("vehicle_key", "vin:VIN32960") + .putMetadata("frame_id", "frame-location-1") + .setRawArchive(RawArchiveRef.newBuilder() + .setUri("archive://gb32960/2026/06/29/frame-location-1.bin") + .setChecksum("sha256:def") + .setSizeBytes(98)) + .setLocation(LocationPayload.newBuilder() + .setLongitude(113.12) + .setLatitude(23.45) + .setAltitudeM(8.0) + .setSpeedKmh(42.5) + .setDirectionDeg(91.0) + .setAlarmFlag(1) + .setStatusFlag(3)) + .build(); + + TdengineLocationRow row = TdengineEnvelopeRows.location(envelope).orElseThrow(); + + assertThat(row.ts()).isEqualTo(Instant.ofEpochMilli(1_772_000_002_000L)); + assertThat(row.factId()).isEqualTo("evt-location-1"); + assertThat(row.frameId()).isEqualTo("frame-location-1"); + assertThat(row.protocol()).isEqualTo("GB32960"); + assertThat(row.vehicleKey()).isEqualTo("vin:VIN32960"); + assertThat(row.longitude()).isEqualTo(113.12); + assertThat(row.latitude()).isEqualTo(23.45); + assertThat(row.speedKmh()).isEqualTo(42.5); + assertThat(row.totalMileageKm()).isNull(); + assertThat(row.rawUri()).isEqualTo("archive://gb32960/2026/06/29/frame-location-1.bin"); + } +} diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchemaTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchemaTest.java new file mode 100644 index 00000000..31945341 --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchemaTest.java @@ -0,0 +1,43 @@ +package com.lingniu.ingest.tdenginehistory; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class TdengineHistorySchemaTest { + + @Test + void bootstrapSqlCreatesRawAndLocationStablesFromSpec() { + TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history"); + + assertThat(schema.bootstrapSql()) + .first() + .isEqualTo("CREATE DATABASE IF NOT EXISTS vehicle_history PRECISION 'ms'"); + assertThat(schema.bootstrapSql()) + .contains("USE vehicle_history"); + assertThat(schema.bootstrapSql().get(2)) + .contains("CREATE STABLE IF NOT EXISTS raw_frames") + .contains("frame_id NCHAR(64)") + .contains("raw_uri NCHAR(512)") + .contains("metadata_json NCHAR(4096)") + .contains("TAGS (") + .contains("protocol NCHAR(16)") + .contains("vehicle_key NCHAR(128)") + .contains("phone NCHAR(32)"); + assertThat(schema.bootstrapSql().get(3)) + .contains("CREATE STABLE IF NOT EXISTS vehicle_locations") + .contains("longitude DOUBLE") + .contains("total_mileage_km DOUBLE") + .contains("raw_uri NCHAR(512)"); + } + + @Test + void childTableNamesAreStableAndSanitized() { + TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history"); + + assertThat(schema.rawFrameTable("JT808", "jt808:g7gps/013800000000")) + .matches("raw_jt808_[0-9a-f]{16}"); + assertThat(schema.locationTable("GB32960", "VIN WITH SPACE")) + .matches("loc_gb32960_[0-9a-f]{16}"); + } +} diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatementsTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatementsTest.java new file mode 100644 index 00000000..a86fa5be --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatementsTest.java @@ -0,0 +1,84 @@ +package com.lingniu.ingest.tdenginehistory; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; + +import static org.assertj.core.api.Assertions.assertThat; + +class TdengineHistoryStatementsTest { + + private final TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history"); + private final TdengineHistoryStatements statements = new TdengineHistoryStatements(schema); + + @Test + void rawFrameBatchStatementCreatesChildTableAndParameterizedInsert() { + TdengineRawFrameRow row = new TdengineRawFrameRow( + Instant.parse("2026-06-29T05:00:01Z"), + "frame-1", + Instant.parse("2026-06-29T05:00:02Z"), + 0x0200, + 0, + Instant.parse("2026-06-29T05:00:01Z"), + "archive://jt808/frame-1.bin", + "sha256:abc", + 67, + "SUCCEEDED", + "", + "10.0.0.1:808", + "{\"auth\":\"passed\"}", + "JT808", + "jt808:g7gps", + "VIN123", + "013800000000"); + + TdengineBatchStatement batch = statements.rawFrame(row); + + assertThat(batch.createChildTableSql()) + .startsWith("CREATE TABLE IF NOT EXISTS raw_jt808_") + .contains(" USING raw_frames TAGS ('JT808', 'jt808:g7gps', 'VIN123', '013800000000')"); + assertThat(batch.insertSql()) + .startsWith("INSERT INTO raw_jt808_") + .contains("(ts, frame_id, received_at, message_id, sub_type, event_time, raw_uri, checksum, raw_size_bytes, parse_status, parse_error, peer, metadata_json)") + .endsWith("VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"); + assertThat(batch.values()) + .containsExactly( + row.ts(), row.frameId(), row.receivedAt(), row.messageId(), row.subType(), + row.eventTime(), row.rawUri(), row.checksum(), row.rawSizeBytes(), row.parseStatus(), + row.parseError(), row.peer(), row.metadataJson()); + } + + @Test + void locationBatchStatementEscapesTags() { + TdengineLocationRow row = new TdengineLocationRow( + Instant.parse("2026-06-29T05:00:01Z"), + "fact-1", + "frame-1", + Instant.parse("2026-06-29T05:00:02Z"), + 113.1, + 23.4, + 8.0, + 42.0, + 90.0, + 1, + 3, + null, + "archive://gb32960/frame-1.bin", + "{}", + "GB32960", + "vin:O'HARE", + "O'HARE", + ""); + + TdengineBatchStatement batch = statements.location(row); + + assertThat(batch.createChildTableSql()) + .startsWith("CREATE TABLE IF NOT EXISTS loc_gb32960_") + .contains(" USING vehicle_locations TAGS ('GB32960', 'vin:O''HARE', 'O''HARE', '')"); + assertThat(batch.values()) + .containsExactly( + row.ts(), row.factId(), row.frameId(), row.receivedAt(), row.longitude(), row.latitude(), + row.altitudeM(), row.speedKmh(), row.directionDeg(), row.alarmFlag(), row.statusFlag(), + row.totalMileageKm(), row.rawUri(), row.metadataJson()); + } +} diff --git a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java new file mode 100644 index 00000000..665abdde --- /dev/null +++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java @@ -0,0 +1,35 @@ +package com.lingniu.ingest.tdenginehistory.config; + +import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema; +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +class TdengineHistoryAutoConfigurationTest { + + private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(TdengineHistoryAutoConfiguration.class)); + + @Test + void staysOffByDefault() { + contextRunner.run(context -> { + assertThat(context).doesNotHaveBean(TdengineHistorySchema.class); + assertThat(context.getBean(TdengineHistoryProperties.class).isEnabled()).isFalse(); + }); + } + + @Test + void createsSchemaWhenEnabled() { + contextRunner + .withPropertyValues( + "lingniu.ingest.tdengine-history.enabled=true", + "lingniu.ingest.tdengine-history.database=vehicle_history_hot") + .run(context -> { + assertThat(context).hasSingleBean(TdengineHistorySchema.class); + assertThat(context.getBean(TdengineHistorySchema.class).bootstrapSql().getFirst()) + .isEqualTo("CREATE DATABASE IF NOT EXISTS vehicle_history_hot PRECISION 'ms'"); + }); + } +} diff --git a/pom.xml b/pom.xml index 6dccb62b..3e20a6fa 100644 --- a/pom.xml +++ b/pom.xml @@ -28,6 +28,7 @@ modules/sinks/sink-archive modules/sinks/raw-archive-store modules/sinks/event-file-store + modules/sinks/tdengine-history-store modules/services/event-history-service modules/services/vehicle-state-service modules/services/vehicle-stat-service @@ -195,6 +196,11 @@ event-file-store ${project.version} + + com.lingniu.ingest + tdengine-history-store + ${project.version} + com.lingniu.ingest event-history-service