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

feat(go): batch nats fast writer tdengine writes

Browse Source
This commit is contained in:
lingniu
2026-07-03 19:17:02 +08:00
parent 2db05d100d
commit 3b694671fa
4 changed files with 127 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -184,6 +184,7 @@ func loadConfig() config {
type fastAppender interface {
AppendAll(context.Context, envelope.FrameEnvelope) error
AppendAllBatch(context.Context, []envelope.FrameEnvelope) error
}
type fastUpdater interface {
@@ -210,23 +211,27 @@ func runFastWorker(ctx context.Context, logger *slog.Logger, registry *metrics.R
time.Sleep(time.Second)
continue
}
fastMessages := make([]*fastMessage, 0, len(msgs))
for _, msg := range msgs {
msg := msg
operationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.OperationWait)
err := processFastMessage(operationCtx, registry, appender, updater, &fastMessage{
subject: msg.Subject,
data: msg.Data,
natsMsg := msg
fastMessages = append(fastMessages, &fastMessage{
subject: natsMsg.Subject,
data: natsMsg.Data,
ack: func() error {
return msg.Ack()
return natsMsg.Ack()
},
})
cancel()
if err != nil {
addFastMetric(registry, msg.Subject, "error")
logger.Error("fast write failed", "subject", msg.Subject, "error", err)
continue
}
addFastMetric(registry, msg.Subject, "ok")
}
operationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.OperationWait)
err = processFastBatch(operationCtx, registry, appender, updater, fastMessages)
cancel()
if err != nil {
addFastMetric(registry, fastBatchSubject(fastMessages), "error")
logger.Error("fast write batch failed", "messages", len(fastMessages), "error", err)
continue
}
for _, msg := range fastMessages {
addFastMetric(registry, msg.subject, "ok")
}
}
}
@@ -237,6 +242,53 @@ type natsPullSubscription interface {
var fastWriterStageDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}
func processFastBatch(ctx context.Context, registry *metrics.Registry, appender fastAppender, updater fastUpdater, messages []*fastMessage) error {
if len(messages) == 0 {
return nil
}
envelopes := make([]envelope.FrameEnvelope, 0, len(messages))
validMessages := make([]*fastMessage, 0, len(messages))
for _, msg := range messages {
var env envelope.FrameEnvelope
if err := json.Unmarshal(msg.data, &env); err != nil {
if msg.ack != nil {
_ = msg.ack()
}
continue
}
envelopes = append(envelopes, env)
validMessages = append(validMessages, msg)
}
if len(envelopes) == 0 {
return nil
}
subject := fastBatchSubject(validMessages)
started := time.Now()
err := appender.AppendAllBatch(ctx, envelopes)
recordFastWriterStageDuration(registry, subject, "tdengine", statusFromError(err), time.Since(started))
if err != nil {
return fmt.Errorf("tdengine batch append: %w", err)
}
for i, env := range envelopes {
msg := validMessages[i]
started = time.Now()
err = updater.FastUpdate(ctx, env)
recordFastWriterStageDuration(registry, msg.subject, "redis", statusFromError(err), time.Since(started))
if err != nil {
return fmt.Errorf("redis fast update: %w", err)
}
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
}
func processFastMessage(ctx context.Context, registry *metrics.Registry, appender fastAppender, updater fastUpdater, msg *fastMessage) error {
var env envelope.FrameEnvelope
if err := json.Unmarshal(msg.data, &env); err != nil {
@@ -268,6 +320,22 @@ func processFastMessage(ctx context.Context, registry *metrics.Registry, appende
return nil
}
func fastBatchSubject(messages []*fastMessage) string {
if len(messages) == 0 {
return "unknown"
}
subject := messages[0].subject
for _, msg := range messages[1:] {
if msg.subject != subject {
return "mixed"
}
}
if strings.TrimSpace(subject) == "" {
return "unknown"
}
return subject
}
func ensureStream(js nats.JetStreamContext, cfg config) error {
stream := &nats.StreamConfig{
Name: cfg.NATSStream,

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

@@ -80,9 +80,44 @@ func TestProcessFastMessageRecordsStageDurationMetrics(t *testing.T) {
}
}
func TestProcessFastBatchAppendsTDengineBatchBeforeRedisAndAck(t *testing.T) {
first := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN001", EventID: "evt-4"}
second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, VIN: "VIN002", EventID: "evt-5"}
firstPayload, err := first.MarshalJSONBytes()
if err != nil {
t.Fatal(err)
}
secondPayload, err := second.MarshalJSONBytes()
if err != nil {
t.Fatal(err)
}
appender := &recordingFastAppender{}
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, appender, updater, msgs); err != nil {
t.Fatalf("processFastBatch() error = %v", err)
}
if appender.count != 0 {
t.Fatalf("AppendAll count = %d, want 0", appender.count)
}
if appender.batchCount != 1 || appender.batchRows != 2 {
t.Fatalf("AppendAllBatch count=%d rows=%d, want count=1 rows=2", appender.batchCount, appender.batchRows)
}
if updater.count != 2 || ackCount != 2 {
t.Fatalf("updates=%d acks=%d, want 2/2", updater.count, ackCount)
}
}
type recordingFastAppender struct {
count int
err error
count int
batchCount int
batchRows int
err error
}
func (a *recordingFastAppender) AppendAll(context.Context, envelope.FrameEnvelope) error {
@@ -90,6 +125,12 @@ func (a *recordingFastAppender) AppendAll(context.Context, envelope.FrameEnvelop
return a.err
}
func (a *recordingFastAppender) AppendAllBatch(_ context.Context, envs []envelope.FrameEnvelope) error {
a.batchCount++
a.batchRows += len(envs)
return a.err
}
type recordingFastUpdater struct {
count int
err error