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

feat(go): qualify tdengine history writes by database

Browse Source
This commit is contained in:
lingniu
2026-07-03 19:39:33 +08:00
parent fb34969602
commit 81e050ec1c
5 changed files with 65 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -72,19 +72,13 @@ func main() {
logger.Error("tdengine ping failed", "error", err)
os.Exit(1)
}
historyWriter := history.NewWriter(tdDB)
historyWriter := history.NewWriterWithDatabase(tdDB, cfg.TDengineDatabase)
if cfg.TDengineEnsureSchema {
if err := historyWriter.EnsureSchema(ctx, cfg.TDengineDatabase); err != nil {
logger.Error("tdengine schema bootstrap failed", "error", err)
os.Exit(1)
}
}
if strings.TrimSpace(cfg.TDengineDatabase) != "" {
if _, err := tdDB.ExecContext(ctx, "USE "+cfg.TDengineDatabase); err != nil {
logger.Error("tdengine use database failed", "database", cfg.TDengineDatabase, "error", err)
os.Exit(1)
}
}
redisClient := redis.NewClient(&redis.Options{
Addr: cfg.RedisAddr,

46
go/vehicle-gateway/internal/history/writer.go
View File

@@ -22,8 +22,9 @@ type Execer interface {
}
type Writer struct {
exec Execer
cache tableCache
exec Execer
database string
cache tableCache
}
const (
@@ -50,6 +51,12 @@ func NewWriter(exec Execer) *Writer {
return &Writer{exec: exec, cache: tableCache{seen: map[string]struct{}{}}}
}
func NewWriterWithDatabase(exec Execer, database string) *Writer {
writer := NewWriter(exec)
writer.database = normalizeIdentifier(database)
return writer
}
func (w *Writer) EnsureSchema(ctx context.Context, database string) error {
for _, statement := range SchemaStatements(database) {
if _, err := w.exec.ExecContext(ctx, statement); err != nil {
@@ -59,6 +66,14 @@ func (w *Writer) EnsureSchema(ctx context.Context, database string) error {
return nil
}
func (w *Writer) qualify(table string) string {
table = normalizeIdentifier(table)
if w.database == "" {
return table
}
return w.database + "." + table
}
func (w *Writer) AppendAll(ctx context.Context, env envelope.FrameEnvelope) error {
if err := w.AppendRawFrame(ctx, env); err != nil {
return err
@@ -87,7 +102,7 @@ func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope)
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint)
VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedFields))))
VALUES (%s)`, w.qualify(table), joinLiterals(rawValues(env, rawHex, rawText, parsedFields))))
if err != nil {
return err
}
@@ -103,7 +118,7 @@ VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedFields))
for _, chunk := range chunks {
if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, event_id, frame_id, received_at, payload_kind, chunk_index, chunk_count, chunk_text)
VALUES (%s)`, chunkTable, joinLiterals(chunkValues(env, chunk)))); err != nil {
VALUES (%s)`, w.qualify(chunkTable), joinLiterals(chunkValues(env, chunk)))); err != nil {
return err
}
}
@@ -144,7 +159,7 @@ func (w *Writer) AppendRawFrameBatch(ctx context.Context, envelopes []envelope.F
if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint)
VALUES %s`, table, strings.Join(rows, ","))); err != nil {
VALUES %s`, w.qualify(table), strings.Join(rows, ","))); err != nil {
return err
}
}
@@ -155,7 +170,7 @@ VALUES %s`, table, strings.Join(rows, ","))); err != nil {
_ = chunkEnvByTable[table]
if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, event_id, frame_id, received_at, payload_kind, chunk_index, chunk_count, chunk_text)
VALUES %s`, table, strings.Join(rows, ","))); err != nil {
VALUES %s`, w.qualify(table), strings.Join(rows, ","))); err != nil {
return err
}
}
@@ -178,7 +193,7 @@ func (w *Writer) AppendLocation(ctx context.Context, env envelope.FrameEnvelope)
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, event_id, received_at, longitude, latitude, altitude_m, speed_kmh,
direction_deg, alarm_flag, status_flag, total_mileage_km)
VALUES (%s)`, table, joinLiterals(locationValues(env, longitude, latitude))))
VALUES (%s)`, w.qualify(table), joinLiterals(locationValues(env, longitude, latitude))))
return err
}
@@ -206,7 +221,7 @@ func (w *Writer) AppendLocationBatch(ctx context.Context, envelopes []envelope.F
if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, event_id, received_at, longitude, latitude, altitude_m, speed_kmh,
direction_deg, alarm_flag, status_flag, total_mileage_km)
VALUES %s`, table, strings.Join(rows, ","))); err != nil {
VALUES %s`, w.qualify(table), strings.Join(rows, ","))); err != nil {
return err
}
}
@@ -222,7 +237,7 @@ func (w *Writer) ensureRawChild(ctx context.Context, table string, stable string
return nil
}
statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s', '%s', '%s', '%s')",
table, stable,
w.qualify(table), w.qualify(stable),
quote(string(env.Protocol)),
quote(env.VehicleKey()),
quote(env.VIN),
@@ -232,7 +247,7 @@ func (w *Writer) ensureRawChild(ctx context.Context, table string, stable string
return err
}
if phone := normalizePhoneTag(env.Phone); phone != "" {
if _, err := w.exec.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s SET TAG phone = '%s'", table, quote(phone))); err != nil {
if _, err := w.exec.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s SET TAG phone = '%s'", w.qualify(table), quote(phone))); err != nil {
return err
}
}
@@ -250,8 +265,9 @@ func (w *Writer) ensureLocationChild(ctx context.Context, table string, env enve
if ok {
return nil
}
statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING vehicle_locations TAGS ('%s', '%s')",
table,
statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s')",
w.qualify(table),
w.qualify("vehicle_locations"),
quote(string(env.Protocol)),
quote(strings.TrimSpace(env.VIN)))
if _, err := w.exec.ExecContext(ctx, statement); err != nil {
@@ -367,6 +383,12 @@ func locationTableName(env envelope.FrameEnvelope) string {
return "loc_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(strings.TrimSpace(env.VIN))
}
func normalizeIdentifier(value string) string {
value = strings.TrimSpace(value)
value = strings.Trim(value, "`")
return value
}
func normalizePhoneTag(phone string) string {
trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0")
if trimmed == "" {

28
go/vehicle-gateway/internal/history/writer_test.go
View File

@@ -261,6 +261,30 @@ func TestWriterAppendsBatchRowsByChildTable(t *testing.T) {
}
}
func TestWriterWithDatabaseQualifiesTDengineTables(t *testing.T) {
exec := &recordingExec{}
writer := NewWriterWithDatabase(exec, "vehicle_ts")
env := sampleEnvelope()
if err := writer.AppendAll(context.Background(), env); err != nil {
t.Fatalf("AppendAll() error = %v", err)
}
for _, want := range []string{
"CREATE TABLE IF NOT EXISTS vehicle_ts.raw_",
"USING vehicle_ts.raw_frames",
"ALTER TABLE vehicle_ts.raw_",
"INSERT INTO vehicle_ts.raw_",
"CREATE TABLE IF NOT EXISTS vehicle_ts.loc_",
"USING vehicle_ts.vehicle_locations",
"INSERT INTO vehicle_ts.loc_",
} {
if !containsSQL(exec.calls, want) {
t.Fatalf("qualified sql missing %q, calls=%v", want, exec.calls)
}
}
}
func TestWriterAppendAllBatchSkipsEmptyBatch(t *testing.T) {
exec := &recordingExec{}
writer := NewWriter(exec)
@@ -338,6 +362,10 @@ func findSQL(calls []execCall, pattern string) string {
return ""
}
func containsSQL(calls []execCall, pattern string) bool {
return findSQL(calls, pattern) != ""
}
func between(value string, start string, end string) string {
startIndex := strings.Index(value, start)
if startIndex < 0 {