124 lines
3.4 KiB
Go
124 lines
3.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
|
|
"github.com/segmentio/kafka-go"
|
|
_ "github.com/taosdata/driver-go/v3/taosWS"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/history"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
|
|
)
|
|
|
|
func main() {
|
|
logger := observability.NewLogger("vehicle-history-writer")
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
cfg := loadConfig()
|
|
db, err := sql.Open(cfg.TDengineDriver, cfg.TDengineDSN)
|
|
if err != nil {
|
|
logger.Error("tdengine open failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
defer db.Close()
|
|
if err := db.PingContext(ctx); err != nil {
|
|
logger.Error("tdengine ping failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
writer := history.NewWriter(db)
|
|
if cfg.EnsureSchema {
|
|
if err := writer.EnsureSchema(ctx, cfg.TDengineDatabase); err != nil {
|
|
logger.Error("tdengine schema bootstrap failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
reader := kafka.NewReader(kafka.ReaderConfig{
|
|
Brokers: cfg.KafkaBrokers,
|
|
GroupID: cfg.KafkaGroup,
|
|
GroupTopics: cfg.KafkaTopics,
|
|
MinBytes: 1,
|
|
MaxBytes: 10e6,
|
|
})
|
|
defer reader.Close()
|
|
|
|
logger.Info("history writer started",
|
|
"driver", cfg.TDengineDriver,
|
|
"group", cfg.KafkaGroup,
|
|
"topics", strings.Join(cfg.KafkaTopics, ","))
|
|
|
|
for {
|
|
message, err := reader.FetchMessage(ctx)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
logger.Error("kafka fetch failed", "error", err)
|
|
continue
|
|
}
|
|
var env envelope.FrameEnvelope
|
|
if err := json.Unmarshal(message.Value, &env); err != nil {
|
|
logger.Warn("skip invalid envelope json", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
|
|
_ = reader.CommitMessages(ctx, message)
|
|
continue
|
|
}
|
|
if err := writer.AppendAll(ctx, env); err != nil {
|
|
logger.Error("tdengine append failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", err)
|
|
continue
|
|
}
|
|
if err := reader.CommitMessages(ctx, message); err != nil {
|
|
logger.Error("kafka commit failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
type config struct {
|
|
KafkaBrokers []string
|
|
KafkaTopics []string
|
|
KafkaGroup string
|
|
TDengineDriver string
|
|
TDengineDSN string
|
|
TDengineDatabase string
|
|
EnsureSchema bool
|
|
}
|
|
|
|
func loadConfig() config {
|
|
return config{
|
|
KafkaBrokers: splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")),
|
|
KafkaTopics: splitCSV(env("KAFKA_TOPICS", "vehicle.raw.gb32960.v1,vehicle.raw.jt808.v1,vehicle.raw.yutong-mqtt.v1")),
|
|
KafkaGroup: env("KAFKA_GROUP", "go-history-writer"),
|
|
TDengineDriver: env("TDENGINE_DRIVER", "taosWS"),
|
|
TDengineDSN: env("TDENGINE_DSN", ""),
|
|
TDengineDatabase: env("TDENGINE_DATABASE", history.DefaultDatabase),
|
|
EnsureSchema: env("TDENGINE_ENSURE_SCHEMA", "true") != "false",
|
|
}
|
|
}
|
|
|
|
func env(key string, fallback string) string {
|
|
value := strings.TrimSpace(os.Getenv(key))
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
return value
|
|
}
|
|
|
|
func splitCSV(value string) []string {
|
|
var out []string
|
|
for _, item := range strings.Split(value, ",") {
|
|
item = strings.TrimSpace(item)
|
|
if item != "" {
|
|
out = append(out, item)
|
|
}
|
|
}
|
|
return out
|
|
}
|