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