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

fix(go): keep fast writer tdengine pool safe by default

Browse Source
This commit is contained in:
lingniu
2026-07-03 19:33:45 +08:00
parent 06866b0fa7
commit fb34969602
4 changed files with 51 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -66,8 +66,8 @@ func main() {
os.Exit(1)
}
defer tdDB.Close()
tdDB.SetMaxOpenConns(1)
tdDB.SetMaxIdleConns(1)
tdDB.SetMaxOpenConns(cfg.TDengineMaxOpenConns)
tdDB.SetMaxIdleConns(cfg.TDengineMaxIdleConns)
if err := tdDB.PingContext(ctx); err != nil {
logger.Error("tdengine ping failed", "error", err)
os.Exit(1)
@@ -117,6 +117,8 @@ func main() {
"filter", cfg.NATSFilter,
"batch_size", cfg.BatchSize,
"workers", cfg.Workers,
"tdengine_max_open_conns", cfg.TDengineMaxOpenConns,
"tdengine_max_idle_conns", cfg.TDengineMaxIdleConns,
"operation_timeout_ms", cfg.OperationWait.Milliseconds())
for i := 0; i < cfg.Workers; i++ {
go runFastWorker(ctx, logger, registry, js, sub, historyWriter, realtimeRepo, cfg)
@@ -141,6 +143,8 @@ type config struct {
TDengineDSN string
TDengineDatabase string
TDengineEnsureSchema bool
TDengineMaxOpenConns int
TDengineMaxIdleConns int
RedisAddr string
RedisUsername string
RedisPassword string
@@ -157,6 +161,9 @@ func loadConfig() config {
env("NATS_SUBJECT_JT808_FIELDS", env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)),
env("NATS_SUBJECT_YUTONG_MQTT_FIELDS", env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)),
}, ",")))
workers := envInt("FAST_WRITER_WORKERS", 8)
maxOpenConns := envInt("FAST_WRITER_TDENGINE_MAX_OPEN_CONNS", 1)
maxIdleConns := envInt("FAST_WRITER_TDENGINE_MAX_IDLE_CONNS", maxOpenConns)
return config{
NATSURL: env("NATS_URL", "nats://127.0.0.1:4222"),
NATSClientName: env("NATS_CLIENT_NAME", "lingniu-nats-fast-writer"),
@@ -169,11 +176,13 @@ func loadConfig() config {
OperationWait: time.Duration(envInt("FAST_WRITER_OPERATION_TIMEOUT_MS", 100)) * time.Millisecond,
AckWait: time.Duration(envInt("NATS_ACK_WAIT_SECONDS", 30)) * time.Second,
StreamMaxAge: time.Duration(envInt("NATS_STREAM_MAX_AGE_HOURS", 24)) * time.Hour,
Workers: envInt("FAST_WRITER_WORKERS", 8),
Workers: workers,
TDengineDriver: env("TDENGINE_DRIVER", "taosWS"),
TDengineDSN: env("TDENGINE_DSN", ""),
TDengineDatabase: env("TDENGINE_DATABASE", history.DefaultDatabase),
TDengineEnsureSchema: env("TDENGINE_ENSURE_SCHEMA", "true") != "false",
TDengineMaxOpenConns: maxOpenConns,
TDengineMaxIdleConns: maxIdleConns,
RedisAddr: env("REDIS_ADDR", "127.0.0.1:6379"),
RedisUsername: env("REDIS_USERNAME", ""),
RedisPassword: env("REDIS_PASSWORD", ""),

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

@@ -197,6 +197,35 @@ func TestRecordFastNATSConsumerInfoMetrics(t *testing.T) {
}
}
func TestLoadConfigDefaultsTDenginePoolToSingleConnection(t *testing.T) {
t.Setenv("FAST_WRITER_WORKERS", "12")
t.Setenv("FAST_WRITER_TDENGINE_MAX_OPEN_CONNS", "")
cfg := loadConfig()
if cfg.TDengineMaxOpenConns != 1 {
t.Fatalf("TDengineMaxOpenConns = %d, want 1", cfg.TDengineMaxOpenConns)
}
if cfg.TDengineMaxIdleConns != 1 {
t.Fatalf("TDengineMaxIdleConns = %d, want 1", cfg.TDengineMaxIdleConns)
}
}
func TestLoadConfigReadsTDenginePoolOverride(t *testing.T) {
t.Setenv("FAST_WRITER_WORKERS", "12")
t.Setenv("FAST_WRITER_TDENGINE_MAX_OPEN_CONNS", "4")
t.Setenv("FAST_WRITER_TDENGINE_MAX_IDLE_CONNS", "2")
cfg := loadConfig()
if cfg.TDengineMaxOpenConns != 4 {
t.Fatalf("TDengineMaxOpenConns = %d, want 4", cfg.TDengineMaxOpenConns)
}
if cfg.TDengineMaxIdleConns != 2 {
t.Fatalf("TDengineMaxIdleConns = %d, want 2", cfg.TDengineMaxIdleConns)
}
}
type recordingFastAppender struct {
count int
batchCount int