runlog.go
1// Package runlog reads and writes a run's events.ndjson: one
2// run.rtw.ci.event per line, the single observable format shared by the
3// terminal renderer, `ci view`, the runner's streams and the appview.
4package runlog
5
6import (
7 "bufio"
8 "bytes"
9 "encoding/json"
10 "fmt"
11 "io"
12 "os"
13 "sync"
14 "time"
15
16 "git.kilimanjaro.io/ci/internal/lex"
17)
18
19// Writer appends events to an events.ndjson file. It is safe for concurrent
20// use (an action's stdout and stderr are written from two goroutines).
21type Writer struct {
22 mu sync.Mutex
23 f *os.File
24 run string
25 seq int
26 sinks []func(lex.RunEvent)
27 now func() time.Time
28}
29
30// Open opens (or creates) path for appending, continuing the sequence of
31// any events already in it, so a resumed run extends its own log.
32func Open(path, run string) (*Writer, error) {
33 n, err := countLines(path)
34 if err != nil {
35 return nil, err
36 }
37 f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
38 if err != nil {
39 return nil, err
40 }
41 return &Writer{f: f, run: run, seq: n, now: time.Now}, nil
42}
43
44func countLines(path string) (int, error) {
45 f, err := os.Open(path)
46 if os.IsNotExist(err) {
47 return 0, nil
48 }
49 if err != nil {
50 return 0, err
51 }
52 defer f.Close()
53 n := 0
54 buf := make([]byte, 64*1024)
55 for {
56 c, err := f.Read(buf)
57 n += bytes.Count(buf[:c], []byte{'\n'})
58 if err == io.EOF {
59 return n, nil
60 }
61 if err != nil {
62 return 0, err
63 }
64 }
65}
66
67// AddSink registers a function called with every event after it is written.
68func (w *Writer) AddSink(f func(lex.RunEvent)) {
69 w.mu.Lock()
70 defer w.mu.Unlock()
71 w.sinks = append(w.sinks, f)
72}
73
74// Emit stamps ev with the next sequence number, the time and the run id, and
75// appends it.
76func (w *Writer) Emit(ev lex.RunEvent) error {
77 w.mu.Lock()
78 defer w.mu.Unlock()
79 w.seq++
80 ev.Seq = w.seq
81 ev.Time = w.now().UTC().Format(time.RFC3339Nano)
82 ev.Run = w.run
83 b, err := json.Marshal(ev)
84 if err != nil {
85 return err
86 }
87 if _, err := w.f.Write(append(b, '\n')); err != nil {
88 return err
89 }
90 for _, s := range w.sinks {
91 s(ev)
92 }
93 return nil
94}
95
96// Close closes the file.
97func (w *Writer) Close() error { return w.f.Close() }
98
99// Lines returns an io.WriteCloser that emits one log event per line written
100// to it, for an action's stdout or stderr. Close flushes a final partial line.
101func (w *Writer) Lines(action string, attempt int, stream string) io.WriteCloser {
102 return &lineWriter{w: w, action: action, attempt: attempt, stream: stream}
103}
104
105type lineWriter struct {
106 w *Writer
107 action string
108 attempt int
109 stream string
110 buf []byte
111}
112
113func (l *lineWriter) Write(p []byte) (int, error) {
114 l.buf = append(l.buf, p...)
115 for {
116 i := bytes.IndexByte(l.buf, '\n')
117 if i < 0 {
118 break
119 }
120 if err := l.emit(l.buf[:i]); err != nil {
121 return 0, err
122 }
123 l.buf = l.buf[i+1:]
124 }
125 return len(p), nil
126}
127
128func (l *lineWriter) emit(line []byte) error {
129 return l.w.Emit(lex.RunEvent{
130 Kind: lex.KindLog,
131 Action: l.action,
132 Attempt: l.attempt,
133 Stream: l.stream,
134 Content: string(bytes.TrimSuffix(line, []byte{'\r'})),
135 })
136}
137
138func (l *lineWriter) Close() error {
139 if len(l.buf) == 0 {
140 return nil
141 }
142 err := l.emit(l.buf)
143 l.buf = nil
144 return err
145}
146
147// Read decodes every event in r.
148func Read(r io.Reader) ([]lex.RunEvent, error) {
149 var out []lex.RunEvent
150 err := Scan(r, func(ev lex.RunEvent) error { out = append(out, ev); return nil })
151 return out, err
152}
153
154// Scan calls f for each event in r, in order.
155func Scan(r io.Reader, f func(lex.RunEvent) error) error {
156 sc := bufio.NewScanner(r)
157 sc.Buffer(make([]byte, 64*1024), 16*1024*1024)
158 line := 0
159 for sc.Scan() {
160 line++
161 if len(bytes.TrimSpace(sc.Bytes())) == 0 {
162 continue
163 }
164 var ev lex.RunEvent
165 if err := json.Unmarshal(sc.Bytes(), &ev); err != nil {
166 return fmt.Errorf("line %d: %w", line, err)
167 }
168 if err := f(ev); err != nil {
169 return err
170 }
171 }
172 return sc.Err()
173}
