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

fix(gateway): guard nats stream retention and writer timeouts

Browse Source
This commit is contained in:
lingniu
2026-07-04 19:00:18 +08:00
parent 504a49a13c
commit 68b260c704
7 changed files with 166 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -132,6 +132,8 @@ type config struct {
OperationWait time.Duration
AckWait time.Duration
StreamMaxAge time.Duration
StreamMaxBytes int64
StreamEnsureWait time.Duration
Workers int
TDengineDriver string
TDengineDSN string
@@ -167,9 +169,11 @@ func loadConfig() config {
NATSSubjects: subjects,
BatchSize: envInt("FAST_WRITER_BATCH_SIZE", 100),
FetchWait: time.Duration(envInt("FAST_WRITER_FETCH_WAIT_MS", 100)) * time.Millisecond,
OperationWait: time.Duration(envInt("FAST_WRITER_OPERATION_TIMEOUT_MS", 100)) * time.Millisecond,
OperationWait: time.Duration(envInt("FAST_WRITER_OPERATION_TIMEOUT_MS", 1000)) * 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,
StreamMaxBytes: envInt64("NATS_STREAM_MAX_BYTES", 20*1024*1024*1024),
StreamEnsureWait: time.Duration(envInt("NATS_STREAM_ENSURE_TIMEOUT_SECONDS", 60)) * time.Second,
Workers: workers,
TDengineDriver: env("TDENGINE_DRIVER", "taosWS"),
TDengineDSN: env("TDENGINE_DSN", ""),
@@ -381,23 +385,27 @@ func fastBatchSubject(messages []*fastMessage) string {
}
func ensureStream(js nats.JetStreamContext, cfg config) error {
ctx, cancel := context.WithTimeout(context.Background(), cfg.StreamEnsureWait)
defer cancel()
opts := []nats.JSOpt{nats.Context(ctx)}
stream := &nats.StreamConfig{
Name: cfg.NATSStream,
Subjects: cfg.NATSSubjects,
Storage: nats.FileStorage,
Retention: nats.LimitsPolicy,
MaxAge: cfg.StreamMaxAge,
MaxBytes: cfg.StreamMaxBytes,
Duplicates: 2 * time.Minute,
}
if _, err := js.StreamInfo(cfg.NATSStream); err == nil {
_, err = js.UpdateStream(stream)
if _, err := js.StreamInfo(cfg.NATSStream, opts...); err == nil {
_, err = js.UpdateStream(stream, opts...)
return err
}
_, err := js.AddStream(stream)
_, err := js.AddStream(stream, opts...)
if err == nil {
return nil
}
_, updateErr := js.UpdateStream(stream)
_, updateErr := js.UpdateStream(stream, opts...)
if updateErr == nil {
return nil
}
@@ -467,6 +475,18 @@ func envInt(key string, fallback int) int {
return parsed
}
func envInt64(key string, fallback int64) int64 {
value := strings.TrimSpace(os.Getenv(key))
if value == "" {
return fallback
}
var parsed int64
if _, err := fmt.Sscanf(value, "%d", &parsed); err != nil || parsed <= 0 {
return fallback
}
return parsed
}
func splitCSV(value string) []string {
var out []string
for _, item := range strings.Split(value, ",") {

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

@@ -6,6 +6,7 @@ import (
"strings"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go"
@@ -243,6 +244,38 @@ func TestLoadConfigDefaultsTDenginePoolToSingleConnection(t *testing.T) {
}
}
func TestLoadConfigDefaultsStreamMaxBytes(t *testing.T) {
cfg := loadConfig()
if got, want := cfg.StreamMaxBytes, int64(20*1024*1024*1024); got != want {
t.Fatalf("StreamMaxBytes = %d, want %d", got, want)
}
if got, want := cfg.StreamEnsureWait, 60*time.Second; got != want {
t.Fatalf("StreamEnsureWait = %v, want %v", got, want)
}
if got, want := cfg.OperationWait, time.Second; got != want {
t.Fatalf("OperationWait = %v, want %v", got, want)
}
}
func TestLoadConfigReadsStreamMaxBytesOverride(t *testing.T) {
t.Setenv("NATS_STREAM_MAX_BYTES", "1073741824")
t.Setenv("NATS_STREAM_ENSURE_TIMEOUT_SECONDS", "90")
t.Setenv("FAST_WRITER_OPERATION_TIMEOUT_MS", "1500")
cfg := loadConfig()
if got, want := cfg.StreamMaxBytes, int64(1073741824); got != want {
t.Fatalf("StreamMaxBytes = %d, want %d", got, want)
}
if got, want := cfg.StreamEnsureWait, 90*time.Second; got != want {
t.Fatalf("StreamEnsureWait = %v, want %v", got, want)
}
if got, want := cfg.OperationWait, 1500*time.Millisecond; got != want {
t.Fatalf("OperationWait = %v, want %v", got, want)
}
}
func TestLoadConfigReadsTDenginePoolOverride(t *testing.T) {
t.Setenv("FAST_WRITER_WORKERS", "12")
t.Setenv("FAST_WRITER_TDENGINE_MAX_OPEN_CONNS", "4")

80
go/vehicle-gateway/cmd/nats-kafka-bridge/main.go
View File

@@ -85,19 +85,21 @@ func main() {
}
type config struct {
NATSURL string
NATSClientName string
NATSStream string
NATSDurable string
NATSFilter string
NATSSubjects []string
KafkaBrokers []string
Route map[string]string
BatchSize int
FetchWait time.Duration
OperationWait time.Duration
AckWait time.Duration
StreamMaxAge time.Duration
NATSURL string
NATSClientName string
NATSStream string
NATSDurable string
NATSFilter string
NATSSubjects []string
KafkaBrokers []string
Route map[string]string
BatchSize int
FetchWait time.Duration
OperationWait time.Duration
AckWait time.Duration
StreamMaxAge time.Duration
StreamMaxBytes int64
StreamEnsureWait time.Duration
}
func loadConfig() config {
@@ -113,19 +115,21 @@ func loadConfig() config {
route[unifiedSubject] = unifiedTopic
}
return config{
NATSURL: env("NATS_URL", "nats://127.0.0.1:4222"),
NATSClientName: env("NATS_CLIENT_NAME", "lingniu-nats-kafka-bridge"),
NATSStream: env("NATS_STREAM", "VEHICLE_INGEST"),
NATSDurable: env("NATS_DURABLE", "vehicle-kafka-bridge"),
NATSFilter: env("NATS_FILTER", "vehicle.>"),
NATSSubjects: splitCSV(env("NATS_STREAM_SUBJECTS", strings.Join(mapKeys(route), ","))),
KafkaBrokers: splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")),
Route: route,
BatchSize: envInt("BRIDGE_BATCH_SIZE", 500),
FetchWait: time.Duration(envInt("BRIDGE_FETCH_WAIT_MS", 1000)) * time.Millisecond,
OperationWait: time.Duration(envInt("BRIDGE_OPERATION_TIMEOUT_MS", 30000)) * time.Millisecond,
AckWait: time.Duration(envInt("NATS_ACK_WAIT_SECONDS", 60)) * time.Second,
StreamMaxAge: time.Duration(envInt("NATS_STREAM_MAX_AGE_HOURS", 24)) * time.Hour,
NATSURL: env("NATS_URL", "nats://127.0.0.1:4222"),
NATSClientName: env("NATS_CLIENT_NAME", "lingniu-nats-kafka-bridge"),
NATSStream: env("NATS_STREAM", "VEHICLE_INGEST"),
NATSDurable: env("NATS_DURABLE", "vehicle-kafka-bridge"),
NATSFilter: env("NATS_FILTER", "vehicle.>"),
NATSSubjects: splitCSV(env("NATS_STREAM_SUBJECTS", strings.Join(mapKeys(route), ","))),
KafkaBrokers: splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")),
Route: route,
BatchSize: envInt("BRIDGE_BATCH_SIZE", 500),
FetchWait: time.Duration(envInt("BRIDGE_FETCH_WAIT_MS", 1000)) * time.Millisecond,
OperationWait: time.Duration(envInt("BRIDGE_OPERATION_TIMEOUT_MS", 30000)) * time.Millisecond,
AckWait: time.Duration(envInt("NATS_ACK_WAIT_SECONDS", 60)) * time.Second,
StreamMaxAge: time.Duration(envInt("NATS_STREAM_MAX_AGE_HOURS", 24)) * time.Hour,
StreamMaxBytes: envInt64("NATS_STREAM_MAX_BYTES", 20*1024*1024*1024),
StreamEnsureWait: time.Duration(envInt("NATS_STREAM_ENSURE_TIMEOUT_SECONDS", 60)) * time.Second,
}
}
@@ -213,23 +217,27 @@ func runBridge(ctx context.Context, logger *slog.Logger, registry *metrics.Regis
}
func ensureStream(js nats.JetStreamContext, cfg config) error {
ctx, cancel := context.WithTimeout(context.Background(), cfg.StreamEnsureWait)
defer cancel()
opts := []nats.JSOpt{nats.Context(ctx)}
stream := &nats.StreamConfig{
Name: cfg.NATSStream,
Subjects: cfg.NATSSubjects,
Storage: nats.FileStorage,
Retention: nats.LimitsPolicy,
MaxAge: cfg.StreamMaxAge,
MaxBytes: cfg.StreamMaxBytes,
Duplicates: 2 * time.Minute,
}
if _, err := js.StreamInfo(cfg.NATSStream); err == nil {
_, err = js.UpdateStream(stream)
if _, err := js.StreamInfo(cfg.NATSStream, opts...); err == nil {
_, err = js.UpdateStream(stream, opts...)
return err
}
_, err := js.AddStream(stream)
_, err := js.AddStream(stream, opts...)
if err == nil {
return nil
}
_, updateErr := js.UpdateStream(stream)
_, updateErr := js.UpdateStream(stream, opts...)
if updateErr == nil {
return nil
}
@@ -362,6 +370,18 @@ func envInt(key string, fallback int) int {
return parsed
}
func envInt64(key string, fallback int64) int64 {
value := strings.TrimSpace(os.Getenv(key))
if value == "" {
return fallback
}
parsed, err := strconv.ParseInt(value, 10, 64)
if err != nil || parsed <= 0 {
return fallback
}
return parsed
}
func splitCSV(value string) []string {
var out []string
for _, item := range strings.Split(value, ",") {

26
go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go
View File

@@ -6,6 +6,7 @@ import (
"errors"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/segmentio/kafka-go"
@@ -177,6 +178,31 @@ func TestLoadConfigDefaultsToGoSubjectRoutes(t *testing.T) {
}
}
func TestLoadConfigDefaultsStreamMaxBytes(t *testing.T) {
cfg := loadConfig()
if got, want := cfg.StreamMaxBytes, int64(20*1024*1024*1024); got != want {
t.Fatalf("StreamMaxBytes = %d, want %d", got, want)
}
if got, want := cfg.StreamEnsureWait, 60*time.Second; got != want {
t.Fatalf("StreamEnsureWait = %v, want %v", got, want)
}
}
func TestLoadConfigReadsStreamMaxBytesOverride(t *testing.T) {
t.Setenv("NATS_STREAM_MAX_BYTES", "1073741824")
t.Setenv("NATS_STREAM_ENSURE_TIMEOUT_SECONDS", "90")
cfg := loadConfig()
if got, want := cfg.StreamMaxBytes, int64(1073741824); got != want {
t.Fatalf("StreamMaxBytes = %d, want %d", got, want)
}
if got, want := cfg.StreamEnsureWait, 90*time.Second; got != want {
t.Fatalf("StreamEnsureWait = %v, want %v", got, want)
}
}
func TestLoadConfigIncludesUnifiedOnlyWhenExplicitlyConfigured(t *testing.T) {
t.Setenv("NATS_SUBJECT_UNIFIED", "vehicle.event.go.unified.v1")
t.Setenv("KAFKA_TOPIC_UNIFIED", "vehicle.event.go.unified.v1")