fix(obs): exclude failed span durations
- commit
- 6d5b59fd87c588d2249143b7cee19ffe91523a19
- author
- brice · · 2026-07-13 03:16 UTC
- parent
- 8d86d44
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +58 -5
M cmd/vev-perf-harness/main.go +8 -1
| @@ -1615,6 +1615,7 @@ type traceRecord struct { | |||
| 1615 | 1615 | Epoch uint64 `json:"epoch"` | |
| 1616 | 1616 | Kind string `json:"kind"` | |
| 1617 | 1617 | Tick int64 `json:"tick"` | |
| 1618 | + | Valid bool `json:"valid"` | |
| 1618 | 1619 | } | |
| 1619 | 1620 | type spanPair struct{ start, end, name string } | |
| 1620 | 1621 | ||
| @@ -1689,7 +1690,13 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) { | |||
| 1689 | 1690 | if r.Tick < start.Tick { | |
| 1690 | 1691 | return nil, errors.New("negative same-process span") | |
| 1691 | 1692 | } | |
| 1692 | - | out = append(out, span{Component: r.Component, Name: pair.name, Samples: []int64{r.Tick - start.Tick}}) | |
| 1693 | + | // Validity affects only measurement eligibility, never structural | |
| 1694 | + | // pairing: every start/end must still match in its exact process, | |
| 1695 | + | // component, and correlation domain. A failed start, failed end, or | |
| 1696 | + | // both is a paired diagnostic fact but contributes no latency sample. | |
| 1697 | + | if start.Valid && r.Valid { | |
| 1698 | + | out = append(out, span{Component: r.Component, Name: pair.name, Samples: []int64{r.Tick - start.Tick}}) | |
| 1699 | + | } | |
| 1693 | 1700 | delete(starts, k) | |
| 1694 | 1701 | } | |
| 1695 | 1702 | } | |
| @@ -1615,6 +1615,7 @@ type traceRecord struct { | |||||
| 1615 | Epoch uint64 `json:"epoch"` | 1615 | Epoch uint64 `json:"epoch"` | ||
| 1616 | Kind string `json:"kind"` | 1616 | Kind string `json:"kind"` | ||
| 1617 | Tick int64 `json:"tick"` | 1617 | Tick int64 `json:"tick"` | ||
| 1618 | + | Valid bool `json:"valid"` | |||
| 1618 | } | 1619 | } | ||
| 1619 | type spanPair struct{ start, end, name string } | 1620 | type spanPair struct{ start, end, name string } | ||
| 1620 | 1621 | ||||
| @@ -1689,7 +1690,13 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) { | |||||
| 1689 | if r.Tick < start.Tick { | 1690 | if r.Tick < start.Tick { | ||
| 1690 | return nil, errors.New("negative same-process span") | 1691 | return nil, errors.New("negative same-process span") | ||
| 1691 | } | 1692 | } | ||
| 1692 | - | out = append(out, span{Component: r.Component, Name: pair.name, Samples: []int64{r.Tick - start.Tick}}) | 1693 | + | // Validity affects only measurement eligibility, never structural |
| 1694 | + | // pairing: every start/end must still match in its exact process, | |||
| 1695 | + | // component, and correlation domain. A failed start, failed end, or | |||
| 1696 | + | // both is a paired diagnostic fact but contributes no latency sample. | |||
| 1697 | + | if start.Valid && r.Valid { | |||
| 1698 | + | out = append(out, span{Component: r.Component, Name: pair.name, Samples: []int64{r.Tick - start.Tick}}) | |||
| 1699 | + | } | |||
| 1693 | delete(starts, k) | 1700 | delete(starts, k) | ||
| 1694 | } | 1701 | } | ||
| 1695 | } | 1702 | } | ||
M cmd/vev-perf-harness/main_test.go +50 -4
| @@ -1277,7 +1277,7 @@ func (l *traceLauncher) Launch(m processMapping, command roleCommand) (launchedP | |||
| 1277 | 1277 | defer f.Close() | |
| 1278 | 1278 | for i, pair := range spanPairs { | |
| 1279 | 1279 | for n, kind := range []string{pair.start, pair.end} { | |
| 1280 | - | b, _ := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5)}) | |
| 1280 | + | b, _ := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5), Valid: true}) | |
| 1281 | 1281 | if _, err := f.Write(append(b, '\n')); err != nil { | |
| 1282 | 1282 | return nil, err | |
| 1283 | 1283 | } | |
| @@ -1303,12 +1303,12 @@ func (l *closeTraceLauncher) Launch(m processMapping, command roleCommand) (laun | |||
| 1303 | 1303 | if err != nil { | |
| 1304 | 1304 | return nil, err | |
| 1305 | 1305 | } | |
| 1306 | - | start := traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_start", Tick: 10} | |
| 1306 | + | start := traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_start", Tick: 10, Valid: true} | |
| 1307 | 1307 | if err := appendTraceRecord(m.TracePath, start); err != nil { | |
| 1308 | 1308 | return nil, err | |
| 1309 | 1309 | } | |
| 1310 | 1310 | return &closeTraceProcess{fakeProcess: p.(*fakeProcess), traceEnd: func() error { | |
| 1311 | - | return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20}) | |
| 1311 | + | return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20, Valid: true}) | |
| 1312 | 1312 | }}, nil | |
| 1313 | 1313 | } | |
| 1314 | 1314 | ||
| @@ -1461,7 +1461,7 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||
| 1461 | 1461 | return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1} | |
| 1462 | 1462 | } | |
| 1463 | 1463 | base := func(kind string, tick int64) traceRecord { | |
| 1464 | - | return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick} | |
| 1464 | + | return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: true} | |
| 1465 | 1465 | } | |
| 1466 | 1466 | for _, records := range [][]traceRecord{ | |
| 1467 | 1467 | {base("diff_end", 1)}, | |
| @@ -1483,3 +1483,49 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||
| 1483 | 1483 | t.Fatalf("spans=%+v err=%v", spans, err) | |
| 1484 | 1484 | } | |
| 1485 | 1485 | } | |
| 1486 | + | ||
| 1487 | + | func TestHarnessExcludesFailedSpanDurationsWhileValidatingPairing(t *testing.T) { | |
| 1488 | + | base := func(kind string, tick int64, valid bool) traceRecord { | |
| 1489 | + | return traceRecord{ProcessID: "one", Component: "adapter", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: valid} | |
| 1490 | + | } | |
| 1491 | + | for _, tc := range []struct { | |
| 1492 | + | name string | |
| 1493 | + | records []traceRecord | |
| 1494 | + | wantSpans int | |
| 1495 | + | wantErr bool | |
| 1496 | + | }{ | |
| 1497 | + | {"successful adapter send is sampled", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, true)}, 1, false}, | |
| 1498 | + | {"failed adapter send end is excluded", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, false)}, 0, false}, | |
| 1499 | + | {"failed adapter receive end is excluded", []traceRecord{base("adapter_receive_start", 10, true), base("adapter_receive_end", 20, false)}, 0, false}, | |
| 1500 | + | {"failed adapter send start is excluded", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_end", 20, true)}, 0, false}, | |
| 1501 | + | {"failed adapter receive boundaries are excluded", []traceRecord{base("adapter_receive_start", 10, false), base("adapter_receive_end", 20, false)}, 0, false}, | |
| 1502 | + | {"failed end without start is rejected", []traceRecord{base("adapter_send_end", 20, false)}, 0, true}, | |
| 1503 | + | {"duplicate failed start is rejected", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_start", 15, false)}, 0, true}, | |
| 1504 | + | {"negative failed span is rejected", []traceRecord{base("adapter_receive_start", 20, false), base("adapter_receive_end", 10, false)}, 0, true}, | |
| 1505 | + | {"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}, | |
| 1506 | + | } { | |
| 1507 | + | t.Run(tc.name, func(t *testing.T) { | |
| 1508 | + | path := filepath.Join(t.TempDir(), "trace.jsonl") | |
| 1509 | + | var lines []string | |
| 1510 | + | for _, record := range tc.records { | |
| 1511 | + | lines = append(lines, mustJSON(record)) | |
| 1512 | + | } | |
| 1513 | + | if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil { | |
| 1514 | + | t.Fatal(err) | |
| 1515 | + | } | |
| 1516 | + | spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}}) | |
| 1517 | + | if tc.wantErr { | |
| 1518 | + | if err == nil { | |
| 1519 | + | t.Fatalf("invalid trace accepted: %+v", tc.records) | |
| 1520 | + | } | |
| 1521 | + | return | |
| 1522 | + | } | |
| 1523 | + | if err != nil || len(spans) != tc.wantSpans { | |
| 1524 | + | t.Fatalf("spans=%+v err=%v, want %d spans", spans, err, tc.wantSpans) | |
| 1525 | + | } | |
| 1526 | + | if tc.wantSpans == 1 && spans[0].Samples[0] != 10 { | |
| 1527 | + | t.Fatalf("successful span=%+v, want 10ns", spans[0]) | |
| 1528 | + | } | |
| 1529 | + | }) | |
| 1530 | + | } | |
| 1531 | + | } | |
| @@ -1277,7 +1277,7 @@ func (l *traceLauncher) Launch(m processMapping, command roleCommand) (launchedP | |||||
| 1277 | defer f.Close() | 1277 | defer f.Close() | ||
| 1278 | for i, pair := range spanPairs { | 1278 | for i, pair := range spanPairs { | ||
| 1279 | for n, kind := range []string{pair.start, pair.end} { | 1279 | for n, kind := range []string{pair.start, pair.end} { | ||
| 1280 | - | b, _ := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5)}) | 1280 | + | b, _ := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5), Valid: true}) |
| 1281 | if _, err := f.Write(append(b, '\n')); err != nil { | 1281 | if _, err := f.Write(append(b, '\n')); err != nil { | ||
| 1282 | return nil, err | 1282 | return nil, err | ||
| 1283 | } | 1283 | } | ||
| @@ -1303,12 +1303,12 @@ func (l *closeTraceLauncher) Launch(m processMapping, command roleCommand) (laun | |||||
| 1303 | if err != nil { | 1303 | if err != nil { | ||
| 1304 | return nil, err | 1304 | return nil, err | ||
| 1305 | } | 1305 | } | ||
| 1306 | - | start := traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_start", Tick: 10} | 1306 | + | start := traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_start", Tick: 10, Valid: true} |
| 1307 | if err := appendTraceRecord(m.TracePath, start); err != nil { | 1307 | if err := appendTraceRecord(m.TracePath, start); err != nil { | ||
| 1308 | return nil, err | 1308 | return nil, err | ||
| 1309 | } | 1309 | } | ||
| 1310 | return &closeTraceProcess{fakeProcess: p.(*fakeProcess), traceEnd: func() error { | 1310 | return &closeTraceProcess{fakeProcess: p.(*fakeProcess), traceEnd: func() error { | ||
| 1311 | - | return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20}) | 1311 | + | return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20, Valid: true}) |
| 1312 | }}, nil | 1312 | }}, nil | ||
| 1313 | } | 1313 | } | ||
| 1314 | 1314 | ||||
| @@ -1461,7 +1461,7 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||||
| 1461 | return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1} | 1461 | return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1} | ||
| 1462 | } | 1462 | } | ||
| 1463 | base := func(kind string, tick int64) traceRecord { | 1463 | base := func(kind string, tick int64) traceRecord { | ||
| 1464 | - | return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick} | 1464 | + | return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: true} |
| 1465 | } | 1465 | } | ||
| 1466 | for _, records := range [][]traceRecord{ | 1466 | for _, records := range [][]traceRecord{ | ||
| 1467 | {base("diff_end", 1)}, | 1467 | {base("diff_end", 1)}, | ||
| @@ -1483,3 +1483,49 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||||
| 1483 | t.Fatalf("spans=%+v err=%v", spans, err) | 1483 | t.Fatalf("spans=%+v err=%v", spans, err) | ||
| 1484 | } | 1484 | } | ||
| 1485 | } | 1485 | } | ||
| 1486 | + | ||||
| 1487 | + | func TestHarnessExcludesFailedSpanDurationsWhileValidatingPairing(t *testing.T) { | |||
| 1488 | + | base := func(kind string, tick int64, valid bool) traceRecord { | |||
| 1489 | + | return traceRecord{ProcessID: "one", Component: "adapter", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: valid} | |||
| 1490 | + | } | |||
| 1491 | + | for _, tc := range []struct { | |||
| 1492 | + | name string | |||
| 1493 | + | records []traceRecord | |||
| 1494 | + | wantSpans int | |||
| 1495 | + | wantErr bool | |||
| 1496 | + | }{ | |||
| 1497 | + | {"successful adapter send is sampled", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, true)}, 1, false}, | |||
| 1498 | + | {"failed adapter send end is excluded", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, false)}, 0, false}, | |||
| 1499 | + | {"failed adapter receive end is excluded", []traceRecord{base("adapter_receive_start", 10, true), base("adapter_receive_end", 20, false)}, 0, false}, | |||
| 1500 | + | {"failed adapter send start is excluded", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_end", 20, true)}, 0, false}, | |||
| 1501 | + | {"failed adapter receive boundaries are excluded", []traceRecord{base("adapter_receive_start", 10, false), base("adapter_receive_end", 20, false)}, 0, false}, | |||
| 1502 | + | {"failed end without start is rejected", []traceRecord{base("adapter_send_end", 20, false)}, 0, true}, | |||
| 1503 | + | {"duplicate failed start is rejected", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_start", 15, false)}, 0, true}, | |||
| 1504 | + | {"negative failed span is rejected", []traceRecord{base("adapter_receive_start", 20, false), base("adapter_receive_end", 10, false)}, 0, true}, | |||
| 1505 | + | {"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}, | |||
| 1506 | + | } { | |||
| 1507 | + | t.Run(tc.name, func(t *testing.T) { | |||
| 1508 | + | path := filepath.Join(t.TempDir(), "trace.jsonl") | |||
| 1509 | + | var lines []string | |||
| 1510 | + | for _, record := range tc.records { | |||
| 1511 | + | lines = append(lines, mustJSON(record)) | |||
| 1512 | + | } | |||
| 1513 | + | if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil { | |||
| 1514 | + | t.Fatal(err) | |||
| 1515 | + | } | |||
| 1516 | + | spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}}) | |||
| 1517 | + | if tc.wantErr { | |||
| 1518 | + | if err == nil { | |||
| 1519 | + | t.Fatalf("invalid trace accepted: %+v", tc.records) | |||
| 1520 | + | } | |||
| 1521 | + | return | |||
| 1522 | + | } | |||
| 1523 | + | if err != nil || len(spans) != tc.wantSpans { | |||
| 1524 | + | t.Fatalf("spans=%+v err=%v, want %d spans", spans, err, tc.wantSpans) | |||
| 1525 | + | } | |||
| 1526 | + | if tc.wantSpans == 1 && spans[0].Samples[0] != 10 { | |||
| 1527 | + | t.Fatalf("successful span=%+v, want 10ns", spans[0]) | |||
| 1528 | + | } | |||
| 1529 | + | }) | |||
| 1530 | + | } | |||
| 1531 | + | } | |||