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

feat: add tdengine history store foundation

Browse Source
This commit is contained in:
lingniu
2026-06-29 13:25:31 +08:00
parent 5a60b8c43c
commit f58fe51cf5
21 changed files with 762 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineBatchStatement.java Normal file
View File

@@ -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<Object> values
) {
public TdengineBatchStatement {
values = values == null ? List.of() : Collections.unmodifiableList(new ArrayList<>(values));
}
}

111
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineEnvelopeRows.java Normal file
View File

@@ -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<TdengineRawFrameRow> rawFrame(VehicleEnvelope envelope) {
if (envelope == null || !envelope.hasRawFrameFact()) {
return Optional.empty();
}
RawFrameFactPayload raw = envelope.getRawFrameFact();
Map<String, String> 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<TdengineLocationRow> 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<String, String> 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 "";
}
}

81
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistorySchema.java Normal file
View File

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

64
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryStatements.java Normal file
View File

@@ -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<Object> values(Object... values) {
return Arrays.asList(values);
}
}

19
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineHistoryWriter.java Normal file
View File

@@ -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<TdengineRawFrameRow> rows) throws IOException;
default void appendLocation(TdengineLocationRow row) throws IOException {
appendLocations(List.of(row));
}
void appendLocations(List<TdengineLocationRow> rows) throws IOException;
}

42
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineIdentifier.java Normal file
View File

@@ -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);
}
}
}

25
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineLocationRow.java Normal file
View File

@@ -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
) {
}

24
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineRawFrameRow.java Normal file
View File

@@ -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
) {
}

23
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java Normal file
View File

@@ -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());
}
}

33
modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryProperties.java Normal file
View File

@@ -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;
}
}

1
modules/sinks/tdengine-history-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
com.lingniu.ingest.tdenginehistory.config.TdengineHistoryAutoConfiguration