From 4c34c1221b3b57e8df38f47f5d89a8c0476b8486 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 08:25:14 +0800 Subject: [PATCH] refactor(go): make raw topics the realtime source --- .../iot-data-platform-principles.md | 6 +-- .../production-data-plane-inventory.md | 6 +-- docs/go-version-data-flow.html | 8 ++-- docs/ops/go-service-observability.md | 2 +- docs/ops/vehicle-ingest-runbook.md | 4 +- go/vehicle-gateway/cmd/gateway/main.go | 5 ++- .../cmd/nats-kafka-bridge/main.go | 30 +++++++++++++- .../cmd/nats-kafka-bridge/main_test.go | 12 +++++- go/vehicle-gateway/cmd/realtime-api/main.go | 3 +- .../cmd/realtime-api/main_test.go | 14 +++---- .../internal/gateway/mqtt_client.go | 13 +++--- .../internal/gateway/mqtt_client_test.go | 36 ++++++++++++++-- .../internal/gateway/tcp_server.go | 15 ++++--- .../internal/gateway/tcp_server_test.go | 41 ++++++++++++++++--- 14 files changed, 151 insertions(+), 44 deletions(-) diff --git a/docs/architecture/iot-data-platform-principles.md b/docs/architecture/iot-data-platform-principles.md index d9a8dd30..e6d11e01 100644 --- a/docs/architecture/iot-data-platform-principles.md +++ b/docs/architecture/iot-data-platform-principles.md @@ -52,8 +52,8 @@ flowchart LR | MySQL | `vehicle_daily_mileage` | Queryable daily mileage | vin, date, protocol, daily mileage, source mileage | temporary vehicle keys, generic metric key/value rows, per-frame raw details | | MySQL | `vehicle_identity_binding` | Manual identity mapping | vin, plate, phone, device id | registration history | | MySQL | `jt808_registration` | JT808 registration and auth trace | phone, device id, plate, auth code, vin match state, first/latest seen | GB32960 or MQTT records | -| Redis | `vehicle:latest:{vehicleKey}` | Latest merged realtime state | cross-protocol latest fields only | full parsed payloads and historical data | -| Redis | `vehicle:realtime-raw:{protocol}:{vehicleKey}` | Latest full realtime protocol state | latest protocol parsed payload | historical data | +| Redis | `vehicle:latest:{vin}` | Latest merged realtime state | cross-protocol latest fields only for VIN-bound vehicles | full parsed payloads, temporary identities, and historical data | +| Redis | `vehicle:realtime-raw:{protocol}:{vin}` | Latest full realtime protocol state | latest protocol parsed payload for VIN-bound vehicles | historical data | ## Event Envelope Rules @@ -91,7 +91,7 @@ Runtime metrics are exposed as Prometheus text from `/metrics`. They are operati Core counters: - `vehicle_gateway_frames_total`: received protocol frames by protocol and parse status. -- `vehicle_gateway_publish_total`: raw and unified publish results by protocol. +- `vehicle_gateway_publish_total`: raw publish results by protocol. Unified publish metrics appear only when the explicit compatibility stream is enabled. - `vehicle_realtime_kafka_messages_total`: realtime consumer messages by topic and status. - `vehicle_realtime_updates_total`: Redis/MySQL realtime projector updates by topic and status. - `vehicle_history_writes_total`: TDengine history writes by topic and status. diff --git a/docs/architecture/production-data-plane-inventory.md b/docs/architecture/production-data-plane-inventory.md index fbe7c106..087f71a3 100644 --- a/docs/architecture/production-data-plane-inventory.md +++ b/docs/architecture/production-data-plane-inventory.md @@ -25,9 +25,9 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。 | 项 | 当前值 | | --- | --- | | Stream | `VEHICLE_INGEST` | -| Subjects | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1`、`vehicle.event.go.unified.v1` | +| Subjects | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` | | Durable consumer | `vehicle-kafka-bridge` | -| 语义 | Gateway 先写 NATS；bridge 写 Kafka 成功后才 ACK NATS | +| 语义 | Gateway 先写 NATS raw subject；bridge 写 Kafka 成功后才 ACK NATS | ### Kafka @@ -38,7 +38,7 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。 | `vehicle.raw.go.gb32960.v1` | 12 | NATS Kafka bridge | history、stat、realtime | | `vehicle.raw.go.jt808.v1` | 12 | NATS Kafka bridge | history、stat、realtime | | `vehicle.raw.go.yutong-mqtt.v1` | 12 | NATS Kafka bridge | history、stat、realtime | -| `vehicle.event.go.unified.v1` | 12 | NATS Kafka bridge | 预留给统一事件消费 | +| `vehicle.event.go.unified.v1` | 12 | 显式打开 `PUBLISH_UNIFIED_ENABLED=true` 后才写入 | 兼容旧统一事件消费；当前最小链路不依赖 | 审计时 Kafka 上仍存在旧 Java/Xinda/telemetry topic，例如 `vehicle.raw.gb32960.v1`、`vehicle.raw.jt808.v1`、`vehicle.event.xinda.v1`、`vehicle.raw.telemetry-input.v1`。这些不属于当前 Go 最小链路，后续如果确认没有消费者依赖，应单独做 Kafka topic 下线清单，不在业务代码里继续引用。 diff --git a/docs/go-version-data-flow.html b/docs/go-version-data-flow.html index 2e414cf7..0d244f68 100644 --- a/docs/go-version-data-flow.html +++ b/docs/go-version-data-flow.html @@ -380,7 +380,7 @@
Subjects -

vehicle.raw.go.gb32960.v1
vehicle.raw.go.jt808.v1
vehicle.raw.go.yutong-mqtt.v1
vehicle.event.go.unified.v1

+

vehicle.raw.go.gb32960.v1
vehicle.raw.go.jt808.v1
vehicle.raw.go.yutong-mqtt.v1

Async + Retry @@ -396,7 +396,7 @@
Kafka Topics -

与 NATS subject 同名：raw 三个协议 + unified 一个实时主题。

+

与 NATS subject 同名：三个协议 RAW topic。vehicle.event.go.unified.v1 仅作为显式兼容开关保留。

失败语义 @@ -416,7 +416,7 @@
realtime-api 消费器 -

消费 unified topic，写 Redis 在线状态、实时合并快照、各协议 realtime-raw。

+

消费三个 RAW topic，写 Redis 在线状态、实时合并快照、各协议 realtime-raw。

HTTP API @@ -433,7 +433,7 @@
2. 拆包TCP 按协议提取完整帧；MQTT 直接按消息处理。
3. 解析生成 parsed 全量结构化数据和 fields 核心字段。
4. 绑定 VIN根据 VIN/phone/device_id/plate 生成 vehicle_key。
-
5. 发 NATS写 RAW subject；同时写 unified subject。
+
5. 发 NATS写协议 RAW subject；unified 只在兼容开关打开时写。
6. 桥接 Kafkabridge 写 Kafka 成功后 ACK NATS。
7. 多路消费历史、实时、统计各自消费 Kafka，互不阻塞。
diff --git a/docs/ops/go-service-observability.md b/docs/ops/go-service-observability.md index ce550738..5320f897 100644 --- a/docs/ops/go-service-observability.md +++ b/docs/ops/go-service-observability.md @@ -35,7 +35,7 @@ curl -fsS http://127.0.0.1:20200/readyz | --- | --- | | `vehicle_gateway_active_connections` | Current TCP connections by protocol. Labels: `protocol`. | | `vehicle_gateway_frames_total` | Protocol frames received and parsed by the gateway. Labels: `protocol`, `status`. | -| `vehicle_gateway_publish_total` | Gateway publish result for raw and unified events. Labels: `protocol`, `kind`, `status`. | +| `vehicle_gateway_publish_total` | Gateway publish result for raw events. Unified appears only when the compatibility stream is explicitly enabled. Labels: `protocol`, `kind`, `status`. | | `vehicle_bridge_messages_total` | NATS messages fetched by the bridge. Labels: `subject`, `status`. | | `vehicle_bridge_kafka_writes_total` | Bridge writes to Kafka. Labels: `topic`, `status`. | | `vehicle_bridge_nats_acks_total` | NATS ack results after Kafka write. Labels: `subject`, `status`. | diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index bd14c38a..bb1550a9 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -33,8 +33,8 @@ | --- | --- | | `lingniu-go-history-writer.service` | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` | | `lingniu-go-stat-writer.service` | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` | -| `lingniu-go-realtime-api.service` | `vehicle.event.go.unified.v1` | -| `lingniu-go-nats-kafka-bridge.service` | NATS subjects 到同名 Kafka topic；包含三条 raw.go topic 和一条 unified topic | +| `lingniu-go-realtime-api.service` | `vehicle.raw.go.gb32960.v1`、`vehicle.raw.go.jt808.v1`、`vehicle.raw.go.yutong-mqtt.v1` | +| `lingniu-go-nats-kafka-bridge.service` | NATS raw subjects 到同名 Kafka topic；`vehicle.event.go.unified.v1` 只在兼容开关打开时使用 | 如果 stat-writer 少消费某个 raw topic，对应协议的每日里程不会进入 `vehicle_daily_mileage`。修正后可能出现短时间 Kafka lag，这是在追补历史 backlog；只要 `vehicle_stat_writes_total` 持续增长且 lag 下降，就是健康状态。 diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index 6982277a..27f4d75c 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -44,6 +44,7 @@ func main() { defer closeResolver() registry := metrics.NewRegistry() health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil, registry)) + publishUnified := envBool("PUBLISH_UNIFIED_ENABLED", false) protocols := []gateway.TCPProtocol{ { @@ -74,6 +75,7 @@ func main() { ReadBufferSize: envInt("TCP_READ_BUFFER_BYTES", 64*1024), IdleTimeout: time.Duration(envInt("TCP_IDLE_TIMEOUT_SECONDS", 180)) * time.Second, MaxConnections: envInt("TCP_MAX_CONNECTIONS", 20_000), + PublishUnified: publishUnified, }) if err != nil { logger.Error("build tcp server failed", "protocol", protocol.Protocol, "error", err) @@ -107,6 +109,7 @@ func main() { Resolver: resolver, Logger: logger, Metrics: registry, + PublishUnified: publishUnified, }) if err != nil { logger.Error("build yutong mqtt client failed", "error", err) @@ -183,7 +186,7 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) }) logger.Info("nats async publish enabled", "queue_size", queueSize, "workers", workers, "publish_timeout_ms", timeout.Milliseconds()) } - logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "unified_subject", natsSinkConfigFromEnv().UnifiedSubject) + logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "unified_subject", natsSinkConfigFromEnv().UnifiedSubject, "publish_unified_enabled", envBool("PUBLISH_UNIFIED_ENABLED", false)) return out, nil } brokers := splitCSV(os.Getenv("KAFKA_BROKERS")) diff --git a/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go b/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go index 379c5046..d13b6a46 100644 --- a/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go +++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go @@ -105,7 +105,9 @@ func loadConfig() config { 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_UNIFIED", env("KAFKA_TOPIC_UNIFIED", topics.Unified)): env("KAFKA_TOPIC_UNIFIED", topics.Unified), + } + if unifiedSubject, unifiedTopic, ok := unifiedRouteFromEnv(); ok { + route[unifiedSubject] = unifiedTopic } return config{ NATSURL: env("NATS_URL", "nats://127.0.0.1:4222"), @@ -124,6 +126,24 @@ func loadConfig() config { } } +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) } @@ -290,6 +310,14 @@ func env(key string, fallback string) string { 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 == "" { diff --git a/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go b/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go index d28e9a1c..cc0e2087 100644 --- a/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go +++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go @@ -120,7 +120,6 @@ func TestLoadConfigDefaultsToGoSubjectRoutes(t *testing.T) { "vehicle.raw.go.gb32960.v1": "vehicle.raw.go.gb32960.v1", "vehicle.raw.go.jt808.v1": "vehicle.raw.go.jt808.v1", "vehicle.raw.go.yutong-mqtt.v1": "vehicle.raw.go.yutong-mqtt.v1", - "vehicle.event.go.unified.v1": "vehicle.event.go.unified.v1", } if len(cfg.Route) != len(want) { t.Fatalf("route len = %d, want %d: %#v", len(cfg.Route), len(want), cfg.Route) @@ -132,6 +131,17 @@ func TestLoadConfigDefaultsToGoSubjectRoutes(t *testing.T) { } } +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") + + cfg := loadConfig() + + if got := cfg.Route["vehicle.event.go.unified.v1"]; got != "vehicle.event.go.unified.v1" { + t.Fatalf("unified route = %q, route=%#v", got, cfg.Route) + } +} + type recordingBridgeWriter struct { messages []kafka.Message err error diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index 848258dc..7701207a 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -190,7 +190,8 @@ func consumeKafka(ctx context.Context, logger interface { } func kafkaTopicsFromEnv() []string { - return splitCSV(env("KAFKA_TOPICS", topics.Unified)) + defaultTopics := strings.Join([]string{topics.RawGB32960, topics.RawJT808, topics.RawYutongMQTT}, ",") + return splitCSV(env("KAFKA_TOPICS", defaultTopics)) } const kafkaMessageOperationTimeout = 30 * time.Second diff --git a/go/vehicle-gateway/cmd/realtime-api/main_test.go b/go/vehicle-gateway/cmd/realtime-api/main_test.go index 3d240a81..0c242924 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main_test.go +++ b/go/vehicle-gateway/cmd/realtime-api/main_test.go @@ -66,15 +66,15 @@ func TestProcessRealtimeMessageRecordsMetrics(t *testing.T) { registry, &contextCheckingRealtimeUpdater{}, &contextCheckingMessageCommitter{}, - kafka.Message{Topic: "vehicle.event.go.unified.v1", Partition: 2, Offset: 10, HighWaterMark: 16, Value: payload}, + kafka.Message{Topic: "vehicle.raw.go.jt808.v1", Partition: 2, Offset: 10, HighWaterMark: 16, Value: payload}, ) text := registry.Render() for _, want := range []string{ - `vehicle_realtime_kafka_messages_total{status="received",topic="vehicle.event.go.unified.v1"} 1`, - `vehicle_realtime_updates_total{status="ok",topic="vehicle.event.go.unified.v1"} 1`, - `vehicle_realtime_kafka_commits_total{status="ok",topic="vehicle.event.go.unified.v1"} 1`, - `vehicle_realtime_kafka_lag{partition="2",topic="vehicle.event.go.unified.v1"} 5`, + `vehicle_realtime_kafka_messages_total{status="received",topic="vehicle.raw.go.jt808.v1"} 1`, + `vehicle_realtime_updates_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`, + `vehicle_realtime_kafka_commits_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`, + `vehicle_realtime_kafka_lag{partition="2",topic="vehicle.raw.go.jt808.v1"} 5`, } { if !strings.Contains(text, want) { t.Fatalf("metrics missing %s:\n%s", want, text) @@ -96,9 +96,9 @@ func TestCompositeRealtimeUpdaterUpdatesBothStores(t *testing.T) { } } -func TestKafkaTopicsFromEnvDefaultsToGoUnifiedTopic(t *testing.T) { +func TestKafkaTopicsFromEnvDefaultsToGoRawTopics(t *testing.T) { got := strings.Join(kafkaTopicsFromEnv(), ",") - want := "vehicle.event.go.unified.v1" + want := "vehicle.raw.go.gb32960.v1,vehicle.raw.go.jt808.v1,vehicle.raw.go.yutong-mqtt.v1" if got != want { t.Fatalf("topics = %q, want %q", got, want) } diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index 84dbd37f..ae24d7a4 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -40,6 +40,7 @@ type MQTTClientConfig struct { Resolver identity.Resolver Logger *slog.Logger Metrics *metrics.Registry + PublishUnified bool } type MQTTClient struct { @@ -220,12 +221,14 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload [] if env.ParseStatus == envelope.ParseBadFrame { return } - if err := c.cfg.Sink.PublishUnified(messageCtx, env); err != nil { - c.recordPublishMetric("unified", "error") - c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err) - return + if c.cfg.PublishUnified { + if err := c.cfg.Sink.PublishUnified(messageCtx, env); err != nil { + c.recordPublishMetric("unified", "error") + c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err) + return + } + c.recordPublishMetric("unified", "ok") } - c.recordPublishMetric("unified", "ok") } func (c *MQTTClient) recordFrameMetric(status envelope.ParseStatus) { diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go index afd6ef44..54ea4e20 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go @@ -19,7 +19,7 @@ import ( "lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics" ) -func TestMQTTClientHandleMessagePublishesRawAndUnified(t *testing.T) { +func TestMQTTClientHandleMessagePublishesOnlyRawByDefault(t *testing.T) { sink := &recordingSink{} client, err := NewMQTTClient(MQTTClientConfig{ EndpointName: "endpoint-a", @@ -39,7 +39,7 @@ func TestMQTTClientHandleMessagePublishesRawAndUnified(t *testing.T) { "data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7} }`)) - if len(sink.raw) != 1 || len(sink.unified) != 1 { + if len(sink.raw) != 1 || len(sink.unified) != 0 { t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) } if sink.raw[0].Protocol != envelope.ProtocolYutongMQTT || sink.raw[0].VIN != "LTEST000000000001" { @@ -72,12 +72,14 @@ func TestMQTTClientRecordsMessageMetrics(t *testing.T) { for _, want := range []string{ `vehicle_gateway_frames_total{protocol="YUTONG_MQTT",status="OK"} 1`, `vehicle_gateway_publish_total{kind="raw",protocol="YUTONG_MQTT",status="ok"} 1`, - `vehicle_gateway_publish_total{kind="unified",protocol="YUTONG_MQTT",status="ok"} 1`, } { if !strings.Contains(text, want) { t.Fatalf("metrics missing %s:\n%s", want, text) } } + if strings.Contains(text, `kind="unified"`) { + t.Fatalf("unified publish metric should not be recorded by default:\n%s", text) + } } func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) { @@ -138,11 +140,37 @@ func TestMQTTClientUsesUncancelledMessageContextForReceivedMessage(t *testing.T) if sink.unifiedCtxErr != nil { t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr) } - if sink.rawCount != 1 || sink.unifiedCount != 1 { + if sink.rawCount != 1 || sink.unifiedCount != 0 { t.Fatalf("raw=%d unified=%d", sink.rawCount, sink.unifiedCount) } } +func TestMQTTClientPublishesUnifiedWhenExplicitlyEnabled(t *testing.T) { + sink := &recordingSink{} + client, err := NewMQTTClient(MQTTClientConfig{ + EndpointName: "endpoint-a", + Broker: "tcp://127.0.0.1:1883", + ClientID: "test-client", + Topics: []string{"/ytforward/shln/+"}, + PublishUnified: true, + Sink: sink, + Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), + }) + if err != nil { + t.Fatalf("NewMQTTClient() error = %v", err) + } + + client.handleMessage(context.Background(), "/ytforward/shln/dev1", []byte(`{ + "device":"LTEST000000000001", + "time":"20260413100000", + "data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7} + }`)) + + if len(sink.raw) != 1 || len(sink.unified) != 1 { + t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) + } +} + func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) { dir := t.TempDir() caPath, certPath, keyPath := writeTestTLSMaterial(t, dir) diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index e7587674..6154b54c 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -41,6 +41,7 @@ type TCPServer struct { readBufferSize int idleTimeout time.Duration maxConnections int + publishUnified bool } type TCPServerConfig struct { @@ -52,6 +53,7 @@ type TCPServerConfig struct { ReadBufferSize int IdleTimeout time.Duration MaxConnections int + PublishUnified bool } const frameOperationTimeout = 30 * time.Second @@ -96,6 +98,7 @@ func NewTCPServer(cfg TCPServerConfig) (*TCPServer, error) { readBufferSize: cfg.ReadBufferSize, idleTimeout: cfg.IdleTimeout, maxConnections: cfg.MaxConnections, + publishUnified: cfg.PublishUnified, }, nil } @@ -226,12 +229,14 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, if env.ParseStatus == envelope.ParseBadFrame { return } - if err := s.sink.PublishUnified(frameCtx, env); err != nil { - s.recordPublishMetric("unified", "error") - s.logger.Error("publish unified failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err) - return + if s.publishUnified { + if err := s.sink.PublishUnified(frameCtx, env); err != nil { + s.recordPublishMetric("unified", "error") + s.logger.Error("publish unified failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err) + return + } + s.recordPublishMetric("unified", "ok") } - s.recordPublishMetric("unified", "ok") if s.protocol.Respond == nil { return } diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index 8a3743b0..0f6dc049 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -16,7 +16,7 @@ import ( "lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808" ) -func TestTCPServerPublishesGoodFrameToRawAndUnified(t *testing.T) { +func TestTCPServerPublishesGoodFrameOnlyToRawByDefault(t *testing.T) { frame, err := hex.DecodeString("7E020000320133077954250001000000000048000301D2C4C707376139000A00E6004F26063016235701040001900C2504000000000202000030011F31010F867E") if err != nil { t.Fatal(err) @@ -36,7 +36,7 @@ func TestTCPServerPublishesGoodFrameToRawAndUnified(t *testing.T) { _ = client.Close() <-done - if len(sink.raw) != 1 || len(sink.unified) != 1 { + if len(sink.raw) != 1 || len(sink.unified) != 0 { t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) } if sink.raw[0].Phone != "13307795425" { @@ -80,12 +80,14 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) { for _, want := range []string{ `vehicle_gateway_frames_total{protocol="JT808",status="OK"} 1`, `vehicle_gateway_publish_total{kind="raw",protocol="JT808",status="ok"} 1`, - `vehicle_gateway_publish_total{kind="unified",protocol="JT808",status="ok"} 1`, } { if !strings.Contains(text, want) { t.Fatalf("metrics missing %s:\n%s", want, text) } } + if strings.Contains(text, `kind="unified"`) { + t.Fatalf("unified publish metric should not be recorded by default:\n%s", text) + } } func TestTCPServerRecordsActiveConnectionGauge(t *testing.T) { @@ -171,8 +173,8 @@ func TestTCPServerWritesProtocolResponseAfterPublish(t *testing.T) { Extract: gb32960.ExtractFrames, Parse: gb32960.ParseFrame, Respond: func(_ []byte, env envelope.FrameEnvelope) ([]byte, bool, error) { - if len(sink.unified) != 1 || sink.unified[0].EventID != env.EventID { - t.Fatalf("response built before publish: raw=%d unified=%d", len(sink.raw), len(sink.unified)) + if len(sink.raw) != 1 || sink.raw[0].EventID != env.EventID { + t.Fatalf("response built before raw publish: raw=%d unified=%d", len(sink.raw), len(sink.unified)) } return []byte("ACK"), true, nil }, @@ -235,11 +237,38 @@ func TestTCPServerUsesUncancelledFrameContextForAlreadyReadFrame(t *testing.T) { if sink.unifiedCtxErr != nil { t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr) } - if sink.rawCount != 1 || sink.unifiedCount != 1 { + if sink.rawCount != 1 || sink.unifiedCount != 0 { t.Fatalf("raw=%d unified=%d", sink.rawCount, sink.unifiedCount) } } +func TestTCPServerPublishesUnifiedWhenExplicitlyEnabled(t *testing.T) { + sink := &recordingSink{} + server := newTestServer(t, TCPProtocol{ + Protocol: envelope.ProtocolJT808, + Addr: ":0", + Extract: jt808.ExtractFrames, + Parse: func(_ []byte, receivedAtMS int64, sourceEndpoint string) (envelope.FrameEnvelope, error) { + return envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0200", + Phone: "13307795425", + SourceEndpoint: sourceEndpoint, + ReceivedAtMS: receivedAtMS, + EventTimeMS: receivedAtMS, + ParseStatus: envelope.ParseOK, + }, nil + }, + }, sink) + server.publishUnified = true + + server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808") + + if len(sink.raw) != 1 || len(sink.unified) != 1 { + t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) + } +} + type contextCheckingResolver struct { ctxErr error }