1package telemetry23import (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 "runtime"16 "slices"17 "sync"18 "time"19)2021// OTLP sends events to an OpenTelemetry Collector (or any OTLP/HTTP22// endpoint) with the JSON encoding, using only the standard library:23// slog events as logs, spans as traces, and the frame summaries and24// operation durations as metrics. A background goroutine batches them25// from bounded queues; when a queue is full the newest items are dropped26// and counted, so the UI never waits on the network.27//28// This file is the exporter's queues and sender; otlpconfig.go says29// where to send, otlplog.go is the slog handler queuing log records,30// otlpmetrics.go builds the metrics and otlpjson.go is the wire format.3132const (33 sigLogs = iota34 sigTraces35 sigMetrics36)3738var signalNames = [3]string{"logs", "traces", "metrics"}3940// Tuning; variables so tests can shrink them.41var (42 // queueLimit bounds each signal's queue (metric windows count as43 // one item each).44 queueLimit = 204845 // batchSize wakes the sender before the next tick.46 batchSize = 51247 // flushEvery is how often queued items are sent.48 flushEvery = 2 * time.Second49)5051type exporter struct {52 urls urls53 headers map[string]string54 gzip bool55 timeout time.Duration56 client *http.Client57 resource resource58 scope scope59 started time.Time6061 mu sync.Mutex62 logs []logRecord63 spans []spanData64 windows []frameWindow65 ops map[string]*durationHist66 opsNew bool // ops changed since they were last sent67 frames int6468 sent [3]int64 // items accepted by the endpoint69 dropped [3]int64 // items dropped because the queue was full70 failed [3]int64 // items in requests that failed71 down bool // the last request failed: the endpoint looks unreachable7273 wake chan struct{}74 done chan struct{}75 stopped chan struct{}76}7778func newExporter(c OTLPConfig, version string) *exporter {79 timeout := c.Timeout80 if timeout <= 0 {81 timeout = defaultTimeout82 }83 var id [16]byte84 randomBytes(id[:])85 res := map[string]string{}86 maps.Copy(res, c.Resource)87 maps.Copy(res, map[string]string{88 "service.name": "012",89 "service.instance.id": hex.EncodeToString(id[:]),90 "os.type": runtime.GOOS,91 "host.arch": runtime.GOARCH,92 "process.runtime.name": "go",93 "process.runtime.version": runtime.Version(),94 })95 if version != "" {96 res["service.version"] = version97 }98 var attrs []keyValue99 for _, k := range slices.Sorted(maps.Keys(res)) {100 attrs = append(attrs, keyValue{Key: k, Value: str(res[k])})101 }102 e := &exporter{103 urls: c.urls(),104 headers: c.Headers,105 gzip: !c.NoCompression,106 timeout: timeout,107 client: &http.Client{Timeout: timeout},108 resource: resource{Attributes: attrs},109 scope: scope{Name: "src.rest/cat/012/internal/telemetry", Version: version},110 started: time.Now(),111 ops: map[string]*durationHist{},112 wake: make(chan struct{}, 1),113 done: make(chan struct{}),114 stopped: make(chan struct{}),115 }116 go e.run()117 return e118}119120func randomBytes(b []byte) {121 for i := 0; i < len(b); i += 8 {122 var w [8]byte123 v := rand.Uint64()124 for j := range w {125 w[j] = byte(v >> (8 * j))126 }127 copy(b[i:], w[:])128 }129}130131// span records a finished operation: a span with ids when traces are on,132// and its duration in the operation histogram when metrics are. attrs is133// not kept.134func (e *exporter) span(name string, start, end time.Time, failed bool, attrs []slog.Attr, ids spanIDs) {135 if e.urls.metrics != "" {136 e.mu.Lock()137 h := e.ops[name]138 if h == nil {139 h = &durationHist{}140 e.ops[name] = h141 }142 h.add(ms(end.Sub(start)))143 e.opsNew = true144 e.mu.Unlock()145 }146 if e.urls.traces == "" {147 return148 }149 sp := spanData{150 TraceID: ids.trace, SpanID: ids.span, ParentSpanID: ids.parent, Name: name, Kind: spanKindInternal,151 StartTimeUnixNano: nanos(start), EndTimeUnixNano: nanos(end),152 Attributes: attrsOf(attrs),153 Status: status{Code: statusOK},154 }155 if failed {156 sp.Status = status{Code: statusError}157 for _, a := range attrs {158 if a.Key == "error" {159 sp.Status.Message = a.Value.String()160 }161 }162 }163 e.mu.Lock()164 ok := len(e.spans) < queueLimit165 if ok {166 e.spans = append(e.spans, sp)167 } else {168 e.dropped[sigTraces]++169 }170 n := len(e.spans)171 e.mu.Unlock()172 e.nudge(n)173}174175func (e *exporter) log(r logRecord) {176 e.mu.Lock()177 ok := len(e.logs) < queueLimit178 if ok {179 e.logs = append(e.logs, r)180 } else {181 e.dropped[sigLogs]++182 }183 n := len(e.logs)184 e.mu.Unlock()185 e.nudge(n)186}187188func (e *exporter) window(w frameWindow) {189 e.mu.Lock()190 defer e.mu.Unlock()191 e.frames += int64(w.render.count)192 if len(e.windows) >= queueLimit {193 e.dropped[sigMetrics]++194 return195 }196 e.windows = append(e.windows, w)197}198199// nudge wakes the sender once a queue holds a batch.200func (e *exporter) nudge(n int) {201 if n < batchSize {202 return203 }204 select {205 case e.wake <- struct{}{}:206 default:207 }208}209210func (e *exporter) run() {211 defer close(e.stopped)212 t := time.NewTicker(flushEvery)213 defer t.Stop()214 for {215 select {216 case <-t.C:217 case <-e.wake:218 case <-e.done:219 // Quitting waits at most finalTimeout for the last flush,220 // and not at all when the endpoint already failed: an221 // inherited OTEL_EXPORTER_OTLP_ENDPOINT may name a collector222 // that's gone, and quitting shouldn't wait on it.223 e.mu.Lock()224 down := e.down225 e.mu.Unlock()226 if !down {227 ctx, cancel := context.WithTimeout(context.Background(), e.finalTimeout())228 e.flush(ctx, true)229 cancel()230 }231 return232 }233 e.flush(context.Background(), false)234 }235}236237// shutdown sends what is queued and stops the sender, waiting at most a238// little over one request timeout.239func (e *exporter) shutdown() {240 close(e.done)241 select {242 case <-e.stopped:243 case <-time.After(e.finalTimeout() + 250*time.Millisecond):244 }245}246247// finalTimeout bounds the flush when 012 quits.248func (e *exporter) finalTimeout() time.Duration { return min(e.timeout, finalFlush) }249250// finalFlush is the most quitting waits for the last flush.251const finalFlush = time.Second252253func (e *exporter) flush(ctx context.Context, final bool) {254 e.mu.Lock()255 logs, spans, windows := e.logs, e.spans, e.windows256 e.logs, e.spans, e.windows = nil, nil, nil257 var ms []metric258 if len(windows) > 0 || e.opsNew || final {259 ms = e.metricsLocked(windows, time.Now())260 e.opsNew = false261 }262 e.mu.Unlock()263264 if len(logs) > 0 {265 e.send(ctx, sigLogs, len(logs), logsRequest{ResourceLogs: []resourceLogs{{266 Resource: e.resource, ScopeLogs: []scopeLogs{{Scope: e.scope, LogRecords: logs}},267 }}})268 }269 if len(spans) > 0 {270 e.send(ctx, sigTraces, len(spans), tracesRequest{ResourceSpans: []resourceSpans{{271 Resource: e.resource, ScopeSpans: []scopeSpans{{Scope: e.scope, Spans: spans}},272 }}})273 }274 if len(ms) > 0 && e.urls.metrics != "" {275 e.send(ctx, sigMetrics, max(len(windows), 1), metricsRequest{ResourceMetrics: []resourceMetrics{{276 Resource: e.resource, ScopeMetrics: []scopeMetrics{{Scope: e.scope, Metrics: ms}},277 }}})278 }279}280281// send posts one request; failures are counted, never retried.282func (e *exporter) send(ctx context.Context, sig, items int, body any) {283 url := [3]string{e.urls.logs, e.urls.traces, e.urls.metrics}[sig]284 err := e.post(ctx, url, body)285 e.mu.Lock()286 e.down = err != nil287 if err != nil {288 e.failed[sig] += int64(items)289 } else {290 e.sent[sig] += int64(items)291 }292 e.mu.Unlock()293}294295func (e *exporter) post(ctx context.Context, url string, body any) error {296 var buf bytes.Buffer297 var w io.Writer = &buf298 var zw *gzip.Writer299 if e.gzip {300 zw = gzip.NewWriter(&buf)301 w = zw302 }303 if err := json.NewEncoder(w).Encode(body); err != nil {304 return err305 }306 if zw != nil {307 if err := zw.Close(); err != nil {308 return err309 }310 }311 req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, &buf)312 if err != nil {313 return err314 }315 req.Header.Set("Content-Type", "application/json")316 if e.gzip {317 req.Header.Set("Content-Encoding", "gzip")318 }319 for k, v := range e.headers {320 req.Header.Set(k, v)321 }322 resp, err := e.client.Do(req)323 if err != nil {324 return err325 }326 defer resp.Body.Close()327 io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10))328 if resp.StatusCode/100 != 2 {329 return fmt.Errorf("OTLP %s: %s", url, resp.Status)330 }331 return nil332}333334// counts reports items sent, dropped and failed per signal.335func (e *exporter) counts() (sent, dropped, failed [3]int64) {336 e.mu.Lock()337 defer e.mu.Unlock()338 return e.sent, e.dropped, e.failed339}