From 1cbe29c5fc313fc8d6ab3f1736bf00d6223ea7b6 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 18:43:15 +0800 Subject: [PATCH] feat(go): generate unique load simulator frames --- docs/ops/100k-capacity-baseline.md | 11 +- .../internal/loadsim/mutator.go | 166 ++++++++++++++++++ .../internal/loadsim/mutator_test.go | 106 +++++++++++ go/vehicle-gateway/internal/loadsim/runner.go | 27 ++- .../internal/loadsim/runner_test.go | 42 +++++ 5 files changed, 340 insertions(+), 12 deletions(-) create mode 100644 go/vehicle-gateway/internal/loadsim/mutator.go create mode 100644 go/vehicle-gateway/internal/loadsim/mutator_test.go diff --git a/docs/ops/100k-capacity-baseline.md b/docs/ops/100k-capacity-baseline.md index a7eb5156..4a141486 100644 --- 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/连接短测。 diff --git a/go/vehicle-gateway/internal/loadsim/mutator.go b/go/vehicle-gateway/internal/loadsim/mutator.go new file mode 100644 index 00000000..97f6c6a2 --- /dev/null +++ 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 +} diff --git a/go/vehicle-gateway/internal/loadsim/mutator_test.go b/go/vehicle-gateway/internal/loadsim/mutator_test.go new file mode 100644 index 00000000..2b6fc6c4 --- /dev/null +++ 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 +} diff --git a/go/vehicle-gateway/internal/loadsim/runner.go b/go/vehicle-gateway/internal/loadsim/runner.go index 6726a2e4..1e737551 100644 --- 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) diff --git a/go/vehicle-gateway/internal/loadsim/runner_test.go b/go/vehicle-gateway/internal/loadsim/runner_test.go index e2d79b35..4823fe5d 100644 --- 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{