Compare commits

...

2 Commits

Author SHA1 Message Date
lingniu
3e19c72915 refactor: store jt808 daily mileage as single metric 2026-07-01 06:25:33 +08:00
lingniu
3d4676adf3 docs: use tdengine checks in split runbook 2026-07-01 06:25:20 +08:00
6 changed files with 165 additions and 63 deletions

View File

@@ -115,7 +115,12 @@ Vehicle history:
```bash
KAFKA_BROKERS=127.0.0.1:9092 \
HTTP_PORT=20200 \
EVENT_FILE_STORE_PATH=./target/split-event-store \
TDENGINE_HISTORY_ENABLED=true \
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
```
@@ -170,14 +175,23 @@ 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
find target/split-event-store -type f | head
find target/split-vehicle-stat -type f | head
```
Use TDengine CLI or a JDBC client to verify history writes:
```sql
USE vehicle_ts;
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';
```
Useful HTTP query checks after events are consumed. For `realtime_001.hex`, the fixture VIN is `LTEST000000000001`; replace `<date-from>`, `<date-to>`, and `<stat-date>` with values that cover the consumed record's event time:
```bash
curl -sS 'http://127.0.0.1:20200/api/event-history/records?protocol=GB32960&dateFrom=<date-from>&dateTo=<date-to>&vin=LTEST000000000001&limit=10'
curl -sS 'http://127.0.0.1:20200/api/event-history/locations?protocol=GB32960&dateFrom=<date-from>&dateTo=<date-to>&vin=LTEST000000000001&limit=10'
curl -sS 'http://127.0.0.1:20200/api/event-history/raw-frames?protocol=GB32960&dateFrom=<date-from>&dateTo=<date-to>&vin=LTEST000000000001&limit=10'
curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/dictionary'
curl -sS 'http://127.0.0.1:20300/api/vehicle-stat/LTEST000000000001/daily?date=<stat-date>'
```
@@ -185,9 +199,10 @@ curl -sS 'http://127.0.0.1:20300/api/vehicle-stat/LTEST000000000001/daily?date=<
Expected E2E result when Kafka and all services are running:
- A GB32960 client receives a binary success ACK only after required Kafka production succeeds, and the captured ACK bytes are inspected.
- `vehicle.raw.gb32960.v1` receives a raw archive reference envelope.
- `vehicle.raw.gb32960.v1` receives a raw frame envelope.
- `vehicle.event.gb32960.v1` receives one or more normalized records.
- `target/split-event-store` receives event-store files after the history service flushes consumed Kafka records, including raw archive reference records.
- TDengine `raw_frames` receives GB32960/JT808 RAW rows with parsed JSON and metadata.
- TDengine location tables receive compact GB32960/JT808 location rows.
- 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.
@@ -226,7 +241,7 @@ Observed on 2026-06-23 in worktree `.worktrees/gb32960-service-split`:
- 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.
- Kafka topic creation, service startup, health checks, Kafka record checks, archive checks, event-store checks, stat output checks, and ACK observation were not run because the local Kafka prerequisite was absent.
- Kafka topic creation, service startup, health checks, Kafka record checks, archive checks, TDengine checks, stat output checks, and ACK observation were not run because the local Kafka prerequisite was absent.
## Latest Local Verification
@@ -238,7 +253,7 @@ Observed on 2026-06-23 in worktree `.worktrees/gb32960-service-split`:
- 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.
- Event-store files: not verified; downstream services were not started without Kafka.
- TDengine history rows: not verified; downstream services were not started without Kafka.
- Analytics output: not verified; downstream services were not started without Kafka.
- ACK behavior: not verified against a live broker in this pass. Unit tests cover the GB32960 ACK boundary and ordering, but an operator should still run the ACK capture command above in a Kafka-backed local or staging environment.

View File

@@ -6,7 +6,7 @@ The analytics app can calculate daily mileage from JT808 telemetry only. It cons
## Storage
There is no separate JT808 daily-mileage table and no Redis mileage state. The current day start total, latest total, and derived daily mileage are all stored in the common JDBC/MySQL metric table `vehicle_stat_metric`.
There is no separate JT808 daily-mileage table and no Redis mileage state. The derived value is stored as one `daily_mileage_km` metric row in the common JDBC/MySQL table `vehicle_stat_metric`; the first and latest GPS total mileage values are kept on that same row as calculation source columns.
The JT808 daily-mileage value is calculated from the GPS total mileage reported in location additional information:
@@ -15,7 +15,7 @@ daily_mileage_km = last_total_mileage_km - first_total_mileage_km
calculation_method = JT808_TOTAL_MILEAGE_DIFF
```
The first valid JT808 location point for a vehicle and local day stores the day start total and writes `daily_mileage_km=0.0`. Later ordered points update the latest total and the derived `daily_mileage_km`. If a later total mileage is lower than the stored start or latest total, the update is ignored and the previous metric remains.
The first valid JT808 location point for a vehicle and local day stores the day start total on the metric row and writes `daily_mileage_km=0.0`. Later ordered points update the latest total on the same row and rewrite the derived `daily_mileage_km`. If a later total mileage is lower than the stored latest total, the previous latest total remains.
## Runtime Settings
@@ -29,4 +29,4 @@ MYSQL_USERNAME=<user>
MYSQL_PASSWORD=<password>
```
Algorithm defaults use the Asia/Shanghai daily boundary. Restart recovery reads the same `vehicle_stat_metric` rows, so no separate mileage state store is required.
Algorithm defaults use the Asia/Shanghai daily boundary. Restart recovery reads the same `daily_mileage_km` metric row, so no separate mileage state store is required.

View File

@@ -125,12 +125,12 @@ calculation_method = JT808_TOTAL_MILEAGE_DIFF
```
- Store the metric in `vehicle_stat_metric` with
`metric_key = daily_mileage_km`.
`metric_key = daily_mileage_km`; the first and latest GPS total mileage used
for the subtraction stay on the same metric row as calculation source columns.
There is no JT808-specific daily mileage table and no in-memory production
state-store mode. Daily start total, latest total, and derived mileage are all
stored in `vehicle_stat_metric`, so restart recovery does not require a separate
Redis mileage state.
state-store mode. Restart recovery reads the same metric row, so it does not
require a separate Redis mileage state.
### Latest State

View File

@@ -107,6 +107,21 @@ class PortainerComposeResourceLimitsTest {
.doesNotContain("Vehicle state can also be enabled from this runtime");
}
@Test
void splitRunbookUsesTdengineHistoryVerificationInsteadOfEventFileStore() throws IOException {
String splitRunbook = Files.readString(repositoryRoot()
.resolve("docs/operations/gb32960-service-split-runbook.md"));
assertThat(splitRunbook)
.contains("TDENGINE_HISTORY_ENABLED=true")
.contains("SELECT COUNT(*) FROM raw_frames")
.contains("SELECT COUNT(*) FROM vehicle_locations")
.doesNotContain("EVENT_FILE_STORE_PATH")
.doesNotContain("target/split-event-store")
.doesNotContain("event-store files")
.doesNotContain("/api/event-history/records?");
}
@Test
void vehicleAnalyticsComposeOnlyExposesStatConsumerEnvironment() throws IOException {
String compose = Files.readString(repositoryRoot().resolve("deploy/portainer/docker-compose.yml"));

View File

@@ -1,9 +1,12 @@
package com.lingniu.ingest.vehiclestat;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.ConnectionCallback;
import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.Date;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.LocalDate;
import java.util.List;
import java.util.Optional;
@@ -15,6 +18,8 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
private static final String DAILY_MILEAGE_UNIT = "km";
private static final String DAILY_MILEAGE_START_TOTAL_KEY = "daily_mileage_start_total_km";
private static final String DAILY_MILEAGE_LATEST_TOTAL_KEY = "daily_mileage_latest_total_km";
private static final String FIRST_TOTAL_COLUMN = "first_total_mileage_km";
private static final String LATEST_TOTAL_COLUMN = "latest_total_mileage_km";
private final JdbcTemplate jdbcTemplate;
@@ -26,17 +31,6 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
ensureSchema();
}
private void saveDailyMileageMetric(VehicleDailyStatResult result) {
if (result == null || result.dailyMileageKm().isEmpty()) {
return;
}
String vin = clean(result.vin());
Date statDate = Date.valueOf(result.statDate());
double value = result.dailyMileageKm().getAsDouble();
String strategy = result.dailyMileageStrategy().name();
upsertMetric(vin, statDate, DAILY_MILEAGE_KEY, value, DAILY_MILEAGE_UNIT, strategy);
}
@Override
public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
String normalizedVin = clean(vin);
@@ -63,34 +57,48 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
String normalizedVin = clean(vin);
Date date = Date.valueOf(statDate);
String strategy = DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF.name();
OptionalDouble existingMin = findMetric(normalizedVin, date, DAILY_MILEAGE_START_TOTAL_KEY);
double minTotalMileage = existingMin.isPresent()
? Math.min(existingMin.getAsDouble(), totalMileageKm)
: totalMileageKm;
if (existingMin.isEmpty() || Double.compare(minTotalMileage, existingMin.getAsDouble()) != 0) {
upsertMetric(normalizedVin, date, DAILY_MILEAGE_START_TOTAL_KEY, minTotalMileage,
DAILY_MILEAGE_UNIT, strategy);
}
OptionalDouble existingMax = findMetric(normalizedVin, date, DAILY_MILEAGE_LATEST_TOTAL_KEY);
double maxTotalMileage = existingMax.isPresent()
? Math.max(existingMax.getAsDouble(), totalMileageKm)
: totalMileageKm;
if (existingMax.isEmpty() || Double.compare(maxTotalMileage, existingMax.getAsDouble()) != 0) {
upsertMetric(normalizedVin, date, DAILY_MILEAGE_LATEST_TOTAL_KEY, maxTotalMileage,
DAILY_MILEAGE_UNIT, strategy);
}
double dailyMileageKm = maxTotalMileage - minTotalMileage;
DailyMileageState existingState = findDailyMileageState(normalizedVin, date)
.orElseGet(() -> legacyDailyMileageState(normalizedVin, date).orElse(null));
double firstTotalMileage = existingState == null || !Double.isFinite(existingState.firstTotalMileageKm())
? totalMileageKm
: existingState.firstTotalMileageKm();
double latestTotalMileage = existingState == null || !Double.isFinite(existingState.latestTotalMileageKm())
? totalMileageKm
: Math.max(existingState.latestTotalMileageKm(), totalMileageKm);
double dailyMileageKm = latestTotalMileage - firstTotalMileage;
upsertDailyMileageMetric(normalizedVin, date, dailyMileageKm, DAILY_MILEAGE_UNIT, strategy,
firstTotalMileage, latestTotalMileage);
deleteLegacyMileageStateMetrics(normalizedVin, date);
VehicleDailyStatResult result = new VehicleDailyStatResult(
normalizedVin,
statDate,
OptionalDouble.of(dailyMileageKm),
DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF);
saveDailyMileageMetric(result);
return Optional.of(result);
}
private Optional<DailyMileageState> findDailyMileageState(String vin, Date statDate) {
List<DailyMileageState> rows = jdbcTemplate.query("""
SELECT first_total_mileage_km, latest_total_mileage_km
FROM vehicle_stat_metric
WHERE vin = ? AND stat_date = ? AND metric_key = ?
""", (rs, rowNum) -> new DailyMileageState(
nullableDouble(rs, FIRST_TOTAL_COLUMN),
nullableDouble(rs, LATEST_TOTAL_COLUMN)), vin, statDate, DAILY_MILEAGE_KEY);
return rows.stream()
.filter(DailyMileageState::hasCompleteState)
.findFirst();
}
private Optional<DailyMileageState> legacyDailyMileageState(String vin, Date statDate) {
OptionalDouble start = findMetric(vin, statDate, DAILY_MILEAGE_START_TOTAL_KEY);
OptionalDouble latest = findMetric(vin, statDate, DAILY_MILEAGE_LATEST_TOTAL_KEY);
if (start.isEmpty() || latest.isEmpty()) {
return Optional.empty();
}
return Optional.of(new DailyMileageState(start.getAsDouble(), latest.getAsDouble()));
}
private OptionalDouble findMetric(String vin, Date statDate, String metricKey) {
List<Double> rows = jdbcTemplate.query("""
SELECT metric_value
@@ -105,25 +113,38 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
return OptionalDouble.of(rows.getFirst());
}
private void upsertMetric(String vin, Date statDate, String metricKey, double value, String unit, String strategy) {
private void upsertDailyMileageMetric(String vin, Date statDate, double value, String unit, String strategy,
double firstTotalMileageKm, double latestTotalMileageKm) {
try {
jdbcTemplate.update("""
INSERT INTO vehicle_stat_metric
(vin, stat_date, metric_key, metric_value, metric_unit, calculation_method)
VALUES (?, ?, ?, ?, ?, ?)
""", vin, statDate, metricKey, value, unit, strategy);
(vin, stat_date, metric_key, metric_value, metric_unit, calculation_method,
first_total_mileage_km, latest_total_mileage_km)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", vin, statDate, DAILY_MILEAGE_KEY, value, unit, strategy,
firstTotalMileageKm, latestTotalMileageKm);
} catch (DuplicateKeyException ex) {
jdbcTemplate.update("""
UPDATE vehicle_stat_metric
SET metric_value = ?,
metric_unit = ?,
calculation_method = ?,
first_total_mileage_km = ?,
latest_total_mileage_km = ?,
updated_at = CURRENT_TIMESTAMP
WHERE vin = ? AND stat_date = ? AND metric_key = ?
""", value, unit, strategy, vin, statDate, metricKey);
""", value, unit, strategy, firstTotalMileageKm, latestTotalMileageKm,
vin, statDate, DAILY_MILEAGE_KEY);
}
}
private void deleteLegacyMileageStateMetrics(String vin, Date statDate) {
jdbcTemplate.update("""
DELETE FROM vehicle_stat_metric
WHERE vin = ? AND stat_date = ? AND metric_key IN (?, ?)
""", vin, statDate, DAILY_MILEAGE_START_TOTAL_KEY, DAILY_MILEAGE_LATEST_TOTAL_KEY);
}
private void ensureSchema() {
jdbcTemplate.execute("""
CREATE TABLE IF NOT EXISTS vehicle_stat_metric (
@@ -133,11 +154,43 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
metric_value DECIMAL(18,6) NULL,
metric_unit VARCHAR(16) NOT NULL,
calculation_method VARCHAR(64) NOT NULL,
first_total_mileage_km DECIMAL(18,6) NULL,
latest_total_mileage_km DECIMAL(18,6) NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (vin, stat_date, metric_key)
)
""");
ensureColumn(FIRST_TOTAL_COLUMN);
ensureColumn(LATEST_TOTAL_COLUMN);
}
private void ensureColumn(String columnName) {
if (columnExists(columnName)) {
return;
}
jdbcTemplate.execute("ALTER TABLE vehicle_stat_metric ADD COLUMN " + columnName + " DECIMAL(18,6) NULL");
}
private boolean columnExists(String columnName) {
ConnectionCallback<Boolean> callback = connection -> {
try (ResultSet columns = connection.getMetaData().getColumns(
connection.getCatalog(), null, "vehicle_stat_metric", columnName)) {
if (columns.next()) {
return true;
}
}
try (ResultSet columns = connection.getMetaData().getColumns(
connection.getCatalog(), null, "VEHICLE_STAT_METRIC", columnName.toUpperCase())) {
return columns.next();
}
};
return Boolean.TRUE.equals(jdbcTemplate.execute(callback));
}
private static double nullableDouble(ResultSet rs, String columnName) throws SQLException {
double value = rs.getDouble(columnName);
return rs.wasNull() ? Double.NaN : value;
}
private static String clean(String value) {
@@ -146,4 +199,10 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
}
return value.trim();
}
private record DailyMileageState(double firstTotalMileageKm, double latestTotalMileageKm) {
boolean hasCompleteState() {
return Double.isFinite(firstTotalMileageKm) && Double.isFinite(latestTotalMileageKm);
}
}
}

View File

@@ -27,13 +27,23 @@ class JdbcVehicleStatMetricRepositoryTest {
.extracting(result -> result.dailyMileageKm().orElseThrow())
.isEqualTo(15.5);
assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM vehicle_stat_metric", Integer.class))
.isEqualTo(3);
.isOne();
assertThat(jdbcTemplate.queryForObject(
"SELECT metric_key FROM vehicle_stat_metric WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'",
String.class)).isEqualTo("daily_mileage_km");
assertThat(jdbcTemplate.queryForObject(
"SELECT calculation_method FROM vehicle_stat_metric WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'",
String.class)).isEqualTo("JT808_TOTAL_MILEAGE_DIFF");
assertThat(jdbcTemplate.queryForObject("""
SELECT first_total_mileage_km
FROM vehicle_stat_metric
WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
""", Double.class)).isEqualTo(100.0);
assertThat(jdbcTemplate.queryForObject("""
SELECT latest_total_mileage_km
FROM vehicle_stat_metric
WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
""", Double.class)).isEqualTo(115.5);
assertThat(jdbcTemplate.queryForObject("""
SELECT COUNT(*)
FROM information_schema.tables
@@ -58,7 +68,7 @@ class JdbcVehicleStatMetricRepositoryTest {
assertThat(second.dailyMileageKm()).hasValue(8.25);
assertThat(jdbcTemplate.queryForObject(
"SELECT COUNT(*) FROM vehicle_stat_metric WHERE vin = 'VIN001' AND stat_date = DATE '2026-07-01'",
Integer.class)).isEqualTo(3);
Integer.class)).isOne();
assertThat(jdbcTemplate.queryForObject("""
SELECT metric_value
FROM vehicle_stat_metric
@@ -70,30 +80,33 @@ class JdbcVehicleStatMetricRepositoryTest {
}
@Test
void recalculatesJt808MileageFromDailyMinAndMaxTotalsWhenSamplesArriveOutOfOrder() {
void recalculatesJt808MileageFromSingleMetricRowAfterRestart() {
JdbcTemplate jdbcTemplate = new JdbcTemplate(new DriverManagerDataSource(
"jdbc:h2:mem:vehicle_stat_metric_jt808_out_of_order;MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1",
"sa",
""));
JdbcVehicleStatMetricRepository repository = new JdbcVehicleStatMetricRepository(jdbcTemplate);
repository.recordDailyMileageSample("VIN001", LocalDate.of(2026, 7, 1), 1008.75).orElseThrow();
VehicleDailyStatResult lowerSample = repository.recordDailyMileageSample(
"VIN001", LocalDate.of(2026, 7, 1), 1000.5).orElseThrow();
VehicleDailyStatResult higherSample = repository.recordDailyMileageSample(
repository.recordDailyMileageSample("VIN001", LocalDate.of(2026, 7, 1), 1000.5).orElseThrow();
JdbcVehicleStatMetricRepository restartedRepository = new JdbcVehicleStatMetricRepository(jdbcTemplate);
VehicleDailyStatResult higherSample = restartedRepository.recordDailyMileageSample(
"VIN001", LocalDate.of(2026, 7, 1), 1010.0).orElseThrow();
assertThat(lowerSample.dailyMileageKm()).hasValue(8.25);
assertThat(higherSample.dailyMileageKm()).hasValue(9.5);
assertThat(jdbcTemplate.queryForObject("""
SELECT metric_value
SELECT COUNT(*)
FROM vehicle_stat_metric
WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_start_total_km'
WHERE vin = 'VIN001' AND stat_date = DATE '2026-07-01'
""", Integer.class)).isOne();
assertThat(jdbcTemplate.queryForObject("""
SELECT first_total_mileage_km
FROM vehicle_stat_metric
WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
""", Double.class)).isEqualTo(1000.5);
assertThat(jdbcTemplate.queryForObject("""
SELECT metric_value
SELECT latest_total_mileage_km
FROM vehicle_stat_metric
WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_latest_total_km'
WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
""", Double.class)).isEqualTo(1010.0);
assertThat(jdbcTemplate.queryForObject("""
SELECT metric_value