Events go out as OTLP logs, spans as root spans with their own trace and span ids (linked from the log record of the same event), and the frame summaries, operation durations and drop counts as metrics. A background goroutine batches from bounded queues, drops and counts on overflow, gzips, and sends what is left on Close within one request timeout. Configured by the standard OTEL_EXPORTER_OTLP_* variables and OTEL_SDK_DISABLED; off, call sites still cost one atomic load.
- commit
- 666f75ec271ade6245e294ae7f9fb9e261f83157
- parent
- dcb7447a48c2b64bff20f723184671aacb120cf0
- author
- Dami <dami.etoile@gmail.com>
| M | internal/ | 39 | +- |
| A | internal/ | 618 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 440 | +++++++++++++++++++++ |
| A | internal/ | 286 | ++++++++++++++ |
| M | internal/ | 144 | +++++-- |
| M | internal/ | 20 | +- |
@@ -53,13 +53,14 @@ func Frame(render, key time.Duration) {
5353 due := now.Sub(f.since) >= window5454 var rs, ks []time.Duration5555 var gauges map[string]int6456+ since := f.since5657 if due {5758 rs, ks, gauges = slices.Clone(f.render), slices.Clone(f.keys), maps.Clone(f.gauges)5859 f.since, f.render, f.keys = now, f.render[:0], f.keys[:0]5960 }6061 f.mu.Unlock()6162 if due {62- summarize(rs, ks, gauges)63+ summarize(since, now, rs, ks, gauges)6364 }6465 Debug("frame", render, slog.Float64("key_ms", ms(key)))6566 }@@ -75,22 +76,36 @@ func Set(name string, v int64) {
7576 frames.mu.Unlock()7677 }7778 78-func summarize(render, keys []time.Duration, gauges map[string]int64) {79+// summarize logs the frames drawn from since to now, and hands them to80+// OTLP as metrics when it is on.81+func summarize(since, now time.Time, render, keys []time.Duration, gauges map[string]int64) {82+ r, k, heap := statsOf(render), statsOf(keys), heapBytes()7983 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()),84+ slog.Int("frames", r.count),85+ slog.Float64("render_p50_ms", r.p50),86+ slog.Float64("render_p95_ms", r.p95),87+ slog.Float64("render_max_ms", r.pmax),88+ slog.Int("keys", k.count),89+ slog.Float64("key_p50_ms", k.p50),90+ slog.Float64("key_p95_ms", k.p95),91+ slog.Float64("key_max_ms", k.pmax),92+ slog.Int64("heap_bytes", heap),8993 }9094 for _, k := range slices.Sorted(maps.Keys(gauges)) {9195 attrs = append(attrs, slog.Int64(k, gauges[k]))9296 }93- log(slog.LevelInfo, "frames", window, attrs)97+ emit(slog.LevelInfo, "frames", window, attrs, false)98+ if _, e := current(); e != nil && e.urls.metrics != "" {99+ e.window(frameWindow{start: since, end: now, render: r, key: k, heap: heap, gauges: gauges})100+ }101+}102+103+func statsOf(ds []time.Duration) stats {104+ s := stats{count: len(ds), p50: ms(percentile(ds, 50)), p95: ms(percentile(ds, 95)), pmax: ms(percentile(ds, 100))}105+ for _, d := range ds {106+ s.sum += ms(d)107+ }108+ return s94109 }95110 96111 // percentile is the p-th percentile of ds (nearest rank), 0 when empty.@@ -0,0 +1,618 @@
1+package telemetry2+3+import (4+ "bytes"5+ "compress/gzip"6+ "context"7+ "encoding/hex"8+ "encoding/json"9+ "fmt"10+ "io"11+ "log/slog"12+ "maps"13+ "math/rand/v2"14+ "net/http"15+ "net/url"16+ "os"17+ "runtime"18+ "slices"19+ "strconv"20+ "strings"21+ "sync"22+ "time"23+)24+25+// OTLP sends events to an OpenTelemetry Collector (or any OTLP/HTTP26+// endpoint) with the JSON encoding, using only the standard library:27+// slog events as logs, spans as traces, and the frame summaries and28+// operation durations as metrics. A background goroutine batches them29+// from bounded queues; when a queue is full the newest items are dropped30+// and counted, so the UI never waits on the network.31+32+// OTLPConfig says where OTLP goes. ConfigFromEnv reads it from the33+// standard OTEL_* variables. Nothing is sent unless an endpoint is set.34+type OTLPConfig struct {35+ // Endpoint is the base URL, like http://localhost:4318, to which36+ // /v1/logs, /v1/traces and /v1/metrics are appended37+ // (OTEL_EXPORTER_OTLP_ENDPOINT, or 012 --otlp).38+ Endpoint string39+ // LogsEndpoint, TracesEndpoint and MetricsEndpoint are full URLs for40+ // one signal each, used as they are41+ // (OTEL_EXPORTER_OTLP_{LOGS,TRACES,METRICS}_ENDPOINT).42+ LogsEndpoint, TracesEndpoint, MetricsEndpoint string43+ // NoLogs, NoTraces and NoMetrics leave a signal out44+ // (OTEL_{LOGS,TRACES,METRICS}_EXPORTER=none).45+ NoLogs, NoTraces, NoMetrics bool46+ // Headers go with every request, e.g. for authentication47+ // (OTEL_EXPORTER_OTLP_HEADERS). They are never logged.48+ Headers map[string]string49+ // Resource adds resource attributes (OTEL_RESOURCE_ATTRIBUTES);50+ // service.name stays 012.51+ Resource map[string]string52+ // Timeout bounds each request (OTEL_EXPORTER_OTLP_TIMEOUT, in53+ // milliseconds); 3 s when zero.54+ Timeout time.Duration55+ // NoCompression sends plain JSON instead of gzip56+ // (OTEL_EXPORTER_OTLP_COMPRESSION=none).57+ NoCompression bool58+ // Disabled turns OTLP off whatever else is set (OTEL_SDK_DISABLED).59+ Disabled bool60+}61+62+const defaultTimeout = 3 * time.Second63+64+// otlpFromEnv reads the OTEL_* variables.65+func otlpFromEnv() OTLPConfig {66+ env := func(signal string) string {67+ return os.Getenv("OTEL_EXPORTER_OTLP_" + signal + "ENDPOINT")68+ }69+ none := func(signal string) bool {70+ return strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_"+signal+"_EXPORTER")), "none")71+ }72+ c := OTLPConfig{73+ Endpoint: env(""),74+ LogsEndpoint: env("LOGS_"),75+ TracesEndpoint: env("TRACES_"),76+ MetricsEndpoint: env("METRICS_"),77+ NoLogs: none("LOGS"),78+ NoTraces: none("TRACES"),79+ NoMetrics: none("METRICS"),80+ Headers: keyValues(os.Getenv("OTEL_EXPORTER_OTLP_HEADERS")),81+ Resource: keyValues(os.Getenv("OTEL_RESOURCE_ATTRIBUTES")),82+ NoCompression: strings.EqualFold(os.Getenv("OTEL_EXPORTER_OTLP_COMPRESSION"), "none"),83+ Disabled: strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_SDK_DISABLED")), "true"),84+ }85+ if ms, err := strconv.Atoi(os.Getenv("OTEL_EXPORTER_OTLP_TIMEOUT")); err == nil && ms > 0 {86+ c.Timeout = time.Duration(ms) * time.Millisecond87+ }88+ return c89+}90+91+// keyValues parses the "k1=v1,k2=v2" lists of OTEL_EXPORTER_OTLP_HEADERS92+// and OTEL_RESOURCE_ATTRIBUTES, whose values are percent-encoded.93+func keyValues(s string) map[string]string {94+ var m map[string]string95+ for kv := range strings.SplitSeq(s, ",") {96+ k, v, ok := strings.Cut(kv, "=")97+ k = strings.TrimSpace(k)98+ if !ok || k == "" {99+ continue100+ }101+ if d, err := url.PathUnescape(strings.TrimSpace(v)); err == nil {102+ v = d103+ }104+ if m == nil {105+ m = map[string]string{}106+ }107+ m[k] = strings.TrimSpace(v)108+ }109+ return m110+}111+112+// urls are where each signal goes; empty leaves it out.113+type urls struct{ logs, traces, metrics string }114+115+func (u urls) any() bool { return u.logs != "" || u.traces != "" || u.metrics != "" }116+117+func (c OTLPConfig) urls() urls {118+ if c.Disabled {119+ return urls{}120+ }121+ base := c.Endpoint122+ if base != "" && !strings.Contains(base, "://") {123+ base = "http://" + base124+ }125+ pick := func(own, path string, off bool) string {126+ switch {127+ case off:128+ return ""129+ case own != "":130+ return own131+ case base != "":132+ return strings.TrimRight(base, "/") + path133+ }134+ return ""135+ }136+ return urls{137+ logs: pick(c.LogsEndpoint, "/v1/logs", c.NoLogs),138+ traces: pick(c.TracesEndpoint, "/v1/traces", c.NoTraces),139+ metrics: pick(c.MetricsEndpoint, "/v1/metrics", c.NoMetrics),140+ }141+}142+143+const (144+ sigLogs = iota145+ sigTraces146+ sigMetrics147+)148+149+var signalNames = [3]string{"logs", "traces", "metrics"}150+151+// Tuning; variables so tests can shrink them.152+var (153+ // queueLimit bounds each signal's queue (metric windows count as154+ // one item each).155+ queueLimit = 2048156+ // batchSize wakes the sender before the next tick.157+ batchSize = 512158+ // flushEvery is how often queued items are sent.159+ flushEvery = 2 * time.Second160+)161+162+// durationBounds are the histogram buckets for operation durations, in163+// milliseconds.164+var durationBounds = []float64{0.5, 1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000}165+166+type durationHist struct {167+ counts []uint64168+ count uint64169+ sum, min, max float64170+}171+172+func (h *durationHist) add(v float64) {173+ if h.counts == nil {174+ h.counts = make([]uint64, len(durationBounds)+1)175+ h.min, h.max = v, v176+ }177+ i, _ := slices.BinarySearch(durationBounds, v)178+ h.counts[i]++179+ h.count++180+ h.sum += v181+ h.min, h.max = min(h.min, v), max(h.max, v)182+}183+184+// frameWindow is one second of frames, from the frame summary.185+type frameWindow struct {186+ start, end time.Time187+ render, key stats188+ heap int64189+ gauges map[string]int64190+}191+192+type stats struct {193+ count int194+ sum float64 // ms195+ p50, p95, pmax float64 // ms196+}197+198+type exporter struct {199+ urls urls200+ headers map[string]string201+ gzip bool202+ timeout time.Duration203+ client *http.Client204+ resource resource205+ scope scope206+ started time.Time207+208+ mu sync.Mutex209+ logs []logRecord210+ spans []spanData211+ windows []frameWindow212+ ops map[string]*durationHist213+ opsNew bool // ops changed since they were last sent214+ frames int64215+ sent [3]int64 // items accepted by the endpoint216+ dropped [3]int64 // items dropped because the queue was full217+ failed [3]int64 // items in requests that failed218+219+ wake chan struct{}220+ done chan struct{}221+ stopped chan struct{}222+}223+224+func newExporter(c OTLPConfig, version string) *exporter {225+ timeout := c.Timeout226+ if timeout <= 0 {227+ timeout = defaultTimeout228+ }229+ var id [16]byte230+ randomBytes(id[:])231+ res := map[string]string{}232+ maps.Copy(res, c.Resource)233+ maps.Copy(res, map[string]string{234+ "service.name": "012",235+ "service.instance.id": hex.EncodeToString(id[:]),236+ "os.type": runtime.GOOS,237+ "host.arch": runtime.GOARCH,238+ "process.runtime.name": "go",239+ "process.runtime.version": runtime.Version(),240+ })241+ if version != "" {242+ res["service.version"] = version243+ }244+ var attrs []keyValue245+ for _, k := range slices.Sorted(maps.Keys(res)) {246+ attrs = append(attrs, keyValue{Key: k, Value: str(res[k])})247+ }248+ e := &exporter{249+ urls: c.urls(),250+ headers: c.Headers,251+ gzip: !c.NoCompression,252+ timeout: timeout,253+ client: &http.Client{Timeout: timeout},254+ resource: resource{Attributes: attrs},255+ scope: scope{Name: "github.com/FelineStateMachine/012/internal/telemetry", Version: version},256+ started: time.Now(),257+ ops: map[string]*durationHist{},258+ wake: make(chan struct{}, 1),259+ done: make(chan struct{}),260+ stopped: make(chan struct{}),261+ }262+ go e.run()263+ return e264+}265+266+func randomBytes(b []byte) {267+ for i := 0; i < len(b); i += 8 {268+ var w [8]byte269+ v := rand.Uint64()270+ for j := range w {271+ w[j] = byte(v >> (8 * j))272+ }273+ copy(b[i:], w[:])274+ }275+}276+277+func newIDs() (traceID, spanID) {278+ var t traceID279+ var s spanID280+ for t.IsZero() {281+ randomBytes(t[:])282+ }283+ for s.IsZero() {284+ randomBytes(s[:])285+ }286+ return t, s287+}288+289+// span records a finished operation: a span when traces are on, and its290+// duration in the operation histogram when metrics are. It returns the291+// span's ids for the log record that goes with it. attrs is not kept.292+func (e *exporter) span(name string, start, end time.Time, failed bool, attrs []slog.Attr) (traceID, spanID) {293+ var t traceID294+ var s spanID295+ if e.urls.metrics != "" {296+ e.mu.Lock()297+ h := e.ops[name]298+ if h == nil {299+ h = &durationHist{}300+ e.ops[name] = h301+ }302+ h.add(ms(end.Sub(start)))303+ e.opsNew = true304+ e.mu.Unlock()305+ }306+ if e.urls.traces == "" {307+ return t, s308+ }309+ t, s = newIDs()310+ sp := spanData{311+ TraceID: t, SpanID: s, Name: name, Kind: spanKindInternal,312+ StartTimeUnixNano: nanos(start), EndTimeUnixNano: nanos(end),313+ Attributes: attrsOf(attrs),314+ Status: status{Code: statusOK},315+ }316+ if failed {317+ sp.Status = status{Code: statusError}318+ for _, a := range attrs {319+ if a.Key == "error" {320+ sp.Status.Message = a.Value.String()321+ }322+ }323+ }324+ e.mu.Lock()325+ ok := len(e.spans) < queueLimit326+ if ok {327+ e.spans = append(e.spans, sp)328+ } else {329+ e.dropped[sigTraces]++330+ }331+ n := len(e.spans)332+ e.mu.Unlock()333+ e.nudge(n)334+ return t, s335+}336+337+func (e *exporter) log(r logRecord) {338+ e.mu.Lock()339+ ok := len(e.logs) < queueLimit340+ if ok {341+ e.logs = append(e.logs, r)342+ } else {343+ e.dropped[sigLogs]++344+ }345+ n := len(e.logs)346+ e.mu.Unlock()347+ e.nudge(n)348+}349+350+func (e *exporter) window(w frameWindow) {351+ e.mu.Lock()352+ defer e.mu.Unlock()353+ e.frames += int64(w.render.count)354+ if len(e.windows) >= queueLimit {355+ e.dropped[sigMetrics]++356+ return357+ }358+ e.windows = append(e.windows, w)359+}360+361+// nudge wakes the sender once a queue holds a batch.362+func (e *exporter) nudge(n int) {363+ if n < batchSize {364+ return365+ }366+ select {367+ case e.wake <- struct{}{}:368+ default:369+ }370+}371+372+func (e *exporter) run() {373+ defer close(e.stopped)374+ t := time.NewTicker(flushEvery)375+ defer t.Stop()376+ for {377+ select {378+ case <-t.C:379+ case <-e.wake:380+ case <-e.done:381+ // The last flush shares one timeout, so quitting 012 waits382+ // at most that long for a collector that doesn't answer.383+ ctx, cancel := context.WithTimeout(context.Background(), e.timeout)384+ e.flush(ctx, true)385+ cancel()386+ return387+ }388+ e.flush(context.Background(), false)389+ }390+}391+392+// shutdown sends what is queued and stops the sender, waiting at most a393+// little over one request timeout.394+func (e *exporter) shutdown() {395+ close(e.done)396+ select {397+ case <-e.stopped:398+ case <-time.After(e.timeout + time.Second):399+ }400+}401+402+func (e *exporter) flush(ctx context.Context, final bool) {403+ e.mu.Lock()404+ logs, spans, windows := e.logs, e.spans, e.windows405+ e.logs, e.spans, e.windows = nil, nil, nil406+ var ms []metric407+ if len(windows) > 0 || e.opsNew || final {408+ ms = e.metricsLocked(windows, time.Now())409+ e.opsNew = false410+ }411+ e.mu.Unlock()412+413+ if len(logs) > 0 {414+ e.send(ctx, sigLogs, len(logs), logsRequest{ResourceLogs: []resourceLogs{{415+ Resource: e.resource, ScopeLogs: []scopeLogs{{Scope: e.scope, LogRecords: logs}},416+ }}})417+ }418+ if len(spans) > 0 {419+ e.send(ctx, sigTraces, len(spans), tracesRequest{ResourceSpans: []resourceSpans{{420+ Resource: e.resource, ScopeSpans: []scopeSpans{{Scope: e.scope, Spans: spans}},421+ }}})422+ }423+ if len(ms) > 0 && e.urls.metrics != "" {424+ e.send(ctx, sigMetrics, max(len(windows), 1), metricsRequest{ResourceMetrics: []resourceMetrics{{425+ Resource: e.resource, ScopeMetrics: []scopeMetrics{{Scope: e.scope, Metrics: ms}},426+ }}})427+ }428+}429+430+// metricsLocked builds the metrics: the queued frame windows as431+// summaries and gauges, and the cumulative counters and histograms as432+// they stand at now.433+func (e *exporter) metricsLocked(windows []frameWindow, now time.Time) []metric {434+ var out []metric435+ if len(windows) > 0 {436+ render := &summary{}437+ key := &summary{}438+ heap := &gauge{}439+ gauges := map[string]*gauge{}440+ for _, w := range windows {441+ render.DataPoints = append(render.DataPoints, w.render.point(w.start, w.end))442+ if w.key.count > 0 {443+ key.DataPoints = append(key.DataPoints, w.key.point(w.start, w.end))444+ }445+ heap.DataPoints = append(heap.DataPoints, intPoint(time.Time{}, w.end, w.heap))446+ for k, v := range w.gauges {447+ g := gauges[k]448+ if g == nil {449+ g = &gauge{}450+ gauges[k] = g451+ }452+ g.DataPoints = append(g.DataPoints, intPoint(time.Time{}, w.end, v))453+ }454+ }455+ out = append(out,456+ metric{Name: "o12.frame.duration", Unit: "ms", Description: "Time to draw a frame (View), per second", Summary: render},457+ metric{Name: "o12.heap", Unit: "By", Description: "Heap held by live and unswept objects", Gauge: heap})458+ if len(key.DataPoints) > 0 {459+ out = append(out, metric{Name: "o12.key.latency", Unit: "ms", Description: "From a key press to the end of its frame, per second", Summary: key})460+ }461+ for _, k := range slices.Sorted(maps.Keys(gauges)) {462+ out = append(out, metric{Name: "o12." + k, Gauge: gauges[k]})463+ }464+ }465+ out = append(out, metric{Name: "o12.frames", Unit: "{frame}", Description: "Frames drawn", Sum: &sum{466+ DataPoints: []numberPoint{intPoint(e.started, now, e.frames)},467+ AggregationTemporality: temporalityCumulative, IsMonotonic: true,468+ }})469+ if len(e.ops) > 0 {470+ h := &histogram{AggregationTemporality: temporalityCumulative}471+ for _, name := range slices.Sorted(maps.Keys(e.ops)) {472+ d := e.ops[name]473+ counts := make([]u64, len(d.counts))474+ for i, c := range d.counts {475+ counts[i] = u64(c)476+ }477+ h.DataPoints = append(h.DataPoints, histogramPoint{478+ Attributes: []keyValue{{Key: "event", Value: str(name)}},479+ StartTimeUnixNano: nanos(e.started), TimeUnixNano: nanos(now),480+ Count: u64(d.count), Sum: d.sum, BucketCounts: counts, ExplicitBounds: durationBounds,481+ Min: d.min, Max: d.max,482+ })483+ }484+ out = append(out, metric{Name: "o12.operation.duration", Unit: "ms", Description: "Duration of operations such as recalc, import and sort", Histogram: h})485+ }486+ dropped := &sum{AggregationTemporality: temporalityCumulative, IsMonotonic: true}487+ for i, n := range e.dropped {488+ dropped.DataPoints = append(dropped.DataPoints, intPoint(e.started, now, n+e.failed[i], keyValue{Key: "signal", Value: str(signalNames[i])}))489+ }490+ out = append(out, metric{Name: "o12.telemetry.dropped", Unit: "{item}", Description: "Telemetry items not delivered: queue full or request failed", Sum: dropped})491+ return out492+}493+494+func (s stats) point(start, end time.Time) summaryPoint {495+ return summaryPoint{496+ StartTimeUnixNano: nanos(start), TimeUnixNano: nanos(end),497+ Count: u64(s.count), Sum: s.sum,498+ QuantileValues: []quantileValue{{0.5, s.p50}, {0.95, s.p95}, {1, s.pmax}},499+ }500+}501+502+// send posts one request; failures are counted, never retried.503+func (e *exporter) send(ctx context.Context, sig, items int, body any) {504+ url := [3]string{e.urls.logs, e.urls.traces, e.urls.metrics}[sig]505+ err := e.post(ctx, url, body)506+ e.mu.Lock()507+ if err != nil {508+ e.failed[sig] += int64(items)509+ } else {510+ e.sent[sig] += int64(items)511+ }512+ e.mu.Unlock()513+}514+515+func (e *exporter) post(ctx context.Context, url string, body any) error {516+ var buf bytes.Buffer517+ var w io.Writer = &buf518+ var zw *gzip.Writer519+ if e.gzip {520+ zw = gzip.NewWriter(&buf)521+ w = zw522+ }523+ if err := json.NewEncoder(w).Encode(body); err != nil {524+ return err525+ }526+ if zw != nil {527+ if err := zw.Close(); err != nil {528+ return err529+ }530+ }531+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, &buf)532+ if err != nil {533+ return err534+ }535+ req.Header.Set("Content-Type", "application/json")536+ if e.gzip {537+ req.Header.Set("Content-Encoding", "gzip")538+ }539+ for k, v := range e.headers {540+ req.Header.Set(k, v)541+ }542+ resp, err := e.client.Do(req)543+ if err != nil {544+ return err545+ }546+ defer resp.Body.Close()547+ io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10))548+ if resp.StatusCode/100 != 2 {549+ return fmt.Errorf("OTLP %s: %s", url, resp.Status)550+ }551+ return nil552+}553+554+// counts reports items sent, dropped and failed per signal.555+func (e *exporter) counts() (sent, dropped, failed [3]int64) {556+ e.mu.Lock()557+ defer e.mu.Unlock()558+ return e.sent, e.dropped, e.failed559+}560+561+// spanKey carries a span's ids from log to the OTLP log handler, so the562+// log record of an operation links to its span.563+type spanKey struct{}564+565+type spanIDs struct {566+ trace traceID567+ span spanID568+}569+570+// otlpHandler is a slog.Handler that queues OTLP log records.571+type otlpHandler struct {572+ e *exporter573+ level slog.Leveler574+ attrs []keyValue575+ prefix string576+}577+578+func (h *otlpHandler) Enabled(_ context.Context, l slog.Level) bool {579+ return l >= h.level.Level()580+}581+582+func (h *otlpHandler) Handle(ctx context.Context, r slog.Record) error {583+ rec := logRecord{584+ TimeUnixNano: nanos(r.Time),585+ ObservedTimeUnixNano: nanos(time.Now()),586+ SeverityNumber: severity(r.Level),587+ SeverityText: r.Level.String(),588+ Body: str(r.Message),589+ Attributes: slices.Clip(h.attrs),590+ }591+ r.Attrs(func(a slog.Attr) bool {592+ rec.Attributes = appendAttr(rec.Attributes, h.prefix, a)593+ return true594+ })595+ if ids, ok := ctx.Value(spanKey{}).(spanIDs); ok {596+ rec.TraceID, rec.SpanID = ids.trace, ids.span597+ }598+ h.e.log(rec)599+ return nil600+}601+602+func (h *otlpHandler) WithAttrs(as []slog.Attr) slog.Handler {603+ h2 := *h604+ h2.attrs = slices.Clip(h.attrs)605+ for _, a := range as {606+ h2.attrs = appendAttr(h2.attrs, h.prefix, a)607+ }608+ return &h2609+}610+611+func (h *otlpHandler) WithGroup(name string) slog.Handler {612+ if name == "" {613+ return h614+ }615+ h2 := *h616+ h2.prefix += name + "."617+ return &h2618+}@@ -0,0 +1,440 @@
1+package telemetry2+3+import (4+ "compress/gzip"5+ "encoding/json"6+ "errors"7+ "io"8+ "log/slog"9+ "maps"10+ "net/http"11+ "net/http/httptest"12+ "os"13+ "regexp"14+ "slices"15+ "strconv"16+ "sync"17+ "testing"18+ "time"19+)20+21+// collector is a fake OTLP/HTTP endpoint that decodes what it gets.22+type collector struct {23+ t testing.TB24+ srv *httptest.Server25+ delay time.Duration26+27+ mu sync.Mutex28+ reqs []request29+}30+31+type request struct {32+ path string33+ header http.Header34+ body map[string]any35+}36+37+func newCollector(t testing.TB) *collector {38+ c := &collector{t: t}39+ c.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {40+ time.Sleep(c.delay)41+ var body io.Reader = r.Body42+ if r.Header.Get("Content-Encoding") == "gzip" {43+ zr, err := gzip.NewReader(r.Body)44+ if err != nil {45+ t.Errorf("%s: gzip: %v", r.URL.Path, err)46+ w.WriteHeader(http.StatusBadRequest)47+ return48+ }49+ body = zr50+ }51+ var m map[string]any52+ if err := json.NewDecoder(body).Decode(&m); err != nil {53+ t.Errorf("%s: %v", r.URL.Path, err)54+ }55+ c.mu.Lock()56+ c.reqs = append(c.reqs, request{r.URL.Path, r.Header.Clone(), m})57+ c.mu.Unlock()58+ w.Header().Set("Content-Type", "application/json")59+ io.WriteString(w, "{}")60+ }))61+ t.Cleanup(c.srv.Close)62+ return c63+}64+65+func (c *collector) requests() []request {66+ c.mu.Lock()67+ defer c.mu.Unlock()68+ return slices.Clone(c.reqs)69+}70+71+// items gathers the records of one signal from every request: log72+// records, spans or metrics, with the resource of the last request.73+func (c *collector) items(path string) (items []map[string]any, res map[string]any) {74+ keys := map[string][3]string{75+ "/v1/logs": {"resourceLogs", "scopeLogs", "logRecords"},76+ "/v1/traces": {"resourceSpans", "scopeSpans", "spans"},77+ "/v1/metrics": {"resourceMetrics", "scopeMetrics", "metrics"},78+ }[path]79+ outer, inner, leaf := keys[0], keys[1], keys[2]80+ for _, r := range c.requests() {81+ if r.path != path {82+ continue83+ }84+ for _, rs := range list(r.body[outer]) {85+ rs := rs.(map[string]any)86+ res = attrMap(c.t, rs["resource"].(map[string]any)["attributes"])87+ for _, ss := range list(rs[inner]) {88+ for _, it := range list(ss.(map[string]any)[leaf]) {89+ items = append(items, it.(map[string]any))90+ }91+ }92+ }93+ }94+ return items, res95+}96+97+func list(v any) []any { l, _ := v.([]any); return l }98+99+// attrMap turns OTLP attributes into key -> AnyValue, checking each100+// value holds exactly one field.101+func attrMap(t testing.TB, v any) map[string]any {102+ m := map[string]any{}103+ for _, kv := range list(v) {104+ kv := kv.(map[string]any)105+ val := kv["value"].(map[string]any)106+ if len(val) != 1 {107+ t.Errorf("attribute %v: value %v", kv["key"], val)108+ }109+ m[kv["key"].(string)] = val110+ }111+ return m112+}113+114+var (115+ hexTrace = regexp.MustCompile(`^[0-9a-f]{32}$`)116+ hexSpan = regexp.MustCompile(`^[0-9a-f]{16}$`)117+ decimal = regexp.MustCompile(`^[0-9]+$`)118+)119+120+// isInt64 checks v is an int64 as the JSON mapping writes it: a string.121+func isInt64(t testing.TB, what string, v any) {122+ t.Helper()123+ if s, ok := v.(string); !ok || !decimal.MatchString(s) {124+ t.Errorf("%s = %#v, want a decimal string", what, v)125+ }126+}127+128+func otlpSetup(t testing.TB, c OTLPConfig) {129+ t.Helper()130+ if _, err := Setup(Config{Version: "v-test", OTLP: c}); err != nil {131+ t.Fatal(err)132+ }133+ t.Cleanup(func() { Close() })134+}135+136+func TestOTLPExport(t *testing.T) {137+ c := newCollector(t)138+ otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL, Headers: map[string]string{"Authorization": "Bearer t"}})139+ Start("import", slog.String("format", "CSV")).End(slog.Int("rows", 42))140+ Start("open").Fail(errors.New("bad file"))141+ Set("cells", 1234)142+ for i := range 5 {143+ Frame(time.Duration(i+1)*time.Millisecond, 3*time.Millisecond)144+ }145+ frames.mu.Lock()146+ frames.since = time.Now().Add(-2 * window)147+ frames.mu.Unlock()148+ Frame(time.Millisecond, 0)149+ if len(c.requests()) != 0 {150+ t.Fatal("sent before a flush")151+ }152+ if err := Close(); err != nil {153+ t.Fatal(err)154+ }155+156+ reqs := c.requests()157+ paths := map[string]bool{}158+ for _, r := range reqs {159+ paths[r.path] = true160+ if r.header.Get("Content-Encoding") != "gzip" || r.header.Get("Content-Type") != "application/json" {161+ t.Errorf("%s headers %v", r.path, r.header)162+ }163+ if r.header.Get("Authorization") != "Bearer t" {164+ t.Errorf("%s: no configured header", r.path)165+ }166+ }167+ if !paths["/v1/logs"] || !paths["/v1/traces"] || !paths["/v1/metrics"] {168+ t.Fatalf("paths %v", paths)169+ }170+171+ logs, res := c.items("/v1/logs")172+ for k, want := range map[string]string{"service.name": "012", "service.version": "v-test", "os.type": "", "host.arch": "", "service.instance.id": ""} {173+ v, ok := res[k].(map[string]any)["stringValue"].(string)174+ if !ok || want != "" && v != want {175+ t.Errorf("resource %s = %v", k, res[k])176+ }177+ }178+ byBody := map[string]map[string]any{}179+ for _, l := range logs {180+ byBody[l["body"].(map[string]any)["stringValue"].(string)] = l181+ isInt64(t, "timeUnixNano", l["timeUnixNano"])182+ isInt64(t, "observedTimeUnixNano", l["observedTimeUnixNano"])183+ }184+ imp := byBody["import"]185+ if imp == nil || byBody["start"] == nil || byBody["open"] == nil || byBody["frames"] == nil || byBody["otlp"] == nil {186+ t.Fatalf("log bodies %v", slices.Collect(maps.Keys(byBody)))187+ }188+ if imp["severityNumber"] != 9.0 || imp["severityText"] != "INFO" || byBody["open"]["severityNumber"] != 13.0 {189+ t.Errorf("severity %v %v", imp["severityNumber"], byBody["open"]["severityNumber"])190+ }191+ ia := attrMap(t, imp["attributes"])192+ if ia["rows"].(map[string]any)["intValue"] != "42" || ia["format"].(map[string]any)["stringValue"] != "CSV" ||193+ ia["event"].(map[string]any)["stringValue"] != "import" {194+ t.Errorf("import attributes %v", ia)195+ }196+ if _, ok := ia["dur_ms"].(map[string]any)["doubleValue"].(float64); !ok {197+ t.Errorf("dur_ms %v", ia["dur_ms"])198+ }199+ if _, ok := ia["service"]; ok {200+ t.Error("service is a log attribute; it belongs in the resource")201+ }202+ if tr, _ := imp["traceId"].(string); !hexTrace.MatchString(tr) {203+ t.Errorf("import log traceId %v", imp["traceId"])204+ }205+ if _, ok := byBody["frames"]["traceId"]; ok {206+ t.Error("the frames summary is not a span, yet its log has a trace id")207+ }208+ oa := attrMap(t, byBody["otlp"]["attributes"])209+ if oa["logs.dropped"].(map[string]any)["intValue"] != "0" {210+ t.Errorf("otlp event %v", oa)211+ }212+213+ spans, _ := c.items("/v1/traces")214+ if len(spans) != 2 {215+ t.Fatalf("%d spans: %v", len(spans), spans)216+ }217+ for _, s := range spans {218+ if !hexTrace.MatchString(s["traceId"].(string)) || !hexSpan.MatchString(s["spanId"].(string)) || s["kind"] != 1.0 {219+ t.Errorf("span ids %v", s)220+ }221+ isInt64(t, "startTimeUnixNano", s["startTimeUnixNano"])222+ isInt64(t, "endTimeUnixNano", s["endTimeUnixNano"])223+ start, _ := strconv.ParseUint(s["startTimeUnixNano"].(string), 10, 64)224+ end, _ := strconv.ParseUint(s["endTimeUnixNano"].(string), 10, 64)225+ if start == 0 || start > end {226+ t.Errorf("span ends before it starts: %v", s)227+ }228+ }229+ if spans[0]["name"] != "import" || spans[0]["traceId"] != imp["traceId"] || spans[0]["spanId"] != imp["spanId"] {230+ t.Errorf("import span %v does not match its log %v", spans[0], imp)231+ }232+ if st := spans[1]["status"].(map[string]any); spans[1]["name"] != "open" || st["code"] != 2.0 || st["message"] != "bad file" {233+ t.Errorf("failed span %v", spans[1])234+ }235+236+ metrics, _ := c.items("/v1/metrics")237+ byName := map[string]map[string]any{}238+ for _, m := range metrics {239+ byName[m["name"].(string)] = m240+ }241+ sumr := byName["o12.frame.duration"]["summary"].(map[string]any)["dataPoints"].([]any)[0].(map[string]any)242+ isInt64(t, "summary count", sumr["count"])243+ if sumr["count"] != "6" || sumr["sum"] != 16.0 || len(list(sumr["quantileValues"])) != 3 {244+ t.Errorf("frame summary %v", sumr)245+ }246+ fr := byName["o12.frames"]["sum"].(map[string]any)247+ if fr["aggregationTemporality"] != 2.0 || fr["isMonotonic"] != true {248+ t.Errorf("frames sum %v", fr)249+ }250+ if p := list(fr["dataPoints"])[0].(map[string]any); p["asInt"] != "6" {251+ t.Errorf("frames %v", p)252+ }253+ if p := list(byName["o12.cells"]["gauge"].(map[string]any)["dataPoints"])[0].(map[string]any); p["asInt"] != "1234" {254+ t.Errorf("cells gauge %v", p)255+ }256+ if p := list(byName["o12.heap"]["gauge"].(map[string]any)["dataPoints"])[0].(map[string]any); p["asInt"] == "0" {257+ t.Errorf("heap gauge %v", p)258+ }259+ hs := byName["o12.operation.duration"]["histogram"].(map[string]any)260+ if len(list(hs["dataPoints"])) != 2 {261+ t.Fatalf("operation histogram %v", hs)262+ }263+ for _, p := range list(hs["dataPoints"]) {264+ p := p.(map[string]any)265+ isInt64(t, "histogram count", p["count"])266+ if len(list(p["bucketCounts"])) != len(list(p["explicitBounds"]))+1 {267+ t.Errorf("buckets %v", p)268+ }269+ for _, b := range list(p["bucketCounts"]) {270+ isInt64(t, "bucket count", b)271+ }272+ }273+ if byName["o12.telemetry.dropped"] == nil {274+ t.Error("no drop counter")275+ }276+}277+278+func TestOTLPOverflowDrops(t *testing.T) {279+ defer func(q, f int, e time.Duration) { queueLimit, batchSize, flushEvery = q, f, e }(queueLimit, batchSize, flushEvery)280+ queueLimit, batchSize, flushEvery = 3, 1000, time.Hour281+ c := newCollector(t)282+ otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL, NoTraces: true, NoMetrics: true})283+ for range 10 {284+ Event("recalc", time.Millisecond) // start + 2 fit285+ }286+ _, e := current()287+ _, dropped, _ := e.counts()288+ if dropped[sigLogs] != 8 {289+ t.Errorf("dropped %d logs, want 8", dropped[sigLogs])290+ }291+ Close()292+ logs, _ := c.items("/v1/logs")293+ if len(logs) != 3 {294+ t.Errorf("%d logs sent, want 3", len(logs))295+ }296+ for _, r := range c.requests() {297+ if r.path != "/v1/logs" {298+ t.Errorf("sent %s with only logs on", r.path)299+ }300+ }301+}302+303+func TestOTLPFlushesInTheBackground(t *testing.T) {304+ defer func(f int) { batchSize = f }(batchSize)305+ batchSize = 4306+ c := newCollector(t)307+ otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL, NoMetrics: true})308+ for range 4 {309+ Event("recalc", time.Millisecond)310+ }311+ deadline := time.Now().Add(5 * time.Second)312+ for len(c.requests()) == 0 {313+ if time.Now().After(deadline) {314+ t.Fatal("a full batch was not sent before Close")315+ }316+ time.Sleep(10 * time.Millisecond)317+ }318+}319+320+func TestOTLPCloseIsBounded(t *testing.T) {321+ c := newCollector(t)322+ c.delay = 2 * time.Second323+ otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL, Timeout: 100 * time.Millisecond})324+ begin := time.Now()325+ for range 100 {326+ Start("sort").End() // never waits on the network327+ }328+ if d := time.Since(begin); d > 100*time.Millisecond {329+ t.Errorf("100 spans took %v", d)330+ }331+ Close()332+ if d := time.Since(begin); d > 1500*time.Millisecond {333+ t.Errorf("Close with a stuck collector took %v", d)334+ }335+}336+337+func TestOTLPOffSendsNothing(t *testing.T) {338+ c := newCollector(t)339+ t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", c.srv.URL)340+ t.Setenv("OTEL_SDK_DISABLED", "true")341+ stop, err := Setup(ConfigFromEnv())342+ if err != nil || Enabled() {343+ t.Fatalf("OTEL_SDK_DISABLED: %v, enabled %v", err, Enabled())344+ }345+ Start("recalc").End()346+ Event("x", time.Second)347+ stop()348+349+ t.Setenv("OTEL_SDK_DISABLED", "")350+ t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "")351+ stop, err = Setup(ConfigFromEnv())352+ if err != nil || Enabled() {353+ t.Fatalf("no endpoint: %v, enabled %v", err, Enabled())354+ }355+ Start("recalc").End()356+ stop()357+358+ // After Close, nothing more goes out.359+ otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL})360+ Close()361+ n := len(c.requests())362+ Start("recalc").End()363+ Frame(time.Millisecond, 0)364+ time.Sleep(50 * time.Millisecond)365+ if len(c.requests()) != n {366+ t.Error("sent after Close")367+ }368+ for _, r := range c.requests()[:n] {369+ if r.path == "/v1/traces" {370+ t.Error("a span was sent though none was recorded while on")371+ }372+ }373+}374+375+func TestOTLPConfigFromEnv(t *testing.T) {376+ t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "localhost:4318/")377+ t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "https://traces.example/v1/t")378+ t.Setenv("OTEL_METRICS_EXPORTER", "none")379+ t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "Authorization=Bearer%20abc, x-team = 012 ,bad")380+ t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "deployment.environment=dev,service.name=other")381+ t.Setenv("OTEL_EXPORTER_OTLP_TIMEOUT", "500")382+ c := ConfigFromEnv().OTLP383+ u := c.urls()384+ want := urls{logs: "http://localhost:4318/v1/logs", traces: "https://traces.example/v1/t"}385+ if u != want {386+ t.Errorf("urls %+v, want %+v", u, want)387+ }388+ if c.Headers["Authorization"] != "Bearer abc" || c.Headers["x-team"] != "012" || len(c.Headers) != 2 {389+ t.Errorf("headers %v", c.Headers)390+ }391+ if c.Timeout != 500*time.Millisecond {392+ t.Errorf("timeout %v", c.Timeout)393+ }394+ e := newExporter(c, "")395+ defer e.shutdown()396+ b, _ := json.Marshal(e.resource)397+ var rm map[string]any398+ if err := json.Unmarshal(b, &rm); err != nil {399+ t.Fatal(err)400+ }401+ res := attrMap(t, rm["attributes"])402+ if res["service.name"].(map[string]any)["stringValue"] != "012" || res["deployment.environment"].(map[string]any)["stringValue"] != "dev" {403+ t.Errorf("resource %v", res)404+ }405+ c.Disabled = true406+ if c.urls().any() {407+ t.Error("disabled but still sending")408+ }409+}410+411+func TestOTLPAndFileTogether(t *testing.T) {412+ c := newCollector(t)413+ path := t.TempDir() + "/log.jsonl"414+ if _, err := Setup(Config{LogPath: path, OTLP: OTLPConfig{Endpoint: c.srv.URL}}); err != nil {415+ t.Fatal(err)416+ }417+ Event("recalc", time.Millisecond, slog.Int("cells", 3))418+ Close()419+ logs, _ := c.items("/v1/logs")420+ if len(logs) != 3 { // start, recalc, otlp421+ t.Errorf("%d OTLP logs", len(logs))422+ }423+ b, err := os.ReadFile(path)424+ if err != nil {425+ t.Fatal(err)426+ }427+ if n := len(regexp.MustCompile(`"service":"012"`).FindAll(b, -1)); n != 3 {428+ t.Errorf("%d file events: %s", n, b)429+ }430+}431+432+// The on-path cost with OTLP, queueing to a local endpoint.433+func BenchmarkSpanOTLP(b *testing.B) {434+ c := newCollector(b)435+ otlpSetup(b, OTLPConfig{Endpoint: c.srv.URL})436+ for b.Loop() {437+ s := Start("recalc", slog.Int("n", 1))438+ s.End(slog.Int("cells", 2))439+ }440+}@@ -0,0 +1,286 @@
1+package telemetry2+3+import (4+ "encoding/hex"5+ "encoding/json"6+ "fmt"7+ "log/slog"8+ "strconv"9+ "time"10+)11+12+// The OTLP/HTTP JSON encoding of logs, traces and metrics: the protobuf13+// messages of opentelemetry-proto under the protobuf JSON mapping, so14+// field names are lowerCamelCase, 64-bit integers are decimal strings,15+// enums are their numbers and trace and span ids are hex. Only the16+// fields 012 sends are declared.17+18+// i64 is a 64-bit integer, written as a JSON string.19+type i64 int6420+21+func (n i64) MarshalJSON() ([]byte, error) {22+ return strconv.AppendQuote(nil, strconv.FormatInt(int64(n), 10)), nil23+}24+25+// u64 is an unsigned 64-bit integer, written as a JSON string.26+type u64 uint6427+28+func (n u64) MarshalJSON() ([]byte, error) {29+ return strconv.AppendQuote(nil, strconv.FormatUint(uint64(n), 10)), nil30+}31+32+func nanos(t time.Time) u64 { return u64(t.UnixNano()) }33+34+type traceID [16]byte35+type spanID [8]byte36+37+func (id traceID) MarshalJSON() ([]byte, error) { return json.Marshal(hex.EncodeToString(id[:])) }38+func (id spanID) MarshalJSON() ([]byte, error) { return json.Marshal(hex.EncodeToString(id[:])) }39+40+func (id traceID) IsZero() bool { return id == traceID{} }41+func (id spanID) IsZero() bool { return id == spanID{} }42+43+// anyValue is an AnyValue holding one scalar.44+type anyValue struct {45+ kind byte // 's', 'i', 'd' or 'b'46+ s string47+ i int6448+ d float6449+ b bool50+}51+52+func str(s string) anyValue { return anyValue{kind: 's', s: s} }53+54+func (v anyValue) MarshalJSON() ([]byte, error) {55+ switch v.kind {56+ case 'i':57+ return fmt.Appendf(nil, `{"intValue":"%d"}`, v.i), nil58+ case 'd':59+ b, err := json.Marshal(v.d)60+ return append(append([]byte(`{"doubleValue":`), b...), '}'), err61+ case 'b':62+ return fmt.Appendf(nil, `{"boolValue":%t}`, v.b), nil63+ }64+ b, err := json.Marshal(v.s)65+ return append(append([]byte(`{"stringValue":`), b...), '}'), err66+}67+68+type keyValue struct {69+ Key string `json:"key"`70+ Value anyValue `json:"value"`71+}72+73+// appendAttr adds a to kvs, flattening groups into dotted keys.74+func appendAttr(kvs []keyValue, prefix string, a slog.Attr) []keyValue {75+ a.Value = a.Value.Resolve()76+ if a.Equal(slog.Attr{}) {77+ return kvs78+ }79+ if a.Value.Kind() == slog.KindGroup {80+ p := prefix81+ if a.Key != "" {82+ p += a.Key + "."83+ }84+ for _, g := range a.Value.Group() {85+ kvs = appendAttr(kvs, p, g)86+ }87+ return kvs88+ }89+ return append(kvs, keyValue{Key: prefix + a.Key, Value: valueOf(a.Value)})90+}91+92+func valueOf(v slog.Value) anyValue {93+ switch v.Kind() {94+ case slog.KindString:95+ return str(v.String())96+ case slog.KindInt64:97+ return anyValue{kind: 'i', i: v.Int64()}98+ case slog.KindUint64:99+ return anyValue{kind: 'i', i: int64(v.Uint64())}100+ case slog.KindFloat64:101+ return anyValue{kind: 'd', d: v.Float64()}102+ case slog.KindBool:103+ return anyValue{kind: 'b', b: v.Bool()}104+ case slog.KindDuration:105+ return anyValue{kind: 'd', d: ms(v.Duration())}106+ case slog.KindTime:107+ return str(v.Time().Format(time.RFC3339Nano))108+ }109+ return str(v.String())110+}111+112+func attrsOf(as []slog.Attr) []keyValue {113+ var kvs []keyValue114+ for _, a := range as {115+ kvs = appendAttr(kvs, "", a)116+ }117+ return kvs118+}119+120+type resource struct {121+ Attributes []keyValue `json:"attributes"`122+}123+124+type scope struct {125+ Name string `json:"name"`126+ Version string `json:"version,omitempty"`127+}128+129+// Logs.130+131+type logsRequest struct {132+ ResourceLogs []resourceLogs `json:"resourceLogs"`133+}134+135+type resourceLogs struct {136+ Resource resource `json:"resource"`137+ ScopeLogs []scopeLogs `json:"scopeLogs"`138+}139+140+type scopeLogs struct {141+ Scope scope `json:"scope"`142+ LogRecords []logRecord `json:"logRecords"`143+}144+145+type logRecord struct {146+ TimeUnixNano u64 `json:"timeUnixNano"`147+ ObservedTimeUnixNano u64 `json:"observedTimeUnixNano"`148+ SeverityNumber int `json:"severityNumber"`149+ SeverityText string `json:"severityText"`150+ Body anyValue `json:"body"`151+ Attributes []keyValue `json:"attributes,omitempty"`152+ TraceID traceID `json:"traceId,omitzero"`153+ SpanID spanID `json:"spanId,omitzero"`154+}155+156+// severity maps a slog level onto an OTLP SeverityNumber: slog's levels157+// are spaced so that Debug, Info, Warn and Error land on DEBUG (5), INFO158+// (9), WARN (13) and ERROR (17).159+func severity(l slog.Level) int { return min(max(int(l)+9, 1), 24) }160+161+// Traces.162+163+type tracesRequest struct {164+ ResourceSpans []resourceSpans `json:"resourceSpans"`165+}166+167+type resourceSpans struct {168+ Resource resource `json:"resource"`169+ ScopeSpans []scopeSpans `json:"scopeSpans"`170+}171+172+type scopeSpans struct {173+ Scope scope `json:"scope"`174+ Spans []spanData `json:"spans"`175+}176+177+const (178+ spanKindInternal = 1179+ statusOK = 1180+ statusError = 2181+)182+183+type spanData struct {184+ TraceID traceID `json:"traceId"`185+ SpanID spanID `json:"spanId"`186+ Name string `json:"name"`187+ Kind int `json:"kind"`188+ StartTimeUnixNano u64 `json:"startTimeUnixNano"`189+ EndTimeUnixNano u64 `json:"endTimeUnixNano"`190+ Attributes []keyValue `json:"attributes,omitempty"`191+ Status status `json:"status"`192+}193+194+type status struct {195+ Code int `json:"code,omitempty"`196+ Message string `json:"message,omitempty"`197+}198+199+// Metrics.200+201+type metricsRequest struct {202+ ResourceMetrics []resourceMetrics `json:"resourceMetrics"`203+}204+205+type resourceMetrics struct {206+ Resource resource `json:"resource"`207+ ScopeMetrics []scopeMetrics `json:"scopeMetrics"`208+}209+210+type scopeMetrics struct {211+ Scope scope `json:"scope"`212+ Metrics []metric `json:"metrics"`213+}214+215+const temporalityCumulative = 2216+217+type metric struct {218+ Name string `json:"name"`219+ Description string `json:"description,omitempty"`220+ Unit string `json:"unit,omitempty"`221+ Gauge *gauge `json:"gauge,omitempty"`222+ Sum *sum `json:"sum,omitempty"`223+ Histogram *histogram `json:"histogram,omitempty"`224+ Summary *summary `json:"summary,omitempty"`225+}226+227+type gauge struct {228+ DataPoints []numberPoint `json:"dataPoints"`229+}230+231+type sum struct {232+ DataPoints []numberPoint `json:"dataPoints"`233+ AggregationTemporality int `json:"aggregationTemporality"`234+ IsMonotonic bool `json:"isMonotonic"`235+}236+237+type numberPoint struct {238+ Attributes []keyValue `json:"attributes,omitempty"`239+ StartTimeUnixNano u64 `json:"startTimeUnixNano,omitempty"`240+ TimeUnixNano u64 `json:"timeUnixNano"`241+ AsInt *i64 `json:"asInt,omitempty"`242+ AsDouble *float64 `json:"asDouble,omitempty"`243+}244+245+func intPoint(start, t time.Time, v int64, attrs ...keyValue) numberPoint {246+ n := i64(v)247+ p := numberPoint{Attributes: attrs, TimeUnixNano: nanos(t), AsInt: &n}248+ if !start.IsZero() {249+ p.StartTimeUnixNano = nanos(start)250+ }251+ return p252+}253+254+type histogram struct {255+ DataPoints []histogramPoint `json:"dataPoints"`256+ AggregationTemporality int `json:"aggregationTemporality"`257+}258+259+type histogramPoint struct {260+ Attributes []keyValue `json:"attributes,omitempty"`261+ StartTimeUnixNano u64 `json:"startTimeUnixNano"`262+ TimeUnixNano u64 `json:"timeUnixNano"`263+ Count u64 `json:"count"`264+ Sum float64 `json:"sum"`265+ BucketCounts []u64 `json:"bucketCounts"`266+ ExplicitBounds []float64 `json:"explicitBounds"`267+ Min float64 `json:"min"`268+ Max float64 `json:"max"`269+}270+271+type summary struct {272+ DataPoints []summaryPoint `json:"dataPoints"`273+}274+275+type summaryPoint struct {276+ StartTimeUnixNano u64 `json:"startTimeUnixNano"`277+ TimeUnixNano u64 `json:"timeUnixNano"`278+ Count u64 `json:"count"`279+ Sum float64 `json:"sum"`280+ QuantileValues []quantileValue `json:"quantileValues"`281+}282+283+type quantileValue struct {284+ Quantile float64 `json:"quantile"`285+ Value float64 `json:"value"`286+}@@ -1,16 +1,17 @@
11 // 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).2+// events written with log/slog. It is off unless a destination is given:3+// a JSON log file (012 --log path, or O12_LOG), one line per event that4+// DuckDB or an OpenTelemetry Collector's filelog receiver can read,5+// and/or an OTLP/HTTP endpoint (012 --otlp URL, or6+// OTEL_EXPORTER_OTLP_ENDPOINT) that gets the events as logs, spans as7+// traces and frame summaries as metrics. Stdout belongs to the terminal8+// UI. See docs/observability.md.89 //910 // Events carry sizes, counts and durations only: never cell contents,1011 // file contents or secrets.1112 //12-// The API is small on purpose, so call sites stay few and a different13-// backend (OTLP) can sit behind it without touching them.13+// The API is small on purpose, so call sites stay few and backends can14+// change behind it without touching them.1415 package telemetry1516 1617 import (@@ -31,71 +32,117 @@ var (
3132 mu sync.Mutex3233 logger = slog.New(slog.DiscardHandler)3334 closer io.Closer35+ exp *exporter // nil unless OTLP is on3436 )3537 3638 // Config says where events go.3739 type Config struct {38- // LogPath is the JSON log file, appended to. Empty turns telemetry39- // off.40+ // LogPath is the JSON log file, appended to. Empty (and no OTLP)41+ // turns telemetry off.4042 LogPath string4143 // Level is the least severe level logged; Info by default. Debug adds4244 // an event for every frame and every command.4345 Level slog.Level44- // Version is 012's version or commit, added to the start event.46+ // Version is 012's version or commit, added to the start event and47+ // sent as the OTLP service.version.4548 Version string49+ // OTLP sends events to an OpenTelemetry endpoint too, or instead.50+ OTLP OTLPConfig4651 }4752 48-// ConfigFromEnv reads O12_LOG and O12_LOG_LEVEL (debug, info, warn).53+// ConfigFromEnv reads O12_LOG, O12_LOG_LEVEL (debug, info, warn) and the54+// standard OTEL_* variables described at OTLPConfig.4955 func ConfigFromEnv() Config {50- c := Config{LogPath: os.Getenv("O12_LOG")}56+ c := Config{LogPath: os.Getenv("O12_LOG"), OTLP: otlpFromEnv()}5157 c.Level.UnmarshalText([]byte(os.Getenv("O12_LOG_LEVEL")))5258 return c5359 }5460 5561 // Setup starts recording as c says. The returned function flushes and56-// closes the log; call it on exit.62+// closes the log and sends what OTLP has queued; call it on exit.5763 func Setup(c Config) (func() error, error) {58- if c.LogPath == "" {64+ u := c.OTLP.urls()65+ if c.LogPath == "" && !u.any() {5966 return func() error { return nil }, nil6067 }61- f, err := os.OpenFile(c.LogPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)62- if err != nil {63- return nil, err68+ var w io.Writer69+ if c.LogPath != "" {70+ f, err := os.OpenFile(c.LogPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)71+ if err != nil {72+ return nil, err73+ }74+ w = f6475 }65- start(f, c)76+ var e *exporter77+ if u.any() {78+ e = newExporter(c.OTLP, c.Version)79+ }80+ start(w, e, c)6681 Logger().Info("start", "event", "start", "version", c.Version, "go", runtime.Version(), "os", runtime.GOOS, "arch", runtime.GOARCH,67- "cpus", runtime.NumCPU(), "pid", os.Getpid())82+ "cpus", runtime.NumCPU(), "pid", os.Getpid(), "file", w != nil, "otlp", e != nil)6883 return Close, nil6984 }7085 71-// start sends events to w; tests use it with a buffer.72-func start(w io.Writer, c Config) {86+// start sends events to w and e, either of which may be nil; tests use87+// it with a buffer.88+func start(w io.Writer, e *exporter, c Config) {7389 mu.Lock()7490 defer mu.Unlock()75- h := slog.NewJSONHandler(w, &slog.HandlerOptions{Level: c.Level})76- logger = slog.New(h).With("service", "012")91+ var hs []slog.Handler92+ if w != nil {93+ // OTLP carries the service in its resource instead.94+ hs = append(hs, slog.NewJSONHandler(w, &slog.HandlerOptions{Level: c.Level}).95+ WithAttrs([]slog.Attr{slog.String("service", "012")}))96+ }97+ if e != nil && e.urls.logs != "" {98+ hs = append(hs, &otlpHandler{e: e, level: c.Level})99+ }100+ switch len(hs) {101+ case 0:102+ logger = slog.New(slog.DiscardHandler)103+ case 1:104+ logger = slog.New(hs[0])105+ default:106+ logger = slog.New(slog.NewMultiHandler(hs...))107+ }77108 slog.SetDefault(logger)109+ closer = nil78110 if cl, ok := w.(io.Closer); ok {79111 closer = cl80112 }113+ exp = e81114 frames.reset(time.Now())82115 on.Store(true)83116 }84117 85-// Close stops recording and closes the log file.118+// Close stops recording, sends what OTLP has queued (waiting at most119+// about one request timeout) and closes the log file.86120 func Close() error {87121 mu.Lock()88- defer mu.Unlock()89122 if !on.Load() {123+ mu.Unlock()90124 return nil91125 }126+ e, l, cl := exp, logger, closer127+ if e != nil {128+ // How OTLP fared so far, for the file log (and the collector).129+ sent, dropped, failed := e.counts()130+ attrs := []slog.Attr{slog.String("event", "otlp")}131+ for i, name := range signalNames {132+ attrs = append(attrs, slog.Group(name, "sent", sent[i], "dropped", dropped[i], "failed", failed[i]))133+ }134+ l.LogAttrs(context.Background(), slog.LevelInfo, "otlp", attrs...)135+ }92136 on.Store(false)93137 logger = slog.New(slog.DiscardHandler)94138 slog.SetDefault(logger)95- if closer != nil {96- err := closer.Close()97- closer = nil98- return err139+ exp, closer = nil, nil140+ mu.Unlock()141+ if e != nil {142+ e.shutdown()143+ }144+ if cl != nil {145+ return cl.Close()99146 }100147 return nil101148 }@@ -111,7 +158,9 @@ func Logger() *slog.Logger {
111158 }112159 113160 // Span times one operation. The zero Span, returned while off, does114-// nothing.161+// nothing. With OTLP on, each span is sent as a root span of its own162+// trace (spans don't nest: call sites pass no context), and the log163+// record of the same event carries its trace and span ids.115164 type Span struct {116165 name string117166 start time.Time@@ -139,7 +188,7 @@ func (s Span) End(attrs ...slog.Attr) {
139188 if s.name == "" || !on.Load() {140189 return141190 }142- Event(s.name, time.Since(s.start), append(s.attrs[:s.n:s.n], attrs...)...)191+ emit(slog.LevelInfo, s.name, time.Since(s.start), append(s.attrs[:s.n:s.n], attrs...), true)143192 }144193 145194 // Fail ends the span with an error, logged at Warn with its message.@@ -152,29 +201,46 @@ func (s Span) Fail(err error, attrs ...slog.Attr) {
152201 return153202 }154203 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)204+ emit(slog.LevelWarn, s.name, time.Since(s.start), attrs, true)156205 }157206 158-// Event logs a finished operation that took d.207+// Event logs a finished operation that took d (a span, for OTLP).159208 func Event(name string, d time.Duration, attrs ...slog.Attr) {160209 if !on.Load() {161210 return162211 }163- log(slog.LevelInfo, name, d, attrs)212+ emit(slog.LevelInfo, name, d, attrs, true)164213 }165214 166215 // Debug logs a finished operation at Debug level, for events too167-// frequent to keep by default.216+// frequent to keep by default (and to send as spans).168217 func Debug(name string, d time.Duration, attrs ...slog.Attr) {169218 if !on.Load() {170219 return171220 }172- log(slog.LevelDebug, name, d, attrs)221+ emit(slog.LevelDebug, name, d, attrs, false)222+}223+224+// current is the logger and the OTLP exporter (nil unless OTLP is on).225+func current() (*slog.Logger, *exporter) {226+ mu.Lock()227+ defer mu.Unlock()228+ return logger, exp173229 }174230 175-func log(level slog.Level, name string, d time.Duration, attrs []slog.Attr) {176- l := Logger()231+// emit logs an event that took d and ended now. With span set and OTLP232+// on, it is also a span and a sample of the operation histogram, and its233+// log record carries the span's ids. attrs is not kept.234+func emit(level slog.Level, name string, d time.Duration, attrs []slog.Attr, span bool) {235+ l, e := current()177236 ctx := context.Background()237+ if span && e != nil {238+ end := time.Now()239+ t, s := e.span(name, end.Add(-d), end, level >= slog.LevelWarn, attrs)240+ if !t.IsZero() {241+ ctx = context.WithValue(ctx, spanKey{}, spanIDs{t, s})242+ }243+ }178244 if !l.Enabled(ctx, level) {179245 return180246 }@@ -14,7 +14,7 @@ import (
1414 // record turns telemetry on into a buffer for the test.1515 func record(t testing.TB, level slog.Level) *bytes.Buffer {1616 var buf bytes.Buffer17- start(&buf, Config{Level: level})17+ start(&buf, nil, Config{Level: level})1818 t.Cleanup(func() { Close() })1919 return &buf2020 }@@ -122,6 +122,20 @@ func TestPercentile(t *testing.T) {
122122 }123123 }124124 125+// Call sites must not allocate while telemetry is off, OTLP compiled in126+// or not.127+func TestOffAllocatesNothing(t *testing.T) {128+ n := testing.AllocsPerRun(100, func() {129+ s := Start("recalc", slog.Int("n", 1))130+ s.End(slog.Int("cells", 2))131+ Event("recalc", time.Millisecond, slog.Int("cells", 2))132+ Frame(time.Millisecond, 0)133+ })134+ if n != 0 {135+ t.Errorf("%v allocations per call while off", n)136+ }137+}138+125139 // The cost at a call site while telemetry is off: what every keystroke126140 // pays for having it compiled in.127141 func BenchmarkSpanOff(b *testing.B) {@@ -139,7 +153,7 @@ func BenchmarkFrameOff(b *testing.B) {
139153 140154 // The cost while on, writing JSON (to io.Discard).141155 func BenchmarkSpanOn(b *testing.B) {142- start(io.Discard, Config{})156+ start(io.Discard, nil, Config{})143157 defer Close()144158 for b.Loop() {145159 s := Start("recalc", slog.Int("n", 1))@@ -148,7 +162,7 @@ func BenchmarkSpanOn(b *testing.B) {
148162 }149163 150164 func BenchmarkFrameOn(b *testing.B) {151- start(io.Discard, Config{})165+ start(io.Discard, nil, Config{})152166 defer Close()153167 for b.Loop() {154168 Frame(time.Millisecond, 0)