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

feat: ingest raw archive envelopes in history

Browse Source
This commit is contained in:
lingniu
2026-06-23 16:58:45 +08:00
parent 66807fdf3d
commit 6b82144f3c
3 changed files with 110 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryEnvelopeRecordMapper.java
View File

@@ -1,7 +1,11 @@
package com.lingniu.ingest.eventhistory;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.eventfilestore.EventFileRecord;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
@@ -21,20 +25,24 @@ import java.util.Map;
public final class TelemetryEnvelopeRecordMapper {
private static final JsonFormat.Printer JSON_PRINTER = JsonFormat.printer();
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
public EventFileRecord toRecord(VehicleEnvelope envelope) {
if (envelope == null) {
throw new IllegalArgumentException("envelope must not be null");
}
if (!envelope.hasTelemetrySnapshot()) {
throw new IllegalArgumentException("envelope telemetry_snapshot is required");
if (envelope.hasTelemetrySnapshot()) {
return telemetryRecord(envelope);
}
if (envelope.hasRawArchive()) {
return rawArchiveRecord(envelope);
}
throw new IllegalArgumentException("envelope telemetry_snapshot or raw_archive is required");
}
private static EventFileRecord telemetryRecord(VehicleEnvelope envelope) {
ProtocolId protocol = protocol(envelope.getSource());
Map<String, String> metadata = new LinkedHashMap<>(envelope.getMetadataMap());
if (protocol == ProtocolId.UNKNOWN && !envelope.getSource().isBlank()) {
// 保留未知 source 原文，避免历史库标准化成 UNKNOWN 后丢失排障线索。
metadata.putIfAbsent("originalSource", envelope.getSource());
}
Map<String, String> metadata = metadata(envelope, protocol);
return new EventFileRecord(
envelope.getEventId(),
protocol,
@@ -48,6 +56,40 @@ public final class TelemetryEnvelopeRecordMapper {
);
}
private static EventFileRecord rawArchiveRecord(VehicleEnvelope envelope) {
ProtocolId protocol = protocol(envelope.getSource());
RawArchiveRef rawArchive = envelope.getRawArchive();
Map<String, String> metadata = metadata(envelope, protocol);
String rawArchiveUri = rawArchive.getUri();
metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, envelope.getEventId());
if (!rawArchiveUri.isBlank()) {
metadata.putIfAbsent(RawArchiveKeys.META_URI, rawArchiveUri);
}
return new EventFileRecord(
envelope.getEventId(),
protocol,
"RAW_ARCHIVE",
envelope.getVin(),
Instant.ofEpochMilli(envelope.getEventTimeMs()),
Instant.ofEpochMilli(envelope.getIngestTimeMs()),
rawArchiveUri,
metadata,
rawArchiveJson(envelope, metadata)
);
}
private static Map<String, String> metadata(VehicleEnvelope envelope, ProtocolId protocol) {
Map<String, String> metadata = new LinkedHashMap<>(envelope.getMetadataMap());
if (!envelope.getProtocolVersion().isBlank()) {
metadata.putIfAbsent("protocolVersion", envelope.getProtocolVersion());
}
if (protocol == ProtocolId.UNKNOWN && !envelope.getSource().isBlank()) {
// 保留未知 source 原文，避免历史库标准化成 UNKNOWN 后丢失排障线索。
metadata.putIfAbsent("originalSource", envelope.getSource());
}
return metadata;
}
private static ProtocolId protocol(String source) {
if (source == null || source.isBlank()) {
return ProtocolId.UNKNOWN;
@@ -66,4 +108,26 @@ public final class TelemetryEnvelopeRecordMapper {
throw new IllegalArgumentException("failed to serialize telemetry_snapshot", e);
}
}
private static String rawArchiveJson(VehicleEnvelope envelope, Map<String, String> metadata) {
Map<String, Object> payload = new LinkedHashMap<>();
RawArchiveRef rawArchive = envelope.getRawArchive();
payload.put("eventId", envelope.getEventId());
payload.put("vin", envelope.getVin());
payload.put("protocol", protocol(envelope.getSource()).name());
if (!envelope.getSource().isBlank()) {
payload.put("source", envelope.getSource());
}
payload.put("eventType", "RAW_ARCHIVE");
payload.put("eventTime", Instant.ofEpochMilli(envelope.getEventTimeMs()).toString());
payload.put("ingestTime", Instant.ofEpochMilli(envelope.getIngestTimeMs()).toString());
payload.put("rawArchiveUri", rawArchive.getUri());
payload.put("rawSizeBytes", rawArchive.getSizeBytes());
payload.put("metadata", metadata);
try {
return OBJECT_MAPPER.writeValueAsString(payload);
} catch (JsonProcessingException e) {
throw new IllegalArgumentException("failed to serialize raw_archive", e);
}
}
}

38
modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java
View File

@@ -5,6 +5,7 @@ import com.lingniu.ingest.api.consumer.EnvelopeIngestor;
import com.lingniu.ingest.eventfilestore.EventFileQuery;
import com.lingniu.ingest.eventfilestore.EventFileRecord;
import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.sink.mq.proto.RawArchiveRef;
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
@@ -55,6 +56,43 @@ class EventHistoryEnvelopeIngestorTest {
assertThat(store.records).isEmpty();
}
@Test
void tryIngestRawArchiveEnvelopeAppendsRawArchiveRecord() {
CapturingStore store = new CapturingStore();
EventHistoryEnvelopeIngestor ingestor =
new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper());
VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
.setSchemaVersion("1.0")
.setEventId("raw-event-1")
.setVin("VINRAW001")
.setSource("GB32960")
.setProtocolVersion("V2016")
.setEventTimeMs(1_782_112_400_000L)
.setIngestTimeMs(1_782_112_401_000L)
.putMetadata("archiveKey", "gb32960/2026/06/23/raw-event-1.bin")
.setRawArchive(RawArchiveRef.newBuilder()
.setUri("archive://gb32960/2026/06/23/raw-event-1.bin")
.setSizeBytes(128)
.build())
.build();
EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
assertThat(store.records).hasSize(1);
EventFileRecord record = store.records.getFirst();
assertThat(record.eventId()).isEqualTo("raw-event-1");
assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
assertThat(record.vin()).isEqualTo("VINRAW001");
assertThat(record.rawArchiveUri()).isEqualTo("archive://gb32960/2026/06/23/raw-event-1.bin");
assertThat(record.metadata())
.containsEntry("protocolVersion", "V2016")
.containsEntry("archiveKey", "gb32960/2026/06/23/raw-event-1.bin");
assertThat(record.payloadJson()).contains("\"rawSizeBytes\":128");
assertThat(record.payloadJson()).doesNotContain("rawBytes");
}
private static VehicleEnvelope envelope(String eventId) {
return VehicleEnvelope.newBuilder()
.setEventId(eventId)

3
modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java
View File

@@ -71,8 +71,7 @@ public final class EnvelopeMapper {
int size = ra.rawBytes() == null ? 0 : ra.rawBytes().length;
String key = rawArchiveKey(ra);
String uri = rawArchiveUri(ra, key);
// RAW envelope 只携带引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。
// 注意 KafkaEventSink 当前仍过滤 RawArchive；这段用于未来放开 raw 引用转发或消费者测试。
// RAW envelope 只携带 URI/size 引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。
b.putMetadata(RawArchiveKeys.META_KEY, key);
b.putMetadata(RawArchiveKeys.META_URI, uri);
b.putMetadata(RawArchiveKeys.META_EVENT_ID, ra.eventId());