diff --git a/internal/websocket/broadcast_projection_bench_test.go b/internal/websocket/broadcast_projection_bench_test.go new file mode 100644 index 0000000000..c83451d6c3 --- /dev/null +++ b/internal/websocket/broadcast_projection_bench_test.go @@ -0,0 +1,68 @@ +package websocket_test + +import ( + "fmt" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rcourtman/pulse-go-rewrite/internal/websocket" +) + +// The fixture models a broadcast after completed ingest. Registry reads/clones, +// identity, health, projection, encoding, deltas and client queues are real; +// persistence, agent ingest, network I/O and installed CPU are not measured. +type broadcastReadStore struct { + *unifiedresources.MonitorAdapter +} + +func (*broadcastReadStore) TryReplaceRegistryForRead(models.StateSnapshot, time.Duration, func() map[unifiedresources.DataSource][]unifiedresources.IngestRecord) bool { + return false +} + +func BenchmarkBroadcastProjection1000Resources(b *testing.B) { + now := time.Now().UTC() + seed := models.StateSnapshot{LastUpdate: now} + for i := 0; i < 1000; i++ { + seed.VMs = append(seed.VMs, models.VM{ + ID: fmt.Sprintf("lab:pve-%d:%d", i%8, i+100), VMID: i + 100, + Name: fmt.Sprintf("vm-%d", i), Node: fmt.Sprintf("pve-%d", i%8), + Instance: "lab", Type: "qemu", Status: "running", CPU: 0.1, + Memory: models.Memory{Total: 4 << 30, Used: 1 << 30}, LastSeen: now, + }) + } + registry := unifiedresources.NewRegistry(nil) + registry.IngestSnapshot(seed) + monitor := &monitoring.Monitor{} + monitor.SetResourceStore(&broadcastReadStore{unifiedresources.NewMonitorAdapter(registry)}) + for _, recipients := range []int{0, 1, 4} { + b.Run(fmt.Sprintf("viewers-%d", recipients), func(b *testing.B) { + sequence := int64(0) + getter := func(string) interface{} { + state := monitor.BuildFrontendState() + sequence++ + state.LastUpdate = sequence + // Change telemetry on a row, not identity/static metadata. + state.Resources[0].CPU.Current = float64(sequence % 100) + state.Resources[0].LastSeen = sequence + return state + } + run, err := websocket.NewBroadcastProjectionProbeForTest(getter, recipients) + if err != nil { + b.Fatal(err) + } + b.ReportAllocs() + b.ResetTimer() + var bytes int + for i := 0; i < b.N; i++ { + bytes, err = run() + if err != nil { + b.Fatal(err) + } + } + b.ReportMetric(float64(bytes), "wire-B/op") + }) + } +} diff --git a/internal/websocket/broadcast_projection_export_test.go b/internal/websocket/broadcast_projection_export_test.go new file mode 100644 index 0000000000..e10bf6c790 --- /dev/null +++ b/internal/websocket/broadcast_projection_export_test.go @@ -0,0 +1,34 @@ +package websocket + +import "fmt" + +// NewBroadcastProjectionProbeForTest exercises the production getter, snapshot, +// per-client delta and queue path without network/compression or timer sleeps. +// It is exported only in the test binary so the fixture can use a real Monitor +// without introducing a monitoring -> websocket import cycle. +func NewBroadcastProjectionProbeForTest(getState func(string) interface{}, recipients int) (func() (int, error), error) { + hub := NewHub(getState) + clients := make([]*Client, recipients) + for i := range clients { + client := &Client{hub: hub, orgID: "default", send: make(chan []byte, 1)} + if _, sent, err := client.queueFullState("initialState", getState("default")); err != nil || !sent { + return nil, fmt.Errorf("initial state: sent=%v error=%v", sent, err) + } + <-client.send + clients[i] = client + hub.clients[client] = true + } + return func() (int, error) { + hub.dispatchStateBroadcast(&Message{Type: "rawData", Data: stateBroadcastRequest{}}, "") + bytes := 0 + for _, client := range clients { + select { + case data := <-client.send: + bytes += len(data) + default: + return 0, fmt.Errorf("broadcast did not queue fresh state") + } + } + return bytes, nil + }, nil +}