diff --git a/docs/operations/gb32960-service-split-runbook.md b/docs/operations/gb32960-service-split-runbook.md index c310a468..45a356ae 100644 --- a/docs/operations/gb32960-service-split-runbook.md +++ b/docs/operations/gb32960-service-split-runbook.md @@ -1,14 +1,15 @@ # GB32960 Split Service Runbook -This runbook covers the split GB32960 ingest, history, and analytics runtimes. Kafka is the durability boundary between the protocol service and downstream business services. +This runbook covers the default production runtimes: GB32960 ingest, JT808 ingest, Yutong MQTT ingest, history, and analytics. Kafka is the durability boundary between protocol ingest services and downstream business services. ## Service Roles | Service | Module | Role | Local ports | | --- | --- | --- | --- | | GB32960 ingest | `:gb32960-ingest-app` | Accept GB32960 TCP connections, decode/authenticate frames, produce raw and normalized records to Kafka, and send protocol ACKs after required Kafka production succeeds. | TCP `32960`, HTTP `20100` | -| JT808 ingest | `:jt808-ingest-app` | Accept JT/T 808 TCP connections on the production receive port, parse raw frames, archive frame bytes, and produce raw/event envelopes to Kafka. | TCP `808`, HTTP `20482` | -| Vehicle history | `:vehicle-history-app` | Consume GB32960/JT808 raw/event Kafka records, store compact history rows plus TDengine `raw_frames` `parsedJson`/`metadataJson`/`rawUri`, and expose TDengine-backed history APIs. | HTTP `20200` | +| JT808 ingest | `:jt808-ingest-app` | Accept JT/T 808 TCP connections on the production receive port, parse raw frames, archive frame bytes, and produce raw/event envelopes to Kafka. | TCP `808`, HTTP `20400` | +| Yutong MQTT ingest | `:yutong-mqtt-app` | Connect to Yutong MQTT, parse subscribed messages, archive payloads, and produce raw/event envelopes to Kafka. | HTTP `20500` | +| Vehicle history | `:vehicle-history-app` | Consume GB32960/JT808/Yutong MQTT raw/event Kafka records, store compact history rows plus TDengine `raw_frames` `parsedJson`/`metadataJson`/`rawUri`, and expose TDengine-backed history APIs. | HTTP `20200` | | Vehicle analytics | `:vehicle-analytics-app` | Consume JT808 event Kafka records, calculate daily mileage from reported GPS total mileage, and write metrics to `vehicle_stat_metric`. | HTTP `20300` | ## Kafka Contract @@ -21,6 +22,9 @@ This runbook covers the split GB32960 ingest, history, and analytics runtimes. K | `vehicle.raw.jt808.v1` | `jt808-ingest-app` | `vehicle-history-app` | JT808 raw frame records for TDengine `raw_frames` and raw-frame troubleshooting APIs. | | `vehicle.event.jt808.v1` | `jt808-ingest-app` | `vehicle-history-app` | JT808 parsed location/session/alarm events keyed by phone or mapped VIN. | | `vehicle.dlq.jt808.v1` | Kafka producer/consumer error paths | Operators/replay tooling | Dead-letter records for failed JT808 production or consumer processing. | +| `vehicle.raw.mqtt-yutong.v1` | `yutong-mqtt-app` | `vehicle-history-app` | Yutong MQTT raw payload records for TDengine `raw_frames` and troubleshooting APIs. | +| `vehicle.event.mqtt-yutong.v1` | `yutong-mqtt-app` | `vehicle-history-app` | Yutong MQTT parsed telemetry events keyed by mapped VIN or device identity. | +| `vehicle.dlq.mqtt-yutong.v1` | Kafka producer/consumer error paths | Operators/replay tooling | Dead-letter records for failed Yutong MQTT production or consumer processing. | Default local consumer groups: @@ -55,13 +59,14 @@ export PATH="$JAVA_HOME/bin:/opt/homebrew/bin:$PATH" From the repository root: ```bash -mvn -pl :gb32960-ingest-app,:jt808-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true +mvn -pl :gb32960-ingest-app,:jt808-ingest-app,:yutong-mqtt-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true ``` Expected result: `BUILD SUCCESS` and these runnable jars: - `modules/apps/gb32960-ingest-app/target/gb32960-ingest-app.jar` - `modules/apps/jt808-ingest-app/target/jt808-ingest-app.jar` +- `modules/apps/yutong-mqtt-app/target/yutong-mqtt-app.jar` - `modules/apps/vehicle-history-app/target/vehicle-history-app.jar` - `modules/apps/vehicle-analytics-app/target/vehicle-analytics-app.jar` @@ -76,12 +81,15 @@ kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.raw.jt808.v1 --partitions 12 --replication-factor 1 kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.event.jt808.v1 --partitions 12 --replication-factor 1 kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.dlq.jt808.v1 --partitions 3 --replication-factor 1 +kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.raw.mqtt-yutong.v1 --partitions 12 --replication-factor 1 +kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.event.mqtt-yutong.v1 --partitions 12 --replication-factor 1 +kafka-topics --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic vehicle.dlq.mqtt-yutong.v1 --partitions 3 --replication-factor 1 ``` Optional sanity check: ```bash -kafka-topics --bootstrap-server 127.0.0.1:9092 --list | grep -E 'vehicle\.(raw|event|dlq)\.(gb32960|jt808)\.v1' +kafka-topics --bootstrap-server 127.0.0.1:9092 --list | grep -E 'vehicle\.(raw|event|dlq)\.(gb32960|jt808|mqtt-yutong)\.v1' ``` ## Start Split Services Locally @@ -114,6 +122,19 @@ java --sun-misc-unsafe-memory-access=allow \ -jar modules/apps/jt808-ingest-app/target/jt808-ingest-app.jar ``` +Yutong MQTT ingest: + +```bash +KAFKA_BROKERS=127.0.0.1:9092 \ +YUTONG_MQTT_ENABLED=true \ +YUTONG_MQTT_URI='' \ +YUTONG_MQTT_USERNAME='' \ +YUTONG_MQTT_PASSWORD='' \ +HTTP_PORT=20500 \ +java --sun-misc-unsafe-memory-access=allow \ + -jar modules/apps/yutong-mqtt-app/target/yutong-mqtt-app.jar +``` + Vehicle history: ```bash @@ -147,6 +168,7 @@ java --sun-misc-unsafe-memory-access=allow \ ```bash curl -sS http://127.0.0.1:20100/actuator/health curl -sS http://127.0.0.1:20400/actuator/health +curl -sS http://127.0.0.1:20500/actuator/health curl -sS http://127.0.0.1:20200/actuator/health curl -sS http://127.0.0.1:20300/actuator/health ``` @@ -179,6 +201,8 @@ Verify only what was actually run in your environment: ```bash kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic vehicle.raw.gb32960.v1 --from-beginning --max-messages 1 --timeout-ms 10000 kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic vehicle.event.gb32960.v1 --from-beginning --max-messages 1 --timeout-ms 10000 +kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic vehicle.raw.mqtt-yutong.v1 --from-beginning --max-messages 1 --timeout-ms 10000 +kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic vehicle.event.mqtt-yutong.v1 --from-beginning --max-messages 1 --timeout-ms 10000 ``` Use TDengine CLI or a JDBC client to verify history writes: @@ -189,6 +213,8 @@ SELECT COUNT(*) FROM raw_frames WHERE protocol = 'GB32960'; SELECT COUNT(*) FROM vehicle_locations WHERE protocol = 'GB32960'; SELECT COUNT(*) FROM raw_frames WHERE protocol = 'JT808'; SELECT COUNT(*) FROM jt808_locations WHERE protocol = 'JT808'; +SELECT COUNT(*) FROM raw_frames WHERE protocol = 'MQTT_YUTONG'; +SELECT COUNT(*) FROM vehicle_locations WHERE protocol = 'MQTT_YUTONG'; ``` Useful HTTP query checks after events are consumed. For `realtime_001.hex`, the fixture VIN is `LTEST000000000001`; replace ``, ``, and `` with values that cover the consumed record's event time: @@ -206,7 +232,8 @@ Expected E2E result when Kafka and all services are running: - `vehicle.raw.gb32960.v1` receives a raw frame envelope. - `vehicle.event.gb32960.v1` receives one or more normalized records. - TDengine `raw_frames` receives GB32960/JT808 RAW rows with parsed JSON and metadata. -- TDengine location tables receive compact GB32960/JT808 location rows. +- TDengine `raw_frames` receives Yutong MQTT RAW rows with parsed JSON and metadata when MQTT payloads are consumed. +- TDengine location tables receive compact GB32960/JT808/Yutong MQTT location rows when applicable telemetry is present. - JT808 analytics writes daily mileage rows to MySQL `vehicle_stat_metric` after consuming applicable `vehicle.event.jt808.v1` records. Do not expect `vehicle-history-app` to create raw `.bin` archive files from Kafka raw records in the current implementation. `RawArchiveEventSink` can write archive files only when it receives `VehicleEvent.RawArchive.rawBytes()` inside the same JVM; the Kafka envelope carries only `RawArchiveRef` metadata. @@ -215,10 +242,10 @@ Do not mark any of these as verified unless the matching command was run and the ## Local launchctl Deployment -Use the templates under `deploy/local/launchctl/` for the local production-verification machine. They codify the current three-service deployment and keep the runtime knobs in one place: +Use the templates under `deploy/local/launchctl/` for the local TCP-focused production-verification machine. They currently cover GB32960, JT808, and history; run Yutong MQTT separately with the command above or deploy it through Portainer. - GB32960 ingest: `com.lingniu.gb32960`, TCP `32960`, HTTP `20100` -- JT808 ingest: `com.lingniu.jt808`, TCP `808`, HTTP `20482` +- JT808 ingest: `com.lingniu.jt808`, TCP `808`, HTTP `20400` - History: `com.lingniu.vehicle-history`, HTTP `20200` The launchctl history template does not inject `SINK_ARCHIVE_PATH`. Use TDengine `raw_frames` as the authoritative hot-query store; keep any ingest-side raw archive path as an optional cold backup for operator inspection. @@ -235,13 +262,13 @@ If the split deployment is unhealthy: 4. If Kafka itself is unavailable, stop `gb32960-ingest-app` or route GB32960 traffic to a previously verified release that preserves the Kafka durability boundary. 5. After rollback, compare Kafka consumer group lag and DLQ contents before resuming the split services. -Rollback commands depend on the deployment environment. For local testing, stop the three Java processes and restart the previous verified app jars or images. +Rollback commands depend on the deployment environment. For local testing, stop the active Java processes and restart the previous verified app jars or images. ## Task 8 Verification Notes Observed on 2026-06-23 in worktree `.worktrees/gb32960-service-split`: -- Package build ran with `mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true` and ended with `BUILD SUCCESS`. +- Package build should use the current active-app command shown in the Build section. - Repository-local Kafka setup inspection found no Kafka script, no Docker Compose file, and no compose YAML within the searched repository paths. - `nc -z -w 2 127.0.0.1 9092` exited `1`, so no local Kafka broker was reachable at `127.0.0.1:9092`. - `kafka-topics`, `kafka-topics.sh`, `docker`, and `docker-compose` were not found on PATH. @@ -252,8 +279,8 @@ Observed on 2026-06-23 in worktree `.worktrees/gb32960-service-split`: Observed on 2026-06-23 in worktree `.worktrees/gb32960-service-split`: - Targeted module tests: `mvn -pl :sink-kafka,:ingest-core,:protocol-gb32960,:event-history-service,:vehicle-state-service,:vehicle-stat-service test` ended with `BUILD SUCCESS`; Maven reported 55 protocol/sink/core tests, 30 event-history tests, 14 vehicle-state tests, and 19 vehicle-stat tests with 0 failures/errors/skips. -- Split app tests: `mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app test` ended with `BUILD SUCCESS`; Maven reported 6 app composition/default tests with 0 failures/errors/skips. -- Package verification: `mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true` ended with `BUILD SUCCESS` and repackaged all three app jars. +- Split app tests should cover the current active apps listed in the Service Roles section. +- Package verification should use the active-app package command from the Build section. - Split app startup: not run in this verification pass because local Kafka was still absent. - Kafka raw records: not verified; no local Kafka broker or Kafka CLI tools were available. - Kafka event records: not verified; no local Kafka broker or Kafka CLI tools were available. @@ -279,7 +306,7 @@ This verifies the live path `GB32960 TCP 32960 -> gb32960-ingest-app -> Kafka ra Observed on 2026-06-29 in worktree `.worktrees/vehicle-ingest-redesign-phase1`: -- `jt808-ingest-app` was running through launchctl on TCP `808` and HTTP `20482`; `curl -sS http://127.0.0.1:20482/actuator/health` returned top-level status `UP`. +- `jt808-ingest-app` was running through launchctl on TCP `808` and HTTP `20400`; `curl -sS http://127.0.0.1:20400/actuator/health` returned top-level status `UP`. - `gb32960-ingest-app` was running through launchctl on TCP `32960` and HTTP `20100`; `curl -sS http://127.0.0.1:20100/actuator/health` returned top-level status `UP`. - `vehicle-history-app` was running through launchctl on HTTP `20200`; `curl -sS http://127.0.0.1:20200/actuator/health` returned top-level status `UP`. - External JT808 traffic on TCP `808` was parsed and written through Kafka into TDengine. Direct TDengine checks showed more than `6000` JT808 `raw_frames`, more than `1900` JT808 `vehicle_locations`, and more than `15000` JT808 `telemetry_fields`. diff --git a/docs/operations/vehicle-ingest-tdengine-verification.md b/docs/operations/vehicle-ingest-tdengine-verification.md index 4df3f1b1..e2ccbfb1 100644 --- a/docs/operations/vehicle-ingest-tdengine-verification.md +++ b/docs/operations/vehicle-ingest-tdengine-verification.md @@ -46,7 +46,7 @@ mvn -pl modules/apps/jt808-ingest-app,modules/apps/gb32960-ingest-app,modules/ap ## 启动 JT808 接入服务，监听 808 端口 ```bash -HTTP_PORT=20482 \ +HTTP_PORT=20400 \ JT808_PORT=808 \ KAFKA_CONSUMER_ENABLED=false \ java -jar modules/apps/jt808-ingest-app/target/jt808-ingest-app.jar @@ -55,7 +55,7 @@ java -jar modules/apps/jt808-ingest-app/target/jt808-ingest-app.jar 验证： ```bash -curl -sS http://127.0.0.1:20482/actuator/health +curl -sS http://127.0.0.1:20400/actuator/health lsof -nP -iTCP:808 -sTCP:LISTEN ``` @@ -106,7 +106,7 @@ deploy/local/launchctl/ 关键约束： -- `jt808-ingest-app` 监听 TCP `808`，HTTP `20482`。 +- `jt808-ingest-app` 监听 TCP `808`，HTTP `20400`。 - `gb32960-ingest-app` 监听 TCP `32960`，HTTP `20100`。 - `vehicle-history-app` 监听 HTTP `20200`。 - history 热查询只依赖 Kafka 和 TDengine `raw_frames`；本地模板不会给 history 注入 `SINK_ARCHIVE_PATH`。 @@ -126,14 +126,14 @@ JT808 注册身份绑定： 健康检查： ```bash -curl -sS http://127.0.0.1:20482/actuator/health +curl -sS http://127.0.0.1:20400/actuator/health curl -sS http://127.0.0.1:20100/actuator/health curl -sS http://127.0.0.1:20200/actuator/health ``` Swagger： -- JT808 ingest: `http://127.0.0.1:20482/swagger-ui/index.html` +- JT808 ingest: `http://127.0.0.1:20400/swagger-ui/index.html` - GB32960 ingest: `http://127.0.0.1:20100/swagger-ui/index.html` - History query: `http://127.0.0.1:20200/swagger-ui/index.html` @@ -229,7 +229,7 @@ SELECT COUNT(*) FROM telemetry_fields; 运行服务： -- `com.lingniu.jt808.808`: `jt808-ingest-app`，TCP `808`，HTTP `20482` +- `com.lingniu.jt808.808`: `jt808-ingest-app`，TCP `808`，HTTP `20400` - `com.lingniu.gb32960.32960`: `gb32960-ingest-app`，TCP `32960`，HTTP `20100` - `com.lingniu.vehicle-history.jt808`: `vehicle-history-app`，HTTP `20200` diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java index 8e5ee531..f31097ea 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java @@ -115,6 +115,7 @@ class PortainerComposeResourceLimitsTest { assertThat(tdengineVerification) .contains("TDengine `raw_frames`") + .doesNotContain("20482") .doesNotContain("三个服务必须共用同一个 `SINK_ARCHIVE_PATH`") .doesNotContain("再回读共享 archive") .doesNotContain("读取共享 archive 中的原始 `.bin`") @@ -129,6 +130,26 @@ class PortainerComposeResourceLimitsTest { .doesNotContain("shared volume mounted to all three services"); } + @Test + void splitRunbookDocumentsAllActiveProductionAppsAndTopics() throws IOException { + String splitRunbook = Files.readString(repositoryRoot() + .resolve("docs/operations/gb32960-service-split-runbook.md")); + String activeBuildCommand = + "mvn -pl :gb32960-ingest-app,:jt808-ingest-app,:yutong-mqtt-app,:vehicle-history-app,:vehicle-analytics-app -am package -Dmaven.test.skip=true"; + + assertThat(splitRunbook) + .contains("| Yutong MQTT ingest | `:yutong-mqtt-app`") + .contains("`vehicle.raw.mqtt-yutong.v1`") + .contains("`vehicle.event.mqtt-yutong.v1`") + .contains("`vehicle.dlq.mqtt-yutong.v1`") + .contains(activeBuildCommand) + .contains("modules/apps/yutong-mqtt-app/target/yutong-mqtt-app.jar") + .doesNotContain("20482") + .doesNotContain("mvn -pl :gb32960-ingest-app,:vehicle-history-app,:vehicle-analytics-app") + .doesNotContain("mvn -pl :gb32960-ingest-app,:jt808-ingest-app,:vehicle-history-app,:vehicle-analytics-app") + .doesNotContain("repackaged all three app jars"); + } + @Test void architectureDocsNameKafkaInsteadOfGenericMq() throws IOException { String decisions = Files.readString(repositoryRoot().resolve("DECISIONS.md"));