Skip to content

Commit c9a59f1

Browse files
author
gluck
committed
telemetry: log every provider round-trip, and retain a queryable tail
An aria stopped answering. `figaro status` said "idle", `queue ls` was empty, and the operator's remaining options were a 10 MB logs.jsonl of OTel envelopes and a trace file that only lands records when a span ENDS. The daemon knew perfectly well that the provider had refused it four times in the last hour. Nothing could ask. Four facts existed on the wire and figaro recorded none of them: - req_bytes was len(bodyBytes), and bodyBytes is only materialized when dumping to disk, so in normal operation it was always 0. "This request was 700 KB" is the first thing you want to know about a refused turn. - Retry-After was recorded nowhere - not a log, not a metric, not a span attribute. It is the entire explanation of the stall: an hour-long value on a 429 is a usage window, not a throttle. - The ratelimit-* headers were dropped, so nothing said WHICH limit was hit. - Aria attribution rode on the byte-dump switch, so unless someone had set figaro_wire_dir every round-trip in the trace was anonymous. So: every round-trip is now LOGGED, as an ordinary slog record with typed attributes, at INFO when healthy and WARN when not. Nothing bespoke; it lands in logs.jsonl like everything else. That leaves the question a log file cannot answer - "what is true NOW" - and the answer is split in two rather than solved by a private ledger. HISTORY -> internal/logring, a bounded slog.Handler. It forwards everything to the inner handler and retains a copy of some of it. Retention never replaces emission: a ring that suppressed logging would be a liability, since it dies with the process, which is exactly when a crash makes you want it most. What it keeps is a predicate - WARN and above by default, plus messages a subsystem opts in by name - so healthy operation copies nothing at all. Measured: 0 B/op and 0 allocs/op on the not-retained path; ~500 B and ~950ns on a kept record, for the mandatory copy of the attributes (a Handler may not retain a Record). 512 records is about a quarter of a megabyte, against a 32 MiB segment cache. It is general on purpose. "The last N warnings from any subsystem" is a facility figaro did not have, and angelus.log - the file everyone opens first - has been stale since May. This is the mechanism that fixes that too. PRESENT -> a small in-flight map in wirelog. A record is a point in time; "still running" is not a record. An outstanding request is current state, and it is precisely what tracing cannot show either, since spans export on end - so the request that is still hanging, the only one an incident is ever about, was invisible by construction. So wirelog keeps exactly that and nothing else: requests that have departed and not returned, with their age. History lives in the log; only the present lives here.
1 parent 3ec3326 commit c9a59f1

10 files changed

Lines changed: 964 additions & 10 deletions

File tree

‎internal/logring/bench_test.go‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package logring
2+
3+
import (
4+
"io"
5+
"log/slog"
6+
"testing"
7+
)
8+
9+
// The two numbers that decide whether this is affordable.
10+
//
11+
// NotRetained is the common case - a log line the policy does not keep - and
12+
// it must cost nothing beyond the predicate. Measured: 0 B/op, 0 allocs/op;
13+
// the nanoseconds are the inner JSON handler's, not ours.
14+
//
15+
// Retained is what a kept record costs: ~500 B and ~950ns over the baseline,
16+
// for the mandatory copy of the attributes (a Handler may not retain a
17+
// Record). At the default capacity of 512 that bounds the ring at roughly a
18+
// quarter of a megabyte, which `figaro doctor mem` reports next to a 32 MiB
19+
// segment cache.
20+
21+
func BenchmarkHandleNotRetained(b *testing.B) {
22+
ring := New(slog.NewJSONHandler(io.Discard, nil), DefaultCapacity, AtLeast(slog.LevelWarn))
23+
log := slog.New(ring)
24+
b.ReportAllocs()
25+
for i := 0; i < b.N; i++ {
26+
log.Info("log opened", "dir", "/some/path", "codec", "jsonl", "sealed", 0)
27+
}
28+
}
29+
30+
func BenchmarkHandleRetained(b *testing.B) {
31+
ring := New(slog.NewJSONHandler(io.Discard, nil), DefaultCapacity, AtLeast(slog.LevelInfo))
32+
log := slog.New(ring)
33+
b.ReportAllocs()
34+
for i := 0; i < b.N; i++ {
35+
log.Info("provider round-trip", "aria", "94f0752b", "method", "POST",
36+
"url", "https://api.anthropic.com/v1/messages", "duration_ms", 11015,
37+
"req_bytes", 729498, "status", 200, "request_id", "req_011Ce3TTTY6NjCj66TBPgp3K")
38+
}
39+
}

‎internal/logring/logring.go‎

Lines changed: 225 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,225 @@
1+
// Package logring keeps a bounded, in-memory tail of slog records so a
2+
// running daemon can be asked what it has been saying, without anyone
3+
// grepping a file.
4+
//
5+
// Why this exists, concretely: an aria stopped answering, and the operator's
6+
// options were `figaro status` (which said "idle"), a 10 MB logs.jsonl of OTel
7+
// envelopes, and a trace file that only lands records when a span ENDS. The
8+
// daemon knew perfectly well that the provider had refused it four times in
9+
// the last hour. Nothing could ask.
10+
//
11+
// The design is deliberately small and deliberately NOT a log sink of its own:
12+
//
13+
// - It is a slog.Handler that forwards EVERYTHING to an inner handler and
14+
// retains a copy of some of it. Retention never replaces emission. A ring
15+
// that suppressed logging would be a liability, since it dies with the
16+
// process - exactly when a crash makes you want it most.
17+
// - It retains slog.Record CLONES, not formatted strings and not JSON. The
18+
// hot path costs one small slice copy; rendering happens when a human
19+
// runs `figaro doctor`, which is to say approximately never.
20+
// - What it retains is decided by a predicate, so the common case (WARN and
21+
// above, a handful of records an hour) costs one integer comparison.
22+
package logring
23+
24+
import (
25+
"context"
26+
"log/slog"
27+
"sync"
28+
"time"
29+
)
30+
31+
// DefaultCapacity is how many records a ring retains. Records are small
32+
// (a header plus five inline attrs; larger ones spill to a slice), so this
33+
// is a couple of hundred kilobytes - immaterial next to the segment cache,
34+
// and reported by `figaro doctor mem` alongside it.
35+
const DefaultCapacity = 512
36+
37+
// Keep decides whether a record is worth retaining. It runs on the logging
38+
// goroutine, so it must be cheap and must not block.
39+
type Keep func(slog.Record) bool
40+
41+
// AtLeast keeps records at or above a level. This is the default policy and
42+
// the reason the ring is free in normal operation: nothing is copied until
43+
// something goes wrong.
44+
func AtLeast(level slog.Level) Keep {
45+
return func(r slog.Record) bool { return r.Level >= level }
46+
}
47+
48+
// WithMessage keeps records whose message matches exactly, at any level. It is
49+
// how a subsystem opts a specific INFO-level stream into retention - provider
50+
// round-trips, say - without lowering the bar for everything else.
51+
func WithMessage(msgs ...string) Keep {
52+
set := make(map[string]struct{}, len(msgs))
53+
for _, m := range msgs {
54+
set[m] = struct{}{}
55+
}
56+
return func(r slog.Record) bool {
57+
_, ok := set[r.Message]
58+
return ok
59+
}
60+
}
61+
62+
// Any keeps a record if any of the given policies would.
63+
func Any(keeps ...Keep) Keep {
64+
return func(r slog.Record) bool {
65+
for _, k := range keeps {
66+
if k != nil && k(r) {
67+
return true
68+
}
69+
}
70+
return false
71+
}
72+
}
73+
74+
// Entry is one retained record, flattened for reading. Seq is monotonic across
75+
// the ring's whole life, so a gap between the oldest retained Seq and the
76+
// newest is exactly how many records retention dropped.
77+
type Entry struct {
78+
Seq uint64 `json:"seq"`
79+
Time time.Time `json:"time"`
80+
Level string `json:"level"`
81+
Msg string `json:"msg"`
82+
Attrs map[string]any `json:"attrs,omitempty"`
83+
}
84+
85+
// Ring is a slog.Handler that forwards to inner and retains a bounded tail.
86+
// The zero value is not usable; call New.
87+
type Ring struct {
88+
inner slog.Handler
89+
keep Keep
90+
prefix []slog.Attr // accumulated by WithAttrs
91+
groups []string // accumulated by WithGroup
92+
93+
// The buffer is shared by every derived handler (WithAttrs/WithGroup
94+
// return a new Ring pointing at the same store), because a ring per
95+
// derived logger would retain nothing useful.
96+
store *store
97+
}
98+
99+
type store struct {
100+
mu sync.Mutex
101+
buf []Entry
102+
next int
103+
seq uint64
104+
}
105+
106+
// New wraps inner. keep decides what is retained; nil means AtLeast(WARN).
107+
// capacity <= 0 means DefaultCapacity.
108+
func New(inner slog.Handler, capacity int, keep Keep) *Ring {
109+
if capacity <= 0 {
110+
capacity = DefaultCapacity
111+
}
112+
if keep == nil {
113+
keep = AtLeast(slog.LevelWarn)
114+
}
115+
return &Ring{
116+
inner: inner,
117+
keep: keep,
118+
store: &store{buf: make([]Entry, 0, capacity)},
119+
}
120+
}
121+
122+
func (r *Ring) Enabled(ctx context.Context, l slog.Level) bool {
123+
return r.inner.Enabled(ctx, l)
124+
}
125+
126+
// Handle forwards first, then retains. Forwarding first means a panic in our
127+
// own bookkeeping cannot cost the caller their log line.
128+
func (r *Ring) Handle(ctx context.Context, rec slog.Record) error {
129+
err := r.inner.Handle(ctx, rec)
130+
if r.keep(rec) {
131+
r.store.add(r.flatten(rec))
132+
}
133+
return err
134+
}
135+
136+
func (r *Ring) WithAttrs(attrs []slog.Attr) slog.Handler {
137+
next := *r
138+
next.inner = r.inner.WithAttrs(attrs)
139+
next.prefix = append(append([]slog.Attr(nil), r.prefix...), attrs...)
140+
return &next
141+
}
142+
143+
func (r *Ring) WithGroup(name string) slog.Handler {
144+
next := *r
145+
next.inner = r.inner.WithGroup(name)
146+
next.groups = append(append([]string(nil), r.groups...), name)
147+
return &next
148+
}
149+
150+
// flatten copies the record's attributes out. This is the one allocation the
151+
// ring costs, and it is mandatory: a Handler may not retain a Record, whose
152+
// overflow attributes live in a slice it does not own.
153+
func (r *Ring) flatten(rec slog.Record) Entry {
154+
e := Entry{
155+
Time: rec.Time,
156+
Level: rec.Level.String(),
157+
Msg: rec.Message,
158+
}
159+
n := len(r.prefix) + rec.NumAttrs()
160+
if n > 0 {
161+
e.Attrs = make(map[string]any, n)
162+
for _, a := range r.prefix {
163+
e.Attrs[a.Key] = a.Value.Resolve().Any()
164+
}
165+
rec.Attrs(func(a slog.Attr) bool {
166+
e.Attrs[a.Key] = a.Value.Resolve().Any()
167+
return true
168+
})
169+
}
170+
return e
171+
}
172+
173+
func (s *store) add(e Entry) {
174+
s.mu.Lock()
175+
defer s.mu.Unlock()
176+
if cap(s.buf) == 0 {
177+
return
178+
}
179+
s.seq++
180+
e.Seq = s.seq
181+
if len(s.buf) < cap(s.buf) {
182+
s.buf = append(s.buf, e)
183+
return
184+
}
185+
s.buf[s.next] = e
186+
s.next = (s.next + 1) % cap(s.buf)
187+
}
188+
189+
// Recent returns retained records oldest first and newest last,
190+
// filtered by match (nil keeps all) and limited to the newest n (n <= 0 for
191+
// everything retained).
192+
func (r *Ring) Recent(n int, match func(Entry) bool) []Entry {
193+
s := r.store
194+
s.mu.Lock()
195+
defer s.mu.Unlock()
196+
if len(s.buf) == 0 {
197+
return nil
198+
}
199+
ordered := make([]Entry, 0, len(s.buf))
200+
if len(s.buf) < cap(s.buf) {
201+
ordered = append(ordered, s.buf...)
202+
} else {
203+
ordered = append(ordered, s.buf[s.next:]...)
204+
ordered = append(ordered, s.buf[:s.next]...)
205+
}
206+
out := ordered[:0:0]
207+
for _, e := range ordered {
208+
if match == nil || match(e) {
209+
out = append(out, e)
210+
}
211+
}
212+
if n > 0 && len(out) > n {
213+
out = out[len(out)-n:]
214+
}
215+
return out
216+
}
217+
218+
// Stats reports what the ring is holding, for `figaro doctor mem`. Every
219+
// other cache in the daemon reports its footprint; this one should too.
220+
func (r *Ring) Stats() (retained, capacity int, seq uint64) {
221+
s := r.store
222+
s.mu.Lock()
223+
defer s.mu.Unlock()
224+
return len(s.buf), cap(s.buf), s.seq
225+
}

0 commit comments

Comments
 (0)