1package telemetry23import (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)2021// collector is a fake OTLP/HTTP endpoint that decodes what it gets.22type collector struct {23 t testing.TB24 srv *httptest.Server25 delay time.Duration2627 mu sync.Mutex28 reqs []request29}3031type request struct {32 path string33 header http.Header34 body map[string]any35}3637func 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}6465func (c *collector) requests() []request {66 c.mu.Lock()67 defer c.mu.Unlock()68 return slices.Clone(c.reqs)69}7071// items gathers the records of one signal from every request: log72// records, spans or metrics, with the resource of the last request.73func (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}9697func list(v any) []any { l, _ := v.([]any); return l }9899// attrMap turns OTLP attributes into key -> AnyValue, checking each100// value holds exactly one field.101func 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}113114var (115 hexTrace = regexp.MustCompile(`^[0-9a-f]{32}$`)116 hexSpan = regexp.MustCompile(`^[0-9a-f]{16}$`)117 decimal = regexp.MustCompile(`^[0-9]+$`)118)119120// isInt64 checks v is an int64 as the JSON mapping writes it: a string.121func 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}127128func 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}135136func 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 }155156 checkRequests(t, c)157 imp := checkLogs(t, c)158 checkSpans(t, c, imp)159 checkMetrics(t, c)160}161162// checkRequests checks every request is gzipped JSON with the configured163// headers, and that all three signals were sent.164func checkRequests(t *testing.T, c *collector) {165 t.Helper()166 reqs := c.requests()167 paths := map[string]bool{}168 for _, r := range reqs {169 paths[r.path] = true170 if r.header.Get("Content-Encoding") != "gzip" || r.header.Get("Content-Type") != "application/json" {171 t.Errorf("%s headers %v", r.path, r.header)172 }173 if r.header.Get("Authorization") != "Bearer t" {174 t.Errorf("%s: no configured header", r.path)175 }176 }177 if !paths["/v1/logs"] || !paths["/v1/traces"] || !paths["/v1/metrics"] {178 t.Fatalf("paths %v", paths)179 }180181}182183// checkLogs checks the resource and the log records, returning the184// import's record for checkSpans.185func checkLogs(t *testing.T, c *collector) map[string]any {186 t.Helper()187 logs, res := c.items("/v1/logs")188 for k, want := range map[string]string{"service.name": "012", "service.version": "v-test", "os.type": "", "host.arch": "", "service.instance.id": ""} {189 v, ok := res[k].(map[string]any)["stringValue"].(string)190 if !ok || want != "" && v != want {191 t.Errorf("resource %s = %v", k, res[k])192 }193 }194 byBody := map[string]map[string]any{}195 for _, l := range logs {196 byBody[l["body"].(map[string]any)["stringValue"].(string)] = l197 isInt64(t, "timeUnixNano", l["timeUnixNano"])198 isInt64(t, "observedTimeUnixNano", l["observedTimeUnixNano"])199 }200 imp := byBody["import"]201 if imp == nil || byBody["start"] == nil || byBody["open"] == nil || byBody["frames"] == nil || byBody["otlp"] == nil {202 t.Fatalf("log bodies %v", slices.Collect(maps.Keys(byBody)))203 }204 if imp["severityNumber"] != 9.0 || imp["severityText"] != "INFO" || byBody["open"]["severityNumber"] != 13.0 {205 t.Errorf("severity %v %v", imp["severityNumber"], byBody["open"]["severityNumber"])206 }207 ia := attrMap(t, imp["attributes"])208 if ia["rows"].(map[string]any)["intValue"] != "42" || ia["format"].(map[string]any)["stringValue"] != "CSV" ||209 ia["event"].(map[string]any)["stringValue"] != "import" {210 t.Errorf("import attributes %v", ia)211 }212 if _, ok := ia["dur_ms"].(map[string]any)["doubleValue"].(float64); !ok {213 t.Errorf("dur_ms %v", ia["dur_ms"])214 }215 if _, ok := ia["service"]; ok {216 t.Error("service is a log attribute; it belongs in the resource")217 }218 if tr, _ := imp["traceId"].(string); !hexTrace.MatchString(tr) {219 t.Errorf("import log traceId %v", imp["traceId"])220 }221 if _, ok := byBody["frames"]["traceId"]; ok {222 t.Error("the frames summary is not a span, yet its log has a trace id")223 }224 oa := attrMap(t, byBody["otlp"]["attributes"])225 if oa["logs.dropped"].(map[string]any)["intValue"] != "0" {226 t.Errorf("otlp event %v", oa)227 }228229 return imp230}231232// checkSpans checks the spans, and that the import's matches its log.233func checkSpans(t *testing.T, c *collector, imp map[string]any) {234 t.Helper()235 spans, _ := c.items("/v1/traces")236 if len(spans) != 2 {237 t.Fatalf("%d spans: %v", len(spans), spans)238 }239 for _, s := range spans {240 if !hexTrace.MatchString(s["traceId"].(string)) || !hexSpan.MatchString(s["spanId"].(string)) || s["kind"] != 1.0 {241 t.Errorf("span ids %v", s)242 }243 isInt64(t, "startTimeUnixNano", s["startTimeUnixNano"])244 isInt64(t, "endTimeUnixNano", s["endTimeUnixNano"])245 start, _ := strconv.ParseUint(s["startTimeUnixNano"].(string), 10, 64)246 end, _ := strconv.ParseUint(s["endTimeUnixNano"].(string), 10, 64)247 if start == 0 || start > end {248 t.Errorf("span ends before it starts: %v", s)249 }250 }251 if spans[0]["name"] != "import" || spans[0]["traceId"] != imp["traceId"] || spans[0]["spanId"] != imp["spanId"] {252 t.Errorf("import span %v does not match its log %v", spans[0], imp)253 }254 if st := spans[1]["status"].(map[string]any); spans[1]["name"] != "open" || st["code"] != 2.0 || st["message"] != "bad file" {255 t.Errorf("failed span %v", spans[1])256 }257258}259260// checkMetrics checks the frame summary, counters, gauges and the261// operation histogram.262func checkMetrics(t *testing.T, c *collector) {263 t.Helper()264 metrics, _ := c.items("/v1/metrics")265 byName := map[string]map[string]any{}266 for _, m := range metrics {267 byName[m["name"].(string)] = m268 }269 sumr := byName["o12.frame.duration"]["summary"].(map[string]any)["dataPoints"].([]any)[0].(map[string]any)270 isInt64(t, "summary count", sumr["count"])271 if sumr["count"] != "6" || sumr["sum"] != 16.0 || len(list(sumr["quantileValues"])) != 3 {272 t.Errorf("frame summary %v", sumr)273 }274 fr := byName["o12.frames"]["sum"].(map[string]any)275 if fr["aggregationTemporality"] != 2.0 || fr["isMonotonic"] != true {276 t.Errorf("frames sum %v", fr)277 }278 if p := list(fr["dataPoints"])[0].(map[string]any); p["asInt"] != "6" {279 t.Errorf("frames %v", p)280 }281 if p := list(byName["o12.cells"]["gauge"].(map[string]any)["dataPoints"])[0].(map[string]any); p["asInt"] != "1234" {282 t.Errorf("cells gauge %v", p)283 }284 if p := list(byName["o12.heap"]["gauge"].(map[string]any)["dataPoints"])[0].(map[string]any); p["asInt"] == "0" {285 t.Errorf("heap gauge %v", p)286 }287 hs := byName["o12.operation.duration"]["histogram"].(map[string]any)288 if len(list(hs["dataPoints"])) != 2 {289 t.Fatalf("operation histogram %v", hs)290 }291 for _, p := range list(hs["dataPoints"]) {292 p := p.(map[string]any)293 isInt64(t, "histogram count", p["count"])294 if len(list(p["bucketCounts"])) != len(list(p["explicitBounds"]))+1 {295 t.Errorf("buckets %v", p)296 }297 for _, b := range list(p["bucketCounts"]) {298 isInt64(t, "bucket count", b)299 }300 }301 if byName["o12.telemetry.dropped"] == nil {302 t.Error("no drop counter")303 }304}305306func TestOTLPOverflowDrops(t *testing.T) {307 defer func(q, f int, e time.Duration) { queueLimit, batchSize, flushEvery = q, f, e }(queueLimit, batchSize, flushEvery)308 queueLimit, batchSize, flushEvery = 3, 1000, time.Hour309 c := newCollector(t)310 otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL, NoTraces: true, NoMetrics: true})311 for range 10 {312 Event("recalc", time.Millisecond) // start + 2 fit313 }314 _, e := current()315 _, dropped, _ := e.counts()316 if dropped[sigLogs] != 8 {317 t.Errorf("dropped %d logs, want 8", dropped[sigLogs])318 }319 Close()320 logs, _ := c.items("/v1/logs")321 if len(logs) != 3 {322 t.Errorf("%d logs sent, want 3", len(logs))323 }324 for _, r := range c.requests() {325 if r.path != "/v1/logs" {326 t.Errorf("sent %s with only logs on", r.path)327 }328 }329}330331func TestOTLPFlushesInTheBackground(t *testing.T) {332 defer func(f int) { batchSize = f }(batchSize)333 batchSize = 4334 c := newCollector(t)335 otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL, NoMetrics: true})336 for range 4 {337 Event("recalc", time.Millisecond)338 }339 deadline := time.Now().Add(5 * time.Second)340 for len(c.requests()) == 0 {341 if time.Now().After(deadline) {342 t.Fatal("a full batch was not sent before Close")343 }344 time.Sleep(10 * time.Millisecond)345 }346}347348func TestOTLPOffSendsNothing(t *testing.T) {349 c := newCollector(t)350 t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", c.srv.URL)351 t.Setenv("OTEL_SDK_DISABLED", "true")352 stop, err := Setup(ConfigFromEnv())353 if err != nil || Enabled() {354 t.Fatalf("OTEL_SDK_DISABLED: %v, enabled %v", err, Enabled())355 }356 Start("recalc").End()357 Event("x", time.Second)358 stop()359360 t.Setenv("OTEL_SDK_DISABLED", "")361 t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "")362 stop, err = Setup(ConfigFromEnv())363 if err != nil || Enabled() {364 t.Fatalf("no endpoint: %v, enabled %v", err, Enabled())365 }366 Start("recalc").End()367 stop()368369 // After Close, nothing more goes out.370 otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL})371 Close()372 n := len(c.requests())373 Start("recalc").End()374 Frame(time.Millisecond, 0)375 time.Sleep(50 * time.Millisecond)376 if len(c.requests()) != n {377 t.Error("sent after Close")378 }379 for _, r := range c.requests()[:n] {380 if r.path == "/v1/traces" {381 t.Error("a span was sent though none was recorded while on")382 }383 }384}385386func TestOTLPConfigFromEnv(t *testing.T) {387 t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "localhost:4318/")388 t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "https://traces.example/v1/t")389 t.Setenv("OTEL_METRICS_EXPORTER", "none")390 t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "Authorization=Bearer%20abc, x-team = 012 ,bad")391 t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "deployment.environment=dev,service.name=other")392 t.Setenv("OTEL_EXPORTER_OTLP_TIMEOUT", "500")393 c := ConfigFromEnv().OTLP394 u := c.urls()395 want := urls{logs: "http://localhost:4318/v1/logs", traces: "https://traces.example/v1/t"}396 if u != want {397 t.Errorf("urls %+v, want %+v", u, want)398 }399 if c.Headers["Authorization"] != "Bearer abc" || c.Headers["x-team"] != "012" || len(c.Headers) != 2 {400 t.Errorf("headers %v", c.Headers)401 }402 if c.Timeout != 500*time.Millisecond {403 t.Errorf("timeout %v", c.Timeout)404 }405 e := newExporter(c, "")406 defer e.shutdown()407 b, _ := json.Marshal(e.resource)408 var rm map[string]any409 if err := json.Unmarshal(b, &rm); err != nil {410 t.Fatal(err)411 }412 res := attrMap(t, rm["attributes"])413 if res["service.name"].(map[string]any)["stringValue"] != "012" || res["deployment.environment"].(map[string]any)["stringValue"] != "dev" {414 t.Errorf("resource %v", res)415 }416 c.Disabled = true417 if c.urls().any() {418 t.Error("disabled but still sending")419 }420}421422func TestOTLPAndFileTogether(t *testing.T) {423 c := newCollector(t)424 path := t.TempDir() + "/log.jsonl"425 if _, err := Setup(Config{LogPath: path, OTLP: OTLPConfig{Endpoint: c.srv.URL}}); err != nil {426 t.Fatal(err)427 }428 Event("recalc", time.Millisecond, slog.Int("cells", 3))429 Close()430 logs, _ := c.items("/v1/logs")431 if len(logs) != 3 { // start, recalc, otlp432 t.Errorf("%d OTLP logs", len(logs))433 }434 b, err := os.ReadFile(path)435 if err != nil {436 t.Fatal(err)437 }438 if n := len(regexp.MustCompile(`"service":"012"`).FindAll(b, -1)); n != 3 {439 t.Errorf("%d file events: %s", n, b)440 }441}442443// The on-path cost with OTLP, queueing to a local endpoint.444func BenchmarkSpanOTLP(b *testing.B) {445 c := newCollector(b)446 otlpSetup(b, OTLPConfig{Endpoint: c.srv.URL})447 for b.Loop() {448 s := Start("recalc", slog.Int("n", 1))449 s.End(slog.Int("cells", 2))450 }451}452453// Nested spans go out with their parent's span id and trace id; roots454// without a parentSpanId.455func TestOTLPNestedSpans(t *testing.T) {456 c := newCollector(t)457 otlpSetup(t, OTLPConfig{Endpoint: c.srv.URL})458 tr := &Trace{}459 cmd := tr.Start("command")460 tr.Begin("recalc")461 tr.End()462 cmd.End()463 Close()464 spans, _ := c.items("/v1/traces")465 if len(spans) != 2 {466 t.Fatalf("%d spans: %v", len(spans), spans)467 }468 recalc, command := spans[0], spans[1]469 if recalc["parentSpanId"] != command["spanId"] || recalc["traceId"] != command["traceId"] || !hexSpan.MatchString(recalc["parentSpanId"].(string)) {470 t.Errorf("recalc %v is not under command %v", recalc, command)471 }472 if _, ok := command["parentSpanId"]; ok {473 t.Errorf("root span with a parent: %v", command)474 }475}