docs: align target architecture with production flow
This commit is contained in:
lingniu
@@ -1,271 +1,187 @@
|
||||
# Target Architecture
|
||||
|
||||
This document defines the target architecture for Lingniu vehicle ingest. The
|
||||
main goal is to keep protocol ingest, raw archive, historical detail storage,
|
||||
realtime state, and statistics as separate responsibilities connected by stable
|
||||
event contracts.
|
||||
This document records the production target for `lingniu-vehicle-ingest`.
|
||||
The project should stay small at runtime: protocol apps ingest and publish,
|
||||
history consumes and indexes, analytics derives metrics, and business systems
|
||||
read through explicit APIs or Kafka.
|
||||
|
||||
## Active Scope
|
||||
|
||||
Active production protocols:
|
||||
|
||||
| Protocol | Runtime app | Event topic | Raw topic |
|
||||
|---|---|---|---|
|
||||
| GB/T 32960 | `gb32960-ingest-app` | `vehicle.event.gb32960.v1` | `vehicle.raw.gb32960.v1` |
|
||||
| JT/T 808 | `jt808-ingest-app` | `vehicle.event.jt808.v1` | `vehicle.raw.jt808.v1` |
|
||||
| Yutong MQTT | `yutong-mqtt-app` | `vehicle.event.mqtt-yutong.v1` | `vehicle.raw.mqtt-yutong.v1` |
|
||||
|
||||
Xinda Push is legacy and is not part of the default production surface.
|
||||
It may remain in source for reference, but new optimization work should target
|
||||
the active protocols above.
|
||||
|
||||
## Goals
|
||||
|
||||
- The ingest service receives protocol messages, archives raw frames, and
|
||||
publishes normalized full-field events to Kafka.
|
||||
- Historical detail storage consumes Kafka events and stores raw parsed event
|
||||
details in Parquet, queried through DuckDB.
|
||||
- Realtime vehicle state consumes Kafka events and writes hot state to Redis.
|
||||
- Statistics consumes Kafka events and calculates daily operational metrics by
|
||||
configurable rules.
|
||||
- Kafka consumers expose non-throwing ingest results for bad protobuf payloads,
|
||||
unsupported envelopes, and downstream failures so one bad record does not
|
||||
block a vehicle partition.
|
||||
- Consumer services expose `EnvelopeConsumerProcessor` beans. Kafka workers or
|
||||
framework listeners should pass topic, partition, offset, key, and value into
|
||||
the processor; `SKIPPED`, `INVALID_ENVELOPE`, and `FAILED` results are routed
|
||||
through `EnvelopeDeadLetterSink`.
|
||||
- `sink-mq` owns Kafka-specific consumer plumbing: `KafkaEnvelopeConsumerFactory`
|
||||
creates one independent Kafka consumer per service processor binding,
|
||||
`KafkaEnvelopeConsumerRunner` manages the long-running poll loop, and
|
||||
`KafkaEnvelopeDeadLetterSink` publishes failed records to the configured DLQ
|
||||
topic with diagnostic headers.
|
||||
- Business consumers depend on internal fields, not protocol-specific field
|
||||
names.
|
||||
- Keep protocol IO, raw archive, history, latest-state, and statistics as
|
||||
separate responsibilities.
|
||||
- Publish normalized Kafka envelopes with VIN as the partition key whenever a
|
||||
VIN is known.
|
||||
- Store complete raw payload metadata and parsed JSON in TDengine `raw_frames`.
|
||||
- Store query-friendly location rows separately from raw payload JSON.
|
||||
- Keep derived statistics in metric tables, not protocol-specific daily tables.
|
||||
- Use Redis only for hot state or streaming calculation state that can be
|
||||
rebuilt from Kafka.
|
||||
- Keep optional compatibility modules outside the production hot path.
|
||||
|
||||
## Non-Goals
|
||||
|
||||
- The ingest service must not write business tables.
|
||||
- The ingest service must not serve historical detail queries.
|
||||
- Parquet history storage must not store derived daily statistics.
|
||||
- Redis hot state must not become the long-term historical store.
|
||||
- Statistics services must not parse GB/T 32960 frames directly.
|
||||
- Ingest apps do not write business tables.
|
||||
- Ingest apps do not serve history query APIs.
|
||||
- Redis is not the long-term historical store.
|
||||
- `event-file-store` Parquet/DuckDB is not the default history store.
|
||||
- `telemetry_fields` parsing is not handled by this project when a separate
|
||||
field parsing service owns that job.
|
||||
|
||||
## Service Boundaries
|
||||
## Runtime Responsibilities
|
||||
|
||||
### Ingest Service
|
||||
### Protocol Apps
|
||||
|
||||
The ingest service owns protocol IO and event publication.
|
||||
Apps:
|
||||
|
||||
- `gb32960-ingest-app`
|
||||
- `jt808-ingest-app`
|
||||
- `yutong-mqtt-app`
|
||||
|
||||
Responsibilities:
|
||||
|
||||
- Accept GB/T 32960, JT/T 808, MQTT, Xinda Push, and future inbound protocols.
|
||||
- Decode frames and acknowledge protocol messages where required.
|
||||
- Emit `RawArchive` before deduplication and rate limiting when raw bytes are
|
||||
available.
|
||||
- Write raw bytes through `sink-archive`.
|
||||
- Map protocol messages into internal telemetry events.
|
||||
- Publish internal events to Kafka with VIN as the partition key.
|
||||
- Accept protocol traffic.
|
||||
- Decode, authenticate, acknowledge, and maintain sessions where the protocol
|
||||
requires it.
|
||||
- Archive raw bytes through `sink-archive` when bytes are available.
|
||||
- Publish parsed event envelopes to Kafka.
|
||||
- Publish raw envelopes to Kafka.
|
||||
- Resolve vehicle identity through the shared MySQL identity binding table.
|
||||
|
||||
The ingest service must not depend on Redis, DuckDB query endpoints, or
|
||||
statistics rule repositories.
|
||||
Protocol apps must stay independent from TDengine history readers, Redis state
|
||||
repositories, and statistic calculators.
|
||||
|
||||
### Event Contract
|
||||
|
||||
The event contract is the only dependency shared by ingest, history, realtime
|
||||
state, and statistics.
|
||||
`sink-mq` owns the Kafka protobuf envelope and consumer plumbing.
|
||||
Consumers should use `EnvelopeConsumerProcessor` so invalid protobuf, skipped
|
||||
envelopes, and downstream failures are converted to structured results and DLQ
|
||||
records instead of blocking a partition.
|
||||
|
||||
Core event identity:
|
||||
Core identity fields:
|
||||
|
||||
| Field | Meaning |
|
||||
|---|---|
|
||||
| `event_id` | Idempotency key for the parsed event |
|
||||
| `event_id` | Parsed event idempotency key |
|
||||
| `trace_id` | Cross-service trace id |
|
||||
| `vin` | Vehicle VIN and Kafka partition key |
|
||||
| `source_protocol` | Protocol source such as `GB32960`; consumers may map unknown future source names to `UNKNOWN` and preserve the original source in metadata |
|
||||
| `protocol_version` | Source protocol version |
|
||||
| `event_type` | Business event type, for example `REALTIME`, `LOCATION`, `ALARM` |
|
||||
| `event_time` | Device collection time |
|
||||
| `ingest_time` | Platform receive time |
|
||||
| `raw_archive_uri` | Optional reference to archived raw bytes |
|
||||
| `vin` | Vehicle id and Kafka partition key |
|
||||
| `source` | Source protocol, for example `GB32960`, `JT808`, `YUTONG_MQTT` |
|
||||
| `event_time_ms` | Device event time |
|
||||
| `ingest_time_ms` | Platform receive time |
|
||||
| `raw_uri` | `archive://...` reference to raw bytes when available |
|
||||
|
||||
Full-field telemetry is represented as internal field values:
|
||||
Full-field telemetry uses `TelemetrySnapshot` fields with stable keys such as
|
||||
`total_mileage_km`, `speed_kmh`, `longitude`, and `latitude`.
|
||||
|
||||
| Field | Meaning |
|
||||
|---|---|
|
||||
| `field_key` | Stable internal field key such as `total_mileage_km` |
|
||||
| `value_type` | `STRING`, `DOUBLE`, `LONG`, `BOOLEAN`, `INSTANT`, `JSON` |
|
||||
| `value` | String representation used for durable serialization |
|
||||
| `unit` | Stable unit when applicable |
|
||||
| `quality` | `GOOD`, `ESTIMATED`, `INVALID`, `MISSING` |
|
||||
| `source_path` | Protocol source path, for traceability |
|
||||
### History App
|
||||
|
||||
The internal field list is defined in
|
||||
`docs/vehicle-telemetry-internal-fields.md`. New protocol-specific fields start
|
||||
as extension fields and are promoted only after a business meaning and unit are
|
||||
stable.
|
||||
|
||||
### Historical Detail Service
|
||||
|
||||
The historical detail service owns raw parsed event detail storage and exports.
|
||||
App: `vehicle-history-app`
|
||||
|
||||
Responsibilities:
|
||||
|
||||
- Consume normalized telemetry events from Kafka.
|
||||
- Store events in Parquet partitioned by protocol and event date.
|
||||
- Query Parquet through DuckDB.
|
||||
- Provide HTTP APIs for date-range query and export.
|
||||
- Support ascending and descending time order.
|
||||
- Support multi-day concatenated export.
|
||||
- Consume active protocol event topics.
|
||||
- Consume active protocol raw topics.
|
||||
- Write raw frame rows into TDengine `raw_frames`.
|
||||
- Write compact location rows into TDengine location tables.
|
||||
- Keep `payloadJson.parsed` on raw rows for full raw inspection.
|
||||
- Expose paged history APIs for raw frames and locations.
|
||||
- Keep specialized APIs off by default unless explicitly enabled.
|
||||
|
||||
Storage partition:
|
||||
Production default:
|
||||
|
||||
- `TDENGINE_HISTORY_ENABLED=true` in deployment.
|
||||
- `EVENT_FILE_STORE_ENABLED=false`.
|
||||
- `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`.
|
||||
|
||||
`event-file-store` remains an optional compatibility path for old low-level
|
||||
record APIs or local investigations. It should not be treated as the primary
|
||||
history backend.
|
||||
|
||||
### Analytics App
|
||||
|
||||
App: `vehicle-analytics-app`
|
||||
|
||||
Responsibilities:
|
||||
|
||||
- Consume JT808 event envelopes.
|
||||
- Calculate 808 daily mileage from the reported GPS total mileage only:
|
||||
|
||||
```text
|
||||
<root>/
|
||||
protocol=GB32960/
|
||||
date=2026-06-22/
|
||||
part-*.parquet
|
||||
daily_mileage_km = last_total_mileage_km - first_total_mileage_km
|
||||
calculation_method = JT808_TOTAL_MILEAGE_DIFF
|
||||
```
|
||||
|
||||
The service stores detail events and field snapshots only. It does not calculate
|
||||
or store daily mileage, daily hydrogen usage, or other derived metrics.
|
||||
- Store the metric in `vehicle_stat_metric` with
|
||||
`metric_key = daily_mileage_km`.
|
||||
- Keep rolling daily calculation state in Redis.
|
||||
|
||||
### Realtime State Service
|
||||
There is no JT808-specific daily mileage table and no in-memory production
|
||||
state-store mode.
|
||||
|
||||
The realtime state service owns hot vehicle state.
|
||||
### Latest State
|
||||
|
||||
Module: `vehicle-state-service`
|
||||
|
||||
Responsibilities:
|
||||
|
||||
- Consume normalized telemetry events from Kafka.
|
||||
- Keep latest vehicle state by VIN in Redis.
|
||||
- Keep latest location by VIN in Redis.
|
||||
- Keep latest safety state and hydrogen leak state by VIN in Redis.
|
||||
- Optionally keep short recent windows for operations screens.
|
||||
- Consume normalized Kafka envelopes.
|
||||
- Maintain latest state, location, and safety snapshots in Redis.
|
||||
- Serve latest-state APIs for operational screens.
|
||||
|
||||
Suggested Redis keys:
|
||||
|
||||
| Key | Value |
|
||||
|---|---|
|
||||
| `vehicle:state:{vin}` | Latest full state snapshot |
|
||||
| `vehicle:location:{vin}` | Latest location snapshot |
|
||||
| `vehicle:safety:{vin}` | Latest safety snapshot |
|
||||
| `vehicle:event:last:{vin}` | Latest event identity and timestamps |
|
||||
|
||||
Redis is for millisecond-level reads and recent state. It is not the historical
|
||||
source of truth.
|
||||
|
||||
Current module: `vehicle-state-service`.
|
||||
|
||||
Implemented boundaries:
|
||||
|
||||
- `VehicleStateEnvelopeIngestor` parses Kafka envelope bytes.
|
||||
- `VehicleStateUpdater` maps full-field telemetry snapshots to Redis JSON
|
||||
payloads.
|
||||
- `VehicleStateRepository` isolates storage from event parsing.
|
||||
- `VehicleStateController` exposes read-only latest state APIs.
|
||||
|
||||
### Statistics Service
|
||||
|
||||
The statistics service owns derived operational metrics.
|
||||
|
||||
Responsibilities:
|
||||
|
||||
- Consume normalized telemetry events from Kafka.
|
||||
- Maintain day-level vehicle point state needed for statistics.
|
||||
- Run scheduled and event-triggered calculations.
|
||||
- Store derived results in the common metric table.
|
||||
|
||||
Current JT808 daily mileage strategy:
|
||||
|
||||
| Strategy | Formula |
|
||||
|---|---|
|
||||
| `JT808_TOTAL_MILEAGE_DIFF` | Current local day last valid JT808 GPS total mileage minus first valid JT808 GPS total mileage |
|
||||
|
||||
The derived value is stored as `metric_key = daily_mileage_km` in
|
||||
`vehicle_stat_metric`; there is no protocol-specific JT808 daily-mileage table.
|
||||
|
||||
Hydrogen and electricity statistics use internal fields from the same event
|
||||
contract. Refueling and charging resets must be handled in statistics logic, not
|
||||
in protocol mappers.
|
||||
|
||||
Current module: `vehicle-stat-service`.
|
||||
|
||||
Implemented boundaries:
|
||||
|
||||
- `VehicleStatEnvelopeIngestor` parses Kafka envelope bytes.
|
||||
- `VehicleStatEventProcessor` extracts statistic source points from internal
|
||||
fields such as `total_mileage_km`.
|
||||
- `VehicleStatRepository` isolates point/result persistence from calculation.
|
||||
- `VehicleStatRuleRepository` isolates per-vehicle calculation rules.
|
||||
- `DailyVehicleStatService` calculates and saves day-level results.
|
||||
Redis state must be rebuildable from Kafka replay and should not be used as the
|
||||
source of truth for historical queries.
|
||||
|
||||
## Data Flow
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
vehicle["Vehicle / Platform"] --> ingest["ingest service"]
|
||||
vehicle["Vehicles / Platforms"] --> ingest["Protocol apps"]
|
||||
ingest --> archive["sink-archive<br/>raw bytes"]
|
||||
ingest --> kafka["Kafka<br/>full-field telemetry events"]
|
||||
kafka --> history["event-history-service"]
|
||||
ingest --> kafka["Kafka<br/>event + raw envelopes"]
|
||||
kafka --> history["vehicle-history-app"]
|
||||
kafka --> analytics["vehicle-analytics-app"]
|
||||
kafka --> state["vehicle-state-service"]
|
||||
kafka --> stat["vehicle-stat-service"]
|
||||
history --> parquet["Parquet files"]
|
||||
history --> duckdb["DuckDB query"]
|
||||
state --> redis["Redis hot state"]
|
||||
stat --> statdb["Statistics DB"]
|
||||
history --> tdengine["TDengine<br/>raw_frames + locations"]
|
||||
analytics --> mysql["MySQL<br/>vehicle_stat_metric"]
|
||||
analytics --> redisCalc["Redis<br/>mileage state"]
|
||||
state --> redisState["Redis<br/>latest state"]
|
||||
```
|
||||
|
||||
## Kafka Topic Boundaries
|
||||
|
||||
Suggested topics:
|
||||
|
||||
| Topic | Producer | Consumer |
|
||||
|---|---|---|
|
||||
| `vehicle.realtime` | protocol ingress | history, realtime state, statistics |
|
||||
| `vehicle.location` | protocol ingress | history, realtime state, statistics |
|
||||
| `vehicle.alarm` | protocol ingress | history, realtime state |
|
||||
| `vehicle.session` | protocol ingress | history |
|
||||
| `vehicle.media.meta` | protocol ingress | history, media tooling |
|
||||
| `vehicle.raw.archive` | optional archive publisher | replay or audit services |
|
||||
| `vehicle.dlq` | any consumer | operations and replay tooling |
|
||||
|
||||
All telemetry event topics use VIN as the partition key to preserve per-vehicle
|
||||
ordering.
|
||||
|
||||
## Failure Handling
|
||||
|
||||
- Ingest must keep receiving traffic when history, Redis, or statistics services
|
||||
are down.
|
||||
- Kafka publish failures go through the existing sink circuit breaker and DLQ
|
||||
strategy.
|
||||
- History consumer commits Kafka offsets only after Parquet append succeeds.
|
||||
- Redis state consumer can replay from Kafka to rebuild hot state.
|
||||
- Statistics consumer stores checkpoints and can recompute a day from historical
|
||||
detail events when rules change.
|
||||
- Raw archive failures are logged and alerted, but parsed event publication
|
||||
remains independent.
|
||||
|
||||
## Implementation Phases
|
||||
|
||||
### Phase 1: Full-Field Contract And Historical Detail
|
||||
|
||||
- Add a full-field telemetry contract to `ingest-api`.
|
||||
- Map GB/T 32960 parsed blocks into internal field snapshots.
|
||||
- Extend Kafka envelope schema for full-field events.
|
||||
- Make `event-file-store` store full-field telemetry records.
|
||||
- Add query and export APIs for date-range historical detail.
|
||||
|
||||
### Phase 2: Redis Hot State
|
||||
|
||||
- Add a Redis-backed realtime state module.
|
||||
- Consume full-field telemetry events.
|
||||
- Store latest state, location, and safety snapshots by VIN.
|
||||
- Provide query APIs for operations screens.
|
||||
|
||||
### Phase 3: Statistics Service
|
||||
|
||||
- Add a statistics module with rule configuration.
|
||||
- Consume full-field telemetry events.
|
||||
- Implement daily mileage strategies.
|
||||
- Add result persistence boundary and calculation tests.
|
||||
- Protocol apps continue accepting traffic when history, analytics, or Redis
|
||||
consumers are unavailable.
|
||||
- Kafka producer failures are handled by sink retry/circuit-breaker behavior and
|
||||
DLQ topics where configured.
|
||||
- History consumers should commit offsets only after TDengine writes succeed.
|
||||
- Analytics state can recover from Redis state or Kafka replay.
|
||||
- Raw archive failures must be observable, but event publication and history
|
||||
indexing remain separate concerns.
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- Ingest has no compile-time dependency on history query controllers, Redis
|
||||
state repositories, or statistics calculators.
|
||||
- Every service boundary communicates through Kafka events or explicit public
|
||||
APIs, not shared mutable state.
|
||||
- GB/T 32960 detail fields are traceable from internal field key back to
|
||||
protocol source path.
|
||||
- Historical query works across multiple days with ascending and descending
|
||||
order.
|
||||
- Redis state can be rebuilt from Kafka replay.
|
||||
- Daily mileage can be calculated by at least the two initial strategies.
|
||||
- Tests cover mapper output, Kafka serialization, Parquet query order, Redis
|
||||
state update, and mileage strategy calculation.
|
||||
- Active production deployment contains GB32960, JT808, Yutong MQTT,
|
||||
vehicle-history, and vehicle-analytics apps.
|
||||
- Xinda Push is absent from default deployment and history consumer bindings.
|
||||
- History APIs read from TDengine by default; Parquet/DuckDB is optional
|
||||
compatibility only.
|
||||
- Raw frame queries can return complete parsed JSON through `payloadJson.parsed`.
|
||||
- Location queries page over compact rows and reference raw records instead of
|
||||
duplicating full raw JSON.
|
||||
- JT808 daily mileage is stored only in `vehicle_stat_metric`.
|
||||
- Runtime state needed for 808 mileage uses Redis, with no production memory
|
||||
mode.
|
||||
- Build validation covers the active modules and their composition tests.
|
||||
|
||||
Reference in New Issue
Block a user