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

refactor: remove telemetry field api from history app

Browse Source
This commit is contained in:
lingniu
2026-07-01 13:53:19 +08:00
parent a5cb0c9705
commit 68f45ee79c
18 changed files with 54 additions and 523 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
deploy/local/launchctl/README.md
View File

@@ -13,9 +13,8 @@
`vehicle-history-app` 的生产高吞吐模板是 TDengine-only：
- 不暴露 DuckDB 或文件型明细库启动变量，避免本地文件索引进入实时消费热路径。
- `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`：默认只写 `raw_frames` 和协议位置表，避免逐字段写放大。
- 通用 RAW 查询默认直接返回 TDengine `raw_frames` 中的 `parsedJson`/`parsedFields`；后续解析器新增字段后，优先通过 replay 或专用解析任务补写 TDengine，而不是让 history 服务依赖本地 archive 挂载。
- 如果需要逐字段趋势宽表，可以显式打开 `TDENGINE_TELEMETRY_FIELDS_ENABLED`；开启后要重新压测 Kafka lag 和 TDengine 写入吞吐。
- 只写 `raw_frames` 和协议位置表，避免逐字段写放大。
- 通用 RAW 查询直接返回 TDengine `raw_frames` 中的 `parsedJson`/`parsedFields`；字段趋势宽表由独立字段解析服务消费 Kafka RAW/事件后维护。
## 生成 plist

2
deploy/local/launchctl/com.lingniu.vehicle-history.plist.template
View File

@@ -60,8 +60,6 @@
<string>16</string>
<key>TDENGINE_INITIALIZE_SCHEMA</key>
<string>true</string>
<key>TDENGINE_TELEMETRY_FIELDS_ENABLED</key>
<string>false</string>
<key>NACOS_CONFIG_ENABLED</key>
<string>false</string>
<key>MANAGEMENT_HEALTH_REDIS_ENABLED</key>

1
deploy/portainer/docker-compose.yml
View File

@@ -175,7 +175,6 @@ services:
TDENGINE_CONNECTION_TIMEOUT_MILLIS: ${TDENGINE_CONNECTION_TIMEOUT_MILLIS:-30000}
TDENGINE_MIN_IDLE: ${TDENGINE_MIN_IDLE:-1}
TDENGINE_MAX_POOL_SIZE: ${TDENGINE_MAX_POOL_SIZE:-32}
TDENGINE_TELEMETRY_FIELDS_ENABLED: ${TDENGINE_TELEMETRY_FIELDS_ENABLED:-false}
ports:
- "${VEHICLE_HISTORY_HTTP_PORT:-20200}:20200"
networks:

1
docs/operations/gb32960-service-split-runbook.md
View File

@@ -145,7 +145,6 @@ TDENGINE_HISTORY_DATABASE=vehicle_ts \
TDENGINE_JDBC_URL='jdbc:TAOS-WS://<tdengine-host>:6041/vehicle_ts' \
TDENGINE_USERNAME=root \
TDENGINE_PASSWORD='<tdengine-password>' \
TDENGINE_TELEMETRY_FIELDS_ENABLED=false \
java --sun-misc-unsafe-memory-access=allow \
-jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar
```

31
docs/operations/vehicle-ingest-tdengine-verification.md
View File

@@ -24,12 +24,11 @@ export TDENGINE_USERNAME=root
export TDENGINE_PASSWORD='<tdengine-password>'
export TDENGINE_MIN_IDLE=0
export TDENGINE_MAX_POOL_SIZE=32
export TDENGINE_TELEMETRY_FIELDS_ENABLED=false
```
`TDENGINE_MIN_IDLE=0` 用于减少冷启动时 TDengine 连接重试日志。TDengine 地址稳定后，再按生产并发调大连接池。
高吞吐生产验收默认保持 `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`，验证 TDengine `raw_frames`、位置表和分页查询闭环。RAW 帧的 `parsedJson`、`metadataJson`、`rawUri` 直接进入 TDengine，GB32960 snapshot/fields 接口基于 `raw_frames` 与当前解析器即时返回结构化结果；接入服务写出的原始 `.bin` 只作为冷备复核材料。需要逐字段趋势宽表时，再显式开启 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true`，并单独压测写入放大。
高吞吐生产验收只验证 TDengine `raw_frames`、位置表和分页查询闭环。RAW 帧的 `parsedJson`、`metadataJson`、`rawUri` 直接进入 TDengine，GB32960 snapshot/fields 接口基于 `raw_frames` 与当前解析器即时返回结构化结果；接入服务写出的原始 `.bin` 只作为冷备复核材料。逐字段趋势宽表由独立字段解析服务消费 Kafka RAW/事件后维护，不由 history-app 写入。
## 构建
@@ -108,7 +107,6 @@ curl -sS http://127.0.0.1:20500/actuator/health/readiness
```bash
HTTP_PORT=20200 \
KAFKA_CONSUMER_ENABLED=true \
TDENGINE_TELEMETRY_FIELDS_ENABLED=false \
java -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar
```
@@ -122,7 +120,7 @@ curl -sS http://127.0.0.1:20200/v3/api-docs \
| grep -E '/api/event-history/locations|/api/event-history/raw-frames'
```
预期：健康检查为 `UP`，OpenAPI 中包含通用位置分页查询和 RAW 帧查询接口。`/api/event-history/telemetry/fields` 只有在 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true` 后才会暴露；默认高吞吐模式不暴露该接口，也不会持续写入逐字段宽表。
预期：健康检查为 `UP`，OpenAPI 中包含通用位置分页查询和 RAW 帧查询接口。history-app 不暴露 `/api/event-history/telemetry/fields`，也不持续写入逐字段宽表。
## 启动 JT808 指标统计服务
@@ -250,22 +248,6 @@ curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/snapshots/fields?vin=
预期：snapshot 返回 `sourceFrames.rawArchiveUri` 和解析后的 `blocks`；字段投影返回所选字段。新增协议字段后，先通过 replay 或专用解析任务补写 `raw_frames.parsedJson`，再基于历史 RAW 结构化结果查询。
## Telemetry Field 查询验收
查询 GB32960 字段历史。该能力需要 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true` 并确认压测后无持续 lag：
```bash
curl -sS 'http://127.0.0.1:20200/api/event-history/telemetry/fields?protocol=GB32960&vin=<vin>&fieldKey=<field-key>&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&limit=10'
```
查询 JT808 字段历史。没有 VIN 映射时，优先使用 `phone`。该能力同样需要 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true`：
```bash
curl -sS 'http://127.0.0.1:20200/api/event-history/telemetry/fields?protocol=JT808&phone=<phone>&fieldKey=location.speed_kmh&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&limit=10'
```
预期：响应包含 `items`，可能包含 `nextCursor`。下一页使用 `nextCursor` 里的 `cursorTs` 和 `cursorId` 查询。
## TDengine 直接检查
使用 TDengine CLI 或 JDBC 客户端检查超级表和子表是否创建：
@@ -276,14 +258,12 @@ SHOW STABLES;
SELECT COUNT(*) FROM raw_frames;
SELECT COUNT(*) FROM vehicle_locations;
SELECT COUNT(*) FROM jt808_locations;
-- 仅在 TDENGINE_TELEMETRY_FIELDS_ENABLED=true 时检查：
SELECT COUNT(*) FROM telemetry_fields;
```
预期：
- `raw_frames`、`vehicle_locations`、`jt808_locations` 存在；有对应协议实时流量后，计数持续增长。
- `telemetry_fields` 表结构会创建；默认高吞吐模式不会持续增长，除非显式开启 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true`。
- 字段趋势宽表不属于 history-app 验收范围，由独立字段解析服务负责。
- 有实时流量后，计数持续增加。
## 失败语义
@@ -359,8 +339,6 @@ python3 tools/jt808_e2e_smoke.py \
USE vehicle_ts;
SELECT COUNT(*) FROM raw_frames WHERE protocol = 'JT808';
SELECT COUNT(*) FROM jt808_locations WHERE protocol = 'JT808';
-- 仅在 TDENGINE_TELEMETRY_FIELDS_ENABLED=true 时复查：
SELECT COUNT(*) FROM telemetry_fields WHERE protocol = 'JT808';
SELECT event_time, vehicle_key, phone, message_id, raw_uri
FROM raw_frames
WHERE protocol = 'JT808'
@@ -399,9 +377,6 @@ GB32960 链路已用正式平台登录和合成实时帧验证：
USE vehicle_ts;
SELECT COUNT(*) FROM raw_frames
WHERE protocol = 'GB32960' AND vin = 'LTEST202606290001';
-- 仅在 TDENGINE_TELEMETRY_FIELDS_ENABLED=true 时复查：
SELECT COUNT(*) FROM telemetry_fields
WHERE protocol = 'GB32960' AND vin = 'LTEST202606290001';
SELECT event_time, vin, raw_uri, frame_id
FROM raw_frames
WHERE protocol = 'GB32960' AND vin = 'LTEST202606290001'

7
docs/target-architecture.md
View File

@@ -41,9 +41,9 @@ is explicitly deployed.
- Ingest apps do not serve history query APIs.
- Redis is not the long-term historical store.
- File-based event indexes are not part of the current build surface.
- `telemetry_fields` parsing is not handled by this project when a separate
field parsing service owns that job; the `telemetry_fields` query API is
also disabled by default unless `TDENGINE_TELEMETRY_FIELDS_ENABLED=true`.
- `telemetry_fields` parsing and field-trend APIs are owned by a separate
field parsing service. This project stores RAW parsed JSON plus compact
location rows and does not expose a telemetry-fields history API.
## Runtime Responsibilities
@@ -108,7 +108,6 @@ Production default:
- `TDENGINE_HISTORY_ENABLED=true` in deployment.
- `EVENT_FILE_STORE_ENABLED=false`.
- `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`.
File-based event indexes have been removed; this app should keep history
queries on TDengine and raw replay on `archive://...` references.

1
modules/apps/vehicle-history-app/src/main/resources/application.yml
View File

@@ -92,7 +92,6 @@ lingniu:
minimum-idle: ${TDENGINE_MIN_IDLE:0}
connection-timeout-millis: ${TDENGINE_CONNECTION_TIMEOUT_MILLIS:5000}
initialization-fail-timeout-millis: ${TDENGINE_INITIALIZATION_FAIL_TIMEOUT_MILLIS:-1}
telemetry-fields-enabled: ${TDENGINE_TELEMETRY_FIELDS_ENABLED:false}
event-history:
enabled: true
api:

3
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java
View File

@@ -287,7 +287,7 @@ class PortainerComposeResourceLimitsTest {
assertThat(compose)
.contains("TDENGINE_HISTORY_ENABLED: ${TDENGINE_HISTORY_ENABLED:-true}")
.contains("TDENGINE_HISTORY_DATABASE: ${TDENGINE_HISTORY_DATABASE:-vehicle_ts}")
.contains("TDENGINE_TELEMETRY_FIELDS_ENABLED: ${TDENGINE_TELEMETRY_FIELDS_ENABLED:-false}")
.doesNotContain("TDENGINE_TELEMETRY_FIELDS_ENABLED")
.doesNotContain("\n TDENGINE_ENABLED:")
.doesNotContain("\n TDENGINE_DATABASE:");
}
@@ -370,6 +370,7 @@ class PortainerComposeResourceLimitsTest {
.doesNotContain("读取共享 archive 中的原始 `.bin`")
.doesNotContain("EVENT_FILE_STORE_ENABLED")
.doesNotContain("EVENT_FILE_STORE_PATH")
.doesNotContain("TDENGINE_TELEMETRY_FIELDS_ENABLED")
.doesNotContain("/api/event-history/telemetry/fields' 会存在")
.doesNotContain("/api/event-history/records");
assertThat(splitRunbook)

6
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppCompositionTest.java
View File

@@ -10,7 +10,6 @@ import com.lingniu.ingest.eventhistory.config.EventHistoryAutoConfiguration;
import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
@@ -111,7 +110,7 @@ class VehicleHistoryAppCompositionTest {
assertThat(context).hasSingleBean(Gb32960DecodedFrameService.class);
assertThat(context).hasSingleBean(LocationHistoryController.class);
assertThat(context).hasSingleBean(RawFrameHistoryController.class);
assertThat(context).doesNotHaveBean(TelemetryFieldHistoryController.class);
assertThat(context).doesNotHaveBean("telemetryFieldHistoryController");
assertThat(context).doesNotHaveBean(Gb32960FrameController.class);
assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class);
assertThat(context).doesNotHaveBean(Gb32960NettyServer.class);
@@ -143,7 +142,6 @@ class VehicleHistoryAppCompositionTest {
"spring.cloud.nacos.config.enabled=false",
"spring.config.import=",
"lingniu.ingest.event-history.enabled=true",
"lingniu.ingest.tdengine-history.telemetry-fields-enabled=false",
"lingniu.ingest.tdengine-history.enabled=false",
"lingniu.ingest.sink.kafka.enabled=false",
"lingniu.ingest.sink.kafka.consumer.enabled=false",
@@ -151,7 +149,7 @@ class VehicleHistoryAppCompositionTest {
.run(context -> {
assertThat(context).hasSingleBean(LocationHistoryController.class);
assertThat(context).hasSingleBean(RawFrameHistoryController.class);
assertThat(context).doesNotHaveBean(TelemetryFieldHistoryController.class);
assertThat(context).doesNotHaveBean("telemetryFieldHistoryController");
});
}

5
modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java
View File

@@ -38,9 +38,6 @@ class VehicleHistoryAppDefaultsTest {
"${TDENGINE_DRIVER_CLASS_NAME:com.taosdata.jdbc.rs.RestfulDriver}")
.containsEntry("lingniu.ingest.tdengine-history.maximum-pool-size", "${TDENGINE_MAX_POOL_SIZE:16}")
.containsEntry("lingniu.ingest.tdengine-history.minimum-idle", "${TDENGINE_MIN_IDLE:0}")
.containsEntry(
"lingniu.ingest.tdengine-history.telemetry-fields-enabled",
"${TDENGINE_TELEMETRY_FIELDS_ENABLED:false}")
.containsEntry("lingniu.ingest.event-history.enabled", true)
.containsEntry("lingniu.ingest.vehicle-state.enabled", false)
.containsEntry("lingniu.ingest.vehicle-stat.enabled", false)
@@ -108,6 +105,8 @@ class VehicleHistoryAppDefaultsTest {
assertThat(properties.stringPropertyNames())
.noneMatch(name -> name.startsWith("lingniu.ingest.event-file-store."));
assertThat(properties.stringPropertyNames())
.noneMatch(name -> name.equals("lingniu.ingest.tdengine-history.telemetry-fields-enabled"));
assertThat(properties.stringPropertyNames())
.noneMatch(name -> name.equals("lingniu.ingest.sink.archive.type")
|| name.equals("lingniu.ingest.sink.archive.path"));

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

@@ -24,19 +24,12 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
private static final Logger log = LoggerFactory.getLogger(EventHistoryEnvelopeIngestor.class);
private final TdengineHistoryWriter tdengineWriter;
private final boolean telemetryFieldsEnabled;
public EventHistoryEnvelopeIngestor(TdengineHistoryWriter tdengineWriter) {
this(tdengineWriter, false);
}
public EventHistoryEnvelopeIngestor(TdengineHistoryWriter tdengineWriter,
boolean telemetryFieldsEnabled) {
if (tdengineWriter == null) {
throw new IllegalArgumentException("tdengineWriter must not be null");
}
this.tdengineWriter = tdengineWriter;
this.telemetryFieldsEnabled = telemetryFieldsEnabled;
}
public void ingest(byte[] kafkaValue) throws IOException {
@@ -129,13 +122,5 @@ public final class EventHistoryEnvelopeIngestor implements EnvelopeBatchIngestor
if (!locations.isEmpty()) {
tdengineWriter.appendLocations(locations);
}
if (telemetryFieldsEnabled) {
var telemetryFields = envelopes.stream()
.flatMap(envelope -> TdengineEnvelopeRows.telemetryFields(envelope).stream())
.toList();
if (!telemetryFields.isEmpty()) {
tdengineWriter.appendTelemetryFields(telemetryFields);
}
}
}
}

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

@@ -3,13 +3,11 @@ package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineLocationRow;
import com.lingniu.ingest.tdenginehistory.TdenginePage;
import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery;
import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
@@ -39,8 +37,6 @@ public final class RawFrameHistoryController {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
private static final int LOCATION_ENRICH_LIMIT = 1;
private static final int TELEMETRY_ENRICH_LIMIT = 500;
private final TdengineHistoryReader reader;
public RawFrameHistoryController(TdengineHistoryReader reader) {
@@ -76,8 +72,6 @@ public final class RawFrameHistoryController {
@RequestParam(defaultValue = "DESC") TdengineQueryOrder order,
@Parameter(description = "返回原始帧数量上限，最大 1000。", example = "100")
@RequestParam(defaultValue = "100") int limit,
@Parameter(description = "是否按 rawUri 补充 location/telemetry 解析字段。默认 false，避免 raw 列表产生额外 TDengine 查询。", example = "false")
@RequestParam(defaultValue = "false") boolean includeParsedFields,
@Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z")
@RequestParam(required = false) String cursorTs,
@Parameter(description = "上一页返回的 nextCursor.id。", example = "frame-xxx")
@@ -101,7 +95,7 @@ public final class RawFrameHistoryController {
order,
limit,
cursor(cursorTs, cursorId)));
return RawFramePageResponse.from(reader, page, includeParsedFields);
return RawFramePageResponse.from(page);
}
private static Integer parseMessageId(String messageId) {
@@ -152,15 +146,13 @@ public final class RawFrameHistoryController {
List<RawFrameResponse> items,
CursorResponse nextCursor) {
private static RawFramePageResponse from(TdengineHistoryReader reader,
TdenginePage<TdengineRawFrameRow> page,
boolean includeParsedFields) throws IOException {
private static RawFramePageResponse from(TdenginePage<TdengineRawFrameRow> page) {
CursorResponse next = page.nextCursor()
.map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker()))
.orElse(null);
List<RawFrameResponse> items = new java.util.ArrayList<>(page.items().size());
for (TdengineRawFrameRow row : page.items()) {
items.add(RawFrameResponse.from(reader, row, includeParsedFields));
items.add(RawFrameResponse.from(row));
}
return new RawFramePageResponse(
items,
@@ -193,9 +185,7 @@ public final class RawFrameHistoryController {
String vin,
String phone) {
private static RawFrameResponse from(TdengineHistoryReader reader,
TdengineRawFrameRow row,
boolean includeParsedFields) throws IOException {
private static RawFrameResponse from(TdengineRawFrameRow row) {
Map<String, Object> metadata = metadata(row.metadataJson());
return new RawFrameResponse(
row.ts().toString(),
@@ -213,7 +203,7 @@ public final class RawFrameHistoryController {
row.metadataJson(),
row.parsedJson(),
metadata,
parsedFields(reader, row, metadata, parsed(row.parsedJson()), includeParsedFields),
parsedFields(metadata, parsed(row.parsedJson())),
row.protocol(),
row.vehicleKey(),
row.vin(),
@@ -231,11 +221,8 @@ public final class RawFrameHistoryController {
}
}
private static Map<String, Object> parsedFields(TdengineHistoryReader reader,
TdengineRawFrameRow row,
Map<String, Object> metadata,
Map<String, Object> parsed,
boolean includeParsedFields) throws IOException {
private static Map<String, Object> parsedFields(Map<String, Object> metadata,
Map<String, Object> parsed) {
Map<String, Object> out = new LinkedHashMap<>();
if (parsed.isEmpty()) {
for (Map.Entry<String, Object> entry : metadata.entrySet()) {
@@ -252,21 +239,6 @@ public final class RawFrameHistoryController {
} else {
out.putAll(parsed);
}
if (!includeParsedFields) {
return immutableMap(out);
}
for (TdengineLocationRow location : reader.queryLocationsByRawUri(
row.protocol(), row.rawUri(), LOCATION_ENRICH_LIMIT)) {
addLocation(out, location);
}
for (TdengineTelemetryFieldRow field : reader.queryTelemetryFieldsByRawUri(
row.protocol(), row.rawUri(), TELEMETRY_ENRICH_LIMIT)) {
out.put(field.fieldKey(), typedValue(field));
out.put(field.fieldKey() + "._type", field.valueType());
if (field.unit() != null && !field.unit().isBlank()) {
out.put(field.fieldKey() + "._unit", field.unit());
}
}
return immutableMap(out);
}
@@ -281,29 +253,6 @@ public final class RawFrameHistoryController {
}
}
private static void addLocation(Map<String, Object> out, TdengineLocationRow row) {
out.put("location.longitude", row.longitude());
out.put("location.latitude", row.latitude());
out.put("location.altitudeM", row.altitudeM());
out.put("location.speedKmh", row.speedKmh());
out.put("location.directionDeg", row.directionDeg());
out.put("location.alarmFlag", row.alarmFlag());
out.put("location.statusFlag", row.statusFlag());
if (row.totalMileageKm() != null) {
out.put("location.totalMileageKm", row.totalMileageKm());
}
}
private static Object typedValue(TdengineTelemetryFieldRow field) {
if (field.valueDouble() != null) {
return field.valueDouble();
}
if (field.valueLong() != null) {
return field.valueLong();
}
return field.valueText();
}
private static Map<String, Object> immutableMap(Map<String, Object> values) {
return Collections.unmodifiableMap(new LinkedHashMap<>(values));
}

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

@@ -1,213 +0,0 @@
package com.lingniu.ingest.eventhistory;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdenginePage;
import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery;
import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ResponseStatusException;
import java.io.IOException;
import java.time.Instant;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@RestController
@ConditionalOnProperty(
name = {
"lingniu.ingest.event-history.enabled",
"lingniu.ingest.tdengine-history.telemetry-fields-enabled"
},
havingValue = "true")
@ConditionalOnBean(TdengineHistoryReader.class)
@RequestMapping("/api/event-history/telemetry")
@Tag(name = "telemetry-field-history-controller", description = "遥测字段历史分页查询接口。")
public final class TelemetryFieldHistoryController {
private static final Map<String, String> JT808_FIELD_ALIASES = Map.of(
"location.longitude", "longitude",
"location.latitude", "latitude",
"location.altitude_m", "altitude_m",
"location.speed_kmh", "speed_kmh",
"location.direction_deg", "direction_deg",
"location.alarm_flag", "location_alarm_flag",
"location.status_flag", "location_status_raw");
private final TdengineHistoryReader reader;
public TelemetryFieldHistoryController(TdengineHistoryReader reader) {
if (reader == null) {
throw new IllegalArgumentException("reader must not be null");
}
this.reader = reader;
}
@GetMapping("/fields")
@Operation(
summary = "查询单车单字段历史值",
description = "按协议、车辆标识、字段 key 和时间范围查询 TDengine telemetry_fields。分页使用 cursorTs + cursorId，不使用 offset。")
public TelemetryFieldPageResponse fields(
@Parameter(description = "协议名，例如 GB32960 或 JT808。", required = true, example = "GB32960")
@RequestParam String protocol,
@Parameter(description = "字段 key，例如 FUEL_CELL_V2016.hydrogenComsuxxx。", required = true)
@RequestParam String fieldKey,
@Parameter(description = "内部车辆键。传入后优先使用。", example = "LNVFC000000000001")
@RequestParam(required = false) String vehicleKey,
@Parameter(description = "VIN。GB32960 常用该字段作为车辆键。", example = "LNVFC000000000001")
@RequestParam(required = false) String vin,
@Parameter(description = "JT808 终端手机号或终端标识；JT808 会自动映射为 jt808:<phone>。", example = "g7gps")
@RequestParam(required = false) String phone,
@Parameter(description = "开始时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:00:00")
@RequestParam String dateFrom,
@Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:10:00")
@RequestParam String dateTo,
@Parameter(description = "排序方向。", example = "DESC")
@RequestParam(defaultValue = "DESC") TdengineQueryOrder order,
@Parameter(description = "返回值数量上限，最大 1000。", example = "100")
@RequestParam(defaultValue = "100") int limit,
@Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z")
@RequestParam(required = false) String cursorTs,
@Parameter(description = "上一页返回的 nextCursor.id。", example = "event-xxx#0")
@RequestParam(required = false) String cursorId) throws IOException {
String normalizedProtocol = require(protocol, "protocol is required for telemetry field query").toUpperCase(Locale.ROOT);
String normalizedFieldKey = normalizeFieldKey(
normalizedProtocol,
require(fieldKey, "fieldKey is required for telemetry field query"));
QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
TdenginePage<TdengineTelemetryFieldRow> page = reader.queryTelemetryFields(new TdengineTelemetryFieldQuery(
normalizedProtocol,
resolveVehicleKey(normalizedProtocol, vehicleKey, vin, phone),
normalizedFieldKey,
range.eventTimeFrom(),
range.eventTimeTo().plusMillis(1),
order,
limit,
cursor(cursorTs, cursorId)));
return TelemetryFieldPageResponse.from(page);
}
private static String normalizeFieldKey(String protocol, String fieldKey) {
if (!"JT808".equals(protocol)) {
return fieldKey;
}
return JT808_FIELD_ALIASES.getOrDefault(fieldKey, fieldKey);
}
private static String resolveVehicleKey(String protocol, String vehicleKey, String vin, String phone) {
String explicitVehicleKey = trimToNull(vehicleKey);
if (explicitVehicleKey != null) {
return explicitVehicleKey;
}
String vinValue = trimToNull(vin);
if (vinValue != null) {
return vinValue;
}
String phoneValue = trimToNull(phone);
if (phoneValue != null) {
if ("JT808".equals(protocol) && !phoneValue.startsWith("jt808:")) {
return "jt808:" + phoneValue;
}
return phoneValue;
}
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "vehicleKey, vin or phone is required");
}
private static String require(String value, String message) {
String trimmed = trimToNull(value);
if (trimmed == null) {
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, message);
}
return trimmed;
}
private static String trimToNull(String value) {
if (value == null || value.isBlank()) {
return null;
}
return value.trim();
}
private static TdenginePageCursor cursor(String cursorTs, String cursorId) {
boolean hasTs = cursorTs != null && !cursorTs.isBlank();
boolean hasId = cursorId != null && !cursorId.isBlank();
if (!hasTs && !hasId) {
return null;
}
if (!hasTs || !hasId) {
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "cursorTs and cursorId must be provided together");
}
return new TdenginePageCursor(Instant.parse(cursorTs.trim()), cursorId.trim());
}
public record TelemetryFieldPageResponse(
List<TelemetryFieldResponse> items,
CursorResponse nextCursor) {
private static TelemetryFieldPageResponse from(TdenginePage<TdengineTelemetryFieldRow> page) {
CursorResponse next = page.nextCursor()
.map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker()))
.orElse(null);
return new TelemetryFieldPageResponse(
page.items().stream().map(TelemetryFieldResponse::from).toList(),
next);
}
}
public record CursorResponse(String ts, String id) {
}
public record TelemetryFieldResponse(
String eventTime,
String receivedAt,
String factId,
String frameId,
String fieldKey,
String valueType,
String valueText,
Double valueDouble,
Long valueLong,
String unit,
String quality,
String sourcePath,
String rawUri,
String metadataJson,
String protocol,
String vehicleKey,
String vin,
String phone) {
private static TelemetryFieldResponse from(TdengineTelemetryFieldRow row) {
return new TelemetryFieldResponse(
row.ts().toString(),
row.receivedAt().toString(),
row.factId(),
row.frameId(),
row.fieldKey(),
row.valueType(),
row.valueText(),
row.valueDouble(),
row.valueLong(),
row.unit(),
row.quality(),
row.sourcePath(),
row.rawUri(),
row.metadataJson(),
row.protocol(),
row.vehicleKey(),
row.vin(),
row.phone());
}
}
}

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

@@ -10,7 +10,6 @@ import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
import com.lingniu.ingest.sink.kafka.KafkaSinkAutoConfiguration;
@@ -53,10 +52,8 @@ public class EventHistoryAutoConfiguration {
@Bean
@ConditionalOnBean(TdengineHistoryWriter.class)
@ConditionalOnMissingBean
public EventHistoryEnvelopeIngestor tdengineEventHistoryEnvelopeIngestor(TdengineHistoryWriter writer,
@Value("${lingniu.ingest.tdengine-history.telemetry-fields-enabled:false}")
boolean telemetryFieldsEnabled) {
return new EventHistoryEnvelopeIngestor(writer, telemetryFieldsEnabled);
public EventHistoryEnvelopeIngestor tdengineEventHistoryEnvelopeIngestor(TdengineHistoryWriter writer) {
return new EventHistoryEnvelopeIngestor(writer);
}
@Bean
@@ -130,14 +127,6 @@ public class EventHistoryAutoConfiguration {
return new Jt808RawFrameHistoryController(reader);
}
@Bean
@ConditionalOnBean(TdengineHistoryReader.class)
@ConditionalOnProperty(prefix = "lingniu.ingest.tdengine-history", name = "telemetry-fields-enabled", havingValue = "true")
@ConditionalOnMissingBean
public TelemetryFieldHistoryController telemetryFieldHistoryController(TdengineHistoryReader reader) {
return new TelemetryFieldHistoryController(reader);
}
static Path archiveRoot(String value) {
if (value == null || value.isBlank()) {
return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive");

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

@@ -121,9 +121,9 @@ class EventHistoryEnvelopeIngestorTest {
}
@Test
void tryIngestCanWriteTelemetryFieldFactsWhenExplicitlyEnabled() {
void tryIngestDoesNotWriteTelemetryFieldFacts() {
CapturingTdengineWriter tdengineWriter = new CapturingTdengineWriter();
EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(tdengineWriter, true);
EventHistoryEnvelopeIngestor ingestor = new EventHistoryEnvelopeIngestor(tdengineWriter);
EnvelopeIngestResult result = ingestor.tryIngest(
jt808LocationEnvelope("jt808-location-1", "frame-jt808-1", "013800000001").toByteArray());
@@ -133,9 +133,7 @@ class EventHistoryEnvelopeIngestorTest {
.containsExactly("frame-jt808-1");
assertThat(tdengineWriter.locations).extracting(TdengineLocationRow::factId)
.containsExactly("jt808-location-1");
assertThat(tdengineWriter.telemetryFields)
.extracting(TdengineTelemetryFieldRow::fieldKey)
.containsExactly("location.speedKmh");
assertThat(tdengineWriter.telemetryFields).isEmpty();
}
@Test

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

@@ -44,7 +44,6 @@ class RawFrameHistoryControllerTest {
"2026-06-29T23:59:59+08:00",
TdengineQueryOrder.DESC,
20,
false,
"2026-06-29T12:00:00Z",
"frame-0");
@@ -74,9 +73,28 @@ class RawFrameHistoryControllerTest {
}
@Test
void enrichesRawFrameWithLocationAndTelemetryFieldsByRawUri() throws Exception {
void returnsParsedJsonWithoutQueryingDerivedTelemetryTables() throws Exception {
Instant ts = Instant.parse("2026-06-29T12:00:01Z");
TdengineRawFrameRow raw = row("frame-1", ts.toString());
TdengineRawFrameRow raw = new TdengineRawFrameRow(
ts,
"frame-1",
ts.plusMillis(500),
0x0200,
0,
ts,
"archive://2026/06/29/JT808/VIN-JT808-1/frame-1.bin",
"sha256:frame-1",
72,
"SUCCEEDED",
"",
"115.231.168.135:43625",
"{\"rawArchiveUri\":\"archive://2026/06/29/JT808/VIN-JT808-1/frame-1.bin\"}",
"{\"location\":{\"longitude\":121.312516,\"latitude\":30.832808,\"speedKmh\":25.0},"
+ "\"extras\":{\"gpsMileageKm\":14504.446}}",
"JT808",
"VIN-JT808-1",
"VIN-JT808-1",
"g7gps");
CapturingReader reader = new CapturingReader(new TdenginePage<>(List.of(raw), Optional.empty()));
reader.locationsByRawUri.add(new TdengineLocationRow(
ts,
@@ -128,19 +146,15 @@ class RawFrameHistoryControllerTest {
"2026-06-29T23:59:59+08:00",
TdengineQueryOrder.DESC,
20,
true,
null,
null);
RawFrameHistoryController.RawFrameResponse item = response.items().getFirst();
assertThat(reader.locationRawUri).isEqualTo(raw.rawUri());
assertThat(reader.telemetryRawUri).isEqualTo(raw.rawUri());
assertThat(reader.locationRawUri).isNull();
assertThat(reader.telemetryRawUri).isNull();
assertThat(item.parsedFields())
.containsEntry("location.longitude", 121.312516)
.containsEntry("location.latitude", 30.832808)
.containsEntry("location.speedKmh", 25.0)
.containsEntry("location.totalMileageKm", 14504.446)
.containsEntry("VEHICLE.speedKmh", 25.0);
.containsKey("location")
.containsKey("extras");
}
@Test
@@ -179,7 +193,6 @@ class RawFrameHistoryControllerTest {
"2026-06-29T23:59:59+08:00",
TdengineQueryOrder.DESC,
20,
false,
null,
null);
@@ -205,7 +218,6 @@ class RawFrameHistoryControllerTest {
"2026-06-30",
TdengineQueryOrder.DESC,
20,
false,
null,
null))
.isInstanceOfSatisfying(ResponseStatusException.class, ex ->
@@ -228,7 +240,6 @@ class RawFrameHistoryControllerTest {
"2026-06-30",
TdengineQueryOrder.DESC,
20,
false,
null,
null))
.isInstanceOfSatisfying(ResponseStatusException.class, ex ->

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

@@ -1,153 +0,0 @@
package com.lingniu.ingest.eventhistory;
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.TdenginePageCursor;
import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
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.springframework.http.HttpStatus;
import org.springframework.web.server.ResponseStatusException;
import java.io.IOException;
import java.time.Instant;
import java.util.List;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class TelemetryFieldHistoryControllerTest {
@Test
void queriesGb32960FieldByVinWithKeysetCursor() throws Exception {
CapturingReader reader = new CapturingReader(new TdenginePage<>(
List.of(row("field-1", "2026-06-29T05:00:01Z")),
Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "field-1"))));
TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(reader);
TelemetryFieldHistoryController.TelemetryFieldPageResponse response = controller.fields(
"GB32960",
"FUEL_CELL_V2016.hydrogenComsuxxx",
null,
"LNVFC000000000001",
null,
"2026-06-29T13:00:00+08:00",
"2026-06-29T13:10:00+08:00",
TdengineQueryOrder.ASC,
100,
"2026-06-29T05:00:00Z",
"field-0");
assertThat(reader.query.protocol()).isEqualTo("GB32960");
assertThat(reader.query.vehicleKey()).isEqualTo("LNVFC000000000001");
assertThat(reader.query.fieldKey()).isEqualTo("FUEL_CELL_V2016.hydrogenComsuxxx");
assertThat(reader.query.order()).isEqualTo(TdengineQueryOrder.ASC);
assertThat(reader.query.limit()).isEqualTo(100);
assertThat(reader.query.cursor()).isEqualTo(new TdenginePageCursor(
Instant.parse("2026-06-29T05:00:00Z"), "field-0"));
assertThat(response.items())
.extracting(TelemetryFieldHistoryController.TelemetryFieldResponse::factId)
.containsExactly("field-1");
assertThat(response.items().getFirst().valueDouble()).isEqualTo(3.14);
assertThat(response.nextCursor()).isEqualTo(new TelemetryFieldHistoryController.CursorResponse(
"2026-06-29T05:00:01Z", "field-1"));
}
@Test
void mapsJt808PhoneToVehicleKeyWhenVehicleKeyIsAbsent() throws Exception {
CapturingReader reader = new CapturingReader(new TdenginePage<>(List.of(), Optional.empty()));
TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(reader);
controller.fields(
"JT808",
"location.speed_kmh",
null,
null,
"g7gps",
"2026-06-29",
"2026-06-29",
TdengineQueryOrder.DESC,
50,
null,
null);
assertThat(reader.query.vehicleKey()).isEqualTo("jt808:g7gps");
assertThat(reader.query.fieldKey()).isEqualTo("speed_kmh");
}
@Test
void rejectsMissingVehicleIdentity() {
TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(
new CapturingReader(new TdenginePage<>(List.of(), Optional.empty())));
assertThatThrownBy(() -> controller.fields(
"GB32960",
"vehicle.speed",
null,
null,
null,
"2026-06-29",
"2026-06-29",
TdengineQueryOrder.DESC,
50,
null,
null))
.isInstanceOfSatisfying(ResponseStatusException.class, ex ->
assertThat(ex.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST));
}
private static TdengineTelemetryFieldRow row(String factId, String ts) {
Instant instant = Instant.parse(ts);
return new TdengineTelemetryFieldRow(
instant,
factId,
"frame-1",
instant.plusMillis(500),
"FUEL_CELL_V2016.hydrogenComsuxxx",
"DOUBLE",
"3.14",
3.14,
null,
"kg/100km",
"GOOD",
"FUEL_CELL_V2016.hydrogenComsuxxx",
"archive://gb32960/frame-1.bin",
"{\"cmd\":\"REALTIME_REPORT\"}",
"GB32960",
"LNVFC000000000001",
"LNVFC000000000001",
"");
}
private static final class CapturingReader implements TdengineHistoryReader {
private final TdenginePage<TdengineTelemetryFieldRow> response;
private TdengineTelemetryFieldQuery query;
private CapturingReader(TdenginePage<TdengineTelemetryFieldRow> response) {
this.response = response;
}
@Override
public TdenginePage<TdengineRawFrameRow> queryRawFrames(TdengineRawFrameQuery query) {
throw new UnsupportedOperationException();
}
@Override
public TdenginePage<TdengineLocationRow> queryLocations(TdengineLocationQuery query) {
throw new UnsupportedOperationException();
}
@Override
public TdenginePage<TdengineTelemetryFieldRow> queryTelemetryFields(TdengineTelemetryFieldQuery query)
throws IOException {
this.query = query;
return response;
}
}
}

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

@@ -10,7 +10,6 @@ import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.RawFrameHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
@@ -119,7 +118,7 @@ class EventHistoryAutoConfigurationTest {
assertThat(context).hasSingleBean(LocationHistoryController.class);
assertThat(context).hasSingleBean(RawFrameHistoryController.class);
assertThat(context).hasSingleBean(ApiExceptionHandler.class);
assertThat(context).doesNotHaveBean(TelemetryFieldHistoryController.class);
assertThat(context).doesNotHaveBean("telemetryFieldHistoryController");
assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class);
assertThat(context).doesNotHaveBean(Jt808RawFrameHistoryController.class);
});
@@ -144,13 +143,13 @@ class EventHistoryAutoConfigurationTest {
}
@Test
void createsTelemetryFieldHistoryControllerWhenTelemetryFieldsAreEnabled() {
void doesNotExposeTelemetryFieldHistoryControllerWhenTelemetryFieldsAreEnabled() {
contextRunner
.withPropertyValues(
"lingniu.ingest.event-history.enabled=true",
"lingniu.ingest.tdengine-history.telemetry-fields-enabled=true")
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
.run(context -> assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class));
.run(context -> assertThat(context).doesNotHaveBean("telemetryFieldHistoryController"));
}
@Test