1// Package observability contains concrete process-local trace sinks.
2package observability
3
4import (
5 "encoding/json"
6 "io"
7 "os"
8 "sync"
9
10 "github.com/bnema/vev/internal/ports"
11)
12
13// NewJSONL opens path as the sole timestamp owner for one OS process. Marks
14// carrying a caller timestamp are rejected rather than rewritten, preventing
15// accidental arithmetic across clock domains.
16func NewJSONL(path string, clock ports.Clock, processID string) (ports.RuntimeObserver, io.Closer, error) {
17 if path == "" || clock == nil || processID == "" {
18 return nil, nil, os.ErrInvalid
19 }
20 f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
21 if err != nil {
22 return nil, nil, err
23 }
24 j := &jsonl{file: f, clock: clock, processID: processID}
25 return j, j, nil
26}
27
28type jsonl struct {
29 mu sync.Mutex
30 file *os.File
31 clock ports.Clock
32 processID string
33 closed bool
34 err error
35}
36
37type record struct {
38 Schema uint16 `json:"schema"`
39 ProcessID string `json:"process_id"`
40 Component string `json:"component"`
41 Scenario string `json:"scenario"`
42 Run uint64 `json:"run"`
43 Sequence uint64 `json:"sequence"`
44 RequestID uint64 `json:"request_id"`
45 Epoch uint64 `json:"epoch"`
46 Kind ports.RuntimeMarkKind `json:"kind"`
47 Tick int64 `json:"tick"`
48 Bytes uint64 `json:"bytes"`
49 Fragments uint64 `json:"fragments"`
50 Retransmits uint64 `json:"retransmits"`
51 Pending uint64 `json:"pending"`
52 AckRTTNanos int64 `json:"ack_rtt_nanos"`
53 Valid bool `json:"valid"`
54}
55
56func (j *jsonl) ObserveRuntime(mark ports.RuntimeMark) {
57 // Tick is a write-once value owned by this adapter. Reject it before any
58 // mutation so callers cannot smuggle a timestamp from their own clock.
59 if j == nil || mark.Tick != 0 {
60 return
61 }
62 mark.ProcessID = j.processID
63 if !ports.RuntimeMarkValid(mark) {
64 return
65 }
66 j.mu.Lock()
67 defer j.mu.Unlock()
68 if j.closed {
69 return
70 }
71 r := record{Schema: mark.Schema, ProcessID: mark.ProcessID, Component: mark.Component, Scenario: mark.Scenario, Run: mark.Run, Sequence: mark.Sequence, RequestID: mark.RequestID, Epoch: mark.Epoch, Kind: mark.Kind, Tick: j.clock.Now().UnixNano(), Bytes: mark.Bytes, Fragments: mark.Fragments, Retransmits: mark.Retransmits, Pending: mark.Pending, AckRTTNanos: mark.AckRTTNanos, Valid: mark.Valid}
72 encoded, err := json.Marshal(r)
73 if err != nil {
74 j.recordError(err)
75 return
76 }
77 if _, err := j.file.Write(append(encoded, '\n')); err != nil {
78 j.recordError(err)
79 }
80}
81
82func (j *jsonl) recordError(err error) {
83 if err != nil && j.err == nil {
84 j.err = err
85 }
86}
87
88func (j *jsonl) Close() error {
89 j.mu.Lock()
90 defer j.mu.Unlock()
91 if j.closed {
92 return nil
93 }
94 j.closed = true
95 closeErr := j.file.Close()
96 if j.err != nil {
97 return j.err
98 }
99 return closeErr
100}