1//go:build linux
2
3package main
4
5import (
6 "bytes"
7 "encoding/json"
8 "errors"
9 "os"
10 "path/filepath"
11 "strings"
12 "testing"
13 "time"
14)
15
16func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
17 fixtures := []struct {
18 name string
19 tr transport
20 rtt time.Duration
21 loss int
22 }{
23 {"baseline", transport{ID: "udp_baseline", Kind: "udp"}, 0, 0},
24 {"25ms", transport{ID: "udp_25ms", Kind: "udp", RTTMS: 25}, 25 * time.Millisecond, 0},
25 {"100ms", transport{ID: "udp_100ms", Kind: "udp", RTTMS: 100}, 100 * time.Millisecond, 0},
26 {"loss0", transport{ID: "udp_loss_0pct", Kind: "udp"}, 0, 0},
27 {"loss1", transport{ID: "udp_loss_1pct", Kind: "udp", LossPercent: 1}, 0, 1},
28 }
29 for _, tc := range fixtures {
30 t.Run(tc.name, func(t *testing.T) {
31 dir := t.TempDir()
32 if err := os.Chmod(dir, 0o700); err != nil {
33 t.Fatal(err)
34 }
35 var configs []udpNetemConfig
36 netem := &fakeUDPNetem{port: 45678}
37 l := &cliLauncher{bin: "/bin/true", netemFactory: func(c udpNetemConfig) (udpNetem, error) {
38 configs = append(configs, c)
39 return netem, nil
40 }}
41 m := processMapping{ProcessID: "udp-peer", TracePath: filepath.Join(dir, "udp.jsonl"), Role: "udp_peer"}
42 if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {
43 t.Fatal(err)
44 }
45 p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy", "work"}, Transport: tc.tr})
46 if err != nil {
47 t.Fatal(err)
48 }
49 if len(configs) != 1 || configs[0].RTT != tc.rtt || configs[0].LossPercent != tc.loss || configs[0].TargetPath != filepath.Join(dir, "udp-peer.target") {
50 t.Fatalf("fixture did not reach netem seam: %+v", configs)
51 }
52 shim, err := os.ReadFile(filepath.Join(dir, "ssh"))
53 if err != nil {
54 t.Fatal(err)
55 }
56 text := string(shim)
57 for _, want := range []string{"_udp-proxy", m.TracePath, m.ProcessID, "udp-peer.target", "VEV-UDP %s %s\\n' 45678", "udp-peer.pid"} {
58 if !strings.Contains(text, want) {
59 t.Errorf("seam does not retain %q:\n%s", want, text)
60 }
61 }
62 if strings.Contains(text, "VEV_PERF_UDP_") {
63 t.Fatalf("ignored vev UDP environment was used instead of netem: %s", text)
64 }
65 // A nonexistent pid is already-cleaned-up; Close must still close the
66 // harness-owned emulator.
67 if err := p.Close(); err != nil {
68 t.Fatal(err)
69 }
70 if !netem.closed {
71 t.Fatal("harness netem was not cleaned up")
72 }
73 })
74 }
75}
76
77func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
78 for _, marks := range [][]harnessMark{
79 {{Sequence: 1, Kind: "terminal_flushed", Tick: 1, Valid: true}},
80 {{Sequence: 1, Kind: "input_injected", Tick: 2, Valid: true}, {Sequence: 1, Kind: "terminal_flushed", Tick: 1, Valid: true}},
81 {{Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}},
82 {{Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Sequence: 1, Kind: "input_injected", Tick: 2, Valid: true}},
83 } {
84 if _, err := pairedSamples(marks); err == nil {
85 t.Fatalf("invalid marks accepted: %+v", marks)
86 }
87 }
88 got, err := pairedSamples([]harnessMark{{Sequence: 9, Kind: "input_injected", Tick: 1, Valid: true}, {Sequence: 9, Kind: "terminal_flushed", Tick: 5, Valid: true}})
89 if err != nil || len(got) != 1 || got[0] != 4 {
90 t.Fatalf("got %v %v", got, err)
91 }
92}
93
94func TestHarnessCollectsOnlyCompleteMeasuredIntervalEvents(t *testing.T) {
95 interval := measuredInterval{Start: 100, End: 200}
96 marks := []harnessMark{
97 {Sequence: 1, Kind: "input_injected", Tick: 10, Valid: true},
98 {Sequence: 1, Kind: "terminal_flushed", Tick: 20, Valid: true}, // warmup
99 {Sequence: 2, Kind: "input_injected", Tick: 100, Valid: true},
100 {Sequence: 2, Kind: "terminal_flushed", Tick: 120, Valid: true},
101 {Sequence: 3, Kind: "input_injected", Tick: 150, Valid: true},
102 {Sequence: 3, Kind: "terminal_flushed", Tick: 201, Valid: true}, // exits interval
103 {Sequence: 4, Kind: "input_injected", Tick: 99, Valid: true},
104 {Sequence: 4, Kind: "terminal_flushed", Tick: 110, Valid: true}, // enters interval
105 {Sequence: 5, Kind: "input_injected", Tick: 180, Valid: true},
106 {Sequence: 5, Kind: "terminal_flushed", Tick: 200, Valid: true}, // end is excluded
107 }
108 events, err := measuredEventSamples(marks, interval)
109 if err != nil {
110 t.Fatal(err)
111 }
112 if len(events) != 1 || events[0].Sequence != 2 || events[0].Latency != 20 {
113 t.Fatalf("measured events=%+v, want only complete sequence 2", events)
114 }
115}
116
117func TestHarnessRejectsInsufficientMeasuredEventSamples(t *testing.T) {
118 marks := make([]harnessMark, 0, 2*(minimumInIntervalEventSamples-1))
119 for i := 0; i < minimumInIntervalEventSamples-1; i++ {
120 marks = append(marks,
121 harnessMark{Sequence: uint64(i + 1), Kind: "input_injected", Tick: int64(i * 100), Valid: true},
122 harnessMark{Sequence: uint64(i + 1), Kind: "terminal_flushed", Tick: int64(i*100 + 10), Valid: true},
123 )
124 }
125 events, err := measuredEventSamples(marks, measuredInterval{Start: 0, End: int64(minimumInIntervalEventSamples * 100)})
126 if err != nil {
127 t.Fatal(err)
128 }
129 if err := requireMinimumEventSamples(events); err == nil {
130 t.Fatal("insufficient measured event samples accepted")
131 }
132}
133
134func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
135 dir := t.TempDir()
136 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
137 if err != nil {
138 t.Fatal(err)
139 }
140 t.Cleanup(func() {
141 if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
142 t.Error(err)
143 }
144 })
145 clock := &fakeClock{}
146 h := defaultHarness()
147 h.clock, h.launcher = clock, insufficientEventLauncher{clock: clock}
148 _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "late", Roles: []string{"daemon", "client"}}, 1, raw)
149 if err == nil || !strings.Contains(err.Error(), "insufficient in-interval event samples") {
150 t.Fatalf("straddling run error=%v, want insufficient in-interval sample rejection", err)
151 }
152 if err := raw.Close(); err != nil {
153 t.Fatal(err)
154 }
155 marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
156 if got, want := len(marks), 4; got != want {
157 t.Fatalf("raw marks=%d, want warmup and straddling pair=%d: %+v", got, want, marks)
158 }
159 if straddling := marks[len(marks)-1]; straddling.Sequence != 2 || straddling.Kind != "terminal_flushed" || !straddling.Valid {
160 t.Fatalf("straddling boundary=%+v", straddling)
161 }
162}
163
164func TestHarnessCalculatesEventCadenceAndMaxGapFromRawPairs(t *testing.T) {
165 events := []eventSample{
166 {Sequence: 1, Injected: 100, Flushed: 110, Latency: 10},
167 {Sequence: 2, Injected: 130, Flushed: 140, Latency: 10},
168 {Sequence: 3, Injected: 180, Flushed: 190, Latency: 10},
169 }
170 cadence, maxGap := eventCadence(events)
171 if cadence.Count != 2 || cadence.P50 != 30 || cadence.P95 != 30 || cadence.P99 != 30 || maxGap != 50 {
172 t.Fatalf("cadence=%+v maxGap=%d", cadence, maxGap)
173 }
174}
175
176func TestHarnessIncludesAllRequiredProcessLocalSpans(t *testing.T) {
177 got := make([]string, 0, len(spanPairs))
178 for _, pair := range spanPairs {
179 got = append(got, pair.name)
180 }
181 want := []string{"capture_duration", "compose_duration", "diff_duration", "queue_wait", "ack_blocked_interval", "emit_duration", "adapter_send_duration", "adapter_receive_duration"}
182 if !equalStrings(got, want) {
183 t.Fatalf("span pairs = %v, want %v", got, want)
184 }
185}
186
187func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {
188 var b bytes.Buffer
189 for _, m := range []harnessMark{{Scenario: "s", Run: 1, Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Scenario: "s", Run: 1, Sequence: 1, Kind: "terminal_flushed", Tick: 2, Valid: true}} {
190 b.WriteString(mustJSON(m))
191 b.WriteByte('\n')
192 }
193 path := filepath.Join(t.TempDir(), "raw.jsonl")
194 if err := os.WriteFile(path, b.Bytes(), 0o600); err != nil {
195 t.Fatal(err)
196 }
197 if got, err := readJSONL(path); err != nil || len(got) != 2 {
198 t.Fatalf("records=%d err=%v", len(got), err)
199 }
200}
201
202func TestHarnessClosesRolesBeforeMergingReceiveSpans(t *testing.T) {
203 dir := t.TempDir()
204 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
205 if err != nil {
206 t.Fatal(err)
207 }
208 t.Cleanup(func() {
209 if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
210 t.Error(err)
211 }
212 })
213 launcher := &closeTraceLauncher{}
214 h := defaultHarness()
215 h.clock, h.launcher = &fakeClock{}, launcher
216 result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}, manifest{}, scenario{ID: "receive-cleanup", Roles: []string{"daemon", "client"}}, 1, raw)
217 if err != nil {
218 t.Fatal(err)
219 }
220 if len(result.Spans) != 2 {
221 t.Fatalf("post-cleanup receive spans=%+v", result.Spans)
222 }
223 for _, s := range result.Spans {
224 if s.Name != "adapter_receive_duration" || len(s.Samples) != 1 || s.Samples[0] != 10 {
225 t.Fatalf("unmatched or wrong receive span: %+v", s)
226 }
227 }
228 for i, p := range launcher.process {
229 if !p.closed {
230 t.Fatalf("role %s was not closed before trace merge", launcher.mappings[i].Role)
231 }
232 }
233}
234
235func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) {
236 base, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
237 if err != nil {
238 t.Fatal(err)
239 }
240 for i := range base.Scenarios {
241 if i > 0 {
242 base.Scenarios[i].InapplicableReason = "fixture limits this test to one public topology"
243 base.Scenarios[i].Roles = nil
244 }
245 }
246 dir := t.TempDir()
247 manifestPath := filepath.Join(dir, "manifest.json")
248 if err := os.WriteFile(manifestPath, []byte(mustJSON(base)), 0o600); err != nil {
249 t.Fatal(err)
250 }
251 l := &traceLauncher{}
252 h := defaultHarness()
253 h.clock = &fakeClock{}
254 h.launcher = l
255 h.gitSHA = func() string { return "test-sha" }
256 if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", filepath.Join(dir, "out"), "--warmup", "1s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {
257 t.Fatal(err)
258 }
259 for _, name := range []string{"raw-harness.jsonl", "runs.json", "summary.json"} {
260 if _, err := os.Stat(filepath.Join(dir, "out", name)); err != nil {
261 t.Errorf("%s: %v", name, err)
262 }
263 }
264 var got summary
265 b, err := os.ReadFile(filepath.Join(dir, "out", "summary.json"))
266 if err != nil {
267 t.Fatal(err)
268 }
269 if err := json.Unmarshal(b, &got); err != nil {
270 t.Fatal(err)
271 }
272 if got.EndToEnd.Count < 10*minimumInIntervalEventSamples || got.Repetitions != 10 || got.RunP50Dispersion.Count != 10 || got.Cadence.Count == 0 || got.MaxGap == 0 || len(got.Spans) == 0 {
273 t.Fatalf("summary=%+v", got)
274 }
275 for _, s := range got.Spans {
276 if s.Distribution.Count < 10 {
277 t.Fatalf("insufficient span summary: %+v", s)
278 }
279 }
280}
281
282func TestHarnessAcceptsConcurrentProductionTraceInterleaving(t *testing.T) {
283 path := filepath.Join(t.TempDir(), "local-client.jsonl")
284 if err := os.WriteFile(path, []byte(boundedLocalConcurrentTrace), 0o600); err != nil {
285 t.Fatal(err)
286 }
287 spans, err := mergeProcessTraces([]processMapping{{
288 ProcessID: "1x4-idle-local-r002-client-02", ClockDomain: "1x4-idle-local-r002-client-02", TracePath: path,
289 Scenario: "1x4-idle-local", Run: 2,
290 }})
291 if err != nil {
292 t.Fatalf("valid concurrent production trace rejected: %v", err)
293 }
294 if len(spans) != 2 || spans[0].Component != "ipc" || spans[0].Name != "adapter_send_duration" || spans[0].Samples[0] != 21900 || spans[1].Component != "ipc" || spans[1].Name != "adapter_receive_duration" || spans[1].Samples[0] != 178550 {
295 t.Fatalf("spans=%+v", spans)
296 }
297}
298
299func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) {
300 write := func(t *testing.T, records ...traceRecord) processMapping {
301 t.Helper()
302 path := filepath.Join(t.TempDir(), "trace.jsonl")
303 f, err := os.Create(path)
304 if err != nil {
305 t.Fatal(err)
306 }
307 for _, r := range records {
308 b, err := json.Marshal(r)
309 if err != nil {
310 t.Fatal(err)
311 }
312 if _, err := f.Write(append(b, '\n')); err != nil {
313 t.Fatal(err)
314 }
315 }
316 if err := f.Close(); err != nil {
317 t.Fatal(err)
318 }
319 return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}
320 }
321 base := func(kind string, tick int64) traceRecord {
322 return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: true}
323 }
324 for _, records := range [][]traceRecord{
325 {base("diff_end", 1)},
326 {base("diff_start", 2), base("diff_end", 1)},
327 {base("diff_start", 1)},
328 {base("diff_start", 1), base("diff_start", 2)},
329 {base("diff_start", 1), func() traceRecord { r := base("diff_end", 2); r.Component = "ipc"; return r }()},
330 {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()},
331 {func() traceRecord { r := base("diff_start", 1); r.Scenario = "other-scenario"; return r }()},
332 {func() traceRecord { r := base("diff_start", 1); r.Run = 2; return r }()},
333 } {
334 if _, err := mergeProcessTraces([]processMapping{write(t, records...)}); err == nil {
335 t.Fatalf("invalid trace accepted: %+v", records)
336 }
337 }
338 m := write(t, base("diff_start", 1), base("diff_end", 3))
339 spans, err := mergeProcessTraces([]processMapping{m})
340 if err != nil || len(spans) != 1 || spans[0].Samples[0] != 2 {
341 t.Fatalf("spans=%+v err=%v", spans, err)
342 }
343}
344
345func TestHarnessExcludesFailedSpanDurationsWhileValidatingPairing(t *testing.T) {
346 base := func(kind string, tick int64, valid bool) traceRecord {
347 return traceRecord{ProcessID: "one", Component: "adapter", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: valid}
348 }
349 for _, tc := range []struct {
350 name string
351 records []traceRecord
352 wantSpans int
353 wantErr bool
354 }{
355 {"successful adapter send is sampled", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, true)}, 1, false},
356 {"failed adapter send end is excluded", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, false)}, 0, false},
357 {"failed adapter receive end is excluded", []traceRecord{base("adapter_receive_start", 10, true), base("adapter_receive_end", 20, false)}, 0, false},
358 {"failed adapter send start is excluded", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_end", 20, true)}, 0, false},
359 {"failed adapter receive boundaries are excluded", []traceRecord{base("adapter_receive_start", 10, false), base("adapter_receive_end", 20, false)}, 0, false},
360 {"failed end without start is rejected", []traceRecord{base("adapter_send_end", 20, false)}, 0, true},
361 {"duplicate failed start is rejected", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_start", 15, false)}, 0, true},
362 {"negative failed span is rejected", []traceRecord{base("adapter_receive_start", 20, false), base("adapter_receive_end", 10, false)}, 0, true},
363 {"cross-component failed end is rejected", []traceRecord{base("adapter_send_start", 10, true), func() traceRecord { r := base("adapter_send_end", 20, false); r.Component = "other"; return r }()}, 0, true},
364 } {
365 t.Run(tc.name, func(t *testing.T) {
366 path := filepath.Join(t.TempDir(), "trace.jsonl")
367 var lines []string
368 for _, record := range tc.records {
369 lines = append(lines, mustJSON(record))
370 }
371 if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
372 t.Fatal(err)
373 }
374 spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}})
375 if tc.wantErr {
376 if err == nil {
377 t.Fatalf("invalid trace accepted: %+v", tc.records)
378 }
379 return
380 }
381 if err != nil || len(spans) != tc.wantSpans {
382 t.Fatalf("spans=%+v err=%v, want %d spans", spans, err, tc.wantSpans)
383 }
384 if tc.wantSpans == 1 && spans[0].Samples[0] != 10 {
385 t.Fatalf("successful span=%+v, want 10ns", spans[0])
386 }
387 })
388 }
389}