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

feat(go): batch redis fast writer projections

Browse Source
This commit is contained in:
lingniu
2026-07-03 19:45:00 +08:00
parent 81e050ec1c
commit 3a0d9bd840
7 changed files with 173 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
go/vehicle-gateway/cmd/nats-fast-writer/main.go
View File

@@ -194,6 +194,10 @@ type fastUpdater interface {
FastUpdate(context.Context, envelope.FrameEnvelope) error
}
type fastBatchUpdater interface {
FastUpdateBatch(context.Context, []envelope.FrameEnvelope) error
}
type fastMessage struct {
subject string
data []byte
@@ -290,6 +294,25 @@ func processFastBatch(ctx context.Context, registry *metrics.Registry, appender
if err != nil {
return fmt.Errorf("tdengine batch append: %w", err)
}
if batchUpdater, ok := updater.(fastBatchUpdater); ok {
started = time.Now()
err = batchUpdater.FastUpdateBatch(ctx, envelopes)
recordFastWriterStageDuration(registry, subject, "redis", statusFromError(err), time.Since(started))
if err != nil {
return fmt.Errorf("redis fast batch update: %w", err)
}
for _, msg := range validMessages {
if msg.ack != nil {
started = time.Now()
err = msg.ack()
recordFastWriterStageDuration(registry, msg.subject, "ack", statusFromError(err), time.Since(started))
if err != nil {
return fmt.Errorf("nats ack: %w", err)
}
}
}
return nil
}
for i, env := range envelopes {
msg := validMessages[i]
started = time.Now()

58
go/vehicle-gateway/cmd/nats-fast-writer/main_test.go
View File

@@ -20,7 +20,7 @@ func TestProcessFastMessageWritesTDengineAndRedisBeforeAck(t *testing.T) {
t.Fatal(err)
}
appender := &recordingFastAppender{}
updater := &recordingFastUpdater{}
updater := &recordingFastSingleUpdater{}
ackCount := 0
msg := &fastMessage{data: payload, ack: func() error {
ackCount++
@@ -95,7 +95,7 @@ func TestProcessFastBatchAppendsTDengineBatchBeforeRedisAndAck(t *testing.T) {
t.Fatal(err)
}
appender := &recordingFastAppender{}
updater := &recordingFastUpdater{}
updater := &recordingFastSingleUpdater{}
ackCount := 0
msgs := []*fastMessage{
{subject: "vehicle.raw.go.jt808.v1", data: firstPayload, ack: func() error { ackCount++; return nil }},
@@ -116,6 +116,38 @@ func TestProcessFastBatchAppendsTDengineBatchBeforeRedisAndAck(t *testing.T) {
}
}
func TestProcessFastBatchUsesRedisBatchUpdaterWhenAvailable(t *testing.T) {
first := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN001", EventID: "evt-batch-1"}
second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN002", EventID: "evt-batch-2"}
firstPayload, err := first.MarshalJSONBytes()
if err != nil {
t.Fatal(err)
}
secondPayload, err := second.MarshalJSONBytes()
if err != nil {
t.Fatal(err)
}
updater := &recordingFastUpdater{}
ackCount := 0
msgs := []*fastMessage{
{subject: "vehicle.raw.go.jt808.v1", data: firstPayload, ack: func() error { ackCount++; return nil }},
{subject: "vehicle.raw.go.jt808.v1", data: secondPayload, ack: func() error { ackCount++; return nil }},
}
if err := processFastBatch(context.Background(), nil, &recordingFastAppender{}, updater, msgs); err != nil {
t.Fatalf("processFastBatch() error = %v", err)
}
if updater.batchCount != 1 || updater.batchRows != 2 {
t.Fatalf("FastUpdateBatch count=%d rows=%d, want count=1 rows=2", updater.batchCount, updater.batchRows)
}
if updater.count != 0 {
t.Fatalf("FastUpdate count=%d, want 0 when batch updater is available", updater.count)
}
if ackCount != 2 {
t.Fatalf("ack count=%d, want 2", ackCount)
}
}
func TestProcessFastBatchExposesPendingMetricsDuringAppend(t *testing.T) {
first := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN001", EventID: "evt-6"}
second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN002", EventID: "evt-7"}
@@ -249,11 +281,29 @@ func (a *recordingFastAppender) AppendAllBatch(_ context.Context, envs []envelop
}
type recordingFastUpdater struct {
count int
err error
count int
batchCount int
batchRows int
err error
}
func (u *recordingFastUpdater) FastUpdate(context.Context, envelope.FrameEnvelope) error {
u.count++
return u.err
}
func (u *recordingFastUpdater) FastUpdateBatch(_ context.Context, envs []envelope.FrameEnvelope) error {
u.batchCount++
u.batchRows += len(envs)
return u.err
}
type recordingFastSingleUpdater struct {
count int
err error
}
func (u *recordingFastSingleUpdater) FastUpdate(context.Context, envelope.FrameEnvelope) error {
u.count++
return u.err
}

43
go/vehicle-gateway/internal/realtime/repository.go
View File

@@ -42,6 +42,36 @@ func (r *Repository) FastUpdate(ctx context.Context, env envelope.FrameEnvelope)
return r.setFastProjection(ctx, vehicleKey, vin, env)
}
func (r *Repository) FastUpdateBatch(ctx context.Context, envs []envelope.FrameEnvelope) error {
if len(envs) == 0 {
return nil
}
pipe := r.client.Pipeline()
queued := 0
for _, env := range envs {
if !envelope.IsRealtimeTelemetryFrame(env) {
continue
}
vin := strings.TrimSpace(env.VIN)
if vin == "" {
continue
}
vehicleKey := strings.TrimSpace(env.VehicleKey())
if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") {
continue
}
if err := r.queueFastProjection(ctx, pipe, vehicleKey, vin, env); err != nil {
return err
}
queued++
}
if queued == 0 {
return nil
}
_, err := pipe.Exec(ctx)
return err
}
func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error {
if !envelope.IsRealtimeTelemetryFrame(env) {
return nil
@@ -373,6 +403,15 @@ func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEn
}
func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, vin string, env envelope.FrameEnvelope) error {
pipe := r.client.Pipeline()
if err := r.queueFastProjection(ctx, pipe, vehicleKey, vin, env); err != nil {
return err
}
_, err := pipe.Exec(ctx)
return err
}
func (r *Repository) queueFastProjection(ctx context.Context, pipe redis.Pipeliner, vehicleKey string, vin string, env envelope.FrameEnvelope) error {
values, types := realtimeKVMapsForEnvelope(env)
eventTimeMS := eventTimeOrReceivedMS(env)
offlineAfterMS := env.ReceivedAtMS + r.cfg.ttl().Milliseconds()
@@ -410,7 +449,6 @@ func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, v
"ttl_seconds": strconv.FormatInt(int64(r.cfg.ttl().Seconds()), 10),
"source_endpoint": env.SourceEndpoint,
}
pipe := r.client.Pipeline()
if len(values) > 0 {
pipe.HSet(ctx, realtimeKVValuesKey(env.Protocol, vin), values)
pipe.HSet(ctx, realtimeKVTypesKey(env.Protocol, vin), types)
@@ -420,8 +458,7 @@ func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, v
pipe.HSet(ctx, onlineStateKey(env.Protocol, vin), state)
pipe.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(env.ReceivedAtMS), Member: onlineMember(env.Protocol, vin)})
pipe.SAdd(ctx, realtimeIndexKey(env.Protocol), vin)
_, err = pipe.Exec(ctx)
return err
return nil
}
func realtimeKVMapsForEnvelope(env envelope.FrameEnvelope) (map[string]any, map[string]any) {

51
go/vehicle-gateway/internal/realtime/repository_test.go
View File

@@ -430,6 +430,57 @@ func TestRepositoryFastUpdateOnlyWritesPermanentKVAndMinuteOnline(t *testing.T)
}
}
func TestRepositoryFastUpdateBatchWritesMultipleRealtimeProjections(t *testing.T) {
repo, closeFn := newTestRepository(t)
defer closeFn()
ctx := context.Background()
err := repo.FastUpdateBatch(ctx, []envelope.FrameEnvelope{
{
Protocol: envelope.ProtocolGB32960,
VIN: "VIN001",
EventTimeMS: 1000,
ReceivedAtMS: 1100,
Parsed: map[string]any{
"data_units": []any{map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}},
},
},
{
Protocol: envelope.ProtocolJT808,
VIN: "VIN002",
EventTimeMS: 2000,
ReceivedAtMS: 2200,
MessageID: "0x0200",
Parsed: map[string]any{"location": map[string]any{"longitude": 121.1, "latitude": 30.2}},
Fields: map[string]any{envelope.FieldLongitude: 121.1, envelope.FieldLatitude: 30.2},
},
})
if err != nil {
t.Fatalf("FastUpdateBatch() error = %v", err)
}
gbValues, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:values").Result()
if err != nil {
t.Fatalf("gb values HGetAll error = %v", err)
}
if gbValues["gb32960.vehicle.soc_percent"] != "88" {
t.Fatalf("gb values = %#v", gbValues)
}
jtValues, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:JT808:VIN002:values").Result()
if err != nil {
t.Fatalf("jt values HGetAll error = %v", err)
}
if jtValues["jt808.location.longitude"] != "121.1" || jtValues["jt808.location.latitude"] != "30.2" {
t.Fatalf("jt values = %#v", jtValues)
}
if repo.client.ZScore(ctx, "vehicle:last_seen", "GB32960:VIN001").Err() != nil {
t.Fatal("last_seen should contain GB32960 VIN001")
}
if repo.client.ZScore(ctx, "vehicle:last_seen", "JT808:VIN002").Err() != nil {
t.Fatal("last_seen should contain JT808 VIN002")
}
}
func TestRepositoryListsOnlineStatusesAndPipelineSummary(t *testing.T) {
repo, closeFn := newTestRepository(t)
defer closeFn()