diff --git a/deploy/local/launchctl/README.md b/deploy/local/launchctl/README.md index bedafe94..8dfec54f 100644 --- a/deploy/local/launchctl/README.md +++ b/deploy/local/launchctl/README.md @@ -8,6 +8,13 @@ 三个服务必须使用同一个 `SINK_ARCHIVE_PATH`。协议接入服务负责落原始 `.bin` 文件，Kafka raw envelope 只携带 `archive://...` 引用；history 服务按同一个 archive root 读取原始帧并提供查询。 +`vehicle-history-app` 的生产高吞吐默认配置是 TDengine-only： + +- `EVENT_FILE_STORE_ENABLED=false`：避免 DuckDB/EventFileStore 进入实时消费热路径。 +- `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`：默认只写 `raw_frames` 和 `vehicle_locations`，避免逐字段写放大。 +- GB32960 snapshot/fields 接口默认通过 TDengine `raw_frames` 索引找到 `rawUri`，再回读共享 archive 中的原始 `.bin` 即时解析；后续解析器新增字段后可直接查询历史 raw。 +- 如果需要兼容旧 `/api/event-history/records` 低层记录接口或逐字段趋势宽表，可以显式打开对应开关；开启后要重新压测 Kafka lag 和 TDengine 写入吞吐。 + ## 生成 plist 在仓库根目录执行： @@ -48,6 +55,13 @@ done 如果生产配置由 Nacos 下发，可以把模板里的 `NACOS_CONFIG_ENABLED` 改成 `true`，并补充 `NACOS_SERVER_ADDR`、`NACOS_NAMESPACE`、`NACOS_GROUP`、`NACOS_USERNAME`、`NACOS_PASSWORD`。端口、Kafka topic、archive root、TDengine 连接建议仍保留为启动环境变量，便于 Portainer、launchctl 和临时压测保持一致。 +history 消费者会按协议和 raw/event 自动拆分 consumer group。设置 `KAFKA_GROUP_HISTORY=vehicle-history-live` 后，实际 group 为： + +- `vehicle-history-live-gb32960-event` +- `vehicle-history-live-gb32960-raw` +- `vehicle-history-live-jt808-event` +- `vehicle-history-live-jt808-raw` + JT808 身份绑定默认使用内存或文件。需要把注册信息维护到 MySQL 时，将 `com.lingniu.jt808.plist.template` 里的 `VEHICLE_IDENTITY_STORE` 改为 `mysql`，并提供 `VEHICLE_IDENTITY_MYSQL_*` 连接参数。服务会自动创建 `vehicle_identity_binding` 和 `vehicle_identity_binding_registration` 两张表：前者保存已确认 VIN 的外部标识绑定，后者保存 0x0100 注册字段；外部系统可把真实 VIN 反写到 registration 表，服务按刷新周期加载后，后续 raw/event 会带真实 VIN。 ## 启停 diff --git a/deploy/local/launchctl/com.lingniu.gb32960.plist.template b/deploy/local/launchctl/com.lingniu.gb32960.plist.template index c13b9b90..cfea6336 100644 --- a/deploy/local/launchctl/com.lingniu.gb32960.plist.template +++ b/deploy/local/launchctl/com.lingniu.gb32960.plist.template @@ -20,6 +20,8 @@ 32960 GB32960_AUTH_ENABLED false + GB32960_PLATFORM_IP_HYUNDAI + 115.29.187.205 KAFKA_ENABLED true KAFKA_BROKERS diff --git a/deploy/local/launchctl/com.lingniu.vehicle-history.plist.template b/deploy/local/launchctl/com.lingniu.vehicle-history.plist.template index b0a18db4..135e76ab 100644 --- a/deploy/local/launchctl/com.lingniu.vehicle-history.plist.template +++ b/deploy/local/launchctl/com.lingniu.vehicle-history.plist.template @@ -26,6 +26,8 @@ vehicle-history-live KAFKA_GROUP_HISTORY vehicle-history-live + KAFKA_CONSUMER_CONCURRENCY + 3 KAFKA_BROKERS 114.55.58.251:9092 KAFKA_TOPIC_GB32960_EVENT @@ -37,7 +39,7 @@ KAFKA_TOPIC_JT808_RAW vehicle.raw.jt808.v1 EVENT_FILE_STORE_ENABLED - true + false EVENT_FILE_STORE_PATH __EVENT_STORE_PATH__ EVENT_FILE_STORE_ZONE_ID @@ -62,6 +64,8 @@ 16 TDENGINE_INITIALIZE_SCHEMA true + TDENGINE_TELEMETRY_FIELDS_ENABLED + false NACOS_CONFIG_ENABLED false MANAGEMENT_HEALTH_REDIS_ENABLED diff --git a/deploy/portainer/docker-compose.yml b/deploy/portainer/docker-compose.yml index ece55207..16ea2247 100644 --- a/deploy/portainer/docker-compose.yml +++ b/deploy/portainer/docker-compose.yml @@ -56,7 +56,7 @@ services: restart: unless-stopped environment: <<: *common-env - HTTP_PORT: 20400 + HTTP_PORT: 20482 JT808_PORT: 808 KAFKA_CONSUMER_ENABLED: "false" KAFKA_NODE_ID: ${KAFKA_NODE_ID_JT808:-jt808-ingest-portainer} @@ -69,7 +69,7 @@ services: VEHICLE_IDENTITY_MYSQL_PASSWORD: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:-} VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL: ${VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL:-60s} ports: - - "${JT808_HTTP_PORT:-20400}:20400" + - "${JT808_HTTP_PORT:-20482}:20482" - "${JT808_TCP_PORT:-808}:808" volumes: - vehicle-history-archive:/archive @@ -89,20 +89,25 @@ services: HTTP_PORT: 20200 KAFKA_CONSUMER_CLIENT_ID_PREFIX: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX_HISTORY:-vehicle-history} KAFKA_GROUP_HISTORY: ${KAFKA_GROUP_HISTORY:-vehicle-history} + KAFKA_CONSUMER_CONCURRENCY: ${KAFKA_CONSUMER_CONCURRENCY:-3} SINK_ARCHIVE_ENABLED: ${SINK_ARCHIVE_ENABLED:-true} SINK_ARCHIVE_PATH: /archive/ - EVENT_FILE_STORE_ENABLED: ${EVENT_FILE_STORE_ENABLED:-true} + # 高吞吐生产热路径默认只写 TDengine。需要 GB32960 snapshot/fields 旧查询时再显式开启。 + EVENT_FILE_STORE_ENABLED: ${EVENT_FILE_STORE_ENABLED:-false} EVENT_FILE_STORE_PATH: /event-store/ EVENT_FILE_STORE_ZONE_ID: ${EVENT_FILE_STORE_ZONE_ID:-Asia/Shanghai} TDENGINE_HISTORY_ENABLED: ${TDENGINE_HISTORY_ENABLED:-true} - TDENGINE_HISTORY_DATABASE: ${TDENGINE_HISTORY_DATABASE:-vehicle_history} - TDENGINE_JDBC_URL: ${TDENGINE_JDBC_URL:-jdbc:TAOS-RS://tdengine:6041/vehicle_history} + TDENGINE_HISTORY_DATABASE: ${TDENGINE_HISTORY_DATABASE:-vehicle_ts} + TDENGINE_JDBC_URL: ${TDENGINE_JDBC_URL:-jdbc:TAOS-WS://tdengine:6041/vehicle_ts} + TDENGINE_DRIVER_CLASS_NAME: ${TDENGINE_DRIVER_CLASS_NAME:-com.taosdata.jdbc.ws.WebSocketDriver} TDENGINE_USERNAME: ${TDENGINE_USERNAME:-root} TDENGINE_PASSWORD: ${TDENGINE_PASSWORD:-taosdata} TDENGINE_MAX_POOL_SIZE: ${TDENGINE_MAX_POOL_SIZE:-32} TDENGINE_MIN_IDLE: ${TDENGINE_MIN_IDLE:-0} TDENGINE_CONNECTION_TIMEOUT_MILLIS: ${TDENGINE_CONNECTION_TIMEOUT_MILLIS:-5000} TDENGINE_INITIALIZATION_FAIL_TIMEOUT_MILLIS: ${TDENGINE_INITIALIZATION_FAIL_TIMEOUT_MILLIS:--1} + # 默认关闭逐字段宽表，避免 telemetry_fields 写放大；需要字段趋势查询时显式打开。 + TDENGINE_TELEMETRY_FIELDS_ENABLED: ${TDENGINE_TELEMETRY_FIELDS_ENABLED:-false} ports: - "${VEHICLE_HISTORY_HTTP_PORT:-20200}:20200" volumes: diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index 10d6506d..73e4d1aa 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -5,7 +5,7 @@ 1. 协议 App 接收 TCP 报文。 2. 协议 App 归档原始帧元数据，并把事件信封发布到 Kafka。 3. `vehicle-history-app` 消费 Kafka。 -4. history 写入 raw、location、telemetry field 到 TDengine。 +4. history 写入 raw、location 到 TDengine；按需开启 telemetry field 逐字段宽表。 5. Swagger/API 通过 TDengine 做历史分页查询。 每一步都必须执行命令并检查输出后，才能标记为已验证。 @@ -17,15 +17,21 @@ ```bash export KAFKA_BROKERS=114.55.58.251:9092 export TDENGINE_HISTORY_ENABLED=true -export TDENGINE_JDBC_URL='jdbc:TAOS-RS://:6041/vehicle_history' +export TDENGINE_HISTORY_DATABASE=vehicle_ts +export TDENGINE_JDBC_URL='jdbc:TAOS-WS://:6041/vehicle_ts' +export TDENGINE_DRIVER_CLASS_NAME=com.taosdata.jdbc.ws.WebSocketDriver export TDENGINE_USERNAME=root export TDENGINE_PASSWORD='' export TDENGINE_MIN_IDLE=0 export TDENGINE_MAX_POOL_SIZE=32 +export EVENT_FILE_STORE_ENABLED=false +export TDENGINE_TELEMETRY_FIELDS_ENABLED=false ``` `TDENGINE_MIN_IDLE=0` 用于减少冷启动时 TDengine 连接重试日志。TDengine 地址稳定后，再按生产并发调大连接池。 +高吞吐生产验收默认关闭 `EVENT_FILE_STORE_ENABLED` 和 `TDENGINE_TELEMETRY_FIELDS_ENABLED`，验证 `raw_frames`、`vehicle_locations`、raw archive 文件和分页查询闭环。GB32960 snapshot/fields 接口会通过 TDengine `raw_frames` 索引找到 `rawUri`，再回读共享 archive 的原始 `.bin` 即时解析；需要兼容旧 `/api/event-history/records` 低层记录接口或逐字段趋势宽表时，再显式开启对应开关，并单独压测写入放大。 + ## 构建 ```bash @@ -41,7 +47,7 @@ mvn -pl modules/apps/jt808-ingest-app,modules/apps/gb32960-ingest-app,modules/ap ## 启动 JT808 接入服务，监听 808 端口 ```bash -HTTP_PORT=20400 \ +HTTP_PORT=20482 \ JT808_PORT=808 \ KAFKA_CONSUMER_ENABLED=false \ java -jar modules/apps/jt808-ingest-app/target/jt808-ingest-app.jar @@ -50,7 +56,7 @@ java -jar modules/apps/jt808-ingest-app/target/jt808-ingest-app.jar 验证： ```bash -curl -sS http://127.0.0.1:20400/actuator/health +curl -sS http://127.0.0.1:20482/actuator/health lsof -nP -iTCP:808 -sTCP:LISTEN ``` @@ -77,8 +83,9 @@ lsof -nP -iTCP:32960 -sTCP:LISTEN ```bash HTTP_PORT=20200 \ KAFKA_CONSUMER_ENABLED=true \ -EVENT_FILE_STORE_ENABLED=true \ +EVENT_FILE_STORE_ENABLED=false \ EVENT_FILE_STORE_PATH=./target/tdengine-verification/event-store \ +TDENGINE_TELEMETRY_FIELDS_ENABLED=false \ SINK_ARCHIVE_PATH=./target/tdengine-verification/archive \ java -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar ``` @@ -88,10 +95,10 @@ java -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar ```bash curl -sS http://127.0.0.1:20200/actuator/health curl -sS http://127.0.0.1:20200/v3/api-docs \ - | grep -E '/api/event-history/jt808/locations|/api/event-history/telemetry/fields' + | grep -E '/api/event-history/jt808/locations|/api/event-history/gb32960/snapshots|/api/event-history/telemetry/fields' ``` -预期：健康检查为 `UP`，OpenAPI 中包含这两个 TDengine 查询接口。 +预期：健康检查为 `UP`，OpenAPI 中包含 JT808 TDengine 分页查询接口和 GB32960 snapshot 接口。`/api/event-history/telemetry/fields` 会存在，但只有在 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true` 后才会持续写入新字段数据。 ## 本机 launchctl 部署 @@ -142,7 +149,12 @@ Swagger： - `vehicle.event.jt808.v1` - `vehicle.raw.jt808.v1` -使用 Kafka CLI 或管理工具检查 topic 是否存在，并确认 `vehicle-history` consumer group 在测试流量后没有持续堆积。 +使用 Kafka CLI 或管理工具检查 topic 是否存在，并确认 history consumer group 在测试流量后没有持续堆积。默认 `KAFKA_GROUP_HISTORY=vehicle-history` 时，实际 group 会拆成： + +- `vehicle-history-gb32960-event` +- `vehicle-history-gb32960-raw` +- `vehicle-history-jt808-event` +- `vehicle-history-jt808-raw` ## JT808 实时转发验收 @@ -162,15 +174,27 @@ curl -sS 'http://127.0.0.1:20200/api/event-history/jt808/locations?phone= 预期：响应包含 `items`，并且每条记录包含 `eventTime`、`phone`、`longitude`、`latitude`、`speedKmh`、`rawUri`、`metadataJson`。 +## GB32960 Snapshot / 字段投影验收 + +默认高吞吐模式下，GB32960 全字段查询不依赖 telemetry_fields 宽表。history 先查 TDengine `raw_frames`，再按 `rawUri` 读取共享 archive 中的原始 `.bin` 并即时解析： + +```bash +curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/snapshots?vin=&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&limit=10' + +curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/snapshots/fields?vin=&fields=VEHICLE.speedKmh,VEHICLE.totalMileageKm,POSITION_V2016.longitude,POSITION_V2016.latitude&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&limit=10' +``` + +预期：snapshot 返回 `sourceFrames.rawArchiveUri` 和解析后的 `blocks`；字段投影返回所选字段。新增协议字段后，只要解析器和字典补齐，就可以基于历史 raw 重新查询。 + ## Telemetry Field 查询验收 -查询 GB32960 字段历史： +查询 GB32960 字段历史。该能力需要 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true` 并确认压测后无持续 lag： ```bash curl -sS 'http://127.0.0.1:20200/api/event-history/telemetry/fields?protocol=GB32960&vin=&fieldKey=&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&limit=10' ``` -查询 JT808 字段历史。没有 VIN 映射时，优先使用 `phone`： +查询 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=&fieldKey=location.speed_kmh&dateFrom=2026-06-29T00:00:00%2B08:00&dateTo=2026-06-29T23:59:59%2B08:00&limit=10' @@ -183,16 +207,18 @@ curl -sS 'http://127.0.0.1:20200/api/event-history/telemetry/fields?protocol=JT8 使用 TDengine CLI 或 JDBC 客户端检查超级表和子表是否创建： ```sql -USE vehicle_history; +USE vehicle_ts; SHOW STABLES; SELECT COUNT(*) FROM raw_frames; SELECT COUNT(*) FROM vehicle_locations; +-- 仅在 TDENGINE_TELEMETRY_FIELDS_ENABLED=true 时检查： SELECT COUNT(*) FROM telemetry_fields; ``` 预期： -- `raw_frames`、`vehicle_locations`、`telemetry_fields` 存在。 +- `raw_frames`、`vehicle_locations` 存在并持续增长。 +- `telemetry_fields` 表结构会创建；默认高吞吐模式不会持续增长，除非显式开启 `TDENGINE_TELEMETRY_FIELDS_ENABLED=true`。 - 有实时流量后，计数持续增加。 ## 失败语义 @@ -217,7 +243,7 @@ JT808 真实转发链路已验证： - `vehicle.raw.jt808.v1` 和 `vehicle.event.jt808.v1` 被 history 消费。 - `raw_frames` 中 `protocol='JT808'` 的真实行数已超过 `6000`。 - `vehicle_locations` 中 `protocol='JT808'` 的位置行数已超过 `1900`。 -- `telemetry_fields` 中 `protocol='JT808'` 的字段行数已超过 `15000`。 +- 在旧的字段宽表开启模式下，`telemetry_fields` 中 `protocol='JT808'` 的字段行数曾超过 `15000`；当前高吞吐默认不要求该表持续增长。 - 最新样例包含终端号 `13079963310`、消息 ID `512`、`archive://2026/06/29/JT808/...` 原始帧引用。 ## JT808 本机 smoke / 轻压测工具 @@ -250,6 +276,7 @@ python3 tools/jt808_e2e_smoke.py \ USE vehicle_ts; SELECT COUNT(*) FROM raw_frames WHERE protocol = 'JT808'; SELECT COUNT(*) FROM vehicle_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 @@ -273,7 +300,7 @@ GB32960 链路已用合成实时帧验证： - TCP `32960` 返回 GB32960 `2323` ACK。 - `raw_frames` 中测试 VIN `LTEST202606290001` 已有 `2` 条 raw 记录。 -- `telemetry_fields` 中测试 VIN `LTEST202606290001` 已有 `34` 条字段记录。 +- 在旧的字段宽表开启模式下，`telemetry_fields` 中测试 VIN `LTEST202606290001` 曾有 `34` 条字段记录；当前默认通过 GB32960 snapshot/fields 即时解析 raw，不要求该表增长。 - latest raw 包含 `archive://2026/06/29/GB32960/LTEST202606290001/...`。 - snapshots API 可查到 `speedKmh=62.4`、`totalMileageKm=223456.7`、`longitude=116.397128`、`latitude=39.916527`。 @@ -283,6 +310,7 @@ 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 @@ -292,7 +320,7 @@ SELECT event_time, vin, raw_uri, frame_id LIMIT 5; ``` -也可以直接运行仓库 smoke 工具验证 TCP `32960`、history records、TDengine 字段、TDengine `raw_frames` 和 raw archive： +也可以直接运行仓库 smoke 工具验证 TCP `32960`、GB32960 snapshot/fields、TDengine `raw_frames` 和 raw archive： ```bash export TDENGINE_REST_URL='http://:6041/rest/sql/vehicle_ts' @@ -306,7 +334,7 @@ python3 tools/gb32960_e2e_smoke.py \ --tdengine-password "$TDENGINE_PASSWORD" ``` -预期输出包含 `records >= 1`、`fieldCounts` 中关键字段均大于 0、`tdengineRawFrames` 等于本次可见唯一 `rawUri` 数量、`archiveChecked >= 1`。 +预期输出包含 `records >= 1`、`fieldCounts` 中关键字段均大于 0、`tdengineRawFrames` 等于本次可见唯一 `rawUri` 数量、`archiveChecked >= 1`。这里的 `records` 字段表示 snapshot 数量，用于兼容早期脚本输出名。 复查 API： diff --git a/modules/apps/gb32960-ingest-app/src/main/resources/application.yml b/modules/apps/gb32960-ingest-app/src/main/resources/application.yml index 0c9bc30d..25836691 100644 --- a/modules/apps/gb32960-ingest-app/src/main/resources/application.yml +++ b/modules/apps/gb32960-ingest-app/src/main/resources/application.yml @@ -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} diff --git a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/auth/Gb32960PlatformAuthorizer.java b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/auth/Gb32960PlatformAuthorizer.java index ed99cec3..9e78ab20 100644 --- a/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/auth/Gb32960PlatformAuthorizer.java +++ b/modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/auth/Gb32960PlatformAuthorizer.java @@ -62,7 +62,11 @@ public final class Gb32960PlatformAuthorizer { if (!Objects.equals(entry.getPassword(), password)) return Result.DENY_BAD_PASSWORD; Set 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; } diff --git a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960AccessServiceTest.java b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960AccessServiceTest.java index d4681cd6..f83da589 100644 --- a/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960AccessServiceTest.java +++ b/modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960AccessServiceTest.java @@ -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); diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java index 53653315..4a0dba6d 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameService.java @@ -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_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 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 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 records = store.query(new EventFileQuery( ProtocolId.GB32960, @@ -118,6 +161,40 @@ public final class Gb32960DecodedFrameService { return out; } + private List 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 page = tdengineReader.queryRawFrames(new TdengineRawFrameQuery( + "GB32960", + vin.trim(), + fromInstant(dateFrom, eventTimeFrom), + toExclusiveInstant(dateTo, eventTimeTo), + tdengineOrder(order), + frameLimit, + null)); + List out = new ArrayList<>(page.items().size()); + LinkedHashSet 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 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> 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 ""; diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java index 339ff4a8..0251a87a 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java @@ -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 store, + ObjectProvider 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 diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java index 2a9681ac..e5f9d07e 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java @@ -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 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 rows; + private TdengineRawFrameQuery lastQuery; + + private CapturingTdengineReader(List rows) { + this.rows = rows; + } + + @Override + public TdenginePage queryRawFrames(TdengineRawFrameQuery query) { + lastQuery = query; + return new TdenginePage<>(rows, java.util.Optional.empty()); + } + + @Override + public TdenginePage queryLocations(TdengineLocationQuery query) { + throw new UnsupportedOperationException(); + } + + @Override + public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) { + throw new UnsupportedOperationException(); + } + } } diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java index baa29aa6..3cf8cba1 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java @@ -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 diff --git a/tools/gb32960_e2e_smoke.py b/tools/gb32960_e2e_smoke.py index 35819dd5..8d884a5e 100644 --- a/tools/gb32960_e2e_smoke.py +++ b/tools/gb32960_e2e_smoke.py @@ -25,7 +25,7 @@ SHANGHAI = dt.timezone(dt.timedelta(hours=8)) DEFAULT_SAMPLE = ( "modules/protocols/protocol-gb32960/src/test/resources/samples/gb32960/realtime_001.hex" ) -DEFAULT_FIELD_KEYS = "speed_kmh,total_mileage_km,longitude,latitude" +DEFAULT_FIELD_KEYS = "VEHICLE.speedKmh,VEHICLE.totalMileageKm,POSITION_V2016.longitude,POSITION_V2016.latitude" @dataclass(frozen=True) @@ -104,7 +104,6 @@ def wait_for_records( ) -> list[dict[str, Any]]: deadline = time.monotonic() + timeout_seconds params = { - "protocol": "GB32960", "vin": vin, "dateFrom": date_from, "dateTo": date_to, @@ -113,7 +112,7 @@ def wait_for_records( } last_items: list[dict[str, Any]] = [] while time.monotonic() < deadline: - response = query_json(history_base_url, "/api/event-history/records", params, request_timeout) + response = query_json(history_base_url, "/api/event-history/gb32960/snapshots", params, request_timeout) last_items = response if isinstance(response, list) else [] if len(last_items) >= minimum: return last_items @@ -132,9 +131,8 @@ def wait_for_fields( ) -> list[dict[str, Any]]: deadline = time.monotonic() + timeout_seconds params = { - "protocol": "GB32960", "vin": vin, - "fieldKey": field_key, + "fields": field_key, "dateFrom": date_from, "dateTo": date_to, "limit": 10, @@ -142,9 +140,9 @@ def wait_for_fields( } last_items: list[dict[str, Any]] = [] while time.monotonic() < deadline: - response = query_json(history_base_url, "/api/event-history/telemetry/fields", params, request_timeout) - last_items = response.get("items") if isinstance(response, dict) else [] - if last_items: + response = query_json(history_base_url, "/api/event-history/gb32960/snapshots/fields", params, request_timeout) + last_items = response if isinstance(response, list) else [] + if any((item.get("fields") or {}).get(field_key) is not None for item in last_items): return last_items time.sleep(1) return last_items @@ -183,9 +181,8 @@ def archive_path(root: pathlib.Path, uri: str) -> pathlib.Path: def check_archive(root: pathlib.Path, records: list[dict[str, Any]], limit: int) -> int: checked = 0 seen: set[str] = set() - for record in records: - uri = record.get("rawArchiveUri") or "" - if not uri or uri in seen: + for uri in raw_uris(records): + if uri in seen: continue seen.add(uri) path = archive_path(root, uri) @@ -207,9 +204,8 @@ def raw_frame_count_sql(info: FrameInfo, records: list[dict[str, Any]]) -> str: def raw_frame_count_sqls(info: FrameInfo, records: list[dict[str, Any]]) -> list[str]: seen: set[str] = set() sqls: list[str] = [] - for record in records: - uri = record.get("rawArchiveUri") or "" - if not uri or uri in seen: + for uri in raw_uris(records): + if uri in seen: continue seen.add(uri) sqls.append(tdengine_smoke.raw_frame_count_sql( @@ -244,7 +240,8 @@ def verify_tdengine_raw_frames( def run(args: argparse.Namespace) -> dict[str, Any]: frame = read_hex_frame(pathlib.Path(args.frame_hex_file)) info = parse_frame_info(frame) - date_from, date_to = query_range(info.event_time, args.window_minutes) + query_center = dt.datetime.now(tz=SHANGHAI) + date_from, date_to = query_range(query_center, args.window_minutes) ack = send_frame(args.tcp_host, args.tcp_port, frame, args.tcp_timeout) if args.require_ack and not valid_ack(ack): @@ -298,6 +295,9 @@ def run(args: argparse.Namespace) -> dict[str, Any]: "vin": info.vin, "command": f"0x{info.command:02x}", "eventTime": info.event_time.isoformat() if info.event_time else None, + "queryTimeSource": "received", + "queryDateFrom": date_from, + "queryDateTo": date_to, "ackBytesHex": ack.hex(), "records": len(records), "fieldCounts": field_counts, @@ -308,11 +308,23 @@ def run(args: argparse.Namespace) -> dict[str, Any]: def first_raw_uri(records: list[dict[str, Any]]) -> str | None: + uris = raw_uris(records) + if uris: + return uris[0] + return None + + +def raw_uris(records: list[dict[str, Any]]) -> list[str]: + out: list[str] = [] for record in records: uri = record.get("rawArchiveUri") - if uri: - return uri - return None + if isinstance(uri, str) and uri: + out.append(uri) + for frame in record.get("sourceFrames") or []: + frame_uri = frame.get("rawArchiveUri") if isinstance(frame, dict) else "" + if frame_uri: + out.append(frame_uri) + return out def parser() -> argparse.ArgumentParser: