1//go:build linux
2
3package main
4
5import (
6 "bufio"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "os"
11 "sort"
12 "strings"
13)
14
15func measuredEventSamples(marks []harnessMark, interval measuredInterval) ([]eventSample, error) {
16 if interval.End < interval.Start {
17 return nil, errors.New("negative measured interval")
18 }
19 starts := map[uint64]int64{}
20 events := make([]eventSample, 0, len(marks)/2)
21 for _, m := range marks {
22 if !m.Valid {
23 return nil, errors.New("invalid harness boundary")
24 }
25 switch m.Kind {
26 case "input_injected":
27 if _, exists := starts[m.Sequence]; exists {
28 return nil, errors.New("duplicate input boundary")
29 }
30 starts[m.Sequence] = m.Tick
31 case "terminal_flushed":
32 start, ok := starts[m.Sequence]
33 if !ok {
34 return nil, errors.New("terminal flush without input pair")
35 }
36 if m.Tick < start {
37 return nil, errors.New("negative harness duration")
38 }
39 delete(starts, m.Sequence)
40 if start >= interval.Start && m.Tick < interval.End {
41 events = append(events, eventSample{Sequence: m.Sequence, Injected: start, Flushed: m.Tick, Latency: m.Tick - start})
42 }
43 default:
44 return nil, errors.New("unknown harness mark")
45 }
46 }
47 if len(starts) != 0 {
48 return nil, errors.New("missing terminal flush pair")
49 }
50 sort.Slice(events, func(i, j int) bool {
51 if events[i].Injected == events[j].Injected {
52 return events[i].Sequence < events[j].Sequence
53 }
54 return events[i].Injected < events[j].Injected
55 })
56 return events, nil
57}
58
59func requireMinimumEventSamples(events []eventSample) error {
60 if len(events) < minimumInIntervalEventSamples {
61 return fmt.Errorf("insufficient in-interval event samples: got %d, need at least %d", len(events), minimumInIntervalEventSamples)
62 }
63 return nil
64}
65
66func eventLatencies(events []eventSample) []int64 {
67 out := make([]int64, len(events))
68 for i, event := range events {
69 out[i] = event.Latency
70 }
71 return out
72}
73
74func eventCadenceSamples(events []eventSample) ([]int64, int64) {
75 if len(events) < 2 {
76 return nil, 0
77 }
78 gaps := make([]int64, 0, len(events)-1)
79 var maxGap int64
80 for i := 1; i < len(events); i++ {
81 gap := events[i].Injected - events[i-1].Injected
82 gaps = append(gaps, gap)
83 if gap > maxGap {
84 maxGap = gap
85 }
86 }
87 return gaps, maxGap
88}
89
90func eventCadence(events []eventSample) (distribution, int64) {
91 gaps, maxGap := eventCadenceSamples(events)
92 if len(gaps) == 0 {
93 return distribution{}, maxGap
94 }
95 return percentiles(gaps), maxGap
96}
97
98func pairedSamples(marks []harnessMark) ([]int64, error) {
99 starts := map[uint64]int64{}
100 var out []int64
101 for _, m := range marks {
102 if !m.Valid {
103 return nil, errors.New("invalid harness boundary")
104 }
105 switch m.Kind {
106 case "input_injected":
107 if _, ok := starts[m.Sequence]; ok {
108 return nil, errors.New("duplicate input boundary")
109 }
110 starts[m.Sequence] = m.Tick
111 case "terminal_flushed":
112 start, ok := starts[m.Sequence]
113 if !ok {
114 return nil, errors.New("terminal flush without input pair")
115 }
116 if m.Tick < start {
117 return nil, errors.New("negative harness duration")
118 }
119 out = append(out, m.Tick-start)
120 delete(starts, m.Sequence)
121 default:
122 return nil, errors.New("unknown harness mark")
123 }
124 }
125 if len(starts) != 0 {
126 return nil, errors.New("missing terminal flush pair")
127 }
128 if len(out) == 0 {
129 return nil, errors.New("insufficient harness samples")
130 }
131 return out, nil
132}
133
134// traceRecord is intentionally local to the command. It mirrors JSONL fields
135// needed for correlation but never exposes a clock to any production component.
136type traceRecord struct {
137 ProcessID string `json:"process_id"`
138 Component string `json:"component"`
139 Scenario string `json:"scenario"`
140 Run uint64 `json:"run"`
141 Sequence uint64 `json:"sequence"`
142 RequestID uint64 `json:"request_id"`
143 Epoch uint64 `json:"epoch"`
144 Kind string `json:"kind"`
145 Tick int64 `json:"tick"`
146 Bytes uint64 `json:"bytes"`
147 Fragments uint64 `json:"fragments"`
148 Retransmits uint64 `json:"retransmits"`
149 Pending uint64 `json:"pending"`
150 AckRTTNanos int64 `json:"ack_rtt_nanos"`
151 Valid bool `json:"valid"`
152}
153
154type spanPair struct{ start, end, name string }
155
156// mergeProcessTraces permits only records from one manifest process in a span.
157// In particular, the correlation key includes process_id before ticks are read.
158func mergeProcessTraces(mappings []processMapping) ([]span, error) {
159 known := map[string]processMapping{}
160 for _, m := range mappings {
161 if m.ProcessID == "" || m.ClockDomain == "" || m.TracePath == "" || known[m.ProcessID].ProcessID != "" {
162 return nil, errors.New("duplicate or incomplete manifest process mapping")
163 }
164 known[m.ProcessID] = m
165 }
166 out := []span{}
167 for _, m := range mappings {
168 f, err := os.Open(m.TracePath)
169 if err != nil {
170 return nil, fmt.Errorf("open manifest trace for %s role: %w", m.Role, err)
171 }
172 var records []traceRecord
173 scan := bufio.NewScanner(f)
174 for scan.Scan() {
175 var r traceRecord
176 if err := json.Unmarshal(scan.Bytes(), &r); err != nil {
177 return nil, closeFile(f, err)
178 }
179 if r.ProcessID != m.ProcessID || r.Scenario != m.Scenario || r.Run != uint64(m.Run) {
180 return nil, closeFile(f, errors.New("trace record identity does not match manifest"))
181 }
182 if r.Scenario == "" || r.Run == 0 || r.Component == "" || r.Sequence == 0 || r.RequestID == 0 || r.Epoch == 0 {
183 return nil, closeFile(f, errors.New("trace record has invalid correlation fields"))
184 }
185 if _, ok := known[r.ProcessID]; !ok {
186 return nil, closeFile(f, errors.New("unknown trace process_id"))
187 }
188 records = append(records, r)
189 }
190 if err := scan.Err(); err != nil {
191 return nil, closeFile(f, err)
192 }
193 if err := f.Close(); err != nil {
194 return nil, err
195 }
196 // A process shares one observer across concurrent components. Correlation
197 // IDs are allocated before a goroutine reaches its first mark, so a later
198 // ID can be serialized before an earlier one. Their first appearance is
199 // consequently not an ordering contract. Only marks in the same component
200 // and exact correlation domain may pair, and each pair's ticks must retain
201 // its own start-before-end ordering.
202 starts := map[string]traceRecord{}
203 traceIncomplete := false
204 for _, r := range records {
205 if r.Component == "observer" && r.Kind == "transport_diagnostic" && !r.Valid {
206 // A bounded producer queue reports that one or more marks were lost,
207 // but not their correlation keys. A later unmatched end can therefore
208 // be the end of a span whose start was dropped. Treat all subsequent
209 // unpaired ends as incomplete evidence rather than rejecting the run.
210 // Starts already observed before the gap are likewise incomplete.
211 starts = map[string]traceRecord{}
212 traceIncomplete = true
213 continue
214 }
215 for _, pair := range spanPairs {
216 key := fmt.Sprintf("%s/%s/%s/%d/%d/%d/%d", r.ProcessID, r.Component, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch)
217 if r.Kind == pair.start {
218 k := pair.name + "/" + key
219 if _, exists := starts[k]; exists {
220 return nil, errors.New("duplicate process-local span start")
221 }
222 starts[k] = r
223 }
224 if r.Kind == pair.end {
225 k := pair.name + "/" + key
226 start, ok := starts[k]
227 if !ok {
228 if traceIncomplete {
229 continue
230 }
231 return nil, errors.New("span end without same-process start")
232 }
233 if r.Tick < start.Tick {
234 return nil, errors.New("negative same-process span")
235 }
236 // Validity affects only measurement eligibility, never structural
237 // pairing: every start/end must still match in its exact process,
238 // component, and correlation domain. A failed start, failed end, or
239 // both is a paired diagnostic fact but contributes no latency sample.
240 if start.Valid && r.Valid {
241 out = append(out, span{Component: r.Component, Name: pair.name, Samples: []int64{r.Tick - start.Tick}})
242 }
243 delete(starts, k)
244 }
245 }
246 }
247 if len(starts) != 0 {
248 keys := make([]string, 0, len(starts))
249 for key := range starts {
250 keys = append(keys, key)
251 }
252 sort.Strings(keys)
253 return nil, fmt.Errorf("missing process-local span pair for %s: %s", m.ProcessID, strings.Join(keys, ", "))
254 }
255 }
256 return out, nil
257}
258
259func summarizeSpans(all []span) []spanSummary {
260 grouped := map[string]*span{}
261 for _, s := range all {
262 k := s.Component + "\x00" + s.Name
263 if grouped[k] == nil {
264 grouped[k] = &span{Component: s.Component, Name: s.Name}
265 }
266 grouped[k].Samples = append(grouped[k].Samples, s.Samples...)
267 }
268 keys := make([]string, 0, len(grouped))
269 for k := range grouped {
270 keys = append(keys, k)
271 }
272 sort.Strings(keys)
273 out := make([]spanSummary, 0, len(keys))
274 for _, k := range keys {
275 s := grouped[k]
276 if len(s.Samples) < minimumRepetitions {
277 out = append(out, spanSummary{Component: s.Component, Name: s.Name, Distribution: distribution{Count: len(s.Samples)}})
278 continue
279 }
280 out = append(out, spanSummary{Component: s.Component, Name: s.Name, Distribution: percentiles(s.Samples)})
281 }
282 return out
283}
284
285func percentiles(samples []int64) distribution {
286 v := append([]int64(nil), samples...)
287 sort.Slice(v, func(i, j int) bool { return v[i] < v[j] })
288 at := func(p int) int64 { return v[(len(v)-1)*p/100] }
289 return distribution{len(v), at(50), at(95), at(99), v[len(v)-1]}
290}
291
292func closeFile(f *os.File, err error) error {
293 return errors.Join(err, f.Close())
294}
295
296func writeJSON(path string, v any) error {
297 f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
298 if err != nil {
299 return err
300 }
301 if err := json.NewEncoder(f).Encode(v); err != nil {
302 return closeFile(f, err)
303 }
304 return f.Close()
305}
306
307func mustJSON(v any) string {
308 b, err := json.Marshal(v)
309 if err != nil {
310 panic(fmt.Sprintf("marshal JSON: %v", err))
311 }
312 return string(b)
313}
314
315func safeName(s string) string {
316 return strings.Map(func(r rune) rune {
317 if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' {
318 return r
319 }
320 return '-'
321 }, s)
322}
323
324func readJSONL(path string) ([]map[string]json.RawMessage, error) {
325 f, err := os.Open(path)
326 if err != nil {
327 return nil, err
328 }
329 var r []map[string]json.RawMessage
330 scan := bufio.NewScanner(f)
331 for scan.Scan() {
332 var v map[string]json.RawMessage
333 if err := json.Unmarshal(scan.Bytes(), &v); err != nil {
334 return nil, closeFile(f, err)
335 }
336 r = append(r, v)
337 }
338 if err := scan.Err(); err != nil {
339 return nil, closeFile(f, err)
340 }
341 if err := f.Close(); err != nil {
342 return nil, err
343 }
344 return r, nil
345}