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

feat: store jt808 daily mileage as metric

Browse Source
This commit is contained in:
lingniu
2026-07-01 01:39:13 +08:00
parent af899d03b7
commit c6a647ce36
9 changed files with 210 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
docs/operations/jt808-daily-mileage-runbook.md
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. Runtime state is stored under the configured JT808 state store so the stream can continue after restart. The derived metric is written through `VehicleStatRepository.saveDailyStat(...)`; the default repository persists it under `VEHICLE_STAT_FILE_PATH` as `daily-stats.tsv`.
There is no separate JT808 daily-mileage table. Runtime state is stored under the configured JT808 state store so the stream can continue after restart. The derived metric is written through `VehicleStatRepository.saveDailyStat(...)`; production uses the common JDBC/MySQL metric table `vehicle_stat_metric`. Local development can force the file fallback with `VEHICLE_STAT_REPOSITORY_TYPE=file`, which writes `daily-stats.tsv` under `VEHICLE_STAT_FILE_PATH`.
The JT808 daily-mileage value is calculated from the GPS total mileage reported in location additional information:
@@ -23,8 +23,11 @@ Set these in Portainer or Nacos, without committing secrets:
```text
KAFKA_TOPIC_JT808_EVENT=vehicle.event.jt808.v1
VEHICLE_STAT_JT808_MILEAGE_ENABLED=true
VEHICLE_STAT_REPOSITORY_TYPE=jdbc
VEHICLE_STAT_JT808_STATE_STORE=redis
VEHICLE_STAT_FILE_PATH=/data/vehicle-stat
MYSQL_JDBC_URL=<jdbc-url>
MYSQL_USERNAME=<user>
MYSQL_PASSWORD=<password>
REDIS_HOST=<host>
REDIS_PORT=6379
REDIS_DATABASE=50

8
docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md
View File

@@ -10,7 +10,8 @@ Consume JT808 Kafka location events, keep rolling daily state, and write the der
- Runtime app: `vehicle-analytics-app`
- State: Redis or in-memory `Jt808MileageStateStore`
- Metric output: `VehicleStatRepository.saveDailyStat(...)`
- Default metric storage: `VEHICLE_STAT_FILE_PATH/daily-stats.tsv`
- Production metric storage: JDBC/MySQL `vehicle_stat_metric`
- Local fallback storage: `VEHICLE_STAT_FILE_PATH/daily-stats.tsv`
- Date boundary: `Asia/Shanghai`
JT808 daily mileage is calculated only from the GPS total mileage reported by JT808 location additional information:
@@ -27,8 +28,11 @@ The stream saves a metric only when at least two ordered location points for the
KAFKA_TOPIC_JT808_EVENT=vehicle.event.jt808.v1
VEHICLE_STAT_ENABLED=true
VEHICLE_STAT_JT808_MILEAGE_ENABLED=true
VEHICLE_STAT_REPOSITORY_TYPE=jdbc
VEHICLE_STAT_JT808_STATE_STORE=redis
VEHICLE_STAT_FILE_PATH=/data/vehicle-stat
MYSQL_JDBC_URL=<jdbc-url>
MYSQL_USERNAME=<user>
MYSQL_PASSWORD=<password>
REDIS_HOST=<host>
REDIS_PORT=6379
REDIS_DATABASE=50

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

@@ -78,6 +78,7 @@ lingniu:
enabled: ${VEHICLE_STATE_ENABLED:false}
vehicle-stat:
enabled: ${VEHICLE_STAT_ENABLED:true}
repository-type: ${VEHICLE_STAT_REPOSITORY_TYPE:jdbc}
file-path: ${VEHICLE_STAT_FILE_PATH:./target/vehicle-stat/}
zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai}
jt808:

1
modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java
View File

@@ -25,6 +25,7 @@ class VehicleAnalyticsAppDefaultsTest {
.containsEntry("lingniu.ingest.event-file-store.enabled", false)
.containsEntry("lingniu.ingest.event-history.enabled", false)
.containsEntry("lingniu.ingest.vehicle-stat.enabled", "${VEHICLE_STAT_ENABLED:true}")
.containsEntry("lingniu.ingest.vehicle-stat.repository-type", "${VEHICLE_STAT_REPOSITORY_TYPE:jdbc}")
.containsEntry("lingniu.ingest.vehicle-state.enabled", "${VEHICLE_STATE_ENABLED:false}")
.containsEntry("lingniu.ingest.sink.mq.consumer.enabled", "${KAFKA_CONSUMER_ENABLED:true}")
.containsEntry(

108
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepository.java Normal file
View File

@@ -0,0 +1,108 @@
package com.lingniu.ingest.vehiclestat;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.Date;
import java.time.LocalDate;
import java.util.List;
import java.util.Optional;
import java.util.OptionalDouble;
public final class JdbcVehicleStatMetricRepository implements VehicleStatRepository {
private static final String DAILY_MILEAGE_KEY = "daily_mileage_km";
private static final String DAILY_MILEAGE_UNIT = "km";
private final JdbcTemplate jdbcTemplate;
public JdbcVehicleStatMetricRepository(JdbcTemplate jdbcTemplate) {
if (jdbcTemplate == null) {
throw new IllegalArgumentException("jdbcTemplate must not be null");
}
this.jdbcTemplate = jdbcTemplate;
ensureSchema();
}
@Override
public void appendMileagePoint(String vin, MileagePoint point) {
// Production metrics are written as final metric rows; raw source points stay in RAW/locations history.
}
@Override
public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return List.of();
}
@Override
public void saveDailyStat(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();
try {
jdbcTemplate.update("""
INSERT INTO vehicle_stat_metric
(vin, stat_date, metric_key, metric_value, metric_unit, calculation_method)
VALUES (?, ?, ?, ?, ?, ?)
""", vin, statDate, DAILY_MILEAGE_KEY, value, DAILY_MILEAGE_UNIT, strategy);
} catch (DuplicateKeyException ex) {
updateDailyStat(vin, statDate, value, strategy);
}
}
@Override
public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
String normalizedVin = clean(vin);
List<VehicleDailyStatResult> rows = jdbcTemplate.query("""
SELECT metric_value, calculation_method
FROM vehicle_stat_metric
WHERE vin = ? AND stat_date = ? AND metric_key = ?
""", (rs, rowNum) -> new VehicleDailyStatResult(
normalizedVin,
statDate,
rs.getBigDecimal("metric_value") == null
? OptionalDouble.empty()
: OptionalDouble.of(rs.getBigDecimal("metric_value").doubleValue()),
DailyMileageStrategy.valueOf(rs.getString("calculation_method"))),
normalizedVin, Date.valueOf(statDate), DAILY_MILEAGE_KEY);
return rows.stream().findFirst();
}
private void updateDailyStat(String vin, Date statDate, double value, String strategy) {
jdbcTemplate.update("""
UPDATE vehicle_stat_metric
SET metric_value = ?,
metric_unit = ?,
calculation_method = ?,
updated_at = CURRENT_TIMESTAMP
WHERE vin = ? AND stat_date = ? AND metric_key = ?
""", value, DAILY_MILEAGE_UNIT, strategy, vin, statDate, DAILY_MILEAGE_KEY);
}
private void ensureSchema() {
jdbcTemplate.execute("""
CREATE TABLE IF NOT EXISTS vehicle_stat_metric (
vin VARCHAR(64) NOT NULL,
stat_date DATE NOT NULL,
metric_key VARCHAR(64) NOT NULL,
metric_value DECIMAL(18,6) NULL,
metric_unit VARCHAR(16) NOT NULL,
calculation_method VARCHAR(64) NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (vin, stat_date, metric_key)
)
""");
}
private static String clean(String value) {
if (value == null || value.isBlank()) {
throw new IllegalArgumentException("vin must not be blank");
}
return value.trim();
}
}

13
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java
View File

@@ -6,6 +6,7 @@ import com.lingniu.ingest.vehiclestat.DailyMileageCalculator;
import com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
import com.lingniu.ingest.vehiclestat.DailyVehicleStatService;
import com.lingniu.ingest.vehiclestat.FileVehicleStatRepository;
import com.lingniu.ingest.vehiclestat.JdbcVehicleStatMetricRepository;
import com.lingniu.ingest.vehiclestat.VehicleStatController;
import com.lingniu.ingest.vehiclestat.VehicleStatEnvelopeIngestor;
import com.lingniu.ingest.vehiclestat.VehicleStatEventSink;
@@ -30,6 +31,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import java.nio.file.Path;
import java.time.Duration;
@@ -44,10 +46,19 @@ import java.time.ZoneId;
@ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat", name = "enabled", havingValue = "true")
public class VehicleStatAutoConfiguration {
@Bean
@ConditionalOnBean(JdbcTemplate.class)
@ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat", name = "repository-type",
havingValue = "jdbc", matchIfMissing = true)
@ConditionalOnMissingBean
public VehicleStatRepository jdbcVehicleStatMetricRepository(JdbcTemplate jdbcTemplate) {
return new JdbcVehicleStatMetricRepository(jdbcTemplate);
}
@Bean
@ConditionalOnMissingBean
public VehicleStatRepository vehicleStatRepository(VehicleStatProperties props) {
// 派生统计默认落本地文件，便于独立开关；主历史库仍由 event-file-store 管理。
// 无 JDBC 指标库时才回落本地文件，主要用于开发和单元验证。
return new FileVehicleStatRepository(Path.of(props.getFilePath()));
}

11
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java
View File

@@ -8,6 +8,9 @@ public class VehicleStatProperties {
/** 派生统计本地文件根目录；不是 32960 RAW archive 或 DuckDB 历史库目录。 */
private String filePath = "./target/vehicle-stat/";
/** 指标仓储类型：有 JDBC 数据源时生产使用 jdbc；本地无数据库时回落 file。 */
private String repositoryType = "jdbc";
/** 统计自然日口径，默认按国内业务使用东八区。 */
private String zoneId = "Asia/Shanghai";
@@ -21,6 +24,14 @@ public class VehicleStatProperties {
this.filePath = filePath;
}
public String getRepositoryType() {
return repositoryType;
}
public void setRepositoryType(String repositoryType) {
this.repositoryType = repositoryType;
}
public String getZoneId() {
return zoneId;
}

44
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepositoryTest.java Normal file
View File

@@ -0,0 +1,44 @@
package com.lingniu.ingest.vehiclestat;
import org.junit.jupiter.api.Test;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import java.time.LocalDate;
import java.util.OptionalDouble;
import static org.assertj.core.api.Assertions.assertThat;
class JdbcVehicleStatMetricRepositoryTest {
@Test
void upsertsDailyMileageIntoCommonMetricTable() {
JdbcTemplate jdbcTemplate = new JdbcTemplate(new DriverManagerDataSource(
"jdbc:h2:mem:vehicle_stat_metric;MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1",
"sa",
""));
JdbcVehicleStatMetricRepository repository = new JdbcVehicleStatMetricRepository(jdbcTemplate);
repository.saveDailyStat(new VehicleDailyStatResult(
"VIN001",
LocalDate.of(2026, 6, 30),
OptionalDouble.of(12.345),
DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST));
repository.saveDailyStat(new VehicleDailyStatResult(
"VIN001",
LocalDate.of(2026, 6, 30),
OptionalDouble.of(15.5),
DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST));
assertThat(repository.findDailyStat("VIN001", LocalDate.of(2026, 6, 30)))
.isPresent()
.get()
.extracting(result -> result.dailyMileageKm().orElseThrow())
.isEqualTo(15.5);
assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM vehicle_stat_metric", Integer.class))
.isEqualTo(1);
assertThat(jdbcTemplate.queryForObject(
"SELECT metric_key FROM vehicle_stat_metric WHERE vin = 'VIN001'",
String.class)).isEqualTo("daily_mileage_km");
}
}

22
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java
View File

@@ -6,6 +6,7 @@ import com.lingniu.ingest.vehiclestat.DailyMileageCalculator;
import com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
import com.lingniu.ingest.vehiclestat.DailyVehicleStatService;
import com.lingniu.ingest.vehiclestat.FileVehicleStatRepository;
import com.lingniu.ingest.vehiclestat.JdbcVehicleStatMetricRepository;
import com.lingniu.ingest.vehiclestat.MileagePoint;
import com.lingniu.ingest.vehiclestat.VehicleStatController;
import com.lingniu.ingest.vehiclestat.VehicleDailyStatResult;
@@ -134,11 +135,32 @@ class VehicleStatAutoConfigurationTest {
"lingniu.ingest.vehicle-stat.jt808.state-store=memory")
.run(context -> {
assertThat(context).hasSingleBean(JdbcTemplate.class);
assertThat(context).hasSingleBean(JdbcVehicleStatMetricRepository.class);
assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class);
assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class);
});
}
@Test
void canForceFileRepositoryEvenWhenJdbcTemplateExists() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
JdbcTemplateAutoConfiguration.class,
VehicleStatAutoConfiguration.class))
.withBean(DataSource.class, () -> new DriverManagerDataSource(
"jdbc:h2:mem:vehicle_stat_file_mode;MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1",
"sa",
""))
.withPropertyValues(
"lingniu.ingest.vehicle-stat.enabled=true",
"lingniu.ingest.vehicle-stat.repository-type=file")
.run(context -> {
assertThat(context).hasSingleBean(VehicleStatRepository.class);
assertThat(context).hasSingleBean(FileVehicleStatRepository.class);
assertThat(context).doesNotHaveBean(JdbcVehicleStatMetricRepository.class);
});
}
private static final class InMemoryVehicleStatRepository implements VehicleStatRepository {
@Override public void appendMileagePoint(String vin, MileagePoint point) {}
@Override public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) { return List.of(); }