feat(go): add nats kafka ingest bridge
This commit is contained in:
lingniu
260
go/vehicle-gateway/cmd/nats-kafka-bridge/main.go
Normal file
260
go/vehicle-gateway/cmd/nats-kafka-bridge/main.go
Normal file
@@ -0,0 +1,260 @@
|
||||
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/observability"
|
||||
)
|
||||
|
||||
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()
|
||||
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, 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
|
||||
}
|
||||
|
||||
func loadConfig() config {
|
||||
route := map[string]string{
|
||||
env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", "vehicle.raw.gb32960.v1")): env("KAFKA_TOPIC_GB32960_RAW", "vehicle.raw.gb32960.v1"),
|
||||
env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", "vehicle.raw.jt808.v1")): env("KAFKA_TOPIC_JT808_RAW", "vehicle.raw.jt808.v1"),
|
||||
env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", "vehicle.raw.yutong-mqtt.v1")): env("KAFKA_TOPIC_YUTONG_MQTT_RAW", "vehicle.raw.yutong-mqtt.v1"),
|
||||
env("NATS_SUBJECT_UNIFIED", env("KAFKA_TOPIC_UNIFIED", "vehicle.event.unified.v1")): env("KAFKA_TOPIC_UNIFIED", "vehicle.event.unified.v1"),
|
||||
}
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
type natsPullSubscription interface {
|
||||
Fetch(int, ...nats.PullOpt) ([]*nats.Msg, 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, sub natsPullSubscription, writer kafkaBatchWriter, cfg config) {
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
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, 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 {
|
||||
stream := &nats.StreamConfig{
|
||||
Name: cfg.NATSStream,
|
||||
Subjects: cfg.NATSSubjects,
|
||||
Storage: nats.FileStorage,
|
||||
Retention: nats.LimitsPolicy,
|
||||
MaxAge: cfg.StreamMaxAge,
|
||||
Duplicates: 2 * time.Minute,
|
||||
}
|
||||
if _, err := js.StreamInfo(cfg.NATSStream); err == nil {
|
||||
_, err = js.UpdateStream(stream)
|
||||
return err
|
||||
}
|
||||
_, err := js.AddStream(stream)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
_, updateErr := js.UpdateStream(stream)
|
||||
if updateErr == nil {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func bridgeBatch(ctx context.Context, writer kafkaBatchWriter, messages []bridgeMessage, route map[string]string) error {
|
||||
if len(messages) == 0 {
|
||||
return nil
|
||||
}
|
||||
kafkaMessages := make([]kafka.Message, 0, len(messages))
|
||||
for _, message := range messages {
|
||||
topic, ok := route[message.subject]
|
||||
if !ok || topic == "" {
|
||||
return fmt.Errorf("kafka topic not configured for nats subject %q", message.subject)
|
||||
}
|
||||
kafkaMessages = append(kafkaMessages, kafkaMessage(topic, message.data))
|
||||
}
|
||||
if err := writer.WriteMessages(ctx, kafkaMessages...); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, message := range messages {
|
||||
if message.ack == nil {
|
||||
continue
|
||||
}
|
||||
if err := message.ack(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
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 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 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
|
||||
}
|
||||
75
go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go
Normal file
75
go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go
Normal file
@@ -0,0 +1,75 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/segmentio/kafka-go"
|
||||
|
||||
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
||||
)
|
||||
|
||||
func TestBridgeBatchWritesKafkaThenAcks(t *testing.T) {
|
||||
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
|
||||
payload, err := json.Marshal(env)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writer := &recordingBridgeWriter{}
|
||||
acked := 0
|
||||
|
||||
err = bridgeBatch(context.Background(), writer, []bridgeMessage{
|
||||
{subject: "vehicle.raw.jt808.v1", data: payload, ack: func() error {
|
||||
acked++
|
||||
return nil
|
||||
}},
|
||||
}, map[string]string{"vehicle.raw.jt808.v1": "vehicle.raw.jt808.v1"})
|
||||
if err != nil {
|
||||
t.Fatalf("bridgeBatch() error = %v", err)
|
||||
}
|
||||
if len(writer.messages) != 1 {
|
||||
t.Fatalf("kafka writes = %d, want 1", len(writer.messages))
|
||||
}
|
||||
if got, want := writer.messages[0].Topic, "vehicle.raw.jt808.v1"; got != want {
|
||||
t.Fatalf("topic = %q, want %q", got, want)
|
||||
}
|
||||
if got, want := string(writer.messages[0].Key), "JT808:13307795425"; got != want {
|
||||
t.Fatalf("key = %q, want %q", got, want)
|
||||
}
|
||||
if acked != 1 {
|
||||
t.Fatalf("acks = %d, want 1", acked)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBridgeBatchDoesNotAckWhenKafkaFails(t *testing.T) {
|
||||
writer := &recordingBridgeWriter{err: errors.New("kafka unavailable")}
|
||||
acked := 0
|
||||
|
||||
err := bridgeBatch(context.Background(), writer, []bridgeMessage{
|
||||
{subject: "vehicle.event.unified.v1", data: []byte(`{"protocol":"JT808"}`), ack: func() error {
|
||||
acked++
|
||||
return nil
|
||||
}},
|
||||
}, map[string]string{"vehicle.event.unified.v1": "vehicle.event.unified.v1"})
|
||||
if err == nil {
|
||||
t.Fatal("bridgeBatch() error is nil, want kafka error")
|
||||
}
|
||||
if acked != 0 {
|
||||
t.Fatalf("acks = %d, want 0", acked)
|
||||
}
|
||||
}
|
||||
|
||||
type recordingBridgeWriter struct {
|
||||
messages []kafka.Message
|
||||
err error
|
||||
}
|
||||
|
||||
func (w *recordingBridgeWriter) WriteMessages(_ context.Context, messages ...kafka.Message) error {
|
||||
if w.err != nil {
|
||||
return w.err
|
||||
}
|
||||
w.messages = append(w.messages, messages...)
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user