403 lines
13 KiB
Go
403 lines
13 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"os/signal"
|
|
"strconv"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/segmentio/kafka-go"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics"
|
|
)
|
|
|
|
func main() {
|
|
logger := observability.NewLogger("nats-kafka-bridge")
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
cfg := loadConfig()
|
|
conn, err := nats.Connect(cfg.NATSURL, nats.Name(cfg.NATSClientName), nats.Timeout(5*time.Second))
|
|
if err != nil {
|
|
logger.Error("nats connect failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
defer conn.Close()
|
|
registry := metrics.NewRegistry()
|
|
health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "nats-kafka-bridge", []health.Check{
|
|
{Name: "nats", Check: func(context.Context) error {
|
|
if conn.Status() != nats.CONNECTED {
|
|
return fmt.Errorf("nats status is %s", conn.Status().String())
|
|
}
|
|
return nil
|
|
}},
|
|
}, registry))
|
|
js, err := conn.JetStream()
|
|
if err != nil {
|
|
logger.Error("nats jetstream init failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
if err := ensureStream(js, cfg); err != nil {
|
|
logger.Error("nats stream ensure failed", "stream", cfg.NATSStream, "error", err)
|
|
os.Exit(1)
|
|
}
|
|
sub, err := js.PullSubscribe(
|
|
cfg.NATSFilter,
|
|
cfg.NATSDurable,
|
|
nats.BindStream(cfg.NATSStream),
|
|
nats.ManualAck(),
|
|
nats.AckWait(cfg.AckWait),
|
|
nats.MaxDeliver(-1),
|
|
)
|
|
if err != nil {
|
|
logger.Error("nats pull consumer init failed", "stream", cfg.NATSStream, "durable", cfg.NATSDurable, "error", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
writer := &kafka.Writer{
|
|
Addr: kafka.TCP(cfg.KafkaBrokers...),
|
|
Balancer: &kafka.Hash{},
|
|
AllowAutoTopicCreation: false,
|
|
RequiredAcks: kafka.RequireAll,
|
|
Async: false,
|
|
}
|
|
defer writer.Close()
|
|
|
|
logger.Info("nats kafka bridge started",
|
|
"nats_url", cfg.NATSURL,
|
|
"stream", cfg.NATSStream,
|
|
"durable", cfg.NATSDurable,
|
|
"filter", cfg.NATSFilter,
|
|
"kafka_brokers", strings.Join(cfg.KafkaBrokers, ","))
|
|
runBridge(ctx, logger, registry, js, sub, writer, cfg)
|
|
}
|
|
|
|
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
|
|
StreamMaxBytes int64
|
|
StreamEnsureWait time.Duration
|
|
}
|
|
|
|
func loadConfig() config {
|
|
route := map[string]string{
|
|
env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)): env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960),
|
|
env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)): env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808),
|
|
env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)): env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT),
|
|
env("NATS_SUBJECT_GB32960_FIELDS", env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960)): env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960),
|
|
env("NATS_SUBJECT_JT808_FIELDS", env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)): env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808),
|
|
env("NATS_SUBJECT_YUTONG_MQTT_FIELDS", env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)): env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT),
|
|
}
|
|
if unifiedSubject, unifiedTopic, ok := unifiedRouteFromEnv(); ok {
|
|
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,
|
|
StreamMaxBytes: envInt64("NATS_STREAM_MAX_BYTES", 20*1024*1024*1024),
|
|
StreamEnsureWait: time.Duration(envInt("NATS_STREAM_ENSURE_TIMEOUT_SECONDS", 60)) * time.Second,
|
|
}
|
|
}
|
|
|
|
func unifiedRouteFromEnv() (string, string, bool) {
|
|
subject, hasSubject := envOptional("NATS_SUBJECT_UNIFIED")
|
|
topic, hasTopic := envOptional("KAFKA_TOPIC_UNIFIED")
|
|
if !hasSubject && !hasTopic {
|
|
return "", "", false
|
|
}
|
|
if subject == "" {
|
|
subject = topic
|
|
}
|
|
if topic == "" {
|
|
topic = subject
|
|
}
|
|
if subject == "" || topic == "" {
|
|
return "", "", false
|
|
}
|
|
return subject, topic, true
|
|
}
|
|
|
|
type natsPullSubscription interface {
|
|
Fetch(int, ...nats.PullOpt) ([]*nats.Msg, error)
|
|
}
|
|
|
|
type natsConsumerInfoReader interface {
|
|
ConsumerInfo(stream string, name string, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
|
|
}
|
|
|
|
type kafkaBatchWriter interface {
|
|
WriteMessages(context.Context, ...kafka.Message) error
|
|
}
|
|
|
|
type bridgeMessage struct {
|
|
subject string
|
|
data []byte
|
|
ack func() error
|
|
}
|
|
|
|
func runBridge(ctx context.Context, logger *slog.Logger, registry *metrics.Registry, infoReader natsConsumerInfoReader, sub natsPullSubscription, writer kafkaBatchWriter, cfg config) {
|
|
var lastConsumerInfoAt time.Time
|
|
for {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
if time.Since(lastConsumerInfoAt) >= time.Duration(envInt("NATS_CONSUMER_METRICS_INTERVAL_SECONDS", 10))*time.Second {
|
|
lastConsumerInfoAt = time.Now()
|
|
if infoReader != nil {
|
|
info, err := infoReader.ConsumerInfo(cfg.NATSStream, cfg.NATSDurable, nats.Context(ctx))
|
|
if err != nil {
|
|
logger.Warn("nats consumer info failed", "stream", cfg.NATSStream, "durable", cfg.NATSDurable, "error", err)
|
|
} else {
|
|
recordNATSConsumerInfoMetrics(registry, cfg, info)
|
|
}
|
|
}
|
|
}
|
|
msgs, err := sub.Fetch(cfg.BatchSize, nats.MaxWait(cfg.FetchWait))
|
|
if err != nil {
|
|
if errors.Is(err, nats.ErrTimeout) {
|
|
continue
|
|
}
|
|
logger.Error("nats fetch failed", "error", err)
|
|
time.Sleep(time.Second)
|
|
continue
|
|
}
|
|
bridgeMessages := make([]bridgeMessage, 0, len(msgs))
|
|
for _, msg := range msgs {
|
|
msg := msg
|
|
bridgeMessages = append(bridgeMessages, bridgeMessage{
|
|
subject: msg.Subject,
|
|
data: msg.Data,
|
|
ack: func() error {
|
|
return msg.Ack()
|
|
},
|
|
})
|
|
}
|
|
operationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.OperationWait)
|
|
err = bridgeBatch(operationCtx, registry, writer, bridgeMessages, cfg.Route)
|
|
cancel()
|
|
if err != nil {
|
|
logger.Error("bridge batch failed", "count", len(bridgeMessages), "error", err)
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
|
|
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, opts...); err == nil {
|
|
_, err = js.UpdateStream(stream, opts...)
|
|
return err
|
|
}
|
|
_, err := js.AddStream(stream, opts...)
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
_, updateErr := js.UpdateStream(stream, opts...)
|
|
if updateErr == nil {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
func bridgeBatch(ctx context.Context, registry *metrics.Registry, writer kafkaBatchWriter, messages []bridgeMessage, route map[string]string) error {
|
|
if len(messages) == 0 {
|
|
return nil
|
|
}
|
|
started := time.Now()
|
|
status := "ok"
|
|
defer func() {
|
|
recordBridgeBatchDuration(registry, status, time.Since(started))
|
|
recordBridgeBatchPending(registry, 0, 0)
|
|
}()
|
|
kafkaMessages := make([]kafka.Message, 0, len(messages))
|
|
for _, message := range messages {
|
|
addBridgeSubjectMetric(registry, "vehicle_bridge_messages_total", message.subject, "received")
|
|
topic, ok := route[message.subject]
|
|
if !ok || topic == "" {
|
|
addBridgeSubjectMetric(registry, "vehicle_bridge_messages_total", message.subject, "route_error")
|
|
status = "error"
|
|
return fmt.Errorf("kafka topic not configured for nats subject %q", message.subject)
|
|
}
|
|
kafkaMessages = append(kafkaMessages, kafkaMessage(topic, message.data))
|
|
}
|
|
recordBridgeBatchPending(registry, len(messages), len(kafkaMessages))
|
|
if err := writer.WriteMessages(ctx, kafkaMessages...); err != nil {
|
|
for _, message := range kafkaMessages {
|
|
addBridgeTopicMetric(registry, "vehicle_bridge_kafka_writes_total", message.Topic, "error")
|
|
}
|
|
status = "error"
|
|
return err
|
|
}
|
|
for _, message := range kafkaMessages {
|
|
addBridgeTopicMetric(registry, "vehicle_bridge_kafka_writes_total", message.Topic, "ok")
|
|
}
|
|
for _, message := range messages {
|
|
if message.ack == nil {
|
|
continue
|
|
}
|
|
if err := message.ack(); err != nil {
|
|
addBridgeSubjectMetric(registry, "vehicle_bridge_nats_acks_total", message.subject, "error")
|
|
status = "error"
|
|
return err
|
|
}
|
|
addBridgeSubjectMetric(registry, "vehicle_bridge_nats_acks_total", message.subject, "ok")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var bridgeBatchDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}
|
|
|
|
func addBridgeSubjectMetric(registry *metrics.Registry, name string, subject string, status string) {
|
|
if registry == nil {
|
|
return
|
|
}
|
|
registry.IncCounter(name, metrics.Labels{"subject": subject, "status": status})
|
|
}
|
|
|
|
func addBridgeTopicMetric(registry *metrics.Registry, name string, topic string, status string) {
|
|
if registry == nil {
|
|
return
|
|
}
|
|
registry.IncCounter(name, metrics.Labels{"topic": topic, "status": status})
|
|
}
|
|
|
|
func recordBridgeBatchPending(registry *metrics.Registry, messages int, kafkaMessages int) {
|
|
if registry == nil {
|
|
return
|
|
}
|
|
registry.SetGauge("vehicle_bridge_batch_pending_messages", nil, float64(messages))
|
|
registry.SetGauge("vehicle_bridge_batch_pending_kafka_messages", nil, float64(kafkaMessages))
|
|
}
|
|
|
|
func recordBridgeBatchDuration(registry *metrics.Registry, status string, elapsed time.Duration) {
|
|
if registry == nil {
|
|
return
|
|
}
|
|
registry.ObserveHistogram("vehicle_bridge_batch_duration_ms_histogram", metrics.Labels{
|
|
"status": status,
|
|
}, bridgeBatchDurationBucketsMS, float64(elapsed.Milliseconds()))
|
|
}
|
|
|
|
func recordNATSConsumerInfoMetrics(registry *metrics.Registry, cfg config, info *nats.ConsumerInfo) {
|
|
if registry == nil || info == nil {
|
|
return
|
|
}
|
|
labels := metrics.Labels{"stream": cfg.NATSStream, "consumer": cfg.NATSDurable}
|
|
registry.SetGauge("vehicle_bridge_nats_consumer_pending", labels, float64(info.NumPending))
|
|
registry.SetGauge("vehicle_bridge_nats_consumer_ack_pending", labels, float64(info.NumAckPending))
|
|
registry.SetGauge("vehicle_bridge_nats_consumer_waiting", labels, float64(info.NumWaiting))
|
|
}
|
|
|
|
func kafkaMessage(topic string, data []byte) kafka.Message {
|
|
var env envelope.FrameEnvelope
|
|
message := kafka.Message{Topic: topic, Value: data}
|
|
if err := json.Unmarshal(data, &env); err == nil {
|
|
message.Key = env.KafkaKey()
|
|
}
|
|
return message
|
|
}
|
|
|
|
func env(key string, fallback string) string {
|
|
value := strings.TrimSpace(os.Getenv(key))
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
return value
|
|
}
|
|
|
|
func envOptional(key string) (string, bool) {
|
|
value, ok := os.LookupEnv(key)
|
|
if !ok {
|
|
return "", false
|
|
}
|
|
return strings.TrimSpace(value), true
|
|
}
|
|
|
|
func envInt(key string, fallback int) int {
|
|
value := strings.TrimSpace(os.Getenv(key))
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
parsed, err := strconv.Atoi(value)
|
|
if err != nil || parsed <= 0 {
|
|
return fallback
|
|
}
|
|
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, ",") {
|
|
item = strings.TrimSpace(item)
|
|
if item != "" {
|
|
out = append(out, item)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func mapKeys(values map[string]string) []string {
|
|
out := make([]string, 0, len(values))
|
|
for key := range values {
|
|
out = append(out, key)
|
|
}
|
|
return out
|
|
}
|