Off unless a log file is given; then spans log their duration and counts as one JSON line each, and frames are summarized once a second (p50, p95 and max render and key-to-frame times, live heap, gauges such as cells and JEV questions waiting). Events carry sizes, counts and durations only, never cell contents. A call site costs 15 ns and no allocation while off (Start copies up to four attributes into the Span rather than keeping the variadic slice, which escaped): BenchmarkSpanOff 15 ns, 0 allocs; BenchmarkFrameOff 1.6 ns. On, a span is about 0.5 us.
- commit
- 595dd8cec9ab097f566a9b8b04887133e4e95dd2
- parent
- 91fd92fa48a4b6f963355ef4fc6a8af8da1657ea
- author
- Dami <dami.etoile@gmail.com>
| A | internal/ | 115 | +++++++++++++++++++ |
| A | internal/ | 186 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 156 | +++++++++++++++++++++++++ |
@@ -0,0 +1,115 @@
1+package telemetry2+3+import (4+ "log/slog"5+ "maps"6+ "runtime/metrics"7+ "slices"8+ "sync"9+ "time"10+)11+12+// Frames come 60 a second, too many to log one by one at Info, so they13+// are summarized once a second with the gauges: a "frames" event with14+// counts, percentiles and the live heap.15+16+// window is how often the summary is written.17+const window = time.Second18+19+// maxSamples bounds the durations kept per window.20+const maxSamples = 409621+22+type frameStats struct {23+ mu sync.Mutex24+ since time.Time25+ render []time.Duration // View and everything the frame took26+ keys []time.Duration // from a key press to its frame27+ gauges map[string]int6428+}29+30+var frames = frameStats{gauges: map[string]int64{}}31+32+func (f *frameStats) reset(now time.Time) {33+ f.mu.Lock()34+ defer f.mu.Unlock()35+ f.since, f.render, f.keys = now, f.render[:0], f.keys[:0]36+}37+38+// Frame records a frame that took render to draw. key is how long ago39+// the key press it answers happened, or 0 when it answers none.40+func Frame(render, key time.Duration) {41+ if !on.Load() {42+ return43+ }44+ now := time.Now()45+ f := &frames46+ f.mu.Lock()47+ if len(f.render) < maxSamples {48+ f.render = append(f.render, render)49+ }50+ if key > 0 && len(f.keys) < maxSamples {51+ f.keys = append(f.keys, key)52+ }53+ due := now.Sub(f.since) >= window54+ var rs, ks []time.Duration55+ var gauges map[string]int6456+ if due {57+ rs, ks, gauges = slices.Clone(f.render), slices.Clone(f.keys), maps.Clone(f.gauges)58+ f.since, f.render, f.keys = now, f.render[:0], f.keys[:0]59+ }60+ f.mu.Unlock()61+ if due {62+ summarize(rs, ks, gauges)63+ }64+ Debug("frame", render, slog.Float64("key_ms", ms(key)))65+}66+67+// Set records a gauge, such as the number of cells or JEV questions68+// waiting, reported with the next frame summary.69+func Set(name string, v int64) {70+ if !on.Load() {71+ return72+ }73+ frames.mu.Lock()74+ frames.gauges[name] = v75+ frames.mu.Unlock()76+}77+78+func summarize(render, keys []time.Duration, gauges map[string]int64) {79+ attrs := []slog.Attr{80+ slog.Int("frames", len(render)),81+ slog.Float64("render_p50_ms", ms(percentile(render, 50))),82+ slog.Float64("render_p95_ms", ms(percentile(render, 95))),83+ slog.Float64("render_max_ms", ms(percentile(render, 100))),84+ slog.Int("keys", len(keys)),85+ slog.Float64("key_p50_ms", ms(percentile(keys, 50))),86+ slog.Float64("key_p95_ms", ms(percentile(keys, 95))),87+ slog.Float64("key_max_ms", ms(percentile(keys, 100))),88+ slog.Int64("heap_bytes", heapBytes()),89+ }90+ for _, k := range slices.Sorted(maps.Keys(gauges)) {91+ attrs = append(attrs, slog.Int64(k, gauges[k]))92+ }93+ log(slog.LevelInfo, "frames", window, attrs)94+}95+96+// percentile is the p-th percentile of ds (nearest rank), 0 when empty.97+func percentile(ds []time.Duration, p int) time.Duration {98+ if len(ds) == 0 {99+ return 0100+ }101+ slices.Sort(ds)102+ i := (len(ds)*p + 99) / 100103+ return ds[max(i-1, 0)]104+}105+106+// heapBytes is the heap held by live and not yet swept objects, read107+// without stopping the world as runtime.ReadMemStats would.108+func heapBytes() int64 {109+ s := []metrics.Sample{{Name: "/memory/classes/heap/objects:bytes"}}110+ metrics.Read(s)111+ if s[0].Value.Kind() != metrics.KindUint64 {112+ return 0113+ }114+ return int64(s[0].Value.Uint64())115+}@@ -0,0 +1,186 @@
1+// Package telemetry records what 012 spends its time on, as structured2+// JSON events written with log/slog to a file (stdout belongs to the3+// terminal UI). It is off unless a log file is given (012 --log path, or4+// O12_LOG); then spans are timed, frame times are summarized once a5+// second, and everything lands in one JSON line per event that DuckDB or6+// an OpenTelemetry Collector's filelog receiver can read (see7+// docs/observability.md).8+//9+// Events carry sizes, counts and durations only: never cell contents,10+// file contents or secrets.11+//12+// The API is small on purpose, so call sites stay few and a different13+// backend (OTLP) can sit behind it without touching them.14+package telemetry15+16+import (17+ "context"18+ "io"19+ "log/slog"20+ "os"21+ "runtime"22+ "sync"23+ "sync/atomic"24+ "time"25+)26+27+// on is read by every call site, so being off costs one atomic load.28+var on atomic.Bool29+30+var (31+ mu sync.Mutex32+ logger = slog.New(slog.DiscardHandler)33+ closer io.Closer34+)35+36+// Config says where events go.37+type Config struct {38+ // LogPath is the JSON log file, appended to. Empty turns telemetry39+ // off.40+ LogPath string41+ // Level is the least severe level logged; Info by default. Debug adds42+ // an event for every frame and every command.43+ Level slog.Level44+ // Version is 012's version or commit, added to the start event.45+ Version string46+}47+48+// ConfigFromEnv reads O12_LOG and O12_LOG_LEVEL (debug, info, warn).49+func ConfigFromEnv() Config {50+ c := Config{LogPath: os.Getenv("O12_LOG")}51+ c.Level.UnmarshalText([]byte(os.Getenv("O12_LOG_LEVEL")))52+ return c53+}54+55+// Setup starts recording as c says. The returned function flushes and56+// closes the log; call it on exit.57+func Setup(c Config) (func() error, error) {58+ if c.LogPath == "" {59+ return func() error { return nil }, nil60+ }61+ f, err := os.OpenFile(c.LogPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)62+ if err != nil {63+ return nil, err64+ }65+ start(f, c)66+ slog.Info("start", "version", c.Version, "go", runtime.Version(), "os", runtime.GOOS, "arch", runtime.GOARCH,67+ "cpus", runtime.NumCPU(), "pid", os.Getpid())68+ return Close, nil69+}70+71+// start sends events to w; tests use it with a buffer.72+func start(w io.Writer, c Config) {73+ mu.Lock()74+ defer mu.Unlock()75+ h := slog.NewJSONHandler(w, &slog.HandlerOptions{Level: c.Level})76+ logger = slog.New(h).With("service", "012")77+ slog.SetDefault(logger)78+ if cl, ok := w.(io.Closer); ok {79+ closer = cl80+ }81+ frames.reset(time.Now())82+ on.Store(true)83+}84+85+// Close stops recording and closes the log file.86+func Close() error {87+ mu.Lock()88+ defer mu.Unlock()89+ if !on.Load() {90+ return nil91+ }92+ on.Store(false)93+ logger = slog.New(slog.DiscardHandler)94+ slog.SetDefault(logger)95+ if closer != nil {96+ err := closer.Close()97+ closer = nil98+ return err99+ }100+ return nil101+}102+103+// Enabled reports whether events are being recorded.104+func Enabled() bool { return on.Load() }105+106+// Logger is the event logger; it discards everything while off.107+func Logger() *slog.Logger {108+ mu.Lock()109+ defer mu.Unlock()110+ return logger111+}112+113+// Span times one operation. The zero Span, returned while off, does114+// nothing.115+type Span struct {116+ name string117+ start time.Time118+ // attrs holds up to four attributes given to Start, in the Span119+ // itself: keeping the variadic slice would move it to the heap at120+ // every call site, telemetry on or off.121+ attrs [4]slog.Attr122+ n int123+}124+125+// Start begins timing an operation named like "recalc" or "import", with126+// up to four attributes known at the start.127+func Start(name string, attrs ...slog.Attr) Span {128+ if !on.Load() {129+ return Span{}130+ }131+ s := Span{name: name, start: time.Now()}132+ s.n = copy(s.attrs[:], attrs)133+ return s134+}135+136+// End logs the operation with its duration in milliseconds and any137+// attributes learned along the way, such as rows read.138+func (s Span) End(attrs ...slog.Attr) {139+ if s.name == "" || !on.Load() {140+ return141+ }142+ Event(s.name, time.Since(s.start), append(s.attrs[:s.n:s.n], attrs...)...)143+}144+145+// Fail ends the span with an error, logged at Warn with its message.146+func (s Span) Fail(err error, attrs ...slog.Attr) {147+ if err == nil {148+ s.End(attrs...)149+ return150+ }151+ if s.name == "" || !on.Load() {152+ return153+ }154+ attrs = append(append(s.attrs[:s.n:s.n], attrs...), slog.String("error", err.Error()))155+ log(slog.LevelWarn, s.name, time.Since(s.start), attrs)156+}157+158+// Event logs a finished operation that took d.159+func Event(name string, d time.Duration, attrs ...slog.Attr) {160+ if !on.Load() {161+ return162+ }163+ log(slog.LevelInfo, name, d, attrs)164+}165+166+// Debug logs a finished operation at Debug level, for events too167+// frequent to keep by default.168+func Debug(name string, d time.Duration, attrs ...slog.Attr) {169+ if !on.Load() {170+ return171+ }172+ log(slog.LevelDebug, name, d, attrs)173+}174+175+func log(level slog.Level, name string, d time.Duration, attrs []slog.Attr) {176+ l := Logger()177+ ctx := context.Background()178+ if !l.Enabled(ctx, level) {179+ return180+ }181+ all := make([]slog.Attr, 0, len(attrs)+2)182+ all = append(all, slog.String("event", name), slog.Float64("dur_ms", ms(d)))183+ l.LogAttrs(ctx, level, name, append(all, attrs...)...)184+}185+186+func ms(d time.Duration) float64 { return float64(d.Microseconds()) / 1000 }@@ -0,0 +1,156 @@
1+package telemetry2+3+import (4+ "bytes"5+ "encoding/json"6+ "errors"7+ "io"8+ "log/slog"9+ "strings"10+ "testing"11+ "time"12+)13+14+// record turns telemetry on into a buffer for the test.15+func record(t testing.TB, level slog.Level) *bytes.Buffer {16+ var buf bytes.Buffer17+ start(&buf, Config{Level: level})18+ t.Cleanup(func() { Close() })19+ return &buf20+}21+22+func events(t *testing.T, buf *bytes.Buffer) []map[string]any {23+ t.Helper()24+ var out []map[string]any25+ for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") {26+ if line == "" {27+ continue28+ }29+ var ev map[string]any30+ if err := json.Unmarshal([]byte(line), &ev); err != nil {31+ t.Fatalf("%q: %v", line, err)32+ }33+ out = append(out, ev)34+ }35+ return out36+}37+38+func TestOffRecordsNothing(t *testing.T) {39+ if Enabled() {40+ t.Fatal("on without Setup")41+ }42+ s := Start("recalc")43+ s.End(slog.Int("cells", 1))44+ Event("x", time.Second)45+ Frame(time.Millisecond, time.Millisecond)46+ Set("cells", 3)47+ if s.name != "" || s.n != 0 {48+ t.Errorf("span while off = %+v", s)49+ }50+ stop, err := Setup(Config{})51+ if err != nil || Enabled() {52+ t.Fatalf("Setup without a path: %v, enabled %v", err, Enabled())53+ }54+ stop()55+}56+57+func TestSpansAreJSONEvents(t *testing.T) {58+ buf := record(t, slog.LevelInfo)59+ s := Start("import", slog.String("format", "CSV"))60+ s.End(slog.Int("rows", 42))61+ Start("open").Fail(errors.New("bad file"))62+ Debug("frame", time.Millisecond) // below the level63+ evs := events(t, buf)64+ if len(evs) != 2 {65+ t.Fatalf("%d events: %s", len(evs), buf)66+ }67+ imp := evs[0]68+ if imp["event"] != "import" || imp["format"] != "CSV" || imp["rows"] != 42.0 || imp["service"] != "012" || imp["level"] != "INFO" {69+ t.Errorf("import event %v", imp)70+ }71+ if _, ok := imp["dur_ms"].(float64); !ok {72+ t.Errorf("no duration in %v", imp)73+ }74+ if evs[1]["level"] != "WARN" || evs[1]["error"] != "bad file" {75+ t.Errorf("failure event %v", evs[1])76+ }77+}78+79+func TestFramesAreSummarized(t *testing.T) {80+ buf := record(t, slog.LevelInfo)81+ Set("cells", 1234)82+ for i := range 10 {83+ Frame(time.Duration(i+1)*time.Millisecond, 0)84+ }85+ Frame(time.Millisecond, 20*time.Millisecond)86+ if buf.Len() > 0 {87+ t.Fatalf("summary before the window closed: %s", buf)88+ }89+ frames.mu.Lock()90+ frames.since = time.Now().Add(-2 * window)91+ frames.mu.Unlock()92+ Frame(time.Millisecond, 0)93+ evs := events(t, buf)94+ if len(evs) != 1 {95+ t.Fatalf("%d events: %s", len(evs), buf)96+ }97+ ev := evs[0]98+ want := map[string]float64{"frames": 12, "render_max_ms": 10, "keys": 1, "key_p95_ms": 20, "cells": 1234}99+ for k, v := range want {100+ if ev[k] != v {101+ t.Errorf("%s = %v, want %v", k, ev[k], v)102+ }103+ }104+ if h, _ := ev["heap_bytes"].(float64); h <= 0 {105+ t.Errorf("heap_bytes = %v", ev["heap_bytes"])106+ }107+}108+109+func TestPercentile(t *testing.T) {110+ ds := []time.Duration{5, 1, 4, 2, 3}111+ tests := []struct {112+ p int113+ want time.Duration114+ }{{0, 1}, {50, 3}, {95, 5}, {100, 5}}115+ for _, tt := range tests {116+ if got := percentile(ds, tt.p); got != tt.want {117+ t.Errorf("p%d = %v, want %v", tt.p, got, tt.want)118+ }119+ }120+ if percentile(nil, 50) != 0 {121+ t.Error("empty percentile")122+ }123+}124+125+// The cost at a call site while telemetry is off: what every keystroke126+// pays for having it compiled in.127+func BenchmarkSpanOff(b *testing.B) {128+ for b.Loop() {129+ s := Start("recalc", slog.Int("n", 1))130+ s.End(slog.Int("cells", 2))131+ }132+}133+134+func BenchmarkFrameOff(b *testing.B) {135+ for b.Loop() {136+ Frame(time.Millisecond, 0)137+ }138+}139+140+// The cost while on, writing JSON (to io.Discard).141+func BenchmarkSpanOn(b *testing.B) {142+ start(io.Discard, Config{})143+ defer Close()144+ for b.Loop() {145+ s := Start("recalc", slog.Int("n", 1))146+ s.End(slog.Int("cells", 2))147+ }148+}149+150+func BenchmarkFrameOn(b *testing.B) {151+ start(io.Discard, Config{})152+ defer Close()153+ for b.Loop() {154+ Frame(time.Millisecond, 0)155+ }156+}