feat(go): generate unique load simulator frames

2026-07-03 18:43:15 +08:00
parent 09f3e68891
commit 1cbe29c5fc
5 changed files with 340 additions and 12 deletions
--- a/docs/ops/100k-capacity-baseline.md
+++ b/docs/ops/100k-capacity-baseline.md
@@ -80,10 +80,8 @@ systemd gateway 已配置 `LimitNOFILE=1048576`，需要持续保持。
 压测入口：

 ```bash
-cd /opt/lingniu-go-native/current/go/vehicle-gateway
-
 # 100 连接 smoke test
-go run ./cmd/load-sim \
+/opt/lingniu-go-native/current/load-sim \
  -protocol jt808 \
  -addr 127.0.0.1:808 \
  -connections 100 \
@@ -94,6 +92,13 @@ go run ./cmd/load-sim \
  -send=false
 ```

+`load-sim` 在 `-send=true` 时会按连接和帧序号生成可被现有解析器解析的唯一协议帧：
+
+- JT808 会变更包头手机号、流水号和 `0x0200` 位置时间，并重新计算转义和校验码。
+- GB32960 会变更 VIN 和实时数据时间，并重新计算 BCC。
+- 低 FPS 全链路压测必须使用该模式，避免重复静态帧导致 event id 冲突。
+- 对生产端口执行 `-send=true` 会写入合成 raw 数据，只能在明确隔离标识和压测窗口后执行。
+
 1. 100 连接 smoke test。
 2. 10,000 连接保持测试。
 3. 10,000 连接 + 1 FPS/连接短测。
--- a/go/vehicle-gateway/internal/loadsim/mutator.go
+++ b/go/vehicle-gateway/internal/loadsim/mutator.go
@@ -0,0 +1,166 @@
+package loadsim
+
+import (
+	"encoding/binary"
+	"fmt"
+	"time"
+)
+
+type FrameFactory struct {
+	protocol Protocol
+	base     []byte
+}
+
+func NewFrameFactory(protocol Protocol, template string) (*FrameFactory, error) {
+	base, err := FrameTemplate(protocol, template)
+	if err != nil {
+		return nil, err
+	}
+	return &FrameFactory{protocol: protocol, base: base}, nil
+}
+
+func (f *FrameFactory) Frame(connectionIndex int, frameIndex int64) ([]byte, error) {
+	switch f.protocol {
+	case ProtocolJT808:
+		return mutateJT808Frame(f.base, connectionIndex, frameIndex)
+	case ProtocolGB32960:
+		return mutateGB32960Frame(f.base, connectionIndex, frameIndex)
+	default:
+		out := append([]byte(nil), f.base...)
+		return out, nil
+	}
+}
+
+func mutateJT808Frame(base []byte, connectionIndex int, frameIndex int64) ([]byte, error) {
+	if len(base) < 2 || base[0] != 0x7e || base[len(base)-1] != 0x7e {
+		return nil, fmt.Errorf("jt808 template must include 0x7e delimiters")
+	}
+	payload, err := jt808Unescape(base[1 : len(base)-1])
+	if err != nil {
+		return nil, err
+	}
+	if len(payload) < 13 {
+		return nil, fmt.Errorf("jt808 template too short: %d", len(payload))
+	}
+	phone := fmt.Sprintf("%012d", 139000000000+connectionIndex%100000000)
+	copy(payload[4:10], encodeBCD(phone, 6))
+	binary.BigEndian.PutUint16(payload[10:12], uint16((int(frameIndex)+connectionIndex)%65536))
+	bodySize := int(binary.BigEndian.Uint16(payload[2:4]) & 0x03ff)
+	bodyStart := 12
+	if len(payload) >= bodyStart+bodySize+1 && bodySize >= 28 {
+		writeJT808Time(payload[bodyStart+22:bodyStart+28], time.Now().Add(time.Duration(frameIndex)*time.Second))
+	}
+	payload[len(payload)-1] = jt808Checksum(payload[:len(payload)-1])
+	escaped := jt808Escape(payload)
+	out := make([]byte, 0, len(escaped)+2)
+	out = append(out, 0x7e)
+	out = append(out, escaped...)
+	out = append(out, 0x7e)
+	return out, nil
+}
+
+func mutateGB32960Frame(base []byte, connectionIndex int, frameIndex int64) ([]byte, error) {
+	if len(base) < 25 || base[0] != '#' || base[1] != '#' {
+		return nil, fmt.Errorf("gb32960 template too short or missing start symbols")
+	}
+	frame := append([]byte(nil), base...)
+	vin := fmt.Sprintf("LNSIM%012d", connectionIndex%1000000000000)
+	copy(frame[4:21], []byte(vin[:17]))
+	bodyLen := int(binary.BigEndian.Uint16(frame[22:24]))
+	bodyStart := 24
+	if len(frame) >= bodyStart+bodyLen+1 && bodyLen >= 6 {
+		writeGB32960Time(frame[bodyStart:bodyStart+6], time.Now().Add(time.Duration(frameIndex)*time.Second))
+	}
+	frame[len(frame)-1] = gb32960BCC(frame[2 : len(frame)-1])
+	return frame, nil
+}
+
+func encodeBCD(value string, size int) []byte {
+	out := make([]byte, size)
+	if len(value)%2 == 1 {
+		value = "0" + value
+	}
+	if len(value) > size*2 {
+		value = value[len(value)-size*2:]
+	}
+	for len(value) < size*2 {
+		value = "0" + value
+	}
+	for i := 0; i < size; i++ {
+		out[i] = (value[i*2]-'0')<<4 | (value[i*2+1] - '0')
+	}
+	return out
+}
+
+func jt808Checksum(data []byte) byte {
+	var out byte
+	for _, value := range data {
+		out ^= value
+	}
+	return out
+}
+
+func jt808Escape(payload []byte) []byte {
+	out := make([]byte, 0, len(payload))
+	for _, value := range payload {
+		switch value {
+		case 0x7e:
+			out = append(out, 0x7d, 0x02)
+		case 0x7d:
+			out = append(out, 0x7d, 0x01)
+		default:
+			out = append(out, value)
+		}
+	}
+	return out
+}
+
+func jt808Unescape(payload []byte) ([]byte, error) {
+	out := make([]byte, 0, len(payload))
+	for i := 0; i < len(payload); i++ {
+		if payload[i] != 0x7d {
+			out = append(out, payload[i])
+			continue
+		}
+		if i+1 >= len(payload) {
+			return nil, fmt.Errorf("invalid jt808 escape at end")
+		}
+		i++
+		switch payload[i] {
+		case 0x01:
+			out = append(out, 0x7d)
+		case 0x02:
+			out = append(out, 0x7e)
+		default:
+			return nil, fmt.Errorf("invalid jt808 escape sequence 0x%02x", payload[i])
+		}
+	}
+	return out, nil
+}
+
+func writeJT808Time(out []byte, value time.Time) {
+	value = value.In(time.FixedZone("CST", 8*3600))
+	year := value.Year() % 100
+	parts := []int{year, int(value.Month()), value.Day(), value.Hour(), value.Minute(), value.Second()}
+	for i, part := range parts {
+		out[i] = byte(part/10)<<4 | byte(part%10)
+	}
+}
+
+func writeGB32960Time(out []byte, value time.Time) {
+	value = value.In(time.FixedZone("CST", 8*3600))
+	out[0] = byte(value.Year() % 100)
+	out[1] = byte(value.Month())
+	out[2] = byte(value.Day())
+	out[3] = byte(value.Hour())
+	out[4] = byte(value.Minute())
+	out[5] = byte(value.Second())
+}
+
+func gb32960BCC(data []byte) byte {
+	var out byte
+	for _, value := range data {
+		out ^= value
+	}
+	return out
+}
--- a/go/vehicle-gateway/internal/loadsim/mutator_test.go
+++ b/go/vehicle-gateway/internal/loadsim/mutator_test.go
@@ -0,0 +1,106 @@
+package loadsim
+
+import (
+	"testing"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808"
+)
+
+func TestFrameFactoryGeneratesUniqueParsableJT808Frames(t *testing.T) {
+	factory, err := NewFrameFactory(ProtocolJT808, "0200")
+	if err != nil {
+		t.Fatalf("NewFrameFactory() error = %v", err)
+	}
+
+	first, err := factory.Frame(0, 0)
+	if err != nil {
+		t.Fatalf("first Frame() error = %v", err)
+	}
+	second, err := factory.Frame(1, 2)
+	if err != nil {
+		t.Fatalf("second Frame() error = %v", err)
+	}
+	if string(first) == string(second) {
+		t.Fatal("variant frames should not be byte-identical")
+	}
+
+	firstEnv := parseJT808Frame(t, first)
+	secondEnv := parseJT808Frame(t, second)
+	if firstEnv.Phone == secondEnv.Phone {
+		t.Fatalf("phones should differ, both %q", firstEnv.Phone)
+	}
+	if firstEnv.Sequence == secondEnv.Sequence {
+		t.Fatalf("sequences should differ, both %d", firstEnv.Sequence)
+	}
+	if firstEnv.StableEventID() == secondEnv.StableEventID() {
+		t.Fatal("stable event ids should differ")
+	}
+}
+
+func TestFrameFactoryGeneratesUniqueParsableGB32960Frames(t *testing.T) {
+	factory, err := NewFrameFactory(ProtocolGB32960, "realtime")
+	if err != nil {
+		t.Fatalf("NewFrameFactory() error = %v", err)
+	}
+
+	first, err := factory.Frame(0, 0)
+	if err != nil {
+		t.Fatalf("first Frame() error = %v", err)
+	}
+	second, err := factory.Frame(1, 2)
+	if err != nil {
+		t.Fatalf("second Frame() error = %v", err)
+	}
+	if string(first) == string(second) {
+		t.Fatal("variant frames should not be byte-identical")
+	}
+
+	firstEnv := parseGB32960Frame(t, first)
+	secondEnv := parseGB32960Frame(t, second)
+	if firstEnv.VIN == secondEnv.VIN {
+		t.Fatalf("VINs should differ, both %q", firstEnv.VIN)
+	}
+	if firstEnv.StableEventID() == secondEnv.StableEventID() {
+		t.Fatal("stable event ids should differ")
+	}
+}
+
+func parseJT808Frame(t *testing.T, frame []byte) parsedEnvelope {
+	t.Helper()
+	frames, remainder, err := jt808.ExtractFrames(frame)
+	if err != nil {
+		t.Fatalf("ExtractFrames() error = %v", err)
+	}
+	if len(remainder) != 0 || len(frames) != 1 {
+		t.Fatalf("unexpected split frames=%d remainder=%x", len(frames), remainder)
+	}
+	env, err := jt808.ParseFrame(frames[0], 1782918600000, "127.0.0.1:808")
+	if err != nil {
+		t.Fatalf("ParseFrame() error = %v", err)
+	}
+	return parsedEnvelope{Phone: env.Phone, VIN: env.VIN, Sequence: env.Sequence, StableEventID: env.StableEventID}
+}
+
+func parseGB32960Frame(t *testing.T, frame []byte) parsedEnvelope {
+	t.Helper()
+	frames, remainder, err := gb32960.ExtractFrames(frame)
+	if err != nil {
+		t.Fatalf("ExtractFrames() error = %v", err)
+	}
+	if len(remainder) != 0 || len(frames) != 1 {
+		t.Fatalf("unexpected split frames=%d remainder=%x", len(frames), remainder)
+	}
+	env, err := gb32960.ParseFrame(frames[0], 1782918600000, "127.0.0.1:32960")
+	if err != nil {
+		t.Fatalf("ParseFrame() error = %v", err)
+	}
+	return parsedEnvelope{Phone: env.Phone, VIN: env.VIN, Sequence: env.Sequence, StableEventID: env.StableEventID}
+}
+
+type parsedEnvelope struct {
+	Phone         string
+	VIN           string
+	Sequence      uint16
+	StableEventID func() string
+}
--- a/go/vehicle-gateway/internal/loadsim/runner.go
+++ b/go/vehicle-gateway/internal/loadsim/runner.go
@@ -33,12 +33,13 @@ func (r Runner) Run(ctx context.Context, cfg Config) (Stats, error) {
 	}).Build(); err != nil {
 		return Stats{}, err
 	}
-	payload, err := FrameTemplate(cfg.Protocol, cfg.Template)
-	if err != nil {
-		return Stats{}, err
-	}
-	if !cfg.SendFrames {
-		payload = nil
+	var factory *FrameFactory
+	if cfg.SendFrames {
+		var err error
+		factory, err = NewFrameFactory(cfg.Protocol, cfg.Template)
+		if err != nil {
+			return Stats{}, err
+		}
 	}
 	dial := r.Dial
 	if dial == nil {
@@ -51,6 +52,7 @@ func (r Runner) Run(ctx context.Context, cfg Config) (Stats, error) {

 	var stats Stats
 	var wg sync.WaitGroup
+	connectionIndex := 0
 	for i, batch := range ConnectionBatches(cfg.Connections, cfg.ConnectRatePerSecond) {
 		if i > 0 {
 			if !sleepOrDone(runCtx, time.Second) {
@@ -68,11 +70,13 @@ func (r Runner) Run(ctx context.Context, cfg Config) (Stats, error) {
 			}
 			atomic.AddInt64(&stats.ConnectionsOpened, 1)
 			wg.Add(1)
+			connIndex := connectionIndex
+			connectionIndex++
 			go func() {
 				defer wg.Done()
 				defer conn.Close()
 				if cfg.SendFrames {
-					writeLoop(runCtx, conn, payload, cfg.SendInterval, &stats)
+					writeLoop(runCtx, conn, factory, connIndex, cfg.SendInterval, &stats)
 					return
 				}
 				<-runCtx.Done()
@@ -84,8 +88,13 @@ func (r Runner) Run(ctx context.Context, cfg Config) (Stats, error) {
 	return stats, nil
 }

-func writeLoop(ctx context.Context, conn net.Conn, payload []byte, interval time.Duration, stats *Stats) {
-	for {
+func writeLoop(ctx context.Context, conn net.Conn, factory *FrameFactory, connectionIndex int, interval time.Duration, stats *Stats) {
+	for frameIndex := int64(0); ; frameIndex++ {
+		payload, err := factory.Frame(connectionIndex, frameIndex)
+		if err != nil {
+			atomic.AddInt64(&stats.WriteErrors, 1)
+			return
+		}
 		_ = conn.SetWriteDeadline(time.Now().Add(3 * time.Second))
 		if _, err := conn.Write(payload); err != nil {
 			atomic.AddInt64(&stats.WriteErrors, 1)
--- a/go/vehicle-gateway/internal/loadsim/runner_test.go
+++ b/go/vehicle-gateway/internal/loadsim/runner_test.go
@@ -54,6 +54,48 @@ func TestRunnerDialsTargetConnectionsAndWritesFrames(t *testing.T) {
 	}
 }

+func TestRunnerWritesVariantFrames(t *testing.T) {
+	writes := make(chan []byte, 8)
+	runner := Runner{
+		Dial: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			return &recordingConn{
+				write: func(p []byte) (int, error) {
+					cp := append([]byte(nil), p...)
+					select {
+					case writes <- cp:
+					default:
+					}
+					return len(p), nil
+				},
+				close: func() error { return nil },
+			}, nil
+		},
+	}
+
+	stats, err := runner.Run(context.Background(), Config{
+		Protocol:             ProtocolJT808,
+		Addr:                 "127.0.0.1:808",
+		Connections:          1,
+		ConnectRatePerSecond: 1000,
+		SendInterval:         time.Millisecond,
+		Duration:             6 * time.Millisecond,
+		Template:             "0200",
+		SendFrames:           true,
+	})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if stats.FramesWritten < 2 {
+		t.Fatalf("FramesWritten = %d, want at least 2", stats.FramesWritten)
+	}
+
+	first := <-writes
+	second := <-writes
+	if string(first) == string(second) {
+		t.Fatal("runner wrote byte-identical frames; want per-frame variants")
+	}
+}
+
 func TestRunnerCanHoldConnectionsWithoutWritingFrames(t *testing.T) {
 	var writes atomic.Int64
 	runner := Runner{